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'); } diff --git a/tests/Commands/QueueWorkTest.php b/tests/Commands/QueueWorkTest.php index af39092..79ab6fa 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,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, + ]); + } }