From a25808836c47744aee454325634b7cf8850af0e6 Mon Sep 17 00:00:00 2001 From: KalimeroMK Date: Tue, 13 Oct 2026 00:08:29 +0200 Subject: [PATCH 1/3] Keep a message the handler refused in the queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `queue:run --limit=1` over two queued messages reported one processed and left the queue empty: `InMemoryAdapter::runExisting()` removed the message that tripped the limit even though the callback returned `false` without handling it. Both shipped adapters already treat `false` as "not consumed" — `queue-redis` deletes only when the callback returned `true`, `queue-amqp` nacks with requeue otherwise — so the stub was the outlier. Spell that contract out on `runExisting()` and `subscribe()`. --- src/Adapter/AdapterInterface.php | 10 ++++++++-- stubs/InMemoryAdapter.php | 8 +++++--- tests/Unit/QueueTest.php | 14 ++++++++++++++ 3 files changed, 27 insertions(+), 5 deletions(-) diff --git a/src/Adapter/AdapterInterface.php b/src/Adapter/AdapterInterface.php index 69491cb1..e2e5286a 100644 --- a/src/Adapter/AdapterInterface.php +++ b/src/Adapter/AdapterInterface.php @@ -10,9 +10,13 @@ interface AdapterInterface { /** - * Returns the first message from the queue if it exists (null otherwise). + * Passes existing messages to the given handler in FIFO order, until the queue is empty or the handler asks + * to stop. * - * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages + * The handler returns `false` for a message it did not consume: that message must stay in the queue, and no + * further messages are passed. Only a message the handler returned `true` for may be removed from the queue. + * + * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages. */ public function runExisting(callable $handlerCallback): void; @@ -32,6 +36,8 @@ public function push(MessageInterface $message): MessageInterface; /** * Listen to the queue and pass messages to the given handler as they come. * + * The handler returns `false` for a message it did not consume, as in {@see runExisting()}. + * * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false if it cannot continue handling messages. */ public function subscribe(callable $handlerCallback): void; diff --git a/stubs/InMemoryAdapter.php b/stubs/InMemoryAdapter.php index 5a0ad198..59fe1dd6 100644 --- a/stubs/InMemoryAdapter.php +++ b/stubs/InMemoryAdapter.php @@ -48,9 +48,11 @@ public function push(MessageInterface $message): MessageInterface public function runExisting(callable $handlerCallback): void { - $result = true; - while (isset($this->messages[$this->current]) && $result === true) { - $result = $handlerCallback($this->messages[$this->current]); + while (isset($this->messages[$this->current])) { + if ($handlerCallback($this->messages[$this->current]) !== true) { + break; + } + unset($this->messages[$this->current]); $this->current++; } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index 4923edca..f09568a1 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -12,6 +12,7 @@ use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\MessageStatus; use Yiisoft\Queue\Stubs\InMemoryAdapter; +use Yiisoft\Queue\Stubs\StubLoop; use Yiisoft\Queue\Tests\TestCase; use function count; @@ -81,5 +82,18 @@ public function testConsumerStopsAtLimit(): void $consumer = new QueueConsumer($this->getWorker(), $this->getLoop(), new NullLogger(), $adapter); self::assertSame(1, $consumer->run(1)); self::assertSame(1, $this->executionTimes); + self::assertCount(1, $adapter->getMessagesList(), 'The message that hits the limit stays in the queue.'); + } + + public function testMessageIsNotConsumedWhenLoopCannotContinue(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $producer->push(new GenericMessage('simple', null)); + $producer->push(new GenericMessage('simple', null)); + $consumer = new QueueConsumer($this->getWorker(), new StubLoop(false), new NullLogger(), $adapter); + + self::assertSame(0, $consumer->run()); + self::assertCount(2, $adapter->getMessagesList()); } } From f0cee051a190b6339c7f647a70cc49d8a40bb73b Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Wed, 28 Oct 2026 01:54:09 +0300 Subject: [PATCH 2/3] Fix #346: Check the loop before processing a message, not after --- CHANGELOG.md | 1 + docs/guide/en/loops.md | 4 ++- src/QueueConsumer.php | 14 ++++---- tests/Unit/QueueTest.php | 78 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 88 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c9589ee..7d0624a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,3 +3,4 @@ ## 1.0.0 under development - Initial release. +- Bug #346: `QueueConsumer` checks the loop before processing a message instead of after, so a message processed just before the loop stops is counted and is not handled again (@samdark) diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index e20bfb17..d74c5ee4 100644 --- a/docs/guide/en/loops.md +++ b/docs/guide/en/loops.md @@ -4,7 +4,9 @@ Yii Queue uses `\Yiisoft\Queue\Cli\LoopInterface` to control long-running execut The loop is evaluated to determine whether it can continue: -- After each processed message (via `QueueConsumer::run()` / `QueueConsumer::listen()`). +- Before each message in `QueueConsumer::run()`. A message taken from the queue after the loop stopped is not processed + and goes back to the queue. +- By the adapter between messages in `QueueConsumer::listen()`. - On each iteration of `queue:listen-all`. When `canContinue()` returns `false`, consuming stops gracefully (as soon as the current message is finished). diff --git a/src/QueueConsumer.php b/src/QueueConsumer.php index ffcfbe7b..2c2b5d6c 100644 --- a/src/QueueConsumer.php +++ b/src/QueueConsumer.php @@ -35,9 +35,10 @@ public function run(int $max = 0): int $this->logger->debug('Start processing queue messages.'); $count = 0; $this->adapter->runExisting(function (MessageInterface $message) use (&$count, $max): bool { - if (($max > 0 && $count >= $max) || !$this->handle($message)) { + if (($max > 0 && $count >= $max) || !$this->loop->canContinue()) { return false; } + $this->worker->process($message, $this->queueName); $count++; return true; }); @@ -52,13 +53,10 @@ public function listen(): void return; } $this->logger->info('Start listening to the queue.'); - $this->adapter->subscribe(fn(MessageInterface $message): bool => $this->handle($message)); + $this->adapter->subscribe(function (MessageInterface $message): bool { + $this->worker->process($message, $this->queueName); + return true; + }); $this->logger->info('Finish listening to the queue.'); } - - private function handle(MessageInterface $message): bool - { - $this->worker->process($message, $this->queueName); - return $this->loop->canContinue(); - } } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index f09568a1..6f028748 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -4,7 +4,9 @@ namespace Yiisoft\Queue\Tests\Unit; +use Closure; use Psr\Log\NullLogger; +use Yiisoft\Queue\Cli\LoopInterface; use Yiisoft\Queue\QueueConsumer; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueProducerInterface; @@ -96,4 +98,80 @@ public function testMessageIsNotConsumedWhenLoopCannotContinue(): void self::assertSame(0, $consumer->run()); self::assertCount(2, $adapter->getMessagesList()); } + + public function testConsumerProcessesLastMessageAtLimitOnceAndLeavesNextOneWaiting(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $producer->push(new GenericMessage('simple', null)); + $producer->push(new GenericMessage('simple', null)); + $third = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))); + $consumer = new QueueConsumer($this->getWorker(), $this->getLoop(), new NullLogger(), $adapter); + + $processed = $consumer->run(2); + + self::assertSame(2, $processed); + self::assertSame(2, $this->executionTimes); + self::assertSame(MessageStatus::WAITING, $adapter->status($third->getId())); + } + + public function testConsumerProcessesMessageLeftAtLimitOnNextRun(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $producer->push(new GenericMessage('simple', null)); + $producer->push(new GenericMessage('simple', null)); + $producer->push(new GenericMessage('simple', null)); + $consumer = new QueueConsumer($this->getWorker(), $this->getLoop(), new NullLogger(), $adapter); + $consumer->run(2); + + $processed = $consumer->run(2); + + self::assertSame(1, $processed); + self::assertSame(3, $this->executionTimes); + } + + public function testConsumerRunReportsMessageProcessedBeforeLoopStops(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $first = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))); + $second = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))); + $consumer = new QueueConsumer($this->getWorker(), $this->createStopAfterFirstExecutionLoop(), new NullLogger(), $adapter); + + $processed = $consumer->run(); + + self::assertSame(1, $processed); + self::assertSame(1, $this->executionTimes); + self::assertSame(MessageStatus::DONE, $adapter->status($first->getId())); + self::assertSame(MessageStatus::WAITING, $adapter->status($second->getId())); + } + + public function testConsumerListenReportsMessageProcessedBeforeLoopStops(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $message = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))); + $consumer = new QueueConsumer($this->getWorker(), $this->createStopAfterFirstExecutionLoop(), new NullLogger(), $adapter); + + $consumer->listen(); + + self::assertSame(1, $this->executionTimes); + self::assertSame(MessageStatus::DONE, $adapter->status($message->getId())); + } + + private function createStopAfterFirstExecutionLoop(): LoopInterface + { + return new class (fn(): int => $this->executionTimes) implements LoopInterface { + /** + * @param Closure(): int $executionTimes + */ + public function __construct(private readonly Closure $executionTimes) {} + + public function canContinue(): bool + { + return ($this->executionTimes)() === 0; + } + }; + } } From a83e7888a22ba0a591d524699f7d187420d8a0c7 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Wed, 28 Oct 2026 01:58:43 +0300 Subject: [PATCH 3/3] Check the loop before processing a message in listen() too --- docs/guide/en/loops.md | 5 ++--- src/QueueConsumer.php | 3 +++ tests/Unit/QueueTest.php | 15 +++++++++++++++ 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/docs/guide/en/loops.md b/docs/guide/en/loops.md index d74c5ee4..c7700b23 100644 --- a/docs/guide/en/loops.md +++ b/docs/guide/en/loops.md @@ -4,9 +4,8 @@ Yii Queue uses `\Yiisoft\Queue\Cli\LoopInterface` to control long-running execut The loop is evaluated to determine whether it can continue: -- Before each message in `QueueConsumer::run()`. A message taken from the queue after the loop stopped is not processed - and goes back to the queue. -- By the adapter between messages in `QueueConsumer::listen()`. +- Before each message in `QueueConsumer::run()` and `QueueConsumer::listen()`. A message taken from the queue after the + loop stopped is not processed and goes back to the queue. - On each iteration of `queue:listen-all`. When `canContinue()` returns `false`, consuming stops gracefully (as soon as the current message is finished). diff --git a/src/QueueConsumer.php b/src/QueueConsumer.php index 2c2b5d6c..6e73da11 100644 --- a/src/QueueConsumer.php +++ b/src/QueueConsumer.php @@ -54,6 +54,9 @@ public function listen(): void } $this->logger->info('Start listening to the queue.'); $this->adapter->subscribe(function (MessageInterface $message): bool { + if (!$this->loop->canContinue()) { + return false; + } $this->worker->process($message, $this->queueName); return true; }); diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index 6f028748..97d0d0cb 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -160,6 +160,21 @@ public function testConsumerListenReportsMessageProcessedBeforeLoopStops(): void self::assertSame(MessageStatus::DONE, $adapter->status($message->getId())); } + public function testConsumerListenStopsBeforeProcessingAnotherMessage(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $first = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))); + $second = IdEnvelope::fromMessage($producer->push(new GenericMessage('simple', null))); + $consumer = new QueueConsumer($this->getWorker(), $this->createStopAfterFirstExecutionLoop(), new NullLogger(), $adapter); + + $consumer->listen(); + + self::assertSame(1, $this->executionTimes); + self::assertSame(MessageStatus::DONE, $adapter->status($first->getId())); + self::assertSame(MessageStatus::WAITING, $adapter->status($second->getId())); + } + private function createStopAfterFirstExecutionLoop(): LoopInterface { return new class (fn(): int => $this->executionTimes) implements LoopInterface {