From a25808836c47744aee454325634b7cf8850af0e6 Mon Sep 17 00:00:00 2001 From: KalimeroMK Date: Tue, 13 Oct 2026 00:08:29 +0200 Subject: [PATCH 1/5] 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 e2b218261b80ef902f10d0f4a564f372d15fc820 Mon Sep 17 00:00:00 2001 From: KalimeroMK Date: Wed, 28 Oct 2026 20:43:43 +0100 Subject: [PATCH 2/5] Check the loop before handling a message A handled message was kept and handled again when the loop stopped right after it. The callback now returns false only for a message it left untouched, which is what the adapter contract says. Drop the FIFO claim and the contract sentence on subscribe(): queue-db orders by priority then id, and queue-amqp's subscribe() acks regardless of the callback result. --- src/Adapter/AdapterInterface.php | 8 +++----- src/QueueConsumer.php | 6 +++++- tests/Unit/QueueTest.php | 23 +++++++++++++++++++++++ 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/src/Adapter/AdapterInterface.php b/src/Adapter/AdapterInterface.php index e2e5286a..cdae0d8c 100644 --- a/src/Adapter/AdapterInterface.php +++ b/src/Adapter/AdapterInterface.php @@ -10,13 +10,13 @@ interface AdapterInterface { /** - * Passes existing messages to the given handler in FIFO order, until the queue is empty or the handler asks - * to stop. + * Passes existing messages to the given handler, in the order the adapter delivers them, until the queue is + * empty or the handler refuses a message. * * 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. + * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false for a message it did not consume. */ public function runExisting(callable $handlerCallback): void; @@ -36,8 +36,6 @@ 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/src/QueueConsumer.php b/src/QueueConsumer.php index ffcfbe7b..c16d1d97 100644 --- a/src/QueueConsumer.php +++ b/src/QueueConsumer.php @@ -58,7 +58,11 @@ public function listen(): void private function handle(MessageInterface $message): bool { + if (!$this->loop->canContinue()) { + return false; + } + $this->worker->process($message, $this->queueName); - return $this->loop->canContinue(); + return true; } } diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php index f09568a1..a0bc23ca 100644 --- a/tests/Unit/QueueTest.php +++ b/tests/Unit/QueueTest.php @@ -8,6 +8,7 @@ use Yiisoft\Queue\QueueConsumer; use Yiisoft\Queue\QueueConsumerInterface; use Yiisoft\Queue\QueueProducerInterface; +use Yiisoft\Queue\Cli\LoopInterface; use Yiisoft\Queue\Message\GenericMessage; use Yiisoft\Queue\Message\IdEnvelope; use Yiisoft\Queue\MessageStatus; @@ -94,6 +95,28 @@ public function testMessageIsNotConsumedWhenLoopCannotContinue(): void $consumer = new QueueConsumer($this->getWorker(), new StubLoop(false), new NullLogger(), $adapter); self::assertSame(0, $consumer->run()); + self::assertSame(0, $this->executionTimes); self::assertCount(2, $adapter->getMessagesList()); } + + public function testMessageHandledBeforeStopIsNotHandledAgain(): void + { + $adapter = new InMemoryAdapter(); + $producer = $this->createQueue($adapter); + $producer->push(new GenericMessage('simple', null)); + $producer->push(new GenericMessage('simple', null)); + $loop = $this->createMock(LoopInterface::class); + $loop->method('canContinue')->willReturn(true, false); + $consumer = new QueueConsumer($this->getWorker(), $loop, new NullLogger(), $adapter); + + self::assertSame(1, $consumer->run()); + self::assertSame(1, $this->executionTimes); + self::assertCount(1, $adapter->getMessagesList()); + + $consumer = new QueueConsumer($this->getWorker(), new StubLoop(), new NullLogger(), $adapter); + + self::assertSame(1, $consumer->run()); + self::assertSame(2, $this->executionTimes, 'The message handled before the stop is not handled again.'); + self::assertCount(0, $adapter->getMessagesList()); + } } From f9668530aa415077e77f29f168412bfec53efc0a Mon Sep 17 00:00:00 2001 From: KalimeroMK Date: Wed, 28 Oct 2026 20:56:02 +0100 Subject: [PATCH 3/5] Do not pause queue:listen-all after the loop asked to stop The pause between iterations ran even when the loop had already stopped, so SIGTERM waited out the full --pause before the command exited. The limit option is read once, and both commands return Command::SUCCESS. --- src/Command/ListenAllCommand.php | 8 ++-- src/Command/RunCommand.php | 2 +- tests/Unit/Command/ListenAllCommandTest.php | 42 ++++++++++++++++++--- 3 files changed, 42 insertions(+), 10 deletions(-) diff --git a/src/Command/ListenAllCommand.php b/src/Command/ListenAllCommand.php index 5ef5d75f..6735ae09 100644 --- a/src/Command/ListenAllCommand.php +++ b/src/Command/ListenAllCommand.php @@ -84,18 +84,20 @@ protected function execute(InputInterface $input, OutputInterface $output): int $pauseSeconds = 1; } + $limit = (int) $input->getOption('limit'); + while ($this->loop->canContinue()) { $hasMessages = false; foreach ($consumers as $consumer) { - $hasMessages = $consumer->run((int) $input->getOption('limit')) > 0 || $hasMessages; + $hasMessages = $consumer->run($limit) > 0 || $hasMessages; } - if (!$hasMessages) { + if (!$hasMessages && $this->loop->canContinue()) { /** @psalm-var 0|positive-int $pauseSeconds */ sleep($pauseSeconds); } } - return 0; + return Command::SUCCESS; } } diff --git a/src/Command/RunCommand.php b/src/Command/RunCommand.php index c6f9a422..648eefed 100644 --- a/src/Command/RunCommand.php +++ b/src/Command/RunCommand.php @@ -60,6 +60,6 @@ protected function execute(InputInterface $input, OutputInterface $output): int $output->writeln("Messages processed: $count."); } - return 0; + return Command::SUCCESS; } } diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php index dd457fff..29247006 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ListenAllCommandTest.php @@ -30,14 +30,44 @@ public function testRunsOnlyConsumerRolesByDefault(): void { $consumer = $this->createMock(QueueConsumerInterface::class); $consumer->expects($this->once())->method('run')->willReturn(0); - $loop = $this->createMock(LoopInterface::class); - $loop->method('canContinue')->willReturn(true, false); $command = new ListenAllCommand(new PredefinedQueueProvider([ 'producer' => ['producer' => new StubQueueProducer()], 'consumer' => ['consumer' => $consumer], - ]), $loop); - $input = new ArrayInput([], $command->getNativeDefinition()); - $input->setOption('pause', 0); - self::assertSame(0, $command->run($input, $this->createMock(OutputInterface::class))); + ]), $this->createStoppingLoop()); + $input = new ArrayInput(['--pause' => 0], $command->getNativeDefinition()); + self::assertSame(Command::SUCCESS, $command->run($input, $this->createMock(OutputInterface::class))); + } + + public function testDoesNotPauseWhenTheLoopCannotContinue(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->method('run')->willReturn(0); + $command = new ListenAllCommand(new PredefinedQueueProvider([ + 'consumer' => ['consumer' => $consumer], + ]), $this->createStoppingLoop()); + $input = new ArrayInput(['--pause' => 3], $command->getNativeDefinition()); + + $start = hrtime(true); + $command->run($input, $this->createMock(OutputInterface::class)); + + self::assertLessThan(1.0, (hrtime(true) - $start) / 1_000_000_000, 'The pause is skipped once the loop asked to stop.'); + } + + /** + * @return LoopInterface A loop which allows a single iteration and then asks to stop. + */ + private function createStoppingLoop(): LoopInterface + { + return new class implements LoopInterface { + private bool $canContinue = true; + + public function canContinue(): bool + { + $canContinue = $this->canContinue; + $this->canContinue = false; + + return $canContinue; + } + }; } } From 4fe6c88f42f7958ea2b3e6336733d87906992987 Mon Sep 17 00:00:00 2001 From: KalimeroMK Date: Wed, 28 Oct 2026 21:24:35 +0100 Subject: [PATCH 4/5] Cover the pause between empty iterations The pause line lost its only (accidental) coverage when the command stopped sleeping on shutdown. --- tests/Unit/Command/ListenAllCommandTest.php | 32 ++++++++++++++------- 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php index 29247006..7053c02d 100644 --- a/tests/Unit/Command/ListenAllCommandTest.php +++ b/tests/Unit/Command/ListenAllCommandTest.php @@ -33,7 +33,7 @@ public function testRunsOnlyConsumerRolesByDefault(): void $command = new ListenAllCommand(new PredefinedQueueProvider([ 'producer' => ['producer' => new StubQueueProducer()], 'consumer' => ['consumer' => $consumer], - ]), $this->createStoppingLoop()); + ]), $this->createLoop(1)); $input = new ArrayInput(['--pause' => 0], $command->getNativeDefinition()); self::assertSame(Command::SUCCESS, $command->run($input, $this->createMock(OutputInterface::class))); } @@ -44,7 +44,7 @@ public function testDoesNotPauseWhenTheLoopCannotContinue(): void $consumer->method('run')->willReturn(0); $command = new ListenAllCommand(new PredefinedQueueProvider([ 'consumer' => ['consumer' => $consumer], - ]), $this->createStoppingLoop()); + ]), $this->createLoop(1)); $input = new ArrayInput(['--pause' => 3], $command->getNativeDefinition()); $start = hrtime(true); @@ -53,20 +53,32 @@ public function testDoesNotPauseWhenTheLoopCannotContinue(): void self::assertLessThan(1.0, (hrtime(true) - $start) / 1_000_000_000, 'The pause is skipped once the loop asked to stop.'); } + public function testPausesWhenThereAreNoMessages(): void + { + $consumer = $this->createMock(QueueConsumerInterface::class); + $consumer->method('run')->willReturn(0); + $command = new ListenAllCommand(new PredefinedQueueProvider([ + 'consumer' => ['consumer' => $consumer], + ]), $this->createLoop(2)); + $input = new ArrayInput(['--pause' => 1], $command->getNativeDefinition()); + + $start = hrtime(true); + $command->run($input, $this->createMock(OutputInterface::class)); + + self::assertGreaterThanOrEqual(1.0, (hrtime(true) - $start) / 1_000_000_000, 'An empty queue is polled once per pause.'); + } + /** - * @return LoopInterface A loop which allows a single iteration and then asks to stop. + * @return LoopInterface A loop which allows the given number of checks and then asks to stop. */ - private function createStoppingLoop(): LoopInterface + private function createLoop(int $continueTimes): LoopInterface { - return new class implements LoopInterface { - private bool $canContinue = true; + return new class ($continueTimes) implements LoopInterface { + public function __construct(private int $continueTimes) {} public function canContinue(): bool { - $canContinue = $this->canContinue; - $this->canContinue = false; - - return $canContinue; + return $this->continueTimes-- > 0; } }; } From 8a4bf0bfacbe76a514d58533e764f99b5a5683bf Mon Sep 17 00:00:00 2001 From: KalimeroMK Date: Wed, 28 Oct 2026 23:44:04 +0100 Subject: [PATCH 5/5] Say on subscribe() too that false means the message was not consumed --- src/Adapter/AdapterInterface.php | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Adapter/AdapterInterface.php b/src/Adapter/AdapterInterface.php index cdae0d8c..ea332ad7 100644 --- a/src/Adapter/AdapterInterface.php +++ b/src/Adapter/AdapterInterface.php @@ -36,7 +36,10 @@ public function push(MessageInterface $message): MessageInterface; /** * Listen to the queue and pass messages to the given handler as they come. * - * @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, as in {@see runExisting()}: that message must + * stay in the queue. + * + * @param callable(MessageInterface): bool $handlerCallback The handler which will handle messages. Returns false for a message it did not consume. */ public function subscribe(callable $handlerCallback): void; }