Keep the message that hits the limit in the queue (#340)

* Keep a message the handler refused in the queue

`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()`.

* 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.

* 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.

* Cover the pause between empty iterations

The pause line lost its only (accidental) coverage when the command
stopped sleeping on shutdown.

* Say on subscribe() too that false means the message was not consumed

Author: KalimeroMK

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 the order the adapter delivers them, until the queue is
+     * empty or the handler refuses a message.
      *
-     * @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 for a message it did not consume.
      */
     public function runExisting(callable $handlerCallback): void;
 
@@ -32,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;
 }

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;
     }
 }

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;
     }
 }

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;
     }
 }

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++;
         }

tests/Unit/Command/ListenAllCommandTest.php

@@ -30,14 +30,56 @@ 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->createLoop(1));
+        $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->createLoop(1));
+        $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.');
+    }
+
+    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 the given number of checks and then asks to stop.
+     */
+    private function createLoop(int $continueTimes): LoopInterface
+    {
+        return new class ($continueTimes) implements LoopInterface {
+            public function __construct(private int $continueTimes) {}
+
+            public function canContinue(): bool
+            {
+                return $this->continueTimes-- > 0;
+            }
+        };
     }
 }

tests/Unit/QueueTest.php

@@ -8,10 +8,12 @@
 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;
 use Yiisoft\Queue\Stubs\InMemoryAdapter;
+use Yiisoft\Queue\Stubs\StubLoop;
 use Yiisoft\Queue\Tests\TestCase;
 
 use function count;
@@ -81,5 +83,40 @@ 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::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());
     }
 }