Keep the message that hits the limit in the queue

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;

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/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());
     }
 }