fix: handle queue work results and isolate chained job dispatching

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

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,92 @@ public function testRunWithChainedQueueSucceed(): void
         ]);
     }
 
+    public function testRunWithChainedJobWhenDoneFailsDoesNotDispatchNextJob(): void
+    {
+        Time::setTestNow('2023-12-19 14:15:16');
+
+        $this->fakeChainedJob();
+
+        $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');
+
+        $this->fakeChainedJob();
+
+        $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');
+
+        $this->fakeChainedJob();
+
+        $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';
@@ -334,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,
+        ]);
+    }
 }