From 2354bdba1ea02bad162d10a8c5a1944597259e30 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Mon, 7 Sep 2026 14:39:07 +0400 Subject: [PATCH 01/10] ref: replace findByIdAndStatus with tryClaimForProcessing in CampaignProcessorMessageHandler --- .../CampaignProcessorMessageHandler.php | 10 ++++- .../Model/Message/MessageMetadata.php | 1 - .../Repository/MessageRepository.php | 26 +++++++++++++ .../Repository/MessageRepositoryTest.php | 38 +++++++++++++++++++ .../CampaignProcessorMessageHandlerTest.php | 25 ++++++------ 5 files changed, 84 insertions(+), 16 deletions(-) diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index ad8d0f48..576c0ae4 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -78,7 +78,7 @@ public function __construct( public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $data): void { - $campaign = $this->messageRepository->findByIdAndStatus($data->getMessageId(), MessageStatus::Submitted); + $campaign = $this->messageRepository->tryClaimForProcessing($data->getMessageId()); if (!$campaign) { $this->logger->warning( $this->translator->trans('Campaign not found or not in submitted status'), @@ -121,7 +121,7 @@ public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $ $this->handleAdminNotifications($campaign, $loadedMessageData, $data->getMessageId()); - $this->updateMessageStatus($campaign, MessageStatus::Prepared); + // Campaign was already atomically claimed into Prepared status above. $subscribers = $this->subscriberProvider->getSubscribersForMessageOrLists($data, $campaign); $this->updateMessageStatus($campaign, MessageStatus::InProcess); @@ -170,6 +170,9 @@ private function updateMessageStatus(Message $message, MessageStatus $status): v if ($status === MessageStatus::InProcess && $message->getMetadata()->getSendStart() === null) { $message->getMetadata()->setSendStart(new DateTime()); } + if ($status === MessageStatus::Sent) { + $message->getMetadata()->setSent(new DateTime()); + } $message->getMetadata()->setStatus($status); $this->entityManager->flush(); } @@ -220,6 +223,9 @@ private function handleEmailSending( htmlPref: $subscriber->hasHtmlEmail(), ); if ($result === null) { + $status = $subscriber->isBlacklisted() ? UserMessageStatus::Excluded : UserMessageStatus::NotSent; + $this->updateUserMessageStatus($userMessage, $status); + return; } [$email, $sentAs] = $result; diff --git a/src/Domain/Messaging/Model/Message/MessageMetadata.php b/src/Domain/Messaging/Model/Message/MessageMetadata.php index c571bd1f..8a6b5d65 100644 --- a/src/Domain/Messaging/Model/Message/MessageMetadata.php +++ b/src/Domain/Messaging/Model/Message/MessageMetadata.php @@ -131,7 +131,6 @@ public function setEntered(?DateTime $entered): self return $this; } - // todo: set sent to the time when it was sent public function setSent(?DateTime $sent): self { $this->sent = $sent; diff --git a/src/Domain/Messaging/Repository/MessageRepository.php b/src/Domain/Messaging/Repository/MessageRepository.php index 13394794..076bfd27 100644 --- a/src/Domain/Messaging/Repository/MessageRepository.php +++ b/src/Domain/Messaging/Repository/MessageRepository.php @@ -158,6 +158,32 @@ public function findByIdAndStatus(int $id, Message\MessageStatus $status): ?Mess ->getOneOrNullResult(); } + /** + * Atomically claims a campaign for processing by flipping its status from Submitted to + * Prepared in a single UPDATE ... WHERE statement, so two concurrent workers can't both + * pass a check-then-act race and process the same campaign. + */ + public function tryClaimForProcessing(int $id): ?Message + { + $connection = $this->getEntityManager()->getConnection(); + $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + + $affected = $connection->executeStatement( + "UPDATE $table SET status = :to WHERE id = :id AND status = :from", + [ + 'to' => Message\MessageStatus::Prepared->value, + 'id' => $id, + 'from' => Message\MessageStatus::Submitted->value, + ] + ); + + if ($affected === 0) { + return null; + } + + return $this->find($id); + } + public function getNonEmptyFields(int $id): array { $message = $this->createQueryBuilder('m') diff --git a/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php b/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php index 7bd83207..baee85b2 100644 --- a/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php +++ b/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php @@ -221,6 +221,44 @@ public function testGetFilteredAfterIdDefaultsToAscendingOrder(): void self::assertSame($second->getId(), $result->getItems()[1]->getId()); } + public function testTryClaimForProcessingClaimsSubmittedCampaign(): void + { + $message = $this->persistMessage(Message\MessageStatus::Submitted, 'Ready to send'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + $claimed = $this->messageRepository->tryClaimForProcessing($id); + + self::assertNotNull($claimed); + self::assertSame($id, $claimed->getId()); + self::assertSame(Message\MessageStatus::Prepared, $claimed->getMetadata()->getStatus()); + } + + public function testTryClaimForProcessingReturnsNullWhenNotSubmitted(): void + { + $message = $this->persistMessage(Message\MessageStatus::Draft, 'Not ready yet'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + self::assertNull($this->messageRepository->tryClaimForProcessing($id)); + } + + public function testTryClaimForProcessingCannotClaimTwice(): void + { + $message = $this->persistMessage(Message\MessageStatus::Submitted, 'Only one winner'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + $firstClaim = $this->messageRepository->tryClaimForProcessing($id); + $secondClaim = $this->messageRepository->tryClaimForProcessing($id); + + self::assertNotNull($firstClaim); + self::assertNull($secondClaim); + } + public function testGetFilteredAfterIdSortsDescendingAndCursorsBackward(): void { $first = $this->persistMessage(Message\MessageStatus::Sent, 'First'); diff --git a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php index 683c215a..1f8424f0 100644 --- a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php +++ b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php @@ -14,7 +14,6 @@ use PhpList\Core\Domain\Messaging\Model\Message; use PhpList\Core\Domain\Messaging\Model\Message\MessageContent; use PhpList\Core\Domain\Messaging\Model\Message\MessageMetadata; -use PhpList\Core\Domain\Messaging\Model\Message\MessageStatus; use PhpList\Core\Domain\Messaging\Repository\MessageRepository; use PhpList\Core\Domain\Messaging\Repository\UserMessageRepository; use PhpList\Core\Domain\Messaging\Service\Builder\EmailBuilder; @@ -101,8 +100,8 @@ public function testInvokeWhenCampaignNotFound(): void $message = new CampaignProcessorMessage(999); $this->messageRepository->expects($this->once()) - ->method('findByIdAndStatus') - ->with(999, MessageStatus::Submitted) + ->method('tryClaimForProcessing') + ->with(999) ->willReturn(null); $this->translator->method('trans')->willReturnCallback(fn(string $msg) => $msg); @@ -122,8 +121,8 @@ public function testInvokeWithNoSubscribers(): void $campaign->method('getId')->willReturn(1); $data = new CampaignProcessorMessage(1); - $this->messageRepository->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -156,8 +155,8 @@ public function testInvokeWithInvalidSubscriberEmail(): void $campaign->method('getId')->willReturn(1); $data = new CampaignProcessorMessage(1); - $this->messageRepository->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -203,8 +202,8 @@ public function testInvokeWithValidSubscriberEmail(): void $campaign->method('getId')->willReturn(1); $data = new CampaignProcessorMessage(1); - $this->messageRepository->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -271,8 +270,8 @@ public function testInvokeWithMailerException(): void $campaign->method('getId')->willReturn(123); $data = new CampaignProcessorMessage(123); - $this->messageRepository->method('findByIdAndStatus') - ->with(123, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(123) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -348,8 +347,8 @@ public function testInvokeWithMultipleSubscribers(): void $data = new CampaignProcessorMessage(1); $this->messageRepository - ->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + ->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService From d84ad2525d5456013897cf7b9b0b299f35ee6631 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Mon, 7 Sep 2026 15:07:08 +0400 Subject: [PATCH 02/10] feat: add exclude list functionality to CampaignProcessorMessageHandler --- .env.dist | 1 + config/parameters.yml | 1 + .../CampaignProcessorMessageHandler.php | 76 +++++-- .../Repository/MessageRepository.php | 2 +- .../Repository/SubscriberRepository.php | 46 ++++ .../Service/Provider/SubscriberProvider.php | 31 ++- .../Repository/SubscriberRepositoryTest.php | 62 ++++++ .../CampaignProcessorMessageHandlerTest.php | 209 +++++++++++++++++- .../Provider/SubscriberProviderTest.php | 43 +++- 9 files changed, 437 insertions(+), 34 deletions(-) diff --git a/.env.dist b/.env.dist index e66a76e4..57924781 100644 --- a/.env.dist +++ b/.env.dist @@ -81,6 +81,7 @@ MESSAGING_MAX_PROCESS_TIME=600 MAX_MAILSIZE=209715200 DEFAULT_MESSAGEAGE=691200 USE_MANUAL_TEXT_PART=0 +USE_LIST_EXCLUDE=0 MESSAGING_BLACKLIST_GRACE_TIME=600 GOOGLE_SENDERID= USE_AMAZONSES=0 diff --git a/config/parameters.yml b/config/parameters.yml index 1edd106d..be4e0307 100644 --- a/config/parameters.yml +++ b/config/parameters.yml @@ -79,6 +79,7 @@ parameters: messaging.max_mail_size: '%env(MAX_MAILSIZE)%' messaging.default_message_age: '%env(DEFAULT_MESSAGEAGE)%' messaging.use_manual_text_part: '%env(USE_MANUAL_TEXT_PART)%' + messaging.use_list_exclude: '%env(bool:USE_LIST_EXCLUDE)%' messaging.blacklist_grace_time: '%env(MESSAGING_BLACKLIST_GRACE_TIME)%' messaging.google_sender_id: '%env(GOOGLE_SENDERID)%' messaging.use_amazon_ses: '%env(USE_AMAZONSES)%' diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index 576c0ae4..134d4f06 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -73,6 +73,7 @@ public function __construct( private readonly MailSizeChecker $mailSizeChecker, private readonly ConfigProvider $configProvider, #[Autowire('%imap_bounce.email%')] private readonly string $bounceEmail, + #[Autowire('%messaging.use_list_exclude%')] private readonly bool $useListExclude = false, ) { } @@ -122,31 +123,16 @@ public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $ $this->handleAdminNotifications($campaign, $loadedMessageData, $data->getMessageId()); // Campaign was already atomically claimed into Prepared status above. - $subscribers = $this->subscriberProvider->getSubscribersForMessageOrLists($data, $campaign); + $excludeListIds = $this->getExcludeListIds($loadedMessageData); + $this->markExcludedSubscribers($campaign, $excludeListIds); + $subscribers = $this->subscriberProvider->getSubscribersForMessageOrLists( + $data, + $campaign, + $excludeListIds + ); $this->updateMessageStatus($campaign, MessageStatus::InProcess); -// if (USE_LIST_EXCLUDE) { -// if (VERBOSE) { -// processQueueOutput(s('looking for users who can be excluded from this mailing')); -// } -// if (count($msgdata['excludelist'])) { -// $query -// = ' select userid' -// .' from '.$GLOBALS['tables']['listuser'] -// .' where listid in ('.implode(',', $msgdata['excludelist']).')'; -// if (VERBOSE) { -// processQueueOutput('Exclude query '.$query); -// } -// $req = Sql_Query($query); -// while ($row = Sql_Fetch_Row($req)) { -// $um = Sql_Query(sprintf('replace into %s (entered,userid,messageid,status) -// values(now(),%d,%d,"excluded")', -// $tables['usermessage'], $row[0], $messageid)); -// } -// } -// } - $stoppedEarly = $this->processSubscribersForCampaign($campaign, $subscribers, $cacheKey); if ($stoppedEarly && $this->requeueHandler->handle($campaign)) { @@ -157,6 +143,52 @@ public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $ $this->updateMessageStatus($campaign, MessageStatus::Sent); } + /** + * Exclude-list IDs are stored via MessageData as an array keyed by list ID e.g. [3 => 1, 7 => 1]. + * + * @return int[] + */ + private function getExcludeListIds(array $loadedMessageData): array + { + if (!$this->useListExclude) { + return []; + } + + $excludeList = $loadedMessageData['excludelist'] ?? []; + if (!is_array($excludeList) || $excludeList === []) { + return []; + } + + return array_values(array_filter(array_map( + static fn (mixed $key): ?int => is_numeric($key) ? (int) $key : null, + array_keys($excludeList) + ), static fn (?int $id): bool => $id !== null)); + } + + /** + * pre-marking of exclude-list members as "excluded" in usermessage before the main send loop runs, + * so there's a persisted audit trail for why a subscriber wasn't sent to. Skips + * subscribers who already have a nontodo UserMessage for this campaign, so a later run + * can't clobber an already-recorded Sent/NotSent/etc. status from an earlier partial run. + */ + private function markExcludedSubscribers(Message $campaign, array $excludeListIds): void + { + if ($excludeListIds === []) { + return; + } + + foreach ($this->subscriberProvider->getExcludedSubscribers($excludeListIds) as $subscriber) { + $existing = $this->userMessageRepository->findByUserAndMessage($subscriber, $campaign); + if ($existing && $existing->getStatus() !== UserMessageStatus::Todo) { + continue; + } + + $userMessage = $existing ?? new UserMessage($subscriber, $campaign); + $userMessage->setStatus(UserMessageStatus::Excluded); + $this->userMessageRepository->save($userMessage); + } + } + private function unconfirmSubscriber(Subscriber $subscriber): void { if ($subscriber->isConfirmed()) { diff --git a/src/Domain/Messaging/Repository/MessageRepository.php b/src/Domain/Messaging/Repository/MessageRepository.php index 076bfd27..443160d5 100644 --- a/src/Domain/Messaging/Repository/MessageRepository.php +++ b/src/Domain/Messaging/Repository/MessageRepository.php @@ -169,7 +169,7 @@ public function tryClaimForProcessing(int $id): ?Message $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); $affected = $connection->executeStatement( - "UPDATE $table SET status = :to WHERE id = :id AND status = :from", + sprintf('UPDATE %s SET status = :to WHERE id = :id AND status = :from', $table), [ 'to' => Message\MessageStatus::Prepared->value, 'id' => $id, diff --git a/src/Domain/Subscription/Repository/SubscriberRepository.php b/src/Domain/Subscription/Repository/SubscriberRepository.php index e9def63e..2f20b334 100644 --- a/src/Domain/Subscription/Repository/SubscriberRepository.php +++ b/src/Domain/Subscription/Repository/SubscriberRepository.php @@ -73,6 +73,52 @@ public function getSubscribersBySubscribedListId(int $listId): array ->getResult(); } + /** + * Same as getSubscribersBySubscribedListId(), but restricted to subscribers who are + * confirmed and not disabled - i.e. eligible to receive a campaign. Blacklisting is + * intentionally not filtered here since it's checked live against UserBlacklistRepository + * at send time instead of the (potentially stale) Subscriber::$blacklisted flag. + * + * @return Subscriber[] + */ + public function getSendableSubscribersBySubscribedListId(int $listId): array + { + return $this->createQueryBuilder('s') + ->innerJoin('s.subscriptions', 'subscription') + ->innerJoin('subscription.subscriberList', 'list') + ->where('list.id = :listId') + ->andWhere('s.confirmed = :confirmed') + ->andWhere('s.disabled = :disabled') + ->setParameter('listId', $listId) + ->setParameter('confirmed', true) + ->setParameter('disabled', false) + ->getQuery() + ->getResult(); + } + + /** + * Returns all subscribers on any of the given lists, regardless of confirmed/disabled + * status - used to resolve campaign exclude-lists, where membership alone is enough + * to suppress a send. + * + * @param int[] $listIds + * @return Subscriber[] + */ + public function getSubscribersBySubscribedListIds(array $listIds): array + { + if ($listIds === []) { + return []; + } + + return $this->createQueryBuilder('s') + ->innerJoin('s.subscriptions', 'subscription') + ->innerJoin('subscription.subscriberList', 'list') + ->where('list.id IN (:listIds)') + ->setParameter('listIds', $listIds) + ->getQuery() + ->getResult(); + } + /** * @return PaginatedResult * @throws InvalidArgumentException diff --git a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php index 758db32e..85308a5b 100644 --- a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php +++ b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php @@ -29,10 +29,15 @@ public function __construct( * * @param CampaignProcessorMessageInterface $data * @param Message $campaign + * @param int[] $excludeListIds List IDs whose members should be suppressed from the send, + * regardless of their confirmed/disabled status. * @return Subscriber[] Array of subscribers */ - public function getSubscribersForMessageOrLists(CampaignProcessorMessageInterface $data, Message $campaign): array - { + public function getSubscribersForMessageOrLists( + CampaignProcessorMessageInterface $data, + Message $campaign, + array $excludeListIds = [], + ): array { if ($data instanceof TestCampaignProcessorMessage) { return $this->subscriberRepository->getByEmails($data->getSubscriberEmails()); } @@ -45,12 +50,32 @@ public function getSubscribersForMessageOrLists(CampaignProcessorMessageInterfac $subscribers = []; foreach ($listIds as $listId) { - $listSubscribers = $this->subscriberRepository->getSubscribersBySubscribedListId($listId); + $listSubscribers = $this->subscriberRepository->getSendableSubscribersBySubscribedListId($listId); foreach ($listSubscribers as $subscriber) { $subscribers[$subscriber->getEmail()] = $subscriber; } } + foreach ($this->getExcludedSubscribers($excludeListIds) as $excluded) { + unset($subscribers[$excluded->getEmail()]); + } + return array_values($subscribers); } + + /** + * Resolves the subscribers on the given exclude-lists, regardless of confirmed/disabled + * status - membership alone is enough to suppress Sand. + * + * @param int[] $excludeListIds + * @return Subscriber[] + */ + public function getExcludedSubscribers(array $excludeListIds): array + { + if ($excludeListIds === []) { + return []; + } + + return $this->subscriberRepository->getSubscribersBySubscribedListIds($excludeListIds); + } } diff --git a/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php b/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php index 2aa89b25..1a091bae 100644 --- a/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php +++ b/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php @@ -8,6 +8,7 @@ use Doctrine\DBAL\Exception\UniqueConstraintViolationException; use Doctrine\ORM\Tools\SchemaTool; use PhpList\Core\Domain\Subscription\Model\Subscriber; +use PhpList\Core\Domain\Subscription\Model\SubscriberList; use PhpList\Core\Domain\Subscription\Model\Subscription; use PhpList\Core\Domain\Subscription\Repository\SubscriberRepository; use PhpList\Core\Domain\Subscription\Repository\SubscriptionRepository; @@ -239,4 +240,65 @@ public function testRemoveRemovesModel() $numberOfModelsAfterRemove = count($this->subscriberRepository->findAll()); self::assertSame(1, $numberOfModelsBeforeRemove - $numberOfModelsAfterRemove); } + + private function subscribe(Subscriber $subscriber, SubscriberList $list): void + { + $subscription = (new Subscription()) + ->setSubscriber($subscriber) + ->setSubscriberList($list); + $this->entityManager->persist($subscription); + } + + public function testGetSendableSubscribersBySubscribedListIdExcludesUnconfirmedAndDisabled(): void + { + $list = (new SubscriberList())->setName('list'); + $this->entityManager->persist($list); + + $confirmed = (new Subscriber('confirmed@example.com'))->setConfirmed(true); + $unconfirmed = (new Subscriber('unconfirmed@example.com'))->setConfirmed(false); + $disabled = (new Subscriber('disabled@example.com'))->setConfirmed(true)->setDisabled(true); + foreach ([$confirmed, $unconfirmed, $disabled] as $subscriber) { + $this->entityManager->persist($subscriber); + $this->subscribe($subscriber, $list); + } + $this->entityManager->flush(); + + $result = $this->subscriberRepository->getSendableSubscribersBySubscribedListId($list->getId()); + + self::assertTrue(in_array($confirmed, $result, true)); + self::assertFalse(in_array($unconfirmed, $result, true)); + self::assertFalse(in_array($disabled, $result, true)); + } + + public function testGetSubscribersBySubscribedListIdsReturnsMembersOfAnyGivenList(): void + { + $listA = (new SubscriberList())->setName('a'); + $listB = (new SubscriberList())->setName('b'); + $listC = (new SubscriberList())->setName('c'); + $this->entityManager->persist($listA); + $this->entityManager->persist($listB); + $this->entityManager->persist($listC); + + $inA = new Subscriber('in-a@example.com'); + $inB = new Subscriber('in-b@example.com'); + $inC = new Subscriber('in-c@example.com'); + $this->entityManager->persist($inA); + $this->entityManager->persist($inB); + $this->entityManager->persist($inC); + $this->subscribe($inA, $listA); + $this->subscribe($inB, $listB); + $this->subscribe($inC, $listC); + $this->entityManager->flush(); + + $result = $this->subscriberRepository->getSubscribersBySubscribedListIds([$listA->getId(), $listB->getId()]); + + self::assertTrue(in_array($inA, $result, true)); + self::assertTrue(in_array($inB, $result, true)); + self::assertFalse(in_array($inC, $result, true)); + } + + public function testGetSubscribersBySubscribedListIdsReturnsEmptyArrayForEmptyInput(): void + { + self::assertSame([], $this->subscriberRepository->getSubscribersBySubscribedListIds([])); + } } diff --git a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php index 1f8424f0..72529cae 100644 --- a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php +++ b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php @@ -14,6 +14,8 @@ use PhpList\Core\Domain\Messaging\Model\Message; use PhpList\Core\Domain\Messaging\Model\Message\MessageContent; use PhpList\Core\Domain\Messaging\Model\Message\MessageMetadata; +use PhpList\Core\Domain\Messaging\Model\Message\UserMessageStatus; +use PhpList\Core\Domain\Messaging\Model\UserMessage; use PhpList\Core\Domain\Messaging\Repository\MessageRepository; use PhpList\Core\Domain\Messaging\Repository\UserMessageRepository; use PhpList\Core\Domain\Messaging\Service\Builder\EmailBuilder; @@ -51,6 +53,9 @@ class CampaignProcessorMessageHandlerTest extends TestCase private MessagePrecacheService|MockObject $precacheService; private CacheInterface|MockObject $cache; private MailerInterface|MockObject $symfonyMailer; + private UserMessageRepository|MockObject $userMessageRepository; + private MaxProcessTimeLimiter|MockObject $timeLimiter; + private RequeueHandler|MockObject $requeueHandler; protected function setUp(): void { @@ -71,7 +76,16 @@ protected function setUp(): void $timeLimiter->method('start'); $timeLimiter->method('shouldStop')->willReturn(false); - $this->handler = new CampaignProcessorMessageHandler( + $this->userMessageRepository = $userMessageRepository; + $this->timeLimiter = $timeLimiter; + $this->requeueHandler = $requeueHandler; + + $this->handler = $this->createHandler(); + } + + private function createHandler(bool $useListExclude = false): CampaignProcessorMessageHandler + { + return new CampaignProcessorMessageHandler( mailer: $this->symfonyMailer, rateLimitedCampaignMailer: $this->mailer, entityManager: $this->entityManager, @@ -79,9 +93,9 @@ protected function setUp(): void messagePreparator: $this->messagePreparator, logger: $this->logger, cache: $this->cache, - userMessageRepository: $userMessageRepository, - timeLimiter: $timeLimiter, - requeueHandler: $requeueHandler, + userMessageRepository: $this->userMessageRepository, + timeLimiter: $this->timeLimiter, + requeueHandler: $this->requeueHandler, translator: $this->translator, subscriberHistoryManager: $this->createMock(SubscriberHistoryManager::class), messageRepository: $this->messageRepository, @@ -92,6 +106,7 @@ protected function setUp(): void mailSizeChecker: $this->createMock(MailSizeChecker::class), configProvider: $this->createMock(ConfigProvider::class), bounceEmail: 'bounce@email.com', + useListExclude: $useListExclude, ); } @@ -147,6 +162,192 @@ public function testInvokeWithNoSubscribers(): void ($this->handler)($data); } + public function testInvokePassesExcludeListIdsFromMessageDataToSubscriberProviderWhenEnabled(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1, 66 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, [55, 66]) + ->willReturn([]); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + + public function testInvokeIgnoresExcludeListWhenUseListExcludeDisabled(): void + { + $handler = $this->createHandler(useListExclude: false); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1, 66 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, []) + ->willReturn([]); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + + public function testInvokeMarksExcludedSubscribersAsExcludedInUserMessage(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $excludedSubscriber = $this->createMock(Subscriber::class); + $excludedSubscriber->method('getEmail')->willReturn('excluded@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getExcludedSubscribers') + ->with([55]) + ->willReturn([$excludedSubscriber]); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, [55]) + ->willReturn([]); + + $this->userMessageRepository->expects($this->once()) + ->method('findByUserAndMessage') + ->with($excludedSubscriber, $campaign) + ->willReturn(null); + + $this->userMessageRepository->expects($this->once()) + ->method('save') + ->with($this->callback( + fn (UserMessage $userMessage): bool => $userMessage->getUser() === $excludedSubscriber + && $userMessage->getStatus() === UserMessageStatus::Excluded + )); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + + public function testInvokeDoesNotOverwriteExistingNonTodoUserMessageWhenMarkingExcluded(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $excludedSubscriber = $this->createMock(Subscriber::class); + $excludedSubscriber->method('getEmail')->willReturn('already-sent@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getExcludedSubscribers') + ->with([55]) + ->willReturn([$excludedSubscriber]); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->willReturn([]); + + $existingUserMessage = $this->createMock(UserMessage::class); + $existingUserMessage->method('getStatus')->willReturn(UserMessageStatus::Sent); + + $this->userMessageRepository->expects($this->once()) + ->method('findByUserAndMessage') + ->with($excludedSubscriber, $campaign) + ->willReturn($existingUserMessage); + + $this->userMessageRepository->expects($this->never()) + ->method('save'); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + public function testInvokeWithInvalidSubscriberEmail(): void { $campaign = $this->createCampaignMock(); diff --git a/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php b/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php index a68576a1..5ba8c29d 100644 --- a/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php +++ b/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php @@ -39,7 +39,7 @@ public function testGetSubscribersForMessageWithNoListsReturnsEmptyArray(): void $this->subscriberRepository ->expects($this->never()) - ->method('getSubscribersBySubscribedListId'); + ->method('getSendableSubscribersBySubscribedListId'); $result = $this->subscriberProvider->getSubscribersForMessageOrLists( $this->createMock(CampaignProcessorMessageInterface::class), @@ -60,7 +60,7 @@ public function testGetSubscribersForMessageWithOneListButNoSubscribersReturnsEm $this->subscriberRepository ->expects($this->once()) - ->method('getSubscribersBySubscribedListId') + ->method('getSendableSubscribersBySubscribedListId') ->with(456) ->willReturn([]); @@ -87,7 +87,7 @@ public function testGetSubscribersForMessageWithOneListAndSubscribersReturnsSubs $this->subscriberRepository ->expects($this->once()) - ->method('getSubscribersBySubscribedListId') + ->method('getSendableSubscribersBySubscribedListId') ->with(456) ->willReturn([$subscriber1, $subscriber2]); @@ -118,7 +118,7 @@ public function testGetSubscribersForMessageWithMultipleListsReturnsUniqueSubscr $this->subscriberRepository ->expects($this->exactly(2)) - ->method('getSubscribersBySubscribedListId') + ->method('getSendableSubscribersBySubscribedListId') ->willReturnMap([ [456, [$subscriber1, $subscriber2]], [789, [$subscriber2, $subscriber3]], @@ -134,4 +134,39 @@ public function testGetSubscribersForMessageWithMultipleListsReturnsUniqueSubscr $this->assertContains($subscriber2, $result); $this->assertContains($subscriber3, $result); } + + public function testGetSubscribersForMessageExcludesSubscribersOnExcludeLists(): void + { + $message = $this->createMock(Message::class); + $message->method('getId')->willReturn(123); + + $this->subscriberListRepository + ->method('getListIdsByMessage') + ->willReturn([456]); + + $subscriber1 = $this->createMock(Subscriber::class); + $subscriber1->method('getEmail')->willReturn('keep@example.am'); + $subscriber2 = $this->createMock(Subscriber::class); + $subscriber2->method('getEmail')->willReturn('exclude@example.am'); + + $this->subscriberRepository + ->method('getSendableSubscribersBySubscribedListId') + ->with(456) + ->willReturn([$subscriber1, $subscriber2]); + + $this->subscriberRepository + ->expects($this->once()) + ->method('getSubscribersBySubscribedListIds') + ->with([789]) + ->willReturn([$subscriber2]); + + $result = $this->subscriberProvider->getSubscribersForMessageOrLists( + $this->createMock(CampaignProcessorMessageInterface::class), + $message, + [789], + ); + + $this->assertCount(1, $result); + $this->assertSame($subscriber1, $result[0]); + } } From 81674172accc15916e93b2eb382cfbaa0d56c571 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Tue, 8 Sep 2026 12:21:45 +0400 Subject: [PATCH 03/10] feat: implement domain rate limiting functionality --- .env.dist | 3 + config/parameters.yml | 3 + config/services/services.yml | 8 ++ .../CampaignProcessorMessageHandler.php | 2 + .../Messaging/Service/DomainRateLimiter.php | 86 +++++++++++++++++++ .../Service/Provider/SubscriberProvider.php | 2 +- .../Service/DomainRateLimiterTest.php | 80 +++++++++++++++++ 7 files changed, 183 insertions(+), 1 deletion(-) create mode 100644 src/Domain/Messaging/Service/DomainRateLimiter.php create mode 100644 tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php diff --git a/.env.dist b/.env.dist index 57924781..09e96ee5 100644 --- a/.env.dist +++ b/.env.dist @@ -82,6 +82,9 @@ MAX_MAILSIZE=209715200 DEFAULT_MESSAGEAGE=691200 USE_MANUAL_TEXT_PART=0 USE_LIST_EXCLUDE=0 +USE_DOMAIN_THROTTLE=0 +DOMAIN_BATCH_SIZE=1 +DOMAIN_BATCH_PERIOD=120 MESSAGING_BLACKLIST_GRACE_TIME=600 GOOGLE_SENDERID= USE_AMAZONSES=0 diff --git a/config/parameters.yml b/config/parameters.yml index be4e0307..e79c6423 100644 --- a/config/parameters.yml +++ b/config/parameters.yml @@ -80,6 +80,9 @@ parameters: messaging.default_message_age: '%env(DEFAULT_MESSAGEAGE)%' messaging.use_manual_text_part: '%env(USE_MANUAL_TEXT_PART)%' messaging.use_list_exclude: '%env(bool:USE_LIST_EXCLUDE)%' + messaging.use_domain_throttle: '%env(bool:USE_DOMAIN_THROTTLE)%' + messaging.domain_batch_size: '%env(int:DOMAIN_BATCH_SIZE)%' + messaging.domain_batch_period: '%env(int:DOMAIN_BATCH_PERIOD)%' messaging.blacklist_grace_time: '%env(MESSAGING_BLACKLIST_GRACE_TIME)%' messaging.google_sender_id: '%env(GOOGLE_SENDERID)%' messaging.use_amazon_ses: '%env(USE_AMAZONSES)%' diff --git a/config/services/services.yml b/config/services/services.yml index 9e527c13..dd2beced 100644 --- a/config/services/services.yml +++ b/config/services/services.yml @@ -73,6 +73,14 @@ services: $mailqueueBatchPeriod: '%messaging.mail_queue_period%' $mailqueueThrottle: '%messaging.mail_queue_throttle%' + PhpList\Core\Domain\Messaging\Service\DomainRateLimiter: + autowire: true + autoconfigure: true + arguments: + $enabled: '%messaging.use_domain_throttle%' + $domainBatchSize: '%messaging.domain_batch_size%' + $domainBatchPeriod: '%messaging.domain_batch_period%' + PhpList\Core\Domain\Common\SystemInfoCollector: autowire: true autoconfigure: true diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index 134d4f06..ec6af2e7 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -25,6 +25,7 @@ use PhpList\Core\Domain\Messaging\Repository\UserMessageRepository; use PhpList\Core\Domain\Messaging\Service\Builder\EmailBuilder; use PhpList\Core\Domain\Messaging\Service\Builder\SystemEmailBuilder; +use PhpList\Core\Domain\Messaging\Service\DomainRateLimiter; use PhpList\Core\Domain\Messaging\Service\Handler\RequeueHandler; use PhpList\Core\Domain\Messaging\Service\MailSizeChecker; use PhpList\Core\Domain\Messaging\Service\MaxProcessTimeLimiter; @@ -72,6 +73,7 @@ public function __construct( private readonly EmailBuilder $campaignEmailBuilder, private readonly MailSizeChecker $mailSizeChecker, private readonly ConfigProvider $configProvider, + private readonly DomainRateLimiter $domainRateLimiter, #[Autowire('%imap_bounce.email%')] private readonly string $bounceEmail, #[Autowire('%messaging.use_list_exclude%')] private readonly bool $useListExclude = false, ) { diff --git a/src/Domain/Messaging/Service/DomainRateLimiter.php b/src/Domain/Messaging/Service/DomainRateLimiter.php new file mode 100644 index 00000000..10374e03 --- /dev/null +++ b/src/Domain/Messaging/Service/DomainRateLimiter.php @@ -0,0 +1,86 @@ + */ + private array $buckets = []; + + public function __construct( + private readonly bool $enabled = false, + private readonly int $domainBatchSize = 1, + private readonly int $domainBatchPeriod = 120, + ) { + } + + /** + * Call before attempting to send to $email. Returns false if that recipient's domain + * has already hit its quota for the current window and the send should be deferred. + */ + public function canSendTo(string $email): bool + { + if (!$this->enabled || $this->domainBatchSize <= 0 || $this->domainBatchPeriod <= 0) { + return true; + } + + $domain = $this->extractDomain($email); + if ($domain === null) { + return true; + } + + return $this->currentBucket($domain)['sent'] < $this->domainBatchSize; + } + + /** + * Call once a send to $email has been attempted, to count it against that domain's quota. + */ + public function recordSend(string $email): void + { + if (!$this->enabled) { + return; + } + + $domain = $this->extractDomain($email); + if ($domain === null) { + return; + } + + $bucket = $this->currentBucket($domain); + $bucket['sent']++; + $this->buckets[$domain] = $bucket; + } + + /** @return array{start: float, sent: int} */ + private function currentBucket(string $domain): array + { + $now = microtime(true); + $bucket = $this->buckets[$domain] ?? ['start' => $now, 'sent' => 0]; + + if ($now - $bucket['start'] >= $this->domainBatchPeriod) { + $bucket = ['start' => $now, 'sent' => 0]; + } + + $this->buckets[$domain] = $bucket; + + return $bucket; + } + + private function extractDomain(string $email): ?string + { + $atPosition = strrpos($email, '@'); + if ($atPosition === false) { + return null; + } + + return strtolower(substr($email, $atPosition + 1)); + } +} diff --git a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php index 85308a5b..31474ef6 100644 --- a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php +++ b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php @@ -65,7 +65,7 @@ public function getSubscribersForMessageOrLists( /** * Resolves the subscribers on the given exclude-lists, regardless of confirmed/disabled - * status - membership alone is enough to suppress Sand. + * status - membership alone is enough to suppress a send. * * @param int[] $excludeListIds * @return Subscriber[] diff --git a/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php new file mode 100644 index 00000000..8a885552 --- /dev/null +++ b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php @@ -0,0 +1,80 @@ +assertTrue($limiter->canSendTo('a@example.com')); + $limiter->recordSend('a@example.com'); + $this->assertTrue($limiter->canSendTo('a@example.com')); + } + + public function testAllowsSendsWhenBatchSizeOrPeriodIsNotPositive(): void + { + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 0, domainBatchPeriod: 120); + $this->assertTrue($limiter->canSendTo('a@example.com')); + + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 0); + $this->assertTrue($limiter->canSendTo('a@example.com')); + } + + public function testBlocksSendsToSameDomainOnceQuotaReached(): void + { + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 2, domainBatchPeriod: 120); + + $this->assertTrue($limiter->canSendTo('first@example.com')); + $limiter->recordSend('first@example.com'); + + $this->assertTrue($limiter->canSendTo('second@example.com')); + $limiter->recordSend('second@example.com'); + + $this->assertFalse($limiter->canSendTo('third@example.com')); + } + + public function testTracksEachDomainIndependently(): void + { + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 120); + + $limiter->recordSend('a@example.com'); + + $this->assertFalse($limiter->canSendTo('b@example.com')); + $this->assertTrue($limiter->canSendTo('c@example.org')); + } + + public function testResetsQuotaAfterPeriodElapses(): void + { + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 0); + + $limiter->recordSend('a@example.com'); + + // domainBatchPeriod = 0 means the window is already elapsed on the very next check + $this->assertTrue($limiter->canSendTo('a@example.com')); + } + + public function testTreatsAddressWithoutAtSignAsUnthrottleable(): void + { + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 120); + + $limiter->recordSend('not-an-email'); + + $this->assertTrue($limiter->canSendTo('not-an-email')); + } + + public function testDomainMatchingIsCaseInsensitive(): void + { + $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 120); + + $limiter->recordSend('first@Example.com'); + + $this->assertFalse($limiter->canSendTo('second@example.COM')); + } +} \ No newline at end of file From 023d26eb8a836fecf3b38af5c1ebaaa2500e29be Mon Sep 17 00:00:00 2001 From: Tatevik Date: Tue, 8 Sep 2026 15:07:16 +0400 Subject: [PATCH 04/10] feat: implement domain rate limiting --- .env.dist | 1 + .env.test | 2 + config/parameters.yml | 1 + config/services/repositories.yml | 4 + config/services/services.yml | 1 + .../CampaignProcessorMessageHandler.php | 7 + .../Messaging/Model/DomainThrottleState.php | 62 +++++++++ .../Model/Dto/DomainThrottleReservation.php | 17 +++ .../Model/Dto/DomainThrottleResult.php | 20 +++ .../DomainThrottleStateRepository.php | 130 ++++++++++++++++++ .../Messaging/Service/DomainRateLimiter.php | 100 +++++++++----- .../Service/Handler/RequeueHandler.php | 16 ++- ...08130000MySqlCreateDomainThrottleTable.php | 46 +++++++ ...130001PostGreCreateDomainThrottleTable.php | 46 +++++++ .../DomainThrottleStateRepositoryTest.php | 86 ++++++++++++ .../CampaignProcessorMessageHandlerTest.php | 50 +++++++ .../Service/DomainRateLimiterTest.php | 127 ++++++++++++----- .../Service/Handler/RequeueHandlerTest.php | 32 ++++- 18 files changed, 671 insertions(+), 77 deletions(-) create mode 100644 src/Domain/Messaging/Model/DomainThrottleState.php create mode 100644 src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php create mode 100644 src/Domain/Messaging/Model/Dto/DomainThrottleResult.php create mode 100644 src/Domain/Messaging/Repository/DomainThrottleStateRepository.php create mode 100644 src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php create mode 100644 src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php create mode 100644 tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php diff --git a/.env.dist b/.env.dist index 09e96ee5..cea77ba2 100644 --- a/.env.dist +++ b/.env.dist @@ -85,6 +85,7 @@ USE_LIST_EXCLUDE=0 USE_DOMAIN_THROTTLE=0 DOMAIN_BATCH_SIZE=1 DOMAIN_BATCH_PERIOD=120 +DOMAIN_AUTO_THROTTLE=0 MESSAGING_BLACKLIST_GRACE_TIME=600 GOOGLE_SENDERID= USE_AMAZONSES=0 diff --git a/.env.test b/.env.test index 88b6c338..a23613db 100644 --- a/.env.test +++ b/.env.test @@ -1,3 +1,5 @@ PHPLIST_DATABASE_DRIVER=pdo_sqlite PHPLIST_DATABASE_PATH=:memory: SEARCH_TRANSPORT_DSN=sync:// +ELASTICSEARCH_ENABLED=false + diff --git a/config/parameters.yml b/config/parameters.yml index e79c6423..6ee93ac1 100644 --- a/config/parameters.yml +++ b/config/parameters.yml @@ -83,6 +83,7 @@ parameters: messaging.use_domain_throttle: '%env(bool:USE_DOMAIN_THROTTLE)%' messaging.domain_batch_size: '%env(int:DOMAIN_BATCH_SIZE)%' messaging.domain_batch_period: '%env(int:DOMAIN_BATCH_PERIOD)%' + messaging.domain_auto_throttle: '%env(bool:DOMAIN_AUTO_THROTTLE)%' messaging.blacklist_grace_time: '%env(MESSAGING_BLACKLIST_GRACE_TIME)%' messaging.google_sender_id: '%env(GOOGLE_SENDERID)%' messaging.use_amazon_ses: '%env(USE_AMAZONSES)%' diff --git a/config/services/repositories.yml b/config/services/repositories.yml index 5ee7eb40..4cb9d01b 100644 --- a/config/services/repositories.yml +++ b/config/services/repositories.yml @@ -185,6 +185,10 @@ services: arguments: - PhpList\Core\Domain\Messaging\Model\Attachment + PhpList\Core\Domain\Messaging\Repository\DomainThrottleStateRepository: + parent: PhpList\Core\Domain\Common\Repository\AbstractRepository + arguments: + - PhpList\Core\Domain\Messaging\Model\DomainThrottleState PhpList\Core\Domain\Messaging\Repository\MessageAttachmentRepository: parent: PhpList\Core\Domain\Common\Repository\AbstractRepository arguments: diff --git a/config/services/services.yml b/config/services/services.yml index dd2beced..4d0b4514 100644 --- a/config/services/services.yml +++ b/config/services/services.yml @@ -80,6 +80,7 @@ services: $enabled: '%messaging.use_domain_throttle%' $domainBatchSize: '%messaging.domain_batch_size%' $domainBatchPeriod: '%messaging.domain_batch_period%' + $autoThrottle: '%messaging.domain_auto_throttle%' PhpList\Core\Domain\Common\SystemInfoCollector: autowire: true diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index ec6af2e7..ee0916d6 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -370,6 +370,13 @@ private function processSubscribersForCampaign(Message $campaign, array $subscri continue; } + if (!$this->domainRateLimiter->attemptSend($subscriber->getEmail())->allowed) { + // Leave no UserMessage record so this subscriber is picked up again on a + // later run, once their domain's throttle window has passed. + $stoppedEarly = true; + continue; + } + $userMessage = $existing ?? new UserMessage($subscriber, $campaign); $userMessage->setStatus(UserMessageStatus::Active); $this->userMessageRepository->save($userMessage); diff --git a/src/Domain/Messaging/Model/DomainThrottleState.php b/src/Domain/Messaging/Model/DomainThrottleState.php new file mode 100644 index 00000000..13035049 --- /dev/null +++ b/src/Domain/Messaging/Model/DomainThrottleState.php @@ -0,0 +1,62 @@ +domain = $domain; + $this->windowStart = $windowStart; + $this->sentCount = $sentCount; + $this->blockedCount = $blockedCount; + } + + public function getDomain(): string + { + return $this->domain; + } + + public function getWindowStart(): int + { + return $this->windowStart; + } + + public function getSentCount(): int + { + return $this->sentCount; + } + + public function getBlockedCount(): int + { + return $this->blockedCount; + } +} diff --git a/src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php b/src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php new file mode 100644 index 00000000..652286fb --- /dev/null +++ b/src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php @@ -0,0 +1,17 @@ +getEntityManager()->getConnection(); + $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + + if ($this->incrementSentIfAllowed($connection, $table, $domain, $windowStart, $batchSize)) { + return new DomainThrottleReservation(allowed: true); + } + + if ($this->rolloverWindow($connection, $table, $domain, $windowStart)) { + return new DomainThrottleReservation(allowed: true); + } + + if ($this->insertFirstRow($connection, $table, $domain, $windowStart)) { + return new DomainThrottleReservation(allowed: true); + } + + // Lost the insert race to another worker; its row may already have room in this + // window, so give the increment one more try before concluding we're blocked. + if ($this->incrementSentIfAllowed($connection, $table, $domain, $windowStart, $batchSize)) { + return new DomainThrottleReservation(allowed: true); + } + + return new DomainThrottleReservation( + allowed: false, + blockedAttempts: $this->incrementBlocked($connection, $table, $domain, $windowStart), + ); + } + + public function resetBlockedCount(string $domain, int $windowStart): void + { + $connection = $this->getEntityManager()->getConnection(); + $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + + $connection->executeStatement( + sprintf('UPDATE %s SET blocked_count = 0 WHERE domain = :domain AND window_start = :window', $table), + ['domain' => $domain, 'window' => $windowStart] + ); + } + + /** @phpstan-impure */ + private function incrementSentIfAllowed( + Connection $connection, + string $table, + string $domain, + int $windowStart, + int $batchSize + ): bool { + $affected = $connection->executeStatement( + sprintf( + 'UPDATE %s SET sent_count = sent_count + 1 + WHERE domain = :domain AND window_start = :window AND sent_count < :batchSize', + $table + ), + ['domain' => $domain, 'window' => $windowStart, 'batchSize' => $batchSize] + ); + + return $affected > 0; + } + + /** @phpstan-impure */ + private function rolloverWindow(Connection $connection, string $table, string $domain, int $windowStart): bool + { + $affected = $connection->executeStatement( + sprintf( + 'UPDATE %s SET window_start = :window, sent_count = 1, blocked_count = 0 + WHERE domain = :domain AND window_start < :window', + $table + ), + ['domain' => $domain, 'window' => $windowStart] + ); + + return $affected > 0; + } + + /** @phpstan-impure */ + private function insertFirstRow(Connection $connection, string $table, string $domain, int $windowStart): bool + { + try { + $connection->executeStatement( + sprintf( + 'INSERT INTO %s (domain, window_start, sent_count, blocked_count) VALUES (:domain, :window, 1, 0)', + $table + ), + ['domain' => $domain, 'window' => $windowStart] + ); + + return true; + } catch (UniqueConstraintViolationException) { + return false; + } + } + + /** @phpstan-impure */ + private function incrementBlocked(Connection $connection, string $table, string $domain, int $windowStart): int + { + $connection->executeStatement( + sprintf( + 'UPDATE %s SET blocked_count = blocked_count + 1 WHERE domain = :domain AND window_start = :window', + $table + ), + ['domain' => $domain, 'window' => $windowStart] + ); + + return (int) $connection->fetchOne( + sprintf('SELECT blocked_count FROM %s WHERE domain = :domain AND window_start = :window', $table), + ['domain' => $domain, 'window' => $windowStart] + ); + } +} diff --git a/src/Domain/Messaging/Service/DomainRateLimiter.php b/src/Domain/Messaging/Service/DomainRateLimiter.php index 10374e03..900e2f81 100644 --- a/src/Domain/Messaging/Service/DomainRateLimiter.php +++ b/src/Domain/Messaging/Service/DomainRateLimiter.php @@ -4,74 +4,98 @@ namespace PhpList\Core\Domain\Messaging\Service; +use PhpList\Core\Domain\Messaging\Model\Dto\DomainThrottleResult; +use PhpList\Core\Domain\Messaging\Repository\DomainThrottleStateRepository; +use Psr\Log\LoggerInterface; + /** - * Limits how many sends go to any single recipient domain within a rolling time window. Unlike - * SendRateLimiter, this never sleeps: it just reports whether a domain is over quota right - * now, so the caller can defer that one recipient to a later run instead of blocking the - * whole batch on one busy domain. State is kept in memory only (not seeded from history) + * Limits how many sends go to any single recipient domain within a fixed time window. + * State is persisted via DomainThrottleStateRepository so the quota is shared across + * concurrent queue-processing workers rather than each keeping its own count. Unlike + * SendRateLimiter, this never blocks the whole batch: it just reports whether a domain + * is over quota right now, so the caller can defer that one recipient to a later run + * instead of stalling on one busy domain. */ class DomainRateLimiter { - /** @var array */ - private array $buckets = []; + /** + * Matches phpList3's threshold for triggering auto-throttle backoff: skip a run of + * blocked attempts before introducing extra delay, so a handful of early blocks + * (normal while a window fills up) don't immediately trigger backoff. + */ + private const AUTO_THROTTLE_ATTEMPT_THRESHOLD = 25; public function __construct( + private readonly DomainThrottleStateRepository $repository, + private readonly LoggerInterface $logger, private readonly bool $enabled = false, private readonly int $domainBatchSize = 1, private readonly int $domainBatchPeriod = 120, + private readonly bool $autoThrottle = false, ) { } /** - * Call before attempting to send to $email. Returns false if that recipient's domain - * has already hit its quota for the current window and the send should be deferred. + * Call before sending to $email. Atomically reserves a send slot for that recipient's + * domain when quota allows; when quota is exhausted, records the blocked attempt and, + * if DOMAIN_AUTO_THROTTLE is enabled and blocked attempts have piled up, sleeps for a + * short backoff before returning. */ - public function canSendTo(string $email): bool + public function attemptSend(string $email): DomainThrottleResult { if (!$this->enabled || $this->domainBatchSize <= 0 || $this->domainBatchPeriod <= 0) { - return true; + return new DomainThrottleResult(allowed: true, domain: null); } $domain = $this->extractDomain($email); if ($domain === null) { - return true; + return new DomainThrottleResult(allowed: true, domain: null); } - return $this->currentBucket($domain)['sent'] < $this->domainBatchSize; - } + $windowStart = intdiv(time(), $this->domainBatchPeriod) * $this->domainBatchPeriod; + $reservation = $this->repository->tryReserveSlot($domain, $windowStart, $this->domainBatchSize); - /** - * Call once a send to $email has been attempted, to count it against that domain's quota. - */ - public function recordSend(string $email): void - { - if (!$this->enabled) { - return; + if ($reservation->allowed) { + return new DomainThrottleResult(allowed: true, domain: $domain); } - $domain = $this->extractDomain($email); - if ($domain === null) { - return; - } + $this->logger->info('Send blocked by domain throttle', [ + 'domain' => $domain, + 'blocked_attempts' => $reservation->blockedAttempts, + 'domain_batch_size' => $this->domainBatchSize, + 'domain_batch_period' => $this->domainBatchPeriod, + ]); - $bucket = $this->currentBucket($domain); - $bucket['sent']++; - $this->buckets[$domain] = $bucket; + return $this->applyAutoThrottleIfDue($domain, $windowStart, $reservation->blockedAttempts); } - /** @return array{start: float, sent: int} */ - private function currentBucket(string $domain): array - { - $now = microtime(true); - $bucket = $this->buckets[$domain] ?? ['start' => $now, 'sent' => 0]; - - if ($now - $bucket['start'] >= $this->domainBatchPeriod) { - $bucket = ['start' => $now, 'sent' => 0]; + private function applyAutoThrottleIfDue( + string $domain, + int $windowStart, + int $blockedAttempts + ): DomainThrottleResult { + if (!$this->autoThrottle || $blockedAttempts <= self::AUTO_THROTTLE_ATTEMPT_THRESHOLD) { + return new DomainThrottleResult(allowed: false, domain: $domain, blockedAttempts: $blockedAttempts); } - $this->buckets[$domain] = $bucket; - - return $bucket; + // Reset the trigger counter so it takes another full run of blocked attempts + // before backoff fires again for this domain/window. + $this->repository->resetBlockedCount($domain, $windowStart); + $delaySeconds = max(1, intdiv($this->domainBatchPeriod, max(1, $this->domainBatchSize * 4))); + + $this->logger->info('Introducing extra delay to reduce domain throttle failures', [ + 'domain' => $domain, + 'delay_seconds' => $delaySeconds, + ]); + sleep($delaySeconds); + + return new DomainThrottleResult( + allowed: false, + domain: $domain, + blockedAttempts: $blockedAttempts, + backoffApplied: true, + backoffSeconds: $delaySeconds, + ); } private function extractDomain(string $email): ?string diff --git a/src/Domain/Messaging/Service/Handler/RequeueHandler.php b/src/Domain/Messaging/Service/Handler/RequeueHandler.php index 3fbca634..0d028023 100644 --- a/src/Domain/Messaging/Service/Handler/RequeueHandler.php +++ b/src/Domain/Messaging/Service/Handler/RequeueHandler.php @@ -14,6 +14,16 @@ class RequeueHandler { + /** + * Fallback delay (minutes) used when a campaign stops early (time limit, domain throttle, + * etc.) but has no explicit requeueInterval configured. requeueInterval/requeueUntil control + * *how long* to wait before resuming, not *whether* to resume: a campaign that stopped early + * must always be retried, mirroring phplist3's unconditional "don't mark sent while anything + * failed/was throttled" guard - it must never be silently marked Sent with recipients still + * unprocessed. requeueUntil remains a legitimate opt-out (a real deadline). + */ + private const DEFAULT_REQUEUE_INTERVAL_MINUTES = 1; + public function __construct( private readonly LoggerInterface $logger, private readonly TranslatorInterface $translator, @@ -24,11 +34,11 @@ public function handle(Message $campaign, ?OutputInterface $output = null): bool { $schedule = $campaign->getSchedule(); $interval = $schedule->getRequeueInterval() ?? 0; - $until = $schedule->getRequeueUntil(); - if ($interval <= 0) { - return false; + $interval = self::DEFAULT_REQUEUE_INTERVAL_MINUTES; } + $until = $schedule->getRequeueUntil(); + $now = new DateTime(); if ($until instanceof DateTime && $now > $until) { return false; diff --git a/src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php b/src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php new file mode 100644 index 00000000..f1cddf57 --- /dev/null +++ b/src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php @@ -0,0 +1,46 @@ +connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof MySQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql( + 'CREATE TABLE phplist_domain_throttle ( + domain VARCHAR(255) NOT NULL, + window_start INT NOT NULL, + sent_count INT NOT NULL DEFAULT 0, + blocked_count INT NOT NULL DEFAULT 0, + PRIMARY KEY (domain) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3' + ); + } + + public function down(Schema $schema): void + { + $platform = $this->connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof MySQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql('DROP TABLE phplist_domain_throttle'); + } +} \ No newline at end of file diff --git a/src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php b/src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php new file mode 100644 index 00000000..f7835465 --- /dev/null +++ b/src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php @@ -0,0 +1,46 @@ +connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof PostgreSQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql( + 'CREATE TABLE phplist_domain_throttle ( + domain VARCHAR(255) NOT NULL, + window_start INT NOT NULL, + sent_count INT NOT NULL DEFAULT 0, + blocked_count INT NOT NULL DEFAULT 0, + PRIMARY KEY (domain) + )' + ); + } + + public function down(Schema $schema): void + { + $platform = $this->connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof PostgreSQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql('DROP TABLE phplist_domain_throttle'); + } +} \ No newline at end of file diff --git a/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php b/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php new file mode 100644 index 00000000..38f32fa8 --- /dev/null +++ b/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php @@ -0,0 +1,86 @@ +loadSchema(); + + $this->repository = self::getContainer()->get(DomainThrottleStateRepository::class); + } + + protected function tearDown(): void + { + $schemaTool = new SchemaTool($this->entityManager); + $schemaTool->dropDatabase(); + parent::tearDown(); + } + + public function testFirstReservationForNewDomainIsAllowed(): void + { + $reservation = $this->repository->tryReserveSlot('example.com', 1000, 1); + + $this->assertTrue($reservation->allowed); + $this->assertSame(0, $reservation->blockedAttempts); + } + + public function testReservationBlockedOnceQuotaReachedInSameWindow(): void + { + $this->assertTrue($this->repository->tryReserveSlot('example.com', 1000, 1)->allowed); + + $second = $this->repository->tryReserveSlot('example.com', 1000, 1); + + $this->assertFalse($second->allowed); + $this->assertSame(1, $second->blockedAttempts); + + $third = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertFalse($third->allowed); + $this->assertSame(2, $third->blockedAttempts); + } + + public function testDomainsAreTrackedIndependently(): void + { + $this->assertTrue($this->repository->tryReserveSlot('a.com', 1000, 1)->allowed); + + $this->assertTrue($this->repository->tryReserveSlot('b.com', 1000, 1)->allowed); + $this->assertFalse($this->repository->tryReserveSlot('a.com', 1000, 1)->allowed); + } + + public function testReservationAllowedAgainAfterWindowRollsOver(): void + { + $this->assertTrue($this->repository->tryReserveSlot('example.com', 1000, 1)->allowed); + $this->assertFalse($this->repository->tryReserveSlot('example.com', 1000, 1)->allowed); + + $nextWindow = $this->repository->tryReserveSlot('example.com', 1120, 1); + + $this->assertTrue($nextWindow->allowed); + } + + public function testResetBlockedCountClearsCounterForCurrentWindow(): void + { + $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->repository->tryReserveSlot('example.com', 1000, 1); + $blocked = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertSame(2, $blocked->blockedAttempts); + + $this->repository->resetBlockedCount('example.com', 1000); + + $afterReset = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertFalse($afterReset->allowed); + $this->assertSame(1, $afterReset->blockedAttempts); + } +} diff --git a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php index 72529cae..e9a9b477 100644 --- a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php +++ b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php @@ -10,6 +10,7 @@ use PhpList\Core\Domain\Configuration\Service\Provider\ConfigProvider; use PhpList\Core\Domain\Messaging\Message\CampaignProcessor\CampaignProcessorMessage; use PhpList\Core\Domain\Messaging\MessageHandler\CampaignProcessor\CampaignProcessorMessageHandler; +use PhpList\Core\Domain\Messaging\Model\Dto\DomainThrottleResult; use PhpList\Core\Domain\Messaging\Model\Dto\MessagePrecacheDto; use PhpList\Core\Domain\Messaging\Model\Message; use PhpList\Core\Domain\Messaging\Model\Message\MessageContent; @@ -20,6 +21,7 @@ use PhpList\Core\Domain\Messaging\Repository\UserMessageRepository; use PhpList\Core\Domain\Messaging\Service\Builder\EmailBuilder; use PhpList\Core\Domain\Messaging\Service\Builder\SystemEmailBuilder; +use PhpList\Core\Domain\Messaging\Service\DomainRateLimiter; use PhpList\Core\Domain\Messaging\Service\Handler\RequeueHandler; use PhpList\Core\Domain\Messaging\Service\MailSizeChecker; use PhpList\Core\Domain\Messaging\Service\MaxProcessTimeLimiter; @@ -56,6 +58,7 @@ class CampaignProcessorMessageHandlerTest extends TestCase private UserMessageRepository|MockObject $userMessageRepository; private MaxProcessTimeLimiter|MockObject $timeLimiter; private RequeueHandler|MockObject $requeueHandler; + private DomainRateLimiter|MockObject $domainRateLimiter; protected function setUp(): void { @@ -79,6 +82,9 @@ protected function setUp(): void $this->userMessageRepository = $userMessageRepository; $this->timeLimiter = $timeLimiter; $this->requeueHandler = $requeueHandler; + $this->domainRateLimiter = $this->createMock(DomainRateLimiter::class); + $this->domainRateLimiter->method('attemptSend') + ->willReturn(new DomainThrottleResult(allowed: true, domain: null)); $this->handler = $this->createHandler(); } @@ -105,6 +111,7 @@ private function createHandler(bool $useListExclude = false): CampaignProcessorM campaignEmailBuilder: $this->createMock(EmailBuilder::class), mailSizeChecker: $this->createMock(MailSizeChecker::class), configProvider: $this->createMock(ConfigProvider::class), + domainRateLimiter: $this->domainRateLimiter, bounceEmail: 'bounce@email.com', useListExclude: $useListExclude, ); @@ -664,6 +671,49 @@ function () use (&$buildCampaignEmailCalls): array { $this->assertCount(2, $buildCampaignEmailCalls); } + public function testInvokeSkipsDomainThrottledSubscriberWithoutCreatingUserMessage(): void + { + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $throttledSubscriber = $this->createMock(Subscriber::class); + $throttledSubscriber->method('getEmail')->willReturn('throttled@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->willReturn([$throttledSubscriber]); + + $this->domainRateLimiter = $this->createMock(DomainRateLimiter::class); + $this->domainRateLimiter->method('attemptSend') + ->willReturn(new DomainThrottleResult(allowed: false, domain: 'example.com', blockedAttempts: 1)); + $handler = $this->createHandler(); + + $this->userMessageRepository->expects($this->never()) + ->method('save'); + + $this->requeueHandler->expects($this->once()) + ->method('handle') + ->with($campaign) + ->willReturn(true); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + /** * Creates a mock for the Message class with content */ diff --git a/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php index 8a885552..82af3ef5 100644 --- a/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php +++ b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php @@ -4,77 +4,136 @@ namespace PhpList\Core\Tests\Unit\Domain\Messaging\Service; +use PhpList\Core\Domain\Messaging\Model\Dto\DomainThrottleReservation; +use PhpList\Core\Domain\Messaging\Repository\DomainThrottleStateRepository; use PhpList\Core\Domain\Messaging\Service\DomainRateLimiter; +use PHPUnit\Framework\MockObject\MockObject; use PHPUnit\Framework\TestCase; +use Psr\Log\LoggerInterface; class DomainRateLimiterTest extends TestCase { + private DomainThrottleStateRepository|MockObject $repository; + private LoggerInterface|MockObject $logger; + + protected function setUp(): void + { + $this->repository = $this->createMock(DomainThrottleStateRepository::class); + $this->logger = $this->createMock(LoggerInterface::class); + } + + private function createLimiter( + bool $enabled = true, + int $domainBatchSize = 1, + int $domainBatchPeriod = 120, + bool $autoThrottle = false, + ): DomainRateLimiter { + return new DomainRateLimiter( + repository: $this->repository, + logger: $this->logger, + enabled: $enabled, + domainBatchSize: $domainBatchSize, + domainBatchPeriod: $domainBatchPeriod, + autoThrottle: $autoThrottle, + ); + } + public function testAllowsSendsWhenDisabled(): void { - $limiter = new DomainRateLimiter(enabled: false, domainBatchSize: 1, domainBatchPeriod: 120); + $this->repository->expects($this->never())->method('tryReserveSlot'); + + $limiter = $this->createLimiter(enabled: false); - $this->assertTrue($limiter->canSendTo('a@example.com')); - $limiter->recordSend('a@example.com'); - $this->assertTrue($limiter->canSendTo('a@example.com')); + $this->assertTrue($limiter->attemptSend('a@example.com')->allowed); } public function testAllowsSendsWhenBatchSizeOrPeriodIsNotPositive(): void { - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 0, domainBatchPeriod: 120); - $this->assertTrue($limiter->canSendTo('a@example.com')); + $this->repository->expects($this->never())->method('tryReserveSlot'); - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 0); - $this->assertTrue($limiter->canSendTo('a@example.com')); + $limiter = $this->createLimiter(domainBatchSize: 0); + $this->assertTrue($limiter->attemptSend('a@example.com')->allowed); + + $limiter = $this->createLimiter(domainBatchPeriod: 0); + $this->assertTrue($limiter->attemptSend('a@example.com')->allowed); } - public function testBlocksSendsToSameDomainOnceQuotaReached(): void + public function testAllowsSendsWhenAddressHasNoAtSign(): void { - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 2, domainBatchPeriod: 120); + $this->repository->expects($this->never())->method('tryReserveSlot'); + + $limiter = $this->createLimiter(); + + $this->assertTrue($limiter->attemptSend('not-an-email')->allowed); + } - $this->assertTrue($limiter->canSendTo('first@example.com')); - $limiter->recordSend('first@example.com'); + public function testDelegatesReservationToRepositoryUsingLowercasedDomain(): void + { + $this->repository->expects($this->once()) + ->method('tryReserveSlot') + ->with('example.com', $this->isType('int'), 1) + ->willReturn(new DomainThrottleReservation(allowed: true)); - $this->assertTrue($limiter->canSendTo('second@example.com')); - $limiter->recordSend('second@example.com'); + $limiter = $this->createLimiter(); + $result = $limiter->attemptSend('first@Example.COM'); - $this->assertFalse($limiter->canSendTo('third@example.com')); + $this->assertTrue($result->allowed); + $this->assertSame('example.com', $result->domain); } - public function testTracksEachDomainIndependently(): void + public function testReturnsBlockedResultWithAttemptsWhenQuotaReached(): void { - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 120); + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 3)); + + $this->logger->expects($this->once()) + ->method('info') + ->with('Send blocked by domain throttle', $this->anything()); - $limiter->recordSend('a@example.com'); + $limiter = $this->createLimiter(); + $result = $limiter->attemptSend('third@example.com'); - $this->assertFalse($limiter->canSendTo('b@example.com')); - $this->assertTrue($limiter->canSendTo('c@example.org')); + $this->assertFalse($result->allowed); + $this->assertSame(3, $result->blockedAttempts); + $this->assertFalse($result->backoffApplied); } - public function testResetsQuotaAfterPeriodElapses(): void + public function testDoesNotBackoffWhenAutoThrottleDisabled(): void { - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 0); + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 999)); + $this->repository->expects($this->never())->method('resetBlockedCount'); - $limiter->recordSend('a@example.com'); + $limiter = $this->createLimiter(autoThrottle: false); + $result = $limiter->attemptSend('third@example.com'); - // domainBatchPeriod = 0 means the window is already elapsed on the very next check - $this->assertTrue($limiter->canSendTo('a@example.com')); + $this->assertFalse($result->backoffApplied); } - public function testTreatsAddressWithoutAtSignAsUnthrottleable(): void + public function testDoesNotBackoffBelowAttemptThreshold(): void { - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 120); + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 5)); + $this->repository->expects($this->never())->method('resetBlockedCount'); - $limiter->recordSend('not-an-email'); + $limiter = $this->createLimiter(autoThrottle: true); + $result = $limiter->attemptSend('third@example.com'); - $this->assertTrue($limiter->canSendTo('not-an-email')); + $this->assertFalse($result->backoffApplied); } - public function testDomainMatchingIsCaseInsensitive(): void + public function testAppliesBackoffAndResetsBlockedCountOnceThresholdExceeded(): void { - $limiter = new DomainRateLimiter(enabled: true, domainBatchSize: 1, domainBatchPeriod: 120); + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 26)); + $this->repository->expects($this->once())->method('resetBlockedCount'); - $limiter->recordSend('first@Example.com'); + // Small batch period/size keeps the resulting sleep() short (~1s) so the test stays fast. + $limiter = $this->createLimiter(domainBatchSize: 1, domainBatchPeriod: 4, autoThrottle: true); + $result = $limiter->attemptSend('third@example.com'); - $this->assertFalse($limiter->canSendTo('second@example.COM')); + $this->assertFalse($result->allowed); + $this->assertTrue($result->backoffApplied); + $this->assertGreaterThanOrEqual(1, $result->backoffSeconds); } -} \ No newline at end of file +} diff --git a/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php b/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php index 495f496e..ddef5190 100644 --- a/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php +++ b/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php @@ -51,12 +51,40 @@ private function createMessage( return new Message($format, $schedule, $metadata, $content, $options, owner: null, template: null); } - public function testReturnsFalseWhenIntervalIsZeroOrNegative(): void + public function testFallsBackToOneMinuteIntervalWhenNoneConfigured(): void { + // requeueInterval controls how long to wait before resuming, not whether to resume: + // a campaign that stopped early must always be retried (mirrors phplist3's + // unconditional "don't mark sent while anything failed/was throttled" guard), so a + // missing/zero interval must not disable requeuing entirely. $handler = new RequeueHandler($this->logger, new Translator('en')); $message = $this->createMessage(0, null, null); - $this->output->expects($this->never())->method('writeln'); + $this->output->expects($this->once())->method('writeln'); + $this->logger->expects($this->once())->method('info'); + + $before = new DateTime(); + $result = $handler->handle($message, $this->output); + $after = new DateTime(); + + $this->assertTrue($result); + $this->assertSame(MessageStatus::Submitted, $message->getMetadata()->getStatus()); + + $embargo = $message->getSchedule()->getEmbargo(); + $this->assertInstanceOf(DateTime::class, $embargo); + + $minExpected = (clone $before)->add(new DateInterval('PT1M')); + $maxExpected = (clone $after)->add(new DateInterval('PT1M')); + $this->assertGreaterThanOrEqual($minExpected->getTimestamp(), $embargo->getTimestamp()); + $this->assertLessThanOrEqual($maxExpected->getTimestamp(), $embargo->getTimestamp()); + } + + public function testStillReturnsFalseWhenNoIntervalConfiguredButRequeueUntilAlreadyPassed(): void + { + $handler = new RequeueHandler($this->logger, new Translator('en')); + $past = (new DateTime())->sub(new DateInterval('PT5M')); + $message = $this->createMessage(0, $past, null); + $this->logger->expects($this->never())->method('info'); $result = $handler->handle($message, $this->output); From 0b8d113fb33decf5d8e2bbb12aa59b97e91ab297 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Tue, 8 Sep 2026 18:22:10 +0400 Subject: [PATCH 05/10] feat: getStuckCampaigns --- .env.dist | 1 + config/parameters.yml | 1 + .../CampaignProcessorMessageHandler.php | 2 +- .../Repository/MessageRepository.php | 71 ++++++++++++++++--- .../Service/Manager/MessageManager.php | 7 ++ .../Service/Manager/MessageManagerTest.php | 18 +++++ 6 files changed, 91 insertions(+), 9 deletions(-) diff --git a/.env.dist b/.env.dist index cea77ba2..a9738072 100644 --- a/.env.dist +++ b/.env.dist @@ -78,6 +78,7 @@ MAILQUEUE_BATCH_SIZE=5 MAILQUEUE_BATCH_PERIOD=5 MAILQUEUE_THROTTLE=5 MESSAGING_MAX_PROCESS_TIME=600 +MESSAGING_STUCK_CAMPAIGN_THRESHOLD=1800 MAX_MAILSIZE=209715200 DEFAULT_MESSAGEAGE=691200 USE_MANUAL_TEXT_PART=0 diff --git a/config/parameters.yml b/config/parameters.yml index 6ee93ac1..772774d1 100644 --- a/config/parameters.yml +++ b/config/parameters.yml @@ -76,6 +76,7 @@ parameters: messaging.mail_queue_period: '%env(MAILQUEUE_BATCH_PERIOD)%' messaging.mail_queue_throttle: '%env(MAILQUEUE_THROTTLE)%' messaging.max_process_time: '%env(MESSAGING_MAX_PROCESS_TIME)%' + messaging.stuck_campaign_threshold: '%env(int:MESSAGING_STUCK_CAMPAIGN_THRESHOLD)%' messaging.max_mail_size: '%env(MAX_MAILSIZE)%' messaging.default_message_age: '%env(DEFAULT_MESSAGEAGE)%' messaging.use_manual_text_part: '%env(USE_MANUAL_TEXT_PART)%' diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index ee0916d6..1cebf5a9 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -268,7 +268,7 @@ private function handleEmailSending( $this->rateLimitedCampaignMailer->send($email); ($this->mailSizeChecker)($campaign, $email, $subscriber->hasHtmlEmail()); $this->updateUserMessageStatus($userMessage, UserMessageStatus::Sent); - $campaign->incrementSentCount($sentAs); + $this->messageRepository->incrementSentCounts($campaign->getId(), $sentAs); } catch (MessageSizeLimitExceededException $e) { // stop after the first message if size is exceeded $this->updateMessageStatus($campaign, MessageStatus::Suspended); diff --git a/src/Domain/Messaging/Repository/MessageRepository.php b/src/Domain/Messaging/Repository/MessageRepository.php index 443160d5..414e27ed 100644 --- a/src/Domain/Messaging/Repository/MessageRepository.php +++ b/src/Domain/Messaging/Repository/MessageRepository.php @@ -4,6 +4,7 @@ namespace PhpList\Core\Domain\Messaging\Repository; +use DateTime; use DateTimeImmutable; use DateTimeInterface; use Doctrine\ORM\AbstractQuery; @@ -11,6 +12,7 @@ use PhpList\Core\Domain\Common\Model\PaginatedResult; use PhpList\Core\Domain\Common\Repository\AbstractRepository; use PhpList\Core\Domain\Common\Repository\Interfaces\PaginatableRepositoryInterface; +use PhpList\Core\Domain\Configuration\Model\OutputFormat; use PhpList\Core\Domain\Messaging\Model\Filter\MessageFilter; use PhpList\Core\Domain\Messaging\Model\Message; use PhpList\Core\Domain\Subscription\Model\SubscriberList; @@ -167,15 +169,17 @@ public function tryClaimForProcessing(int $id): ?Message { $connection = $this->getEntityManager()->getConnection(); $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + $now = new DateTime(); - $affected = $connection->executeStatement( - sprintf('UPDATE %s SET status = :to WHERE id = :id AND status = :from', $table), - [ - 'to' => Message\MessageStatus::Prepared->value, - 'id' => $id, - 'from' => Message\MessageStatus::Submitted->value, - ] - ); + $sql = sprintf('UPDATE %s SET status = :to, modified = :now WHERE id = :id AND status = :from', $table); + $params = [ + 'to' => Message\MessageStatus::Prepared->value, + 'now' => $now->format('Y-m-d H:i:s'), + 'id' => $id, + 'from' => Message\MessageStatus::Submitted->value, + ]; + + $affected = $connection->executeStatement($sql, $params); if ($affected === 0) { return null; @@ -184,6 +188,57 @@ public function tryClaimForProcessing(int $id): ?Message return $this->find($id); } + /** + * Returns campaigns stuck in Prepared/InProcess whose row hasn't been touched since + * $staleBefore, i.e. candidates for tryClaimForProcessing's stale-reclaim path. Callers + * are expected to re-dispatch a CampaignProcessorMessage for each, since nothing else + * automatically resumes a campaign that isn't in Submitted status. + * + * @return Message[] + */ + public function getStuckInProcessing(DateTimeImmutable $staleBefore): array + { + return $this->createQueryBuilder('m') + ->where('m.metadata.status IN (:statuses)') + ->andWhere('m.updatedAt < :staleBefore') + ->setParameter('statuses', [ + Message\MessageStatus::Prepared->value, + Message\MessageStatus::InProcess->value, + ]) + ->setParameter('staleBefore', $staleBefore) + ->getQuery() + ->getResult(); + } + + /** + * Atomically increments a campaign's processed/format-sent counters directly in the + * database (bypassing the entity's in-memory incrementSentCount()), so concurrent + * updates to the same campaign can't lose an update the way a read-modify-write via + * the entity manager could. Also bumps `modified`, since this is the liveness signal + * tryClaimForProcessing's stale-reclaim relies on. + */ + public function incrementSentCounts(int $messageId, OutputFormat $sentAs): void + { + $formatField = match ($sentAs) { + OutputFormat::Html => 'm.format.asHtml', + OutputFormat::Text => 'm.format.asText', + OutputFormat::Pdf => 'm.format.asPdf', + OutputFormat::TextAndHtml => 'm.format.asTextAndHtml', + OutputFormat::TextAndPdf => 'm.format.asTextAndPdf', + }; + + $this->createQueryBuilder('m') + ->update() + ->set('m.metadata.processed', 'm.metadata.processed + 1') + ->set($formatField, $formatField . ' + 1') + ->set('m.updatedAt', ':now') + ->where('m.id = :id') + ->setParameter('now', new DateTime()) + ->setParameter('id', $messageId) + ->getQuery() + ->execute(); + } + public function getNonEmptyFields(int $id): array { $message = $this->createQueryBuilder('m') diff --git a/src/Domain/Messaging/Service/Manager/MessageManager.php b/src/Domain/Messaging/Service/Manager/MessageManager.php index 7ed34594..bd7b9315 100644 --- a/src/Domain/Messaging/Service/Manager/MessageManager.php +++ b/src/Domain/Messaging/Service/Manager/MessageManager.php @@ -4,6 +4,7 @@ namespace PhpList\Core\Domain\Messaging\Service\Manager; +use DateTimeImmutable; use PhpList\Core\Domain\Identity\Model\Administrator; use PhpList\Core\Domain\Messaging\Model\Dto\MessageContext; use PhpList\Core\Domain\Messaging\Model\Dto\MessageDtoInterface; @@ -101,6 +102,12 @@ public function getMessagesByOwner(Administrator $owner): array return $this->messageRepository->getByOwnerId($owner->getId()); } + /** @return Message[] */ + public function getStuckCampaigns(DateTimeImmutable $staleBefore): array + { + return $this->messageRepository->getStuckInProcessing($staleBefore); + } + private function canBeSubmitted(Message $message): bool { return $message->getListMessages()->count() > 0 diff --git a/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php b/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php index 0021ae87..6e7bcfe3 100644 --- a/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php +++ b/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php @@ -5,6 +5,7 @@ namespace PhpList\Core\Tests\Unit\Domain\Messaging\Service\Manager; use DateTime; +use DateTimeImmutable; use InvalidArgumentException; use PhpList\Core\Domain\Identity\Model\Administrator; use PhpList\Core\Domain\Messaging\Model\ListMessage; @@ -241,4 +242,21 @@ public function testUpdateStatusSetsSubmittedWhenRequiredFieldsAndListArePresent $this->assertSame($message, $updated); $this->assertSame(Message\MessageStatus::Submitted, $message->getMetadata()->getStatus()); } + + public function testGetStuckCampaignsDelegatesToRepository(): void + { + $messageRepository = $this->createMock(MessageRepository::class); + $messageBuilder = $this->createMock(MessageBuilder::class); + $manager = new MessageManager($messageRepository, $messageBuilder); + + $staleBefore = new DateTimeImmutable('-30 minutes'); + $stuckMessage = $this->createMock(Message::class); + + $messageRepository->expects($this->once()) + ->method('getStuckInProcessing') + ->with($staleBefore) + ->willReturn([$stuckMessage]); + + $this->assertSame([$stuckMessage], $manager->getStuckCampaigns($staleBefore)); + } } From ba33ad21ba36da5ed64c3a458f08b16e0ce2efc5 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Wed, 9 Sep 2026 17:32:23 +0400 Subject: [PATCH 06/10] feat: enhance exclusion logic for campaign subscribers --- .../CampaignProcessorMessageHandler.php | 28 ++++++-- .../Service/Provider/SubscriberProvider.php | 45 +++++++++++-- .../CampaignProcessorMessageHandlerTest.php | 66 +++++++++++++++++++ 3 files changed, 130 insertions(+), 9 deletions(-) diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index 1cebf5a9..97331a65 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -126,7 +126,7 @@ public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $ // Campaign was already atomically claimed into Prepared status above. $excludeListIds = $this->getExcludeListIds($loadedMessageData); - $this->markExcludedSubscribers($campaign, $excludeListIds); + $this->markExcludedSubscribers($campaign, $data, $excludeListIds); $subscribers = $this->subscriberProvider->getSubscribersForMessageOrLists( $data, $campaign, @@ -172,14 +172,34 @@ private function getExcludeListIds(array $loadedMessageData): array * so there's a persisted audit trail for why a subscriber wasn't sent to. Skips * subscribers who already have a nontodo UserMessage for this campaign, so a later run * can't clobber an already-recorded Sent/NotSent/etc. status from an earlier partial run. + * Only campaign recipients (i.e. subscribers who'd otherwise be sent this campaign) are + * marked, since a subscriber on an exclude list who isn't a campaign recipient anyway + * shouldn't get an exclusion record. */ - private function markExcludedSubscribers(Message $campaign, array $excludeListIds): void - { + private function markExcludedSubscribers( + Message $campaign, + CampaignProcessorMessage|SyncCampaignProcessorMessage $data, + array $excludeListIds, + ): void { if ($excludeListIds === []) { return; } - foreach ($this->subscriberProvider->getExcludedSubscribers($excludeListIds) as $subscriber) { + $excludedSubscribers = $this->subscriberProvider->getExcludedSubscribers($excludeListIds); + if ($excludedSubscribers === []) { + return; + } + + $sendableSubscribers = $this->subscriberProvider->getSendableSubscribersForMessageOrLists( + $data, + $campaign + ); + + foreach ($excludedSubscribers as $subscriber) { + if (!isset($sendableSubscribers[$subscriber->getEmail()])) { + continue; + } + $existing = $this->userMessageRepository->findByUserAndMessage($subscriber, $campaign); if ($existing && $existing->getStatus() !== UserMessageStatus::Todo) { continue; diff --git a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php index 31474ef6..3d5b50fa 100644 --- a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php +++ b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php @@ -42,6 +42,45 @@ public function getSubscribersForMessageOrLists( return $this->subscriberRepository->getByEmails($data->getSubscriberEmails()); } + $subscribers = $this->getSendableSubscribersByListMembership($data, $campaign); + + foreach ($this->getExcludedSubscribers($excludeListIds) as $excluded) { + unset($subscribers[$excluded->getEmail()]); + } + + return array_values($subscribers); + } + + /** + * Resolves the campaign's sendable recipients by list membership (confirmed, not disabled), + * before any list-based exclusion is applied. Used to determine which excluded subscribers + * are actually campaign recipients, so exclusion records aren't created for non-recipients. + * + * @return array Subscribers keyed by email + */ + public function getSendableSubscribersForMessageOrLists( + CampaignProcessorMessageInterface $data, + Message $campaign, + ): array { + if ($data instanceof TestCampaignProcessorMessage) { + $subscribers = []; + foreach ($this->subscriberRepository->getByEmails($data->getSubscriberEmails()) as $subscriber) { + $subscribers[$subscriber->getEmail()] = $subscriber; + } + + return $subscribers; + } + + return $this->getSendableSubscribersByListMembership($data, $campaign); + } + + /** + * @return array Subscribers keyed by email + */ + private function getSendableSubscribersByListMembership( + CampaignProcessorMessageInterface $data, + Message $campaign, + ): array { if (count($data->getListIds()) > 0) { $listIds = $data->getListIds(); } else { @@ -56,11 +95,7 @@ public function getSubscribersForMessageOrLists( } } - foreach ($this->getExcludedSubscribers($excludeListIds) as $excluded) { - unset($subscribers[$excluded->getEmail()]); - } - - return array_values($subscribers); + return $subscribers; } /** diff --git a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php index e9a9b477..482509ec 100644 --- a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php +++ b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php @@ -277,6 +277,11 @@ public function testInvokeMarksExcludedSubscribersAsExcludedInUserMessage(): voi ->with([55]) ->willReturn([$excludedSubscriber]); + $this->subscriberProvider->expects($this->once()) + ->method('getSendableSubscribersForMessageOrLists') + ->with($data, $campaign) + ->willReturn(['excluded@example.com' => $excludedSubscriber]); + $this->subscriberProvider->expects($this->once()) ->method('getSubscribersForMessageOrLists') ->with($data, $campaign, [55]) @@ -334,6 +339,11 @@ public function testInvokeDoesNotOverwriteExistingNonTodoUserMessageWhenMarkingE ->with([55]) ->willReturn([$excludedSubscriber]); + $this->subscriberProvider->expects($this->once()) + ->method('getSendableSubscribersForMessageOrLists') + ->with($data, $campaign) + ->willReturn(['already-sent@example.com' => $excludedSubscriber]); + $this->subscriberProvider->expects($this->once()) ->method('getSubscribersForMessageOrLists') ->willReturn([]); @@ -355,6 +365,62 @@ public function testInvokeDoesNotOverwriteExistingNonTodoUserMessageWhenMarkingE $handler($data); } + public function testInvokeDoesNotMarkExcludedSubscriberWhoIsNotACampaignRecipient(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $nonRecipientExcludedSubscriber = $this->createMock(Subscriber::class); + $nonRecipientExcludedSubscriber->method('getEmail')->willReturn('not-a-recipient@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getExcludedSubscribers') + ->with([55]) + ->willReturn([$nonRecipientExcludedSubscriber]); + + $this->subscriberProvider->expects($this->once()) + ->method('getSendableSubscribersForMessageOrLists') + ->with($data, $campaign) + ->willReturn([]); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, [55]) + ->willReturn([]); + + $this->userMessageRepository->expects($this->never()) + ->method('findByUserAndMessage'); + + $this->userMessageRepository->expects($this->never()) + ->method('save'); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + public function testInvokeWithInvalidSubscriberEmail(): void { $campaign = $this->createCampaignMock(); From 69e8c3335fb8035909b92ae95ac7c78c32ec2e9e Mon Sep 17 00:00:00 2001 From: Tatevik Date: Wed, 9 Sep 2026 17:46:16 +0400 Subject: [PATCH 07/10] fix: getStatus method to handle null status --- src/Domain/Messaging/Model/UserMessage.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Domain/Messaging/Model/UserMessage.php b/src/Domain/Messaging/Model/UserMessage.php index 93b457f3..6c03ca1d 100644 --- a/src/Domain/Messaging/Model/UserMessage.php +++ b/src/Domain/Messaging/Model/UserMessage.php @@ -66,9 +66,9 @@ public function getViewed(): ?DateTime return $this->viewed; } - public function getStatus(): ?UserMessageStatus + public function getStatus(): UserMessageStatus { - return UserMessageStatus::from($this->status); + return $this->status !== null ? UserMessageStatus::from($this->status) : UserMessageStatus::Todo; } public function setViewed(?DateTime $viewed): self From 913e823431a430b1a566c3f122450dab2b8215a3 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Wed, 9 Sep 2026 18:13:26 +0400 Subject: [PATCH 08/10] feat: update guidelines for atomic conditional-UPDATE reservation patterns --- .coderabbit.yaml | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/.coderabbit.yaml b/.coderabbit.yaml index c343b8fc..ab0a966f 100644 --- a/.coderabbit.yaml +++ b/.coderabbit.yaml @@ -35,7 +35,22 @@ reviews: - Still prefer keeping this logic well-encapsulated (e.g. in dedicated services/repos), not scattered across unrelated domain objects. - - ⚠️ For non-DynamicListAttr code: + - ✅ **Relaxed rule for atomic conditional-UPDATE reservation/claim patterns**: + - Some Domain repositories intentionally use `$connection->executeStatement(...)` to run a + conditional `UPDATE ... WHERE ` (checking the affected-row count) or a guarded + `INSERT` (catching `UniqueConstraintViolationException`) as a portable, lock-free way to + atomically claim/reserve a row across concurrent workers - e.g. + `MessageRepository::tryClaimForProcessing` and + `DomainThrottleStateRepository::tryReserveSlot`/`resetBlockedCount`. + - This is not a domain-purity violation: it involves no `flush()`, transaction, or DDL, and + avoids vendor-specific upsert/locking syntax so it stays portable across MySQL/PostgreSQL/SQLite. + - Do *not* flag `executeStatement()` DML calls that follow this guarded-UPDATE/affected-rows + or catch-`UniqueConstraintViolationException`-on-INSERT pattern against the repository's own + entity table. + - Still flag any DBAL write that is unconditional, unrelated to claim/reservation semantics, or + that mutates a different entity's table than the repository owns. + + - ⚠️ For other non-DynamicListAttr code: - If code is invoking actual table-creation, DDL execution, or schema synchronization, then request moving that to the Infrastructure or Application layer (e.g. MessageHandler). - Repositories in Domain should be abstractions without side effects; they should express *intent*, From 83bc8bdc0c703fe113fbce62172cd4e3f7b7f3d9 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Wed, 9 Sep 2026 18:55:10 +0400 Subject: [PATCH 09/10] feat: add stuck campaign threshold to tryClaimForProcessing method --- .../CampaignProcessorMessageHandler.php | 7 ++- .../Repository/MessageRepository.php | 22 ++++++++- .../Repository/MessageRepositoryTest.php | 46 ++++++++++++++++++ .../CampaignProcessorMessageHandlerTest.php | 47 +++++++++++++------ 4 files changed, 105 insertions(+), 17 deletions(-) diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index 97331a65..1e44e45d 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php @@ -76,12 +76,17 @@ public function __construct( private readonly DomainRateLimiter $domainRateLimiter, #[Autowire('%imap_bounce.email%')] private readonly string $bounceEmail, #[Autowire('%messaging.use_list_exclude%')] private readonly bool $useListExclude = false, + #[Autowire('%messaging.stuck_campaign_threshold%')] private readonly int $stuckCampaignThresholdSeconds = 0, ) { } public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $data): void { - $campaign = $this->messageRepository->tryClaimForProcessing($data->getMessageId()); + // todo: recheck this stuckCampaignThresholdSeconds logic + $campaign = $this->messageRepository->tryClaimForProcessing( + $data->getMessageId(), + $this->stuckCampaignThresholdSeconds + ); if (!$campaign) { $this->logger->warning( $this->translator->trans('Campaign not found or not in submitted status'), diff --git a/src/Domain/Messaging/Repository/MessageRepository.php b/src/Domain/Messaging/Repository/MessageRepository.php index 414e27ed..c5698775 100644 --- a/src/Domain/Messaging/Repository/MessageRepository.php +++ b/src/Domain/Messaging/Repository/MessageRepository.php @@ -164,14 +164,20 @@ public function findByIdAndStatus(int $id, Message\MessageStatus $status): ?Mess * Atomically claims a campaign for processing by flipping its status from Submitted to * Prepared in a single UPDATE ... WHERE statement, so two concurrent workers can't both * pass a check-then-act race and process the same campaign. + * + * When $staleAfterSeconds > 0, the same atomic UPDATE also reclaims a row stuck in + * Prepared/InProcess whose `modified` is older than that threshold (crashed/killed worker, + * or a handler that threw before requeuing). Staleness is re-checked against the row's + * current `modified` at the moment of this UPDATE, not pre-computed by the caller, so a + * worker that's merely slow (and keeps bumping `modified` via incrementSentCounts()) can't + * be claimed out from under itself by a second, concurrent dispatch. */ - public function tryClaimForProcessing(int $id): ?Message + public function tryClaimForProcessing(int $id, int $staleAfterSeconds = 0): ?Message { $connection = $this->getEntityManager()->getConnection(); $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); $now = new DateTime(); - $sql = sprintf('UPDATE %s SET status = :to, modified = :now WHERE id = :id AND status = :from', $table); $params = [ 'to' => Message\MessageStatus::Prepared->value, 'now' => $now->format('Y-m-d H:i:s'), @@ -179,6 +185,18 @@ public function tryClaimForProcessing(int $id): ?Message 'from' => Message\MessageStatus::Submitted->value, ]; + $claimCondition = 'status = :from'; + if ($staleAfterSeconds > 0) { + $claimCondition = '(status = :from OR (status IN (:prepared, :inProcess) AND modified < :staleBefore))'; + $params['prepared'] = Message\MessageStatus::Prepared->value; + $params['inProcess'] = Message\MessageStatus::InProcess->value; + $params['staleBefore'] = (clone $now) + ->modify(sprintf('-%d seconds', $staleAfterSeconds)) + ->format('Y-m-d H:i:s'); + } + + $sql = sprintf('UPDATE %s SET status = :to, modified = :now WHERE id = :id AND %s', $table, $claimCondition); + $affected = $connection->executeStatement($sql, $params); if ($affected === 0) { diff --git a/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php b/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php index baee85b2..a384457e 100644 --- a/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php +++ b/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php @@ -259,6 +259,52 @@ public function testTryClaimForProcessingCannotClaimTwice(): void self::assertNull($secondClaim); } + public function testTryClaimForProcessingReclaimsStalePreparedCampaignWhenThresholdGiven(): void + { + $message = $this->persistMessage(Message\MessageStatus::Prepared, 'Stuck in prepared'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->backdateModified($id, 3600); + $this->entityManager->clear(); + + $claimed = $this->messageRepository->tryClaimForProcessing($id, staleAfterSeconds: 1800); + + self::assertNotNull($claimed); + self::assertSame(Message\MessageStatus::Prepared, $claimed->getMetadata()->getStatus()); + } + + public function testTryClaimForProcessingDoesNotReclaimRecentlyTouchedPreparedCampaign(): void + { + $message = $this->persistMessage(Message\MessageStatus::Prepared, 'Still alive'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + self::assertNull($this->messageRepository->tryClaimForProcessing($id, staleAfterSeconds: 1800)); + } + + public function testTryClaimForProcessingIgnoresStalePreparedCampaignWithoutThreshold(): void + { + $message = $this->persistMessage(Message\MessageStatus::Prepared, 'Stuck but no threshold given'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->backdateModified($id, 3600); + $this->entityManager->clear(); + + self::assertNull($this->messageRepository->tryClaimForProcessing($id)); + } + + private function backdateModified(int $id, int $secondsAgo): void + { + $table = $this->entityManager->getClassMetadata(Message::class)->getTableName(); + $modified = (new DateTime())->modify(sprintf('-%d seconds', $secondsAgo)); + + $this->entityManager->getConnection()->executeStatement( + sprintf('UPDATE %s SET modified = :modified WHERE id = :id', $table), + ['modified' => $modified->format('Y-m-d H:i:s'), 'id' => $id] + ); + } + public function testGetFilteredAfterIdSortsDescendingAndCursorsBackward(): void { $first = $this->persistMessage(Message\MessageStatus::Sent, 'First'); diff --git a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php index 482509ec..fc7d5802 100644 --- a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php +++ b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php @@ -89,8 +89,10 @@ protected function setUp(): void $this->handler = $this->createHandler(); } - private function createHandler(bool $useListExclude = false): CampaignProcessorMessageHandler - { + private function createHandler( + bool $useListExclude = false, + int $stuckCampaignThresholdSeconds = 0, + ): CampaignProcessorMessageHandler { return new CampaignProcessorMessageHandler( mailer: $this->symfonyMailer, rateLimitedCampaignMailer: $this->mailer, @@ -114,6 +116,7 @@ private function createHandler(bool $useListExclude = false): CampaignProcessorM domainRateLimiter: $this->domainRateLimiter, bounceEmail: 'bounce@email.com', useListExclude: $useListExclude, + stuckCampaignThresholdSeconds: $stuckCampaignThresholdSeconds, ); } @@ -123,7 +126,7 @@ public function testInvokeWhenCampaignNotFound(): void $this->messageRepository->expects($this->once()) ->method('tryClaimForProcessing') - ->with(999) + ->with(999, 0) ->willReturn(null); $this->translator->method('trans')->willReturnCallback(fn(string $msg) => $msg); @@ -135,6 +138,22 @@ public function testInvokeWhenCampaignNotFound(): void ($this->handler)($message); } + public function testInvokePassesStuckCampaignThresholdToTryClaimForProcessing(): void + { + $handler = $this->createHandler(stuckCampaignThresholdSeconds: 1800); + + $message = new CampaignProcessorMessage(999); + + $this->messageRepository->expects($this->once()) + ->method('tryClaimForProcessing') + ->with(999, 1800) + ->willReturn(null); + + $this->translator->method('trans')->willReturnCallback(fn(string $msg) => $msg); + + $handler($message); + } + public function testInvokeWithNoSubscribers(): void { $campaign = $this->createCampaignMock(); @@ -144,7 +163,7 @@ public function testInvokeWithNoSubscribers(): void $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -180,7 +199,7 @@ public function testInvokePassesExcludeListIdsFromMessageDataToSubscriberProvide $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); @@ -217,7 +236,7 @@ public function testInvokeIgnoresExcludeListWhenUseListExcludeDisabled(): void $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); @@ -254,7 +273,7 @@ public function testInvokeMarksExcludedSubscribersAsExcludedInUserMessage(): voi $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); @@ -316,7 +335,7 @@ public function testInvokeDoesNotOverwriteExistingNonTodoUserMessageWhenMarkingE $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); @@ -376,7 +395,7 @@ public function testInvokeDoesNotMarkExcludedSubscriberWhoIsNotACampaignRecipien $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); @@ -430,7 +449,7 @@ public function testInvokeWithInvalidSubscriberEmail(): void $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -477,7 +496,7 @@ public function testInvokeWithValidSubscriberEmail(): void $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -545,7 +564,7 @@ public function testInvokeWithMailerException(): void $data = new CampaignProcessorMessage(123); $this->messageRepository->method('tryClaimForProcessing') - ->with(123) + ->with(123, 0) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -622,7 +641,7 @@ public function testInvokeWithMultipleSubscribers(): void $this->messageRepository ->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $this->precacheService @@ -746,7 +765,7 @@ public function testInvokeSkipsDomainThrottledSubscriberWithoutCreatingUserMessa $data = new CampaignProcessorMessage(1); $this->messageRepository->method('tryClaimForProcessing') - ->with(1) + ->with(1, 0) ->willReturn($campaign); $this->precacheService->expects($this->once()) From cd3f6b22afdd7c09cd8145879189634557144ce5 Mon Sep 17 00:00:00 2001 From: Tatevik Date: Wed, 9 Sep 2026 19:07:23 +0400 Subject: [PATCH 10/10] feat: implement atomic reset for blocked count in DomainThrottleStateRepository --- .../DomainThrottleStateRepository.php | 22 +++++++++++++++---- .../Messaging/Service/DomainRateLimiter.php | 15 ++++++++++--- .../DomainThrottleStateRepositoryTest.php | 18 ++++++++++++++- .../Service/DomainRateLimiterTest.php | 21 +++++++++++++++++- 4 files changed, 67 insertions(+), 9 deletions(-) diff --git a/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php b/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php index a1d38673..4baa3907 100644 --- a/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php +++ b/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php @@ -47,15 +47,29 @@ public function tryReserveSlot(string $domain, int $windowStart, int $batchSize) ); } - public function resetBlockedCount(string $domain, int $windowStart): void + /** + * Atomically claims the auto-throttle trigger for the specified domain/window. + * + * The row's blocked_count is reset to 0 only if it is still greater than $threshold at the moment + * the UPDATE executes. The worker whose UPDATE successfully performs that reset receives true and is + * considered to have claimed the trigger. Concurrent workers racing to claim the same trigger + * will see no rows updated once the count has already been reset and will receive false. + */ + public function resetBlockedCount(string $domain, int $windowStart, int $threshold): bool { $connection = $this->getEntityManager()->getConnection(); $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); - $connection->executeStatement( - sprintf('UPDATE %s SET blocked_count = 0 WHERE domain = :domain AND window_start = :window', $table), - ['domain' => $domain, 'window' => $windowStart] + $affected = $connection->executeStatement( + sprintf( + 'UPDATE %s SET blocked_count = 0 + WHERE domain = :domain AND window_start = :window AND blocked_count > :threshold', + $table + ), + ['domain' => $domain, 'window' => $windowStart, 'threshold' => $threshold] ); + + return $affected > 0; } /** @phpstan-impure */ diff --git a/src/Domain/Messaging/Service/DomainRateLimiter.php b/src/Domain/Messaging/Service/DomainRateLimiter.php index 900e2f81..8d4634b7 100644 --- a/src/Domain/Messaging/Service/DomainRateLimiter.php +++ b/src/Domain/Messaging/Service/DomainRateLimiter.php @@ -78,9 +78,18 @@ private function applyAutoThrottleIfDue( return new DomainThrottleResult(allowed: false, domain: $domain, blockedAttempts: $blockedAttempts); } - // Reset the trigger counter so it takes another full run of blocked attempts - // before backoff fires again for this domain/window. - $this->repository->resetBlockedCount($domain, $windowStart); + // Concurrent workers can all observe blockedAttempts over the threshold at once; only + // the one that atomically claims the reset applies the backoff delay, so the rest + // continue instead of all sleeping for the same trigger. + $claimed = $this->repository->resetBlockedCount( + $domain, + $windowStart, + self::AUTO_THROTTLE_ATTEMPT_THRESHOLD + ); + if (!$claimed) { + return new DomainThrottleResult(allowed: false, domain: $domain, blockedAttempts: $blockedAttempts); + } + $delaySeconds = max(1, intdiv($this->domainBatchPeriod, max(1, $this->domainBatchSize * 4))); $this->logger->info('Introducing extra delay to reduce domain throttle failures', [ diff --git a/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php b/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php index 38f32fa8..08a6c6a0 100644 --- a/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php +++ b/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php @@ -77,10 +77,26 @@ public function testResetBlockedCountClearsCounterForCurrentWindow(): void $blocked = $this->repository->tryReserveSlot('example.com', 1000, 1); $this->assertSame(2, $blocked->blockedAttempts); - $this->repository->resetBlockedCount('example.com', 1000); + $claimed = $this->repository->resetBlockedCount('example.com', 1000, threshold: 1); + $this->assertTrue($claimed); $afterReset = $this->repository->tryReserveSlot('example.com', 1000, 1); $this->assertFalse($afterReset->allowed); $this->assertSame(1, $afterReset->blockedAttempts); } + + public function testResetBlockedCountDoesNotClaimWhenCountAtOrBelowThreshold(): void + { + $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->repository->tryReserveSlot('example.com', 1000, 1); + $blocked = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertSame(2, $blocked->blockedAttempts); + + $claimed = $this->repository->resetBlockedCount('example.com', 1000, threshold: 2); + $this->assertFalse($claimed); + + $afterAttempt = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertFalse($afterAttempt->allowed); + $this->assertSame(3, $afterAttempt->blockedAttempts); + } } diff --git a/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php index 82af3ef5..e4b3d587 100644 --- a/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php +++ b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php @@ -126,7 +126,10 @@ public function testAppliesBackoffAndResetsBlockedCountOnceThresholdExceeded(): { $this->repository->method('tryReserveSlot') ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 26)); - $this->repository->expects($this->once())->method('resetBlockedCount'); + $this->repository->expects($this->once()) + ->method('resetBlockedCount') + ->with('example.com', $this->isType('int'), 25) + ->willReturn(true); // Small batch period/size keeps the resulting sleep() short (~1s) so the test stays fast. $limiter = $this->createLimiter(domainBatchSize: 1, domainBatchPeriod: 4, autoThrottle: true); @@ -136,4 +139,20 @@ public function testAppliesBackoffAndResetsBlockedCountOnceThresholdExceeded(): $this->assertTrue($result->backoffApplied); $this->assertGreaterThanOrEqual(1, $result->backoffSeconds); } + + public function testDoesNotBackoffWhenLosingTheResetRaceToAnotherWorker(): void + { + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 26)); + $this->repository->expects($this->once()) + ->method('resetBlockedCount') + ->willReturn(false); + + $limiter = $this->createLimiter(domainBatchSize: 1, domainBatchPeriod: 4, autoThrottle: true); + $result = $limiter->attemptSend('third@example.com'); + + $this->assertFalse($result->allowed); + $this->assertFalse($result->backoffApplied); + $this->assertSame(0, $result->backoffSeconds); + } }