From 9f8ad838918a3fc4008dd70f54ab9aa6e12bb8ad Mon Sep 17 00:00:00 2001 From: Bogdan Date: Tue, 10 Nov 2026 21:40:54 +0100 Subject: [PATCH 1/3] fix: handle queue work results and isolate chained job dispatching --- src/Commands/QueueWork.php | 68 ++++++++++++++++++++++++++++++++++---- 1 file changed, 62 insertions(+), 6 deletions(-) diff --git a/src/Commands/QueueWork.php b/src/Commands/QueueWork.php index d3a8848..ac5e07f 100644 --- a/src/Commands/QueueWork.php +++ b/src/Commands/QueueWork.php @@ -302,6 +302,7 @@ private function handleWork(QueueJob $work, QueueConfig $config, ?int $tries, ?i $payload = $work->payload; $payloadMetadata = null; + $completed = false; // Emit job processing started event QueueEventManager::jobProcessingStarted( @@ -325,7 +326,16 @@ private function handleWork(QueueJob $work, QueueConfig $config, ?int $tries, ?i $job->process(); // Mark as done - service('queue')->done($work); + if (! service('queue')->done($work)) { + log_message('error', sprintf( + 'Queue: failed to mark job %s from queue %s as done with the %s handler.', + $work->id, + $work->queue, + service('queue')->name(), + )); + } else { + $completed = true; + } // Emit job processing completed event QueueEventManager::jobProcessingCompleted( @@ -340,12 +350,31 @@ private function handleWork(QueueJob $work, QueueConfig $config, ?int $tries, ?i CLI::write('The processing of this job was successful', 'green'); - // Check chained jobs - $this->processNextJobInChain($payloadMetadata); + if (! $completed) { + return; + } + + // Chain failures must not turn an already completed job into a + // failure or trigger a retry of the current job. + try { + $this->processNextJobInChain($payloadMetadata); + } catch (Throwable $err) { + log_message('error', 'Queue: failed to process the next job in the chain: ' . $err->getMessage()); + CLI::write('The next job in the chain could not be queued.', 'red'); + } } catch (Throwable $err) { if (isset($job) && ++$work->attempts < ($tries ?? $job->getTries())) { // Schedule for later - service('queue')->later($work, $retryAfter ?? $job->getRetryAfter()); + $retryAfter ??= $job->getRetryAfter(); + + if (! service('queue')->later($work, $retryAfter)) { + log_message('error', sprintf( + 'Queue: failed to reschedule job %s from queue %s with the %s handler.', + $work->id, + $work->queue, + service('queue')->name(), + )); + } } else { // Mark as failed QueueEventManager::jobFailed( @@ -359,7 +388,14 @@ private function handleWork(QueueJob $work, QueueConfig $config, ?int $tries, ?i ], ); - service('queue')->failed($work, $err, $config->keepFailedJobs); + if (! service('queue')->failed($work, $err, $config->keepFailedJobs)) { + log_message('error', sprintf( + 'Queue: failed to mark job %s from queue %s as failed with the %s handler.', + $work->id, + $work->queue, + service('queue')->name(), + )); + } } CLI::write('The processing of this job failed', 'red'); } finally { @@ -396,13 +432,33 @@ private function processNextJobInChain(PayloadMetadata $payloadMetadata): void $nextPayload->setChainedJobs($payloadMetadata->getChainedJobs()); } - service('queue')->push( + $result = service('queue')->push( $nextPayload->getQueue(), $nextPayload->getJob(), $nextPayload->getData(), $nextPayload->getMetadata(), ); + if (! $result->getStatus()) { + $error = $result->getError() ?? 'Unknown error.'; + + log_message('error', sprintf( + 'Queue: failed to place chained job %s in the queue %s: %s', + $nextPayload->getJob(), + $nextPayload->getQueue(), + $error, + )); + + CLI::write(sprintf( + 'Chained job: %s could not be placed in the queue: %s. %s', + $nextPayload->getJob(), + $nextPayload->getQueue(), + $error, + ), 'red'); + + return; + } + CLI::write(sprintf('Chained job: %s has been placed in the queue: %s', $nextPayload->getJob(), $nextPayload->getQueue()), 'green'); } From fdd2df41d9b5be4115225b33952e222f36af86aa Mon Sep 17 00:00:00 2001 From: Bogdan Date: Tue, 10 Nov 2026 21:59:49 +0100 Subject: [PATCH 2/3] test: add tests for done failure, chained dispatch error isolation, and chained push failure warning --- tests/Commands/QueueWorkTest.php | 158 +++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/tests/Commands/QueueWorkTest.php b/tests/Commands/QueueWorkTest.php index af39092..3cb31f8 100644 --- a/tests/Commands/QueueWorkTest.php +++ b/tests/Commands/QueueWorkTest.php @@ -16,8 +16,11 @@ use CodeIgniter\Cache\CacheInterface; use CodeIgniter\Config\Services; use CodeIgniter\I18n\Time; +use CodeIgniter\Queue\Handlers\DatabaseHandler; use CodeIgniter\Queue\Models\QueueJobModel; +use CodeIgniter\Queue\QueuePushResult; use CodeIgniter\Test\Filters\CITestStreamFilter; +use RuntimeException; use Tests\Support\CLITestCase; /** @@ -175,6 +178,161 @@ public function testRunWithChainedQueueSucceed(): void ]); } + public function testRunWithChainedJobWhenDoneFailsDoesNotDispatchNextJob(): void + { + Time::setTestNow('2023-12-19 14:15:16'); + + fake(QueueJobModel::class, [ + 'connection' => 'database', + 'queue' => 'test', + 'payload' => [ + 'job' => 'success', + 'data' => ['key' => 'value'], + 'metadata' => [ + 'queue' => 'test', + 'chainedJobs' => [ + [ + 'job' => 'success', + 'data' => ['key3' => 'value3'], + 'metadata' => [ + 'queue' => 'queue', + ], + ], + ], + ], + ], + 'priority' => 'default', + 'status' => 0, + 'attempts' => 0, + 'available_at' => 1_702_977_074, + ]); + + $mockHandler = $this->getMockBuilder(DatabaseHandler::class) + ->setConstructorArgs([config('Queue')]) + ->onlyMethods(['done', 'push']) + ->getMock(); + $mockHandler->method('done')->willReturn(false); + $mockHandler->expects($this->never())->method('push'); + + Services::injectMock('queue', $mockHandler); + + CITestStreamFilter::registration(); + CITestStreamFilter::addOutputFilter(); + + $this->assertNotFalse(command('queue:work test sleep 1 --stop-when-empty')); + $output = $this->parseOutput(CITestStreamFilter::$buffer); + + CITestStreamFilter::removeOutputFilter(); + + $this->assertStringContainsString('The processing of this job was successful', $output); + $this->assertStringNotContainsString('Chained job:', $output); + $this->assertLogContains('error', 'failed to mark job'); + $this->dontSeeInDatabase('queue_jobs', ['queue' => 'queue']); + } + + public function testRunWithChainedJobWhenDispatchThrowsExceptionDoesNotFailOrRetryCurrentJob(): void + { + Time::setTestNow('2023-12-19 14:15:16'); + + fake(QueueJobModel::class, [ + 'connection' => 'database', + 'queue' => 'test', + 'payload' => [ + 'job' => 'success', + 'data' => ['key' => 'value'], + 'metadata' => [ + 'queue' => 'test', + 'chainedJobs' => [ + [ + 'job' => 'success', + 'data' => ['key3' => 'value3'], + 'metadata' => [ + 'queue' => 'queue', + ], + ], + ], + ], + ], + 'priority' => 'default', + 'status' => 0, + 'attempts' => 0, + 'available_at' => 1_702_977_074, + ]); + + $mockHandler = $this->getMockBuilder(DatabaseHandler::class) + ->setConstructorArgs([config('Queue')]) + ->onlyMethods(['push', 'failed', 'later']) + ->getMock(); + $mockHandler->method('push')->willThrowException(new RuntimeException('Push service unavailable')); + $mockHandler->expects($this->never())->method('failed'); + $mockHandler->expects($this->never())->method('later'); + + Services::injectMock('queue', $mockHandler); + + CITestStreamFilter::registration(); + CITestStreamFilter::addOutputFilter(); + + $this->assertNotFalse(command('queue:work test sleep 1 --stop-when-empty')); + $output = $this->parseOutput(CITestStreamFilter::$buffer); + + CITestStreamFilter::removeOutputFilter(); + + $this->assertStringContainsString('The processing of this job was successful', $output); + $this->assertStringContainsString('The next job in the chain could not be queued.', $output); + $this->assertStringNotContainsString('The processing of this job failed', $output); + $this->assertLogContains('error', 'Queue: failed to process the next job in the chain: Push service unavailable'); + } + + public function testRunWithChainedJobWhenPushReturnsFailureShowsWarning(): void + { + Time::setTestNow('2023-12-19 14:15:16'); + + fake(QueueJobModel::class, [ + 'connection' => 'database', + 'queue' => 'test', + 'payload' => [ + 'job' => 'success', + 'data' => ['key' => 'value'], + 'metadata' => [ + 'queue' => 'test', + 'chainedJobs' => [ + [ + 'job' => 'success', + 'data' => ['key3' => 'value3'], + 'metadata' => [ + 'queue' => 'queue', + ], + ], + ], + ], + ], + 'priority' => 'default', + 'status' => 0, + 'attempts' => 0, + 'available_at' => 1_702_977_074, + ]); + + $mockHandler = $this->getMockBuilder(DatabaseHandler::class) + ->setConstructorArgs([config('Queue')]) + ->onlyMethods(['push']) + ->getMock(); + $mockHandler->method('push')->willReturn(QueuePushResult::failure('Simulated push error message')); + + Services::injectMock('queue', $mockHandler); + + CITestStreamFilter::registration(); + CITestStreamFilter::addOutputFilter(); + + $this->assertNotFalse(command('queue:work test sleep 1 --stop-when-empty')); + $output = $this->parseOutput(CITestStreamFilter::$buffer); + + CITestStreamFilter::removeOutputFilter(); + + $this->assertStringContainsString('The processing of this job was successful', $output); + $this->assertStringContainsString('Chained job: success could not be placed in the queue: queue. Simulated push error message', $output); + $this->assertLogContains('error', 'Queue: failed to place chained job success in the queue queue: Simulated push error message'); + } + public function testRunWithTaskLock(): void { $lockKey = 'test_lock_key'; From 22286fdc1f461ca7da41abf1dc5b9d6f56b115c2 Mon Sep 17 00:00:00 2001 From: Bogdan Date: Wed, 11 Nov 2026 01:02:38 +0100 Subject: [PATCH 3/3] test: extract duplicated chained job fixture in QueueWorkTest --- tests/Commands/QueueWorkTest.php | 106 ++++++++++--------------------- 1 file changed, 34 insertions(+), 72 deletions(-) diff --git a/tests/Commands/QueueWorkTest.php b/tests/Commands/QueueWorkTest.php index 3cb31f8..79ab6fa 100644 --- a/tests/Commands/QueueWorkTest.php +++ b/tests/Commands/QueueWorkTest.php @@ -182,30 +182,7 @@ public function testRunWithChainedJobWhenDoneFailsDoesNotDispatchNextJob(): void { Time::setTestNow('2023-12-19 14:15:16'); - fake(QueueJobModel::class, [ - 'connection' => 'database', - 'queue' => 'test', - 'payload' => [ - 'job' => 'success', - 'data' => ['key' => 'value'], - 'metadata' => [ - 'queue' => 'test', - 'chainedJobs' => [ - [ - 'job' => 'success', - 'data' => ['key3' => 'value3'], - 'metadata' => [ - 'queue' => 'queue', - ], - ], - ], - ], - ], - 'priority' => 'default', - 'status' => 0, - 'attempts' => 0, - 'available_at' => 1_702_977_074, - ]); + $this->fakeChainedJob(); $mockHandler = $this->getMockBuilder(DatabaseHandler::class) ->setConstructorArgs([config('Queue')]) @@ -234,30 +211,7 @@ public function testRunWithChainedJobWhenDispatchThrowsExceptionDoesNotFailOrRet { Time::setTestNow('2023-12-19 14:15:16'); - fake(QueueJobModel::class, [ - 'connection' => 'database', - 'queue' => 'test', - 'payload' => [ - 'job' => 'success', - 'data' => ['key' => 'value'], - 'metadata' => [ - 'queue' => 'test', - 'chainedJobs' => [ - [ - 'job' => 'success', - 'data' => ['key3' => 'value3'], - 'metadata' => [ - 'queue' => 'queue', - ], - ], - ], - ], - ], - 'priority' => 'default', - 'status' => 0, - 'attempts' => 0, - 'available_at' => 1_702_977_074, - ]); + $this->fakeChainedJob(); $mockHandler = $this->getMockBuilder(DatabaseHandler::class) ->setConstructorArgs([config('Queue')]) @@ -287,30 +241,7 @@ public function testRunWithChainedJobWhenPushReturnsFailureShowsWarning(): void { Time::setTestNow('2023-12-19 14:15:16'); - fake(QueueJobModel::class, [ - 'connection' => 'database', - 'queue' => 'test', - 'payload' => [ - 'job' => 'success', - 'data' => ['key' => 'value'], - 'metadata' => [ - 'queue' => 'test', - 'chainedJobs' => [ - [ - 'job' => 'success', - 'data' => ['key3' => 'value3'], - 'metadata' => [ - 'queue' => 'queue', - ], - ], - ], - ], - ], - 'priority' => 'default', - 'status' => 0, - 'attempts' => 0, - 'available_at' => 1_702_977_074, - ]); + $this->fakeChainedJob(); $mockHandler = $this->getMockBuilder(DatabaseHandler::class) ->setConstructorArgs([config('Queue')]) @@ -492,4 +423,35 @@ public function testLockClearedOnFailure(): void $this->assertSame('Starting a new job: failure, with ID: 1', $this->getLine(3)); $this->assertSame('The processing of this job failed', $this->getLine(4)); } + + /** + * Seed a job in the "test" queue that chains one job into the "queue" queue. + */ + private function fakeChainedJob(): void + { + fake(QueueJobModel::class, [ + 'connection' => 'database', + 'queue' => 'test', + 'payload' => [ + 'job' => 'success', + 'data' => ['key' => 'value'], + 'metadata' => [ + 'queue' => 'test', + 'chainedJobs' => [ + [ + 'job' => 'success', + 'data' => ['key3' => 'value3'], + 'metadata' => [ + 'queue' => 'queue', + ], + ], + ], + ], + ], + 'priority' => 'default', + 'status' => 0, + 'attempts' => 0, + 'available_at' => 1_702_977_074, + ]); + } }