Fix #34: Return refused message to the queue and stop handling

.github/workflows/build.yml

@@ -28,5 +28,6 @@ jobs:
         ['ubuntu-latest', 'windows-latest']
       php: >-
         ['8.1', '8.2', '8.3', '8.4', '8.5']
+      extensions: pdo, pdo_sqlite
     secrets:
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

CHANGELOG.md

@@ -3,3 +3,4 @@
 ## 1.0.0 under development
 
 - Initial release.
+- Bug #34: Return a message refused by the handler to the queue and stop instead of reserving the remaining messages (@samdark)

src/Adapter.php

@@ -138,17 +138,20 @@ public function withChannel(BackedEnum|string $channel): self
     /**
      * Listens queue and runs each job.
      *
-     * @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 if
+     * it did not process the message and cannot continue handling messages. Such a message is returned to the queue.
      * @param bool $repeat whether to continue listening when queue is empty.
      * @param non-negative-int $timeout number of seconds to sleep before next iteration.
      */
     public function run(callable $handlerCallback, bool $repeat, int $timeout = 0): void
     {
         while ($this->loop->canContinue()) {
             if ($payload = $this->reserve()) {
-                if ($handlerCallback($this->serializer->unserialize($payload['job']))) {
-                    $this->release($payload);
+                if (!$handlerCallback($this->serializer->unserialize($payload['job']))) {
+                    $this->cancelReservation($payload);
+                    break;
                 }
+                $this->release($payload);
                 continue;
             }
             if (!$repeat) {
@@ -230,6 +233,30 @@ protected function release($payload): void
         }
     }
 
+    /**
+     * Returns a message the handler did not process back to the waiting list as it was before {@see reserve()}.
+     *
+     * The update is skipped if the reservation expired and the message was reserved again in the meantime.
+     *
+     * @param array $payload
+     */
+    private function cancelReservation(array $payload): void
+    {
+        $attempt = (int) $payload['attempt'] - 1;
+
+        $this->db->createCommand()->update(
+            $this->tableName,
+            [
+                'reserved_at' => null,
+                'attempt' => $attempt > 0 ? $attempt : null,
+            ],
+            [
+                'id' => $payload['id'],
+                'reserved_at' => $payload['reserved_at'],
+            ],
+        )->execute();
+    }
+
     /**
      * Moves expired messages into waiting list.
      */

tests/AdapterRunTest.php

@@ -0,0 +1,159 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Db\Tests;
+
+use PHPUnit\Framework\TestCase;
+use Yiisoft\Db\Connection\ConnectionInterface;
+use Yiisoft\Db\Query\Query;
+use Yiisoft\Db\Sqlite\Connection;
+use Yiisoft\Db\Sqlite\Driver;
+use Yiisoft\Mutex\MutexFactoryInterface;
+use Yiisoft\Mutex\MutexInterface;
+use Yiisoft\Queue\Cli\SimpleLoop;
+use Yiisoft\Queue\Db\Adapter;
+use Yiisoft\Queue\Message\GenericMessage;
+use Yiisoft\Queue\Message\IdEnvelope;
+use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\Message\Serializer\JsonMessageEncoder;
+use Yiisoft\Queue\Message\Serializer\MessageSerializer;
+use Yiisoft\Queue\MessageStatus;
+
+use function array_slice;
+use function count;
+
+final class AdapterRunTest extends TestCase
+{
+    private ConnectionInterface $db;
+
+    protected function setUp(): void
+    {
+        $this->db = new Connection(new Driver('sqlite::memory:'));
+        $this->db->createCommand()->createTable('{{%queue}}', [
+            'id' => 'integer PRIMARY KEY AUTOINCREMENT',
+            'channel' => 'varchar(255) NOT NULL',
+            'job' => 'blob NOT NULL',
+            'pushed_at' => 'integer NOT NULL',
+            'ttr' => 'integer NOT NULL',
+            'delay' => 'integer NOT NULL DEFAULT 0',
+            'priority' => 'integer NOT NULL DEFAULT 1024',
+            'reserved_at' => 'integer DEFAULT NULL',
+            'attempt' => 'integer DEFAULT NULL',
+            'done_at' => 'integer DEFAULT NULL',
+        ])->execute();
+    }
+
+    public function testRunExistingStopsAndReturnsMessageWhenHandlerRefuses(): void
+    {
+        $adapter = $this->createAdapter();
+        $ids = $this->pushMessages($adapter, 5);
+
+        $handled = $this->runExisting($adapter, 2);
+
+        self::assertSame([1, 2], $handled);
+        foreach (array_slice($ids, 2) as $id) {
+            self::assertSame(MessageStatus::WAITING, $adapter->status($id));
+            self::assertNull($this->getRow($id)['attempt']);
+        }
+    }
+
+    public function testRunExistingContinuesWithRefusedMessageOnNextRun(): void
+    {
+        $adapter = $this->createAdapter();
+        $this->pushMessages($adapter, 5);
+
+        $this->runExisting($adapter, 2);
+        $handled = $this->runExisting($adapter, 2);
+
+        self::assertSame([3, 4], $handled);
+    }
+
+    public function testRunExistingKeepsAttemptOfRetriedMessageWhenHandlerRefuses(): void
+    {
+        $adapter = $this->createAdapter();
+        [$id] = $this->pushMessages($adapter, 1);
+        $this->db->createCommand()->update('{{%queue}}', ['attempt' => 2], ['id' => $id])->execute();
+
+        $this->runExisting($adapter, 0);
+
+        $row = $this->getRow($id);
+        self::assertNull($row['reserved_at']);
+        self::assertSame(2, (int) $row['attempt']);
+    }
+
+    public function testSubscribeStopsAndReturnsMessageWhenHandlerRefuses(): void
+    {
+        $adapter = $this->createAdapter();
+        [$id] = $this->pushMessages($adapter, 1);
+
+        $adapter->subscribe(static fn(): bool => false);
+
+        self::assertSame(MessageStatus::WAITING, $adapter->status($id));
+        self::assertNull($this->getRow($id)['attempt']);
+    }
+
+    public function testRunExistingReleasesAcceptedMessages(): void
+    {
+        $adapter = $this->createAdapter();
+        $ids = $this->pushMessages($adapter, 2);
+
+        $handled = $this->runExisting($adapter, 2);
+
+        self::assertSame([1, 2], $handled);
+        self::assertNull($this->getRow($ids[0]));
+        self::assertNull($this->getRow($ids[1]));
+    }
+
+    /**
+     * @return list<int|string>
+     */
+    private function pushMessages(Adapter $adapter, int $count): array
+    {
+        $ids = [];
+        for ($i = 1; $i <= $count; $i++) {
+            $ids[] = IdEnvelope::fromMessage($adapter->push(new GenericMessage('job', $i)))->getId();
+        }
+
+        return $ids;
+    }
+
+    /**
+     * Runs existing messages, accepting at most `$limit` of them and refusing the next one.
+     *
+     * @return list<mixed> Payloads of the accepted messages.
+     */
+    private function runExisting(Adapter $adapter, int $limit): array
+    {
+        $handled = [];
+        $adapter->runExisting(static function (MessageInterface $message) use (&$handled, $limit): bool {
+            if (count($handled) >= $limit) {
+                return false;
+            }
+            $handled[] = $message->getPayload();
+            return true;
+        });
+
+        return $handled;
+    }
+
+    private function getRow(int|string $id): ?array
+    {
+        return (new Query($this->db))->from('{{%queue}}')->where(['id' => $id])->one();
+    }
+
+    private function createAdapter(): Adapter
+    {
+        $mutex = $this->createMock(MutexInterface::class);
+        $mutex->method('acquire')->willReturn(true);
+        $mutexFactory = $this->createMock(MutexFactoryInterface::class);
+        $mutexFactory->method('create')->willReturn($mutex);
+
+        return new Adapter(
+            $this->db,
+            new MessageSerializer(new JsonMessageEncoder()),
+            new SimpleLoop(),
+            $mutexFactory,
+        );
+    }
+}