From a269e55b59d659e0e9a65f95d5165a9743ac496c Mon Sep 17 00:00:00 2001 From: smarcet Date: Fri, 4 Sep 2026 00:02:21 -0300 Subject: [PATCH 1/4] fix(speakers): retry a killed bulk-send chunk once and resume without re-emailing ProcessSpeakersEmailRequestJob had tries = 1 and timeout = 0: a chunk whose worker was killed mid-run (rolling deploy, OOM, scale-down) was re-served and failed without ever running again, and the failed() hook's own advice (re-send with should_resend=false) silently drops every speaker who ever received that email type from any earlier campaign, since the guard it relies on carries no date. - Bound the job's runtime (timeout 1200s) strictly below every queue connection's retry_after (1800s) and the db-fallback worker's --timeout (1400s), extracted into a reusable ResumableChunkJob trait alongside tries = 2 and a 300s backoff, so a retried attempt can never run concurrently with a still-live earlier one. - SpeakerService::triggerSendEmails stamps each chunk with dispatched_at once per run and strips any caller-supplied resume_since. - On a retry, the job sets resume_since = dispatched_at; the per-speaker closure in SpeakerService::sendEmails skips only speakers whose proof for this email type was written since that instant, before any promo-code or assistance side effect. should_resend passes through untouched - it answers a different question (skip anyone with any historical proof) and stacks with the resume check rather than replacing it. - PresentationSpeaker::hasAnnouncementEmailTypeSentSince adds the dated proof check; SpeakerActionsEmailStrategy::getAnnouncementType exposes the flow_event -> type mapping so the service can resolve it before process(). - Corrected the failed() operator hint, which previously recommended should_resend=false unconditionally - now a warning, since that guard also silently skips every speaker from any earlier, unrelated campaign. --- .../ProcessSpeakersEmailRequestJob.php | 35 ++- app/Jobs/Emails/Traits/ResumableChunkJob.php | 87 ++++++ .../Summit/Speakers/PresentationSpeaker.php | 24 ++ app/Services/Model/Imp/SpeakerService.php | 38 +++ .../Imp/Traits/ParametrizedSendEmails.php | 17 ++ .../SpeakerActionsEmailStrategy.php | 68 +++-- ...sSpeakersEmailRequestJobFailedHookTest.php | 6 +- ...ocessSpeakersEmailRequestJobResumeTest.php | 121 ++++++++ tests/SpeakerServiceResumeSendEmailsTest.php | 284 ++++++++++++++++++ 9 files changed, 636 insertions(+), 44 deletions(-) create mode 100644 app/Jobs/Emails/Traits/ResumableChunkJob.php create mode 100644 tests/ProcessSpeakersEmailRequestJobResumeTest.php create mode 100644 tests/SpeakerServiceResumeSendEmailsTest.php diff --git a/app/Jobs/Emails/PresentationSubmissions/ProcessSpeakersEmailRequestJob.php b/app/Jobs/Emails/PresentationSubmissions/ProcessSpeakersEmailRequestJob.php index 0510e85a5..e35381f4c 100644 --- a/app/Jobs/Emails/PresentationSubmissions/ProcessSpeakersEmailRequestJob.php +++ b/app/Jobs/Emails/PresentationSubmissions/ProcessSpeakersEmailRequestJob.php @@ -12,6 +12,7 @@ * limitations under the License. **/ use App\Jobs\Emails\PresentationSubmissions\SelectionProcess\PresentationSpeakerSelectionProcessExcerptEmail; +use App\Jobs\Emails\Traits\ResumableChunkJob; use App\Jobs\Utils\JobDispatcher; use App\Services\utils\IEmailExcerptService; use Illuminate\Bus\Queueable; @@ -32,11 +33,10 @@ */ final class ProcessSpeakersEmailRequestJob implements ShouldQueue { - public $timeout = 0; - - public $tries = 1; - - use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + // $timeout/$tries/$backoff and the resume-on-retry mechanics come from ResumableChunkJob - + // see that trait's doc comment for why timeout must stay below every retry_after / worker + // --timeout this job can run under. + use Dispatchable, InteractsWithQueue, Queueable, SerializesModels, ResumableChunkJob; /** * @var int @@ -85,24 +85,35 @@ public function handle ) ); + // ResumableChunkJob::activateResumeIfRetrying(): resume, not resend. On a retry it sets + // resume_since = dispatched_at in $this->payload so the service skips only the speakers + // whose proof for this email type was written by THIS run. should_resend is left exactly + // as it arrived - it answers a different question (does the operator want to re-email + // speakers with a proof from any earlier campaign?) and the two filters stack. + $this->activateResumeIfRetrying(); + $filter = !is_null($this->filter) ? FilterParser::parse($this->filter, \services\model\ISpeakerFilterFields::OPERATORS) : null; $service->sendEmails($this->summit_id, $this->payload, $filter); } /** - * Invoked by the queue worker once this job is marked failed. With tries = 1 that includes a - * chunk whose worker was killed mid-run: the job sits reserved until the connection's - * retry_after elapses, is re-served, and is failed without re-running. Nothing else reports + * Invoked by the queue worker once this job is marked failed - with tries = 2 and a resume + * check on the retry (see handle()), that means BOTH attempts failed: a chunk whose worker + * was merely killed mid-run (rolling deploy, OOM, scale-down) is re-served and automatically + * resumed once, skipping only the speakers this run already reached. This hook only fires + * when that automatic resume itself also failed to finish the chunk. Nothing else reports * that loss - the outcome excerpt is only sent when sendEmails() runs to completion - so * without this hook a dead chunk leaves no trace beyond a queue_failed_jobs row. * * Log the chunk's speaker ids at error, and when the operator asked for an outcome e-mail * send one naming them, so the chunk can be re-sent by id. The chunk is processed one speaker * per transaction, so a worker killed mid-run has already e-mailed (and written the "already - * sent" proof for) the speakers before the kill: the ids are an upper bound on what was lost, - * not confirmed misses, and both messages say so and tell the operator to re-send with - * should_resend=false so the resend guard skips the speakers that already have a proof. The + * sent" proof for) some of the speakers before the kill: the ids are an upper bound on what + * was lost, not confirmed misses, and both messages say so. should_resend=false is NOT a + * blanket recommendation here: hasAnnouncementEmailTypeSent (the check it gates) carries no + * date, so it also skips every speaker with a proof from any EARLIER, unrelated campaign of + * the same type - the hint below warns about that rather than recommending it outright. The * excerpt goes through JobDispatcher::withDbFallback (primary, then the database queue, then * an inline run) and is best-effort: a failure there must not mask the original failure. * @@ -114,7 +125,7 @@ public function failed(\Throwable $e): void $flow_event = $this->payload['email_flow_event'] ?? ''; $ids_list = implode(', ', $speaker_ids); - $resend_hint = "Re-send these ids with should_resend=false so the speakers already e-mailed are skipped"; + $resend_hint = "Both automatic attempts are exhausted. A manual re-send of these ids with should_resend=false skips any speaker who already has a proof of this email type - not just from this run, but from ANY earlier campaign of the same type - so use it to avoid duplicating what this run already sent, never to re-run a deliberate second campaign, or the first campaign's speakers get silently skipped too"; if (isset($this->payload['promo_code_spec'])) { // AutomaticMultiSpeakerPromoCodeStrategy::getPromoCode() generates a fresh code on every // call, before the resend guard runs, so should_resend=false does not prevent this one. diff --git a/app/Jobs/Emails/Traits/ResumableChunkJob.php b/app/Jobs/Emails/Traits/ResumableChunkJob.php new file mode 100644 index 000000000..17b2e2087 --- /dev/null +++ b/app/Jobs/Emails/Traits/ResumableChunkJob.php @@ -0,0 +1,87 @@ +payload to the + * receiving service. + * + * A chunk with no dispatched_at (queued by a pod running a version of the composing job from + * before this trait existed) is left untouched on a retry - it retries as a full re-run rather + * than an incorrect resume computed from a missing timestamp. That is a bounded, accepted risk + * (one duplicate chunk) limited to a single rolling-deploy window. + */ +trait ResumableChunkJob +{ + // 1200s: strictly below every queue connection's retry_after (1800 on both redis and + // database, config/queue.php) and the database-fallback worker's --timeout=1400 + // (fn-docker/summit-api/php-entry-point.sh). Composing jobs sharing this trait must not + // override this to a value at or above the lowest retry_after / worker --timeout they run + // under, or the concurrency guarantee above no longer holds. + public $timeout = 1200; + + public $tries = 2; + + public $backoff = 300; + + /** + * Mutates $this->payload in place: on a retry (attempts() > 1) with a dispatched_at already + * stamped, sets resume_since = dispatched_at and logs a warning. First attempt, or a chunk + * with no dispatched_at, leaves the payload untouched. + */ + private function activateResumeIfRetrying(): void + { + if ($this->attempts() <= 1 || !isset($this->payload['dispatched_at'])) { + return; + } + + $this->payload['resume_since'] = $this->payload['dispatched_at']; + + Log::warning + ( + sprintf + ( + "%s::handle attempt %s: resuming chunk from dispatched_at %s", + static::class, + $this->attempts(), + date('c', $this->payload['dispatched_at']) + ) + ); + } +} diff --git a/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php b/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php index cdcae06d5..87aa7db52 100644 --- a/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php +++ b/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php @@ -2309,6 +2309,30 @@ public function hasAnnouncementEmailTypeSent(Summit $summit, string $type): bool return $this->announcement_summit_emails->matching($criteria)->count() > 0; } + /** + * Same criteria as hasAnnouncementEmailTypeSent, additionally bounded to a proof written at + * or after $since. Used to resume a retried bulk send: it answers "did THIS run already + * reach this speaker" rather than "has this speaker ever received this type", which is what + * makes it safe to combine with should_resend on a retry without silently swallowing a + * deliberate second campaign of the same type. + * + * @param Summit $summit + * @param string $type + * @param \DateTime $since + * @return bool + */ + public function hasAnnouncementEmailTypeSentSince(Summit $summit, string $type, \DateTime $since): bool + { + $criteria = Criteria::create(); + + $criteria + ->where(Criteria::expr()->eq('summit', $summit)) + ->andWhere(Criteria::expr()->eq('type', $type)) + ->andWhere(Criteria::expr()->gte('send_date', $since)); + + return $this->announcement_summit_emails->matching($criteria)->count() > 0; + } + /** * @return bool diff --git a/app/Services/Model/Imp/SpeakerService.php b/app/Services/Model/Imp/SpeakerService.php index e47018633..90bd788e4 100644 --- a/app/Services/Model/Imp/SpeakerService.php +++ b/app/Services/Model/Imp/SpeakerService.php @@ -1288,11 +1288,19 @@ public function triggerSendEmails(Summit $summit, array $payload, $filter = null // deduplicated). withDbFallback() fails over to the database queue (and runs sync on a // double failure) so the loop completes. Same pattern as // PresentationSubmissionReopenService::notify's per-recipient dispatch loop. + // Stamped once per run (not per chunk) so every chunk's resume check, if the chunk is + // ever retried, compares against the same instant this run started. + $dispatched_at = time(); $chunk_nbr = 1; foreach (array_chunk($ids, $process_jon_chunk_size) as $chunk) { $chunkPayload = $payload; $chunkPayload['speaker_ids'] = $chunk; + $chunkPayload['dispatched_at'] = $dispatched_at; unset($chunkPayload['excluded_speaker_ids']); + // resume_since is set only by ProcessSpeakersEmailRequestJob::handle() on a retry - + // getJsonPayload() returns the raw request body and no validation rule declares this + // key, so a caller-supplied value must never reach a first-attempt chunk. + unset($chunkPayload['resume_since']); Log::debug ( @@ -1387,6 +1395,36 @@ function throw new EntityNotFoundException('Speaker not found'); } + // Resume, not resend: set only by ProcessSpeakersEmailRequestJob::handle() + // on a retry (attempts() > 1). A speaker whose proof for this email type + // was written at or after resume_since was already reached by this run + // before the kill - skip before any side effect (promo code, assistance) + // so a retry never leaves an orphan promo code for a speaker it must not + // re-email. should_resend is independent and evaluated later, unchanged, + // inside SpeakerActionsEmailStrategy::process(). + $resume_since = $payload['resume_since'] ?? null; + if (!is_null($resume_since)) { + $announcement_type = $email_strategy->getAnnouncementType(); + if + ( + !is_null($announcement_type) && + $speaker->hasAnnouncementEmailTypeSentSince($summit, $announcement_type, new \DateTime('@' . $resume_since)) + ) { + if (!is_null($onDispatchInfo)) { + $onDispatchInfo + ( + sprintf + ( + "Speaker %s (%s) already processed by this run before the retry, skipped.", + $speaker->getEmail(), + $speaker->getId() + ) + ); + } + return; + } + } + // try to get or auto-build a promo code $promo_code_strategy = $this->promo_code_strategy_factory->createStrategy($summit, $payload); diff --git a/app/Services/Model/Imp/Traits/ParametrizedSendEmails.php b/app/Services/Model/Imp/Traits/ParametrizedSendEmails.php index d2662c4bf..654cfc8da 100644 --- a/app/Services/Model/Imp/Traits/ParametrizedSendEmails.php +++ b/app/Services/Model/Imp/Traits/ParametrizedSendEmails.php @@ -127,6 +127,23 @@ private function _sendEmails( sprintf("Processing EMAIL %s for %s %s", $flow_event, $getRootEntityName(), $root_entity_id) ); + // Set only by ProcessSpeakersEmailRequestJob::handle() on a retry (SpeakerService is the + // only caller of _sendEmails whose payload ever carries this key); announces the resume + // so the operator's excerpt reads the same as the "already has an email of type" lines it + // sits next to. Must stay after the header above - anything added before + // EmailExcerpt::clearReport() (top of this method) is wiped. + if (isset($payload['resume_since'])) { + EmailExcerpt::addInfoMessage + ( + sprintf + ( + "RETRY: resuming this run; %s already processed since %s are skipped", + $subject, + date('c', $payload['resume_since']) + ) + ); + } + $root_entity = $this->tx_service->transaction(function () use($root_entity_id, $getRootEntity){ return $getRootEntity($root_entity_id); }); diff --git a/app/Services/Model/Strategies/EmailActions/SpeakerActionsEmailStrategy.php b/app/Services/Model/Strategies/EmailActions/SpeakerActionsEmailStrategy.php index 0459450ce..5acbc80f0 100644 --- a/app/Services/Model/Strategies/EmailActions/SpeakerActionsEmailStrategy.php +++ b/app/Services/Model/Strategies/EmailActions/SpeakerActionsEmailStrategy.php @@ -80,8 +80,6 @@ public function process(PresentationSpeaker ): void { try { - $type = null; - Log::debug ( sprintf @@ -163,35 +161,15 @@ public function process(PresentationSpeaker ) ); - switch ($this->flow_event) { - case PresentationSpeakerSelectionProcessAcceptedAlternateEmail::EVENT_SLUG: - $type = SpeakerAnnouncementSummitEmail::TypeAcceptedAlternate; - break; - case PresentationSpeakerSelectionProcessAcceptedOnlyEmail::EVENT_SLUG: - $type = SpeakerAnnouncementSummitEmail::TypeAccepted; - break; - case PresentationSpeakerSelectionProcessAcceptedRejectedEmail::EVENT_SLUG: - $type = SpeakerAnnouncementSummitEmail::TypeAcceptedRejected; - break; - case PresentationSpeakerSelectionProcessAlternateOnlyEmail::EVENT_SLUG: - $type = SpeakerAnnouncementSummitEmail::TypeAlternate; - break; - case PresentationSpeakerSelectionProcessAlternateRejectedEmail::EVENT_SLUG: - $type = SpeakerAnnouncementSummitEmail::TypeAlternateRejected; - break; - case PresentationSpeakerSelectionProcessRejectedOnlyEmail::EVENT_SLUG: - $type = SpeakerAnnouncementSummitEmail::TypeRejected; - break; - default: - if (!is_null($onSuccess)) { - $onSuccess - ( - $speaker->getEmail(), - IEmailExcerptService::EmailLineType, - SpeakerAnnouncementSummitEmail::TypeNone - ); - } - break; + $type = $this->getAnnouncementType(); + + if (is_null($type) && !is_null($onSuccess)) { + $onSuccess + ( + $speaker->getEmail(), + IEmailExcerptService::EmailLineType, + SpeakerAnnouncementSummitEmail::TypeNone + ); } if (!is_null($type)) { @@ -280,4 +258,32 @@ public function process(PresentationSpeaker $onError($ex->getMessage()); } } + + /** + * Maps this strategy's flow_event to the SpeakerAnnouncementSummitEmail type it announces, or + * null when the flow_event carries no announcement type (process()'s TypeNone branch). Public + * so SpeakerService::sendEmails can resolve the type ahead of process() to evaluate a retry's + * resume check before any side effect, without duplicating this mapping. + * + * @return string|null + */ + public function getAnnouncementType(): ?string + { + switch ($this->flow_event) { + case PresentationSpeakerSelectionProcessAcceptedAlternateEmail::EVENT_SLUG: + return SpeakerAnnouncementSummitEmail::TypeAcceptedAlternate; + case PresentationSpeakerSelectionProcessAcceptedOnlyEmail::EVENT_SLUG: + return SpeakerAnnouncementSummitEmail::TypeAccepted; + case PresentationSpeakerSelectionProcessAcceptedRejectedEmail::EVENT_SLUG: + return SpeakerAnnouncementSummitEmail::TypeAcceptedRejected; + case PresentationSpeakerSelectionProcessAlternateOnlyEmail::EVENT_SLUG: + return SpeakerAnnouncementSummitEmail::TypeAlternate; + case PresentationSpeakerSelectionProcessAlternateRejectedEmail::EVENT_SLUG: + return SpeakerAnnouncementSummitEmail::TypeAlternateRejected; + case PresentationSpeakerSelectionProcessRejectedOnlyEmail::EVENT_SLUG: + return SpeakerAnnouncementSummitEmail::TypeRejected; + default: + return null; + } + } } \ No newline at end of file diff --git a/tests/ProcessSpeakersEmailRequestJobFailedHookTest.php b/tests/ProcessSpeakersEmailRequestJobFailedHookTest.php index cfdfdc781..a9347e8c0 100644 --- a/tests/ProcessSpeakersEmailRequestJobFailedHookTest.php +++ b/tests/ProcessSpeakersEmailRequestJobFailedHookTest.php @@ -89,7 +89,11 @@ public function testFailedChunkWithOutcomeRecipientSendsExcerptNamingTheUnproces // part of it: the line must not present the ids as confirmed misses, and it must tell // the operator how to re-send without duplicating those. $this->assertStringContainsString('up to 3 of them may not have been processed', $errorLines[0]); - $this->assertStringContainsString('should_resend=false', $errorLines[0], 'the ERROR line must tell the operator to re-send with should_resend=false'); + // Both attempts (the original run plus the automatic resume) are exhausted once this + // hook fires - should_resend=false is a warning against silently skipping speakers + // from other campaigns, not a blanket instruction, since it carries no date. + $this->assertStringContainsString('should_resend=false', $errorLines[0], 'the ERROR line must warn the operator about should_resend=false on a manual re-send'); + $this->assertStringContainsString('automatic', $errorLines[0], 'the ERROR line must say the automatic resume already ran and also failed'); $this->assertStringNotContainsString('promo code', $errorLines[0], 'no promo-code caveat when the send carries no promo_code_spec'); $this->assertEmpty( array_filter($lines, fn($l) => str_starts_with($l, 'Email type')), diff --git a/tests/ProcessSpeakersEmailRequestJobResumeTest.php b/tests/ProcessSpeakersEmailRequestJobResumeTest.php new file mode 100644 index 000000000..f49268d12 --- /dev/null +++ b/tests/ProcessSpeakersEmailRequestJobResumeTest.php @@ -0,0 +1,121 @@ + + * 1) with a dispatched_at already stamped in the payload, handle() must add resume_since = + * dispatched_at before calling sendEmails(), and must leave should_resend exactly as it arrived + * in the payload - the resume check and should_resend are independent filters that stack, not + * one replacing the other. + * + * attempts() is driven via InteractsWithQueue::setJob() with a mocked + * Illuminate\Contracts\Queue\Job, since $this->attempts() returns 1 whenever no job instance is + * set (dispatchSync, direct calls) and there is no other way to simulate a second delivery + * without a real queue connection. + * + * Class ProcessSpeakersEmailRequestJobResumeTest + */ +class ProcessSpeakersEmailRequestJobResumeTest extends ProtectedApiTestCase +{ + private function jobWithAttempts(array $payload, int $attempts): ProcessSpeakersEmailRequestJob + { + $job = new ProcessSpeakersEmailRequestJob(1, $payload, null); + + $queueJob = Mockery::mock(QueueJobContract::class); + $queueJob->shouldReceive('attempts')->andReturn($attempts); + $job->setJob($queueJob); + + return $job; + } + + public function testHandleOnSecondAttemptSetsResumeSinceAndKeepsShouldResendFalse(): void + { + $payload = [ + 'email_flow_event' => 'SUMMIT_SUBMISSIONS_PRESENTATION_SPEAKER_ACCEPTED_ALTERNATE', + 'speaker_ids' => [11, 22], + 'dispatched_at' => 1700000000, + 'should_resend' => false, + ]; + $job = $this->jobWithAttempts($payload, 2); + + $service = Mockery::mock(ISpeakerService::class); + $service->shouldReceive('sendEmails')->once()->withArgs(function ($summit_id, $sentPayload) { + return ($sentPayload['resume_since'] ?? null) === 1700000000 + && array_key_exists('should_resend', $sentPayload) + && $sentPayload['should_resend'] === false; + }); + + $job->handle($service); + } + + public function testHandleOnSecondAttemptKeepsShouldResendTrue(): void + { + $payload = [ + 'email_flow_event' => 'SUMMIT_SUBMISSIONS_PRESENTATION_SPEAKER_ACCEPTED_ALTERNATE', + 'speaker_ids' => [11, 22], + 'dispatched_at' => 1700000000, + 'should_resend' => true, + ]; + $job = $this->jobWithAttempts($payload, 2); + + $service = Mockery::mock(ISpeakerService::class); + $service->shouldReceive('sendEmails')->once()->withArgs(function ($summit_id, $sentPayload) { + return ($sentPayload['resume_since'] ?? null) === 1700000000 + && $sentPayload['should_resend'] === true; + }); + + $job->handle($service); + } + + public function testHandleOnFirstAttemptSetsNoResumeSince(): void + { + $payload = [ + 'email_flow_event' => 'SUMMIT_SUBMISSIONS_PRESENTATION_SPEAKER_ACCEPTED_ALTERNATE', + 'speaker_ids' => [11, 22], + 'dispatched_at' => 1700000000, + ]; + $job = $this->jobWithAttempts($payload, 1); + + $service = Mockery::mock(ISpeakerService::class); + $service->shouldReceive('sendEmails')->once()->withArgs(function ($summit_id, $sentPayload) { + return !array_key_exists('resume_since', $sentPayload); + }); + + $job->handle($service); + } + + public function testHandleOnSecondAttemptWithoutDispatchedAtSetsNoResumeSince(): void + { + // A chunk queued by a pod running the previous version of this job (before dispatched_at + // existed) must not attempt a resume it cannot correctly compute - it retries as a full + // re-run instead. This covers the deploy-window gap documented in the plan's Fix Approach. + $payload = [ + 'email_flow_event' => 'SUMMIT_SUBMISSIONS_PRESENTATION_SPEAKER_ACCEPTED_ALTERNATE', + 'speaker_ids' => [11, 22], + ]; + $job = $this->jobWithAttempts($payload, 2); + + $service = Mockery::mock(ISpeakerService::class); + $service->shouldReceive('sendEmails')->once()->withArgs(function ($summit_id, $sentPayload) { + return !array_key_exists('resume_since', $sentPayload); + }); + + $job->handle($service); + } +} diff --git a/tests/SpeakerServiceResumeSendEmailsTest.php b/tests/SpeakerServiceResumeSendEmailsTest.php new file mode 100644 index 000000000..8523ee8dc --- /dev/null +++ b/tests/SpeakerServiceResumeSendEmailsTest.php @@ -0,0 +1,284 @@ + AbstractEmailJob::$to_email), so walk the hierarchy the same way + * ProcessSpeakersEmailRequestJobFailedHookTest::jobProperty() does. + */ + private function jobProperty(object $job, string $name) + { + $reflection = new ReflectionObject($job); + while ($reflection && !$reflection->hasProperty($name)) { + $reflection = $reflection->getParentClass(); + } + $property = $reflection->getProperty($name); + $property->setAccessible(true); + return $property->getValue($job); + } + + /** + * @return PresentationSpeaker + */ + private function newFixtureSpeaker(string $prefix): PresentationSpeaker + { + // A member is required so getEmail() (member wins, else registration request, else null) + // resolves to a distinct, real address per speaker - the pushed job's to_email is how + // these tests tell which speaker a dispatch targeted (the job exposes no getSpeaker()). + // str_random suffix avoids a duplicate Member.Email collision across separate phpunit + // invocations - InsertSummitTestData's cleanup does not roll back the Member table. + $member = new Member(); + $member->setEmail("resume-test+{$prefix}-" . str_random(8) . "@test.com"); + $member->setActive(true); + $member->setFirstName("Resume"); + $member->setLastName("Member {$prefix}"); + $member->setEmailVerified(true); + $member->setUserExternalId(mt_rand()); + self::$em->persist($member); + + $speaker = new PresentationSpeaker(); + $speaker->setFirstName("Resume"); + $speaker->setLastName("Speaker {$prefix}"); + $speaker->setMember($member); + self::$em->persist($speaker); + + $presentation = new \models\summit\Presentation(); + self::$summit->addEvent($presentation); + $presentation->setTitle("Resume test presentation {$prefix}"); + $presentation->setAbstract("Abstract {$prefix}"); + $presentation->setCategory(self::$defaultTrack); + $presentation->setType(self::$defaultPresentationType); + $presentation->setStartDate(new \DateTime('now', new \DateTimeZone('UTC'))); + $presentation->setEndDate(new \DateTime('+1 hour', new \DateTimeZone('UTC'))); + $presentation->addSpeaker($speaker); + self::$em->persist($presentation); + + return $speaker; + } + + /** + * @param PresentationSpeaker $speaker + * @param \DateTime|null $backdateTo when null, the proof is stamped "now" (markAsSent()); + * otherwise its send_date is force-set via reflection, since production code has no setter + * that backdates a proof (markAsSent() always stamps "now" by design). + */ + private function givenSpeakerHasProof(PresentationSpeaker $speaker, ?\DateTime $backdateTo): void + { + $proof = new SpeakerAnnouncementSummitEmail(); + $proof->setType(self::EMAIL_TYPE); + $speaker->addAnnouncementSummitEmail($proof); + self::$summit->addAnnouncementSummitEmail($proof); + + if (is_null($backdateTo)) { + $proof->markAsSent(); + } else { + $prop = new ReflectionProperty(SpeakerAnnouncementSummitEmail::class, 'send_date'); + $prop->setAccessible(true); + $prop->setValue($proof, $backdateTo); + } + } + + public function testResumedRunSkipsOnlySpeakerWithProofSinceDispatch(): void + { + Queue::fake(); + + $speakerA = $this->newFixtureSpeaker('a-since-dispatch'); + $speakerB = $this->newFixtureSpeaker('b-before-dispatch'); + + $dispatchedAt = time() - 600; + + $this->givenSpeakerHasProof($speakerA, null); // proof written "now", i.e. after dispatch + $this->givenSpeakerHasProof($speakerB, new \DateTime('-30 days', new \DateTimeZone('UTC'))); + + self::$em->flush(); + + $this->service()->sendEmails(self::$summit->getId(), [ + 'email_flow_event' => self::FLOW_EVENT, + 'speaker_ids' => [$speakerA->getId(), $speakerB->getId()], + 'dispatched_at' => $dispatchedAt, + 'resume_since' => $dispatchedAt, + ], null); + + Queue::assertPushed(PresentationSpeakerSelectionProcessAcceptedAlternateEmail::class, 1); + Queue::assertPushed(PresentationSpeakerSelectionProcessAcceptedAlternateEmail::class, function ($job) use ($speakerB) { + return $this->jobProperty($job, 'to_email') === $speakerB->getEmail(); + }); + + $skippedLines = array_values(array_filter( + EmailExcerpt::getReport(), + fn($line) => str_contains($line['message'] ?? '', $speakerA->getEmail()) + )); + $this->assertCount( + 1, + $skippedLines, + 'exactly one excerpt line must name the resume-skipped speaker A' + ); + } + + public function testNonResumedRunWithShouldResendDefaultEmailsBoth(): void + { + Queue::fake(); + + $speakerA = $this->newFixtureSpeaker('a-no-resume'); + $speakerB = $this->newFixtureSpeaker('b-no-resume'); + + $this->givenSpeakerHasProof($speakerA, null); + $this->givenSpeakerHasProof($speakerB, new \DateTime('-30 days', new \DateTimeZone('UTC'))); + + self::$em->flush(); + + $this->service()->sendEmails(self::$summit->getId(), [ + 'email_flow_event' => self::FLOW_EVENT, + 'speaker_ids' => [$speakerA->getId(), $speakerB->getId()], + ], null); + + Queue::assertPushed(PresentationSpeakerSelectionProcessAcceptedAlternateEmail::class, 2); + } + + public function testShouldResendFalseSkipsBothRegardlessOfResumeSince(): void + { + Queue::fake(); + + $speakerA = $this->newFixtureSpeaker('a-should-resend-false'); + $speakerB = $this->newFixtureSpeaker('b-should-resend-false'); + + $dispatchedAt = time() - 600; + + $this->givenSpeakerHasProof($speakerA, null); + $this->givenSpeakerHasProof($speakerB, new \DateTime('-30 days', new \DateTimeZone('UTC'))); + + self::$em->flush(); + + $this->service()->sendEmails(self::$summit->getId(), [ + 'email_flow_event' => self::FLOW_EVENT, + 'speaker_ids' => [$speakerA->getId(), $speakerB->getId()], + 'dispatched_at' => $dispatchedAt, + 'resume_since' => $dispatchedAt, + 'should_resend' => false, + ], null); + + Queue::assertNotPushed(PresentationSpeakerSelectionProcessAcceptedAlternateEmail::class); + } + + public function testResumedRunNeverInvokesGetPromoCodeForSkippedSpeaker(): void + { + Queue::fake(); + + $speakerA = $this->newFixtureSpeaker('a-promo-skip'); + $speakerB = $this->newFixtureSpeaker('b-promo-send'); + + $dispatchedAt = time() - 600; + + $this->givenSpeakerHasProof($speakerA, null); + $this->givenSpeakerHasProof($speakerB, new \DateTime('-30 days', new \DateTimeZone('UTC'))); + + self::$em->flush(); + + // A strict shouldReceive(...)->once()->withArgs(...) expectation would look right but + // never actually fail this test: SpeakerService::sendEmails wraps the per-speaker closure + // in catch (\Exception $ex) { Log::warning($ex); ... }, and Mockery's + // NoMatchingExpectationException extends \Exception - a mismatched call gets swallowed + // and logged, not surfaced as a test failure. Recording calls and asserting afterward + // sidesteps that; verified empirically by removing the resume-skip fix. + $calledWith = []; + $promoCodeStrategy = Mockery::mock(IPromoCodeStrategy::class); + $promoCodeStrategy->shouldReceive('getPromoCode') + ->andReturnUsing(function (PresentationSpeaker $speaker) use (&$calledWith) { + $calledWith[] = $speaker->getId(); + return null; + }); + + $promoCodeStrategyFactory = Mockery::mock(IPromoCodeStrategyFactory::class); + $promoCodeStrategyFactory->shouldReceive('createStrategy')->andReturn($promoCodeStrategy); + App::instance(IPromoCodeStrategyFactory::class, $promoCodeStrategyFactory); + // ISpeakerService is bound as a singleton (ModelServicesProvider): once resolved, its + // constructor-injected IPromoCodeStrategyFactory is fixed. Forget the cached instance so + // service() below rebuilds it against the rebind above, or this mock is silently ignored + // and the real factory (and real promo-code side effects) runs instead. + App::forgetInstance(ISpeakerService::class); + + $this->service()->sendEmails(self::$summit->getId(), [ + 'email_flow_event' => self::FLOW_EVENT, + 'speaker_ids' => [$speakerA->getId(), $speakerB->getId()], + 'dispatched_at' => $dispatchedAt, + 'resume_since' => $dispatchedAt, + 'promo_code_spec' => ['type' => 'automatic'], + ], null); + + $this->assertEquals( + [$speakerB->getId()], + $calledWith, + 'getPromoCode() must be called exactly once, for speaker B only - never for the resume-skipped speaker A' + ); + } +} From c7cf75e6a71b484ff640b91f488497bfcad9adc4 Mon Sep 17 00:00:00 2001 From: smarcet Date: Fri, 4 Sep 2026 00:16:20 -0300 Subject: [PATCH 2/4] test(speakers): cover the two untested branches of the resume-skip check - An unmapped email_flow_event (getAnnouncementType() returns null) must not be skipped by the resume check - the "!is_null($announcement_type) &&" guard short-circuits before calling hasAnnouncementEmailTypeSentSince(), whose $type parameter is non-nullable string; dropping that guard would throw an uncaught TypeError (extends \Error, not caught by sendEmails' catch (\Exception)) instead of just letting the speaker through. - The resume-skip's early return happens before generateSpeakerAssistance(), not just before getPromoCode() - asserted by spying on IPresentationSpeakerSummitAssistanceConfirmationRequestRepository::getBySpeaker(). newFixtureSpeaker() gains an optional $published flag (default false, no change to existing call sites) so a fixture speaker's presentation can satisfy hasAcceptedPresentations()'s DQL (p.published = 1) and make generateSpeakerAssistance() actually reach the repository instead of short-circuiting to null before it. --- tests/SpeakerServiceResumeSendEmailsTest.php | 104 ++++++++++++++++++- 1 file changed, 103 insertions(+), 1 deletion(-) diff --git a/tests/SpeakerServiceResumeSendEmailsTest.php b/tests/SpeakerServiceResumeSendEmailsTest.php index 8523ee8dc..21336a944 100644 --- a/tests/SpeakerServiceResumeSendEmailsTest.php +++ b/tests/SpeakerServiceResumeSendEmailsTest.php @@ -13,6 +13,7 @@ **/ use App\Jobs\Emails\PresentationSubmissions\SelectionProcess\PresentationSpeakerSelectionProcessAcceptedAlternateEmail; +use App\Models\Foundation\Summit\Repositories\IPresentationSpeakerSummitAssistanceConfirmationRequestRepository; use App\Services\Model\Strategies\PromoCodes\IPromoCodeStrategy; use App\Services\Model\Strategies\PromoCodes\IPromoCodeStrategyFactory; use App\Services\Utils\Facades\EmailExcerpt; @@ -88,9 +89,14 @@ private function jobProperty(object $job, string $name) } /** + * @param string $prefix + * @param bool $published when true, publish()es the fixture presentation so + * PresentationSpeaker::hasAcceptedPresentations() (p.published = 1 satisfies its DQL OR + * clause) returns true for it - needed to make SpeakerService::generateSpeakerAssistance() + * proceed past its has_accepted/has_alternate guard instead of short-circuiting to null. * @return PresentationSpeaker */ - private function newFixtureSpeaker(string $prefix): PresentationSpeaker + private function newFixtureSpeaker(string $prefix, bool $published = false): PresentationSpeaker { // A member is required so getEmail() (member wins, else registration request, else null) // resolves to a distinct, real address per speaker - the pushed job's to_email is how @@ -123,6 +129,10 @@ private function newFixtureSpeaker(string $prefix): PresentationSpeaker $presentation->addSpeaker($speaker); self::$em->persist($presentation); + if ($published) { + $presentation->publish(); + } + return $speaker; } @@ -281,4 +291,96 @@ public function testResumedRunNeverInvokesGetPromoCodeForSkippedSpeaker(): void 'getPromoCode() must be called exactly once, for speaker B only - never for the resume-skipped speaker A' ); } + + public function testResumeCheckDoesNotSkipWhenFlowEventHasNoAnnouncementType(): void + { + // getAnnouncementType() returns null for a flow_event outside the six known SLUGs. + // hasAnnouncementEmailTypeSentSince(Summit, string $type, ...) takes a non-nullable + // string $type - if the "!is_null($announcement_type) &&" short-circuit in the resume + // check were ever dropped, this scenario would throw a TypeError (not caught by + // SpeakerService::sendEmails' catch (\Exception $ex), since TypeError extends \Error), + // failing the whole chunk instead of just skipping one speaker. Proven here by seeding a + // speaker with a proof that WOULD trigger a skip if the type resolved, and asserting the + // resume check let it through to getPromoCode() instead of returning early. + Queue::fake(); + + $speaker = $this->newFixtureSpeaker('unmapped-flow-event'); + + $dispatchedAt = time() - 600; + + $this->givenSpeakerHasProof($speaker, null); // proof written "now", after dispatch + + self::$em->flush(); + + $calledWith = []; + $promoCodeStrategy = Mockery::mock(IPromoCodeStrategy::class); + $promoCodeStrategy->shouldReceive('getPromoCode') + ->andReturnUsing(function (PresentationSpeaker $s) use (&$calledWith) { + $calledWith[] = $s->getId(); + return null; + }); + + $promoCodeStrategyFactory = Mockery::mock(IPromoCodeStrategyFactory::class); + $promoCodeStrategyFactory->shouldReceive('createStrategy')->andReturn($promoCodeStrategy); + App::instance(IPromoCodeStrategyFactory::class, $promoCodeStrategyFactory); + App::forgetInstance(ISpeakerService::class); + + $this->service()->sendEmails(self::$summit->getId(), [ + 'email_flow_event' => 'SUMMIT_SUBMISSIONS_PRESENTATION_SPEAKER_SOME_UNMAPPED_EVENT', + 'speaker_ids' => [$speaker->getId()], + 'dispatched_at' => $dispatchedAt, + 'resume_since' => $dispatchedAt, + 'promo_code_spec' => ['type' => 'automatic'], + ], null); + + $this->assertEquals( + [$speaker->getId()], + $calledWith, + 'an unmapped flow_event must not be skipped by the resume check - getAnnouncementType() returning null must short-circuit the check, not crash or silently skip' + ); + Queue::assertNotPushed(PresentationSpeakerSelectionProcessAcceptedAlternateEmail::class); + } + + public function testResumeCheckSkipsBeforeGeneratingSpeakerAssistance(): void + { + Queue::fake(); + + // generateSpeakerAssistance() only reaches the assistance repository when the speaker has + // an accepted or alternate presentation (SpeakerService.php's has_accepted_presentations / + // has_alternate_presentations guard) - published: true is the real production path that + // makes hasAcceptedPresentations() true (p.published = 1 satisfies its DQL OR clause). + $skippedSpeaker = $this->newFixtureSpeaker('assistance-skipped', published: true); + $sentSpeaker = $this->newFixtureSpeaker('assistance-sent', published: true); + self::$em->flush(); + + $dispatchedAt = time() - 600; + $this->givenSpeakerHasProof($skippedSpeaker, null); // proof after dispatch -> resume-skipped + // $sentSpeaker has no proof at all -> proceeds normally, reaching generateSpeakerAssistance(). + + self::$em->flush(); + + $calledWith = []; + $assistanceRepository = Mockery::mock(IPresentationSpeakerSummitAssistanceConfirmationRequestRepository::class); + $assistanceRepository->shouldReceive('getBySpeaker') + ->andReturnUsing(function (PresentationSpeaker $speaker, $summit) use (&$calledWith) { + $calledWith[] = $speaker->getId(); + return null; + }); + $assistanceRepository->shouldReceive('existByHash')->andReturn(false); + App::instance(IPresentationSpeakerSummitAssistanceConfirmationRequestRepository::class, $assistanceRepository); + App::forgetInstance(ISpeakerService::class); + + $this->service()->sendEmails(self::$summit->getId(), [ + 'email_flow_event' => self::FLOW_EVENT, + 'speaker_ids' => [$skippedSpeaker->getId(), $sentSpeaker->getId()], + 'dispatched_at' => $dispatchedAt, + 'resume_since' => $dispatchedAt, + ], null); + + $this->assertEquals( + [$sentSpeaker->getId()], + $calledWith, + 'generateSpeakerAssistance() must not run for the resume-skipped speaker - the return happens before it, same as before getPromoCode()' + ); + } } From cab6991cef2a3928d34679b6ff20528936311092 Mon Sep 17 00:00:00 2001 From: smarcet Date: Fri, 4 Sep 2026 10:38:42 -0300 Subject: [PATCH 3/4] fix(speakers): correct the $announcement_summit_emails PHPDoc collection type @var SpeakerAnnouncementSummitEmail[] described it as a plain array; at runtime it's a Doctrine Collection (implements Selectable), which is why matching() already worked on it. PHPStan flagged the call as "Cannot call method matching() on array" for both the pre-existing hasAnnouncementEmailTypeSent() and the new hasAnnouncementEmailTypeSentSince(). Docblock-only change, no behavior change. --- app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php b/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php index 87aa7db52..35e9a9005 100644 --- a/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php +++ b/app/Models/Foundation/Summit/Speakers/PresentationSpeaker.php @@ -199,7 +199,7 @@ class PresentationSpeaker extends SilverstripeBaseModel protected $active_involvements; /** - * @var SpeakerAnnouncementSummitEmail[] + * @var \Doctrine\Common\Collections\Collection */ #[ORM\OneToMany(targetEntity: \SpeakerAnnouncementSummitEmail::class, mappedBy: 'speaker', cascade: ['persist'], orphanRemoval: true, fetch: 'EXTRA_LAZY')] private $announcement_summit_emails; From 01d8093dc937157090f623d35b45b1e57c7ed1fc Mon Sep 17 00:00:00 2001 From: smarcet Date: Fri, 4 Sep 2026 12:02:37 -0300 Subject: [PATCH 4/4] test(speakers): assert chunk timeout stays below every queue retry_after Deep review of #598 flagged that ResumableChunkJob's core safety argument - job timeout strictly below every queue connection's retry_after, so a retried attempt can never overlap a still-live earlier one - had no regression coverage, despite the PR description claiming it did. A future change to DB_QUEUE_RETRY_AFTER/ REDIS_RETRY_AFTER, or to ResumableChunkJob::$timeout, could silently reintroduce concurrent execution of the same chunk. Add testTimeoutStaysStrictlyBelowRetryAfterForEveryQueueConnection to assert $job->timeout against both the database and redis connections' retry_after. Red-green verified by temporarily setting ResumableChunkJob::$timeout to 1800 (equal to retry_after) and confirming the assertion fails, then restoring it. --- ...ocessSpeakersEmailRequestJobResumeTest.php | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/tests/ProcessSpeakersEmailRequestJobResumeTest.php b/tests/ProcessSpeakersEmailRequestJobResumeTest.php index f49268d12..fdf0503cd 100644 --- a/tests/ProcessSpeakersEmailRequestJobResumeTest.php +++ b/tests/ProcessSpeakersEmailRequestJobResumeTest.php @@ -118,4 +118,29 @@ public function testHandleOnSecondAttemptWithoutDispatchedAtSetsNoResumeSince(): $job->handle($service); } + + /** + * Regression guard for ResumableChunkJob's core safety argument: the job's own $timeout must + * stay strictly below every queue connection's retry_after, or a retried attempt (tries=2) + * could be re-served to a second worker while the first is still running it - true concurrent + * execution of the same chunk, which the resume_since check cannot protect against. Nothing + * else in this test suite asserts this relationship, so a future change to + * DB_QUEUE_RETRY_AFTER/REDIS_RETRY_AFTER (or to ResumableChunkJob::$timeout) that violates it + * would otherwise go uncaught. + */ + public function testTimeoutStaysStrictlyBelowRetryAfterForEveryQueueConnection(): void + { + $job = new ProcessSpeakersEmailRequestJob(1, [], null); + + $this->assertLessThan( + config('queue.connections.database.retry_after'), + $job->timeout, + 'job timeout must stay strictly below the database queue retry_after, or a retried attempt can run concurrently with a still-live earlier attempt' + ); + $this->assertLessThan( + config('queue.connections.redis.retry_after'), + $job->timeout, + 'job timeout must stay strictly below the redis queue retry_after, or a retried attempt can run concurrently with a still-live earlier attempt' + ); + } }