From 3688c738c74bddc8890d8405ced836bb8c8ea330 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Wed, 28 Oct 2026 01:49:59 +0300 Subject: [PATCH 1/2] Fix #34: Return refused message to the queue and stop handling --- CHANGELOG.md | 1 + src/Adapter.php | 33 +++++++- tests/AdapterRunTest.php | 159 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 190 insertions(+), 3 deletions(-) create mode 100644 tests/AdapterRunTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fce813..5454993 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/src/Adapter.php b/src/Adapter.php index 65841ff..8c49cd5 100644 --- a/src/Adapter.php +++ b/src/Adapter.php @@ -138,7 +138,8 @@ 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. */ @@ -146,9 +147,11 @@ public function run(callable $handlerCallback, bool $repeat, int $timeout = 0): { 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. */ diff --git a/tests/AdapterRunTest.php b/tests/AdapterRunTest.php new file mode 100644 index 0000000..177e9b3 --- /dev/null +++ b/tests/AdapterRunTest.php @@ -0,0 +1,159 @@ +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 + */ + 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 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, + ); + } +} From 86f20fd94d89f61ddfd72f61aa736fd883b8b226 Mon Sep 17 00:00:00 2001 From: Alexander Makarov Date: Thu, 29 Oct 2026 17:49:01 +0300 Subject: [PATCH 2/2] Enable pdo_sqlite for PHPUnit on all runners AdapterRunTest uses an SQLite database, and the Windows PHP builds do not load pdo_sqlite by default. --- .github/workflows/build.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 0516d51..cdfa61d 100644 --- a/.github/workflows/build.yml +++ b/.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 }}