diff --git a/src/QueueWorker.php b/src/QueueWorker.php index 77cd335..069ffbb 100644 --- a/src/QueueWorker.php +++ b/src/QueueWorker.php @@ -420,6 +420,17 @@ protected function executeJobWithTimeout(JobInterface $job): void */ protected function executeJob(JobInterface $job): void { + $app = app(); + + // Each job is its own execution: callbacks it registers with defer() + // run when it finishes, after success or failure, and never leak + // into the next job handled by this worker. + if (is_object($app) && method_exists($app, 'runInDeferScope')) { + $app->runInDeferScope(static fn() => $job->handle()); + + return; + } + $job->handle(); } diff --git a/tests/Support/claim_worker.php b/tests/Support/claim_worker.php index 090d056..8f4430f 100644 --- a/tests/Support/claim_worker.php +++ b/tests/Support/claim_worker.php @@ -10,6 +10,10 @@ * job and kills itself with SIGKILL without acknowledging it. */ +// Deprecation notices printed by third-party packages while the autoloader loads +// (an old Mockery on PHP 8.5, for example) are noise here; real warnings still show. +error_reporting(E_ALL & ~E_DEPRECATED); + require __DIR__ . '/../../vendor/autoload.php'; use Doppar\Queue\Drivers\DatabaseDriver; diff --git a/tests/Support/slow_job_worker.php b/tests/Support/slow_job_worker.php index a8ff697..20e40fa 100644 --- a/tests/Support/slow_job_worker.php +++ b/tests/Support/slow_job_worker.php @@ -7,6 +7,10 @@ * Usage: php slow_job_worker.php */ +// Deprecation notices printed by third-party packages while the autoloader loads +// (an old Mockery on PHP 8.5, for example) are noise here; real warnings still show. +error_reporting(E_ALL & ~E_DEPRECATED); + require __DIR__ . '/../../vendor/autoload.php'; use Doppar\Queue\QueueManager; diff --git a/tests/Unit/QueueWorkerTest.php b/tests/Unit/QueueWorkerTest.php index 375f251..6fea57e 100644 --- a/tests/Unit/QueueWorkerTest.php +++ b/tests/Unit/QueueWorkerTest.php @@ -549,4 +549,92 @@ public function extend(ReservedJob $job, int $seconds): bool $this->assertStringContainsString('Could not renew job lease: redis timeout', $this->output); } + + // ------------------------------------------------------------------ + // Deferred cleanup + // ------------------------------------------------------------------ + + /** + * Install a container that implements the framework's defer scope API. + * + * @return object{log: array} + */ + private function installDeferAwareContainer(): object + { + $container = new class extends MockContainer { + public array $log = []; + + public int $scopes = 0; + + private array $stack = []; + + public function defer(callable $callback): static + { + if ($this->stack === []) { + throw new \LogicException('defer() outside a scope'); + } + + $this->stack[array_key_last($this->stack)][] = $callback; + + return $this; + } + + public function runInDeferScope(callable $callback): mixed + { + $this->scopes++; + $this->stack[] = []; + + try { + return $callback(); + } finally { + foreach (array_reverse(array_pop($this->stack)) as $cleanup) { + $cleanup(); + } + } + } + }; + + Container::setInstance($container); + $container->singleton(QueueManager::class, fn() => $this->manager); + $container->alias(QueueManager::class, 'queue.worker'); + + return $container; + } + + public function testEachJobRunsInItsOwnDeferScope(): void + { + $container = $this->installDeferAwareContainer(); + + RecordingJob::$during = function (RecordingJob $job) use ($container) { + $label = $job->label; + $container->defer(function () use ($container, $label) { $container->log[] = "cleanup-{$label}"; }); + $container->log[] = "run-{$label}"; + }; + + $this->manager->push(new RecordingJob('a')); + $this->manager->push(new RecordingJob('b')); + + $this->runOne(); + $this->runOne(); + + $this->assertSame(2, $container->scopes); + $this->assertSame(['run-a', 'cleanup-a', 'run-b', 'cleanup-b'], $container->log); + } + + public function testDeferredCallbacksRunWhenAJobFailsBeforeItIsReleased(): void + { + $container = $this->installDeferAwareContainer(); + + RecordingJob::$during = function () use ($container) { + $container->defer(function () use ($container) { $container->log[] = 'cleanup'; }); + + throw new \RuntimeException('job failed'); + }; + + $this->manager->push(new RecordingJob('a')); + $this->runOne(); + + $this->assertSame(['cleanup'], $container->log); + $this->assertSame(1, $this->driver()->stats('default')['delayed'] + $this->driver()->countFailed()); + } }