From 36f52a50693d535c92ce9692e7f3d4a1dd62b46e Mon Sep 17 00:00:00 2001 From: Arif Hoque Date: Fri, 16 Oct 2026 23:09:54 +0600 Subject: [PATCH] support defer and execution context for each job --- src/QueueWorker.php | 66 ++++++++++++ tests/Mock/Jobs/DeferringFailedJob.php | 38 +++++++ tests/Unit/QueueWorkerTest.php | 138 ++++++++++++++++++++++++- 3 files changed, 238 insertions(+), 4 deletions(-) create mode 100644 tests/Mock/Jobs/DeferringFailedJob.php diff --git a/src/QueueWorker.php b/src/QueueWorker.php index 069ffbb..5c20bd1 100644 --- a/src/QueueWorker.php +++ b/src/QueueWorker.php @@ -234,6 +234,31 @@ protected function processNextJob(string $queue): void * @return void */ protected function processJob(ReservedJob $queueJob): void + { + $app = app(); + + $lifecycle = fn() => $this->runJobLifecycle($queueJob); + + if (is_object($app) && method_exists($app, 'runInDeferScope')) { + $lifecycle = fn() => $app->runInDeferScope(fn() => $this->runJobLifecycle($queueJob)); + } + + if (is_object($app) && method_exists($app, 'runInContext')) { + $app->runInContext('worker', $this->executionMetadata($queueJob), $lifecycle); + + return; + } + + $lifecycle(); + } + + /** + * Run the full lifecycle of a reserved job + * + * @param ReservedJob $queueJob + * @return void + */ + protected function runJobLifecycle(ReservedJob $queueJob): void { $job = null; @@ -242,6 +267,11 @@ protected function processJob(ReservedJob $queueJob): void $job = $this->manager->unserializeJob($queueJob->payload); $job->attempts = $queueJob->attempts; + $this->stampExecution([ + 'job' => $job::class, + 'job_id' => $job->getJobId(), + ]); + // A job whose lease expired mid-run comes back with its attempts // already spent; do not run it again. $maxAttempts = max(1, $job->tries()); @@ -301,6 +331,42 @@ protected function processJob(ReservedJob $queueJob): void } } + /** + * Describe a reserved job for its worker execution context + * + * @param ReservedJob $queueJob + * @return array + */ + protected function executionMetadata(ReservedJob $queueJob): array + { + return [ + 'queue' => $queueJob->queue, + 'queue_job_id' => $queueJob->id, + 'attempt' => $queueJob->attempts, + 'connection' => $this->connection, + ]; + } + + /** + * Add metadata to the active execution context, when the framework has one. + * + * @param array $metadata + * @return void + */ + protected function stampExecution(array $metadata): void + { + $app = app(); + + if ( + is_object($app) + && method_exists($app, 'stampContext') + && method_exists($app, 'hasExecutionContext') + && $app->hasExecutionContext() + ) { + $app->stampContext(array_filter($metadata, static fn($value) => $value !== null)); + } + } + /** * Execute a job with timeout protection using process forking. * diff --git a/tests/Mock/Jobs/DeferringFailedJob.php b/tests/Mock/Jobs/DeferringFailedJob.php new file mode 100644 index 0000000..c15b3df --- /dev/null +++ b/tests/Mock/Jobs/DeferringFailedJob.php @@ -0,0 +1,38 @@ +stack === []) { - throw new \LogicException('defer() outside a scope'); + if (array_key_last($this->stack) === 0) { + $this->leakedToOuterScope++; } $this->stack[array_key_last($this->stack)][] = $callback; @@ -617,10 +624,133 @@ public function testEachJobRunsInItsOwnDeferScope(): void $this->runOne(); $this->runOne(); - $this->assertSame(2, $container->scopes); + // One scope for running the job body and one around its whole lifecycle. + $this->assertSame(4, $container->scopes); $this->assertSame(['run-a', 'cleanup-a', 'run-b', 'cleanup-b'], $container->log); } + /** + * The worker only opens contexts when the installed framework has them. + */ + private function requireExecutionContexts(): void + { + if (!method_exists(Container::class, 'runInContext')) { + $this->markTestSkipped('The installed framework has no execution context support.'); + } + } + + /** + * Collect the final state of every context as it ends. + * + * @return \ArrayObject + */ + private function recordEndedContexts(object $container): \ArrayObject + { + $ended = new \ArrayObject(); + $container->contextEnding(function ($context) use ($ended) { + $ended[] = $context; + }); + + return $ended; + } + + public function testEachJobRunsInItsOwnWorkerContextWithJobMetadata(): void + { + $this->requireExecutionContexts(); + $container = $this->installDeferAwareContainer(); + $ended = $this->recordEndedContexts($container); + + $this->manager->push(new RecordingJob('a')); + $this->manager->push(new RecordingJob('b')); + + $this->runOne(); + $this->runOne(); + + $this->assertCount(2, $ended); + + [$first, $second] = $ended->getArrayCopy(); + + $this->assertSame('worker', $first->type()); + $this->assertNotSame($first->id(), $second->id()); + $this->assertSame('default', $first->get('queue')); + $this->assertSame(1, $first->get('attempt')); + $this->assertSame(RecordingJob::class, $first->get('job')); + $this->assertNotEmpty($first->get('job_id')); + $this->assertNotSame($first->get('queue_job_id'), $second->get('queue_job_id')); + $this->assertFalse($container->hasExecutionContext(), 'no context outlives the job'); + } + + public function testExecutionScopedStateIsNotSharedBetweenSequentialJobs(): void + { + $this->requireExecutionContexts(); + $container = $this->installDeferAwareContainer(); + $container->bindExecution('job.state', fn() => new \stdClass()); + $seen = []; + + RecordingJob::$during = function (RecordingJob $job) use ($container, &$seen) { + $state = $container->get('job.state'); + $state->count = ($state->count ?? 0) + 1; + $seen[$job->label] = ['object' => spl_object_id($state), 'count' => $state->count]; + }; + + $this->manager->push(new RecordingJob('a')); + $this->manager->push(new RecordingJob('b')); + + $this->runOne(); + $this->runOne(); + + $this->assertSame(1, $seen['a']['count']); + $this->assertSame(1, $seen['b']['count'], 'the second job must start from fresh state'); + } + + public function testDeferInsideFailedIsIsolatedToTheJobInsteadOfTheWorkerScope(): void + { + $container = $this->installDeferAwareContainer(); + + DeferringFailedJob::$inFailed = function () use ($container) { + $container->defer(function () use ($container) { $container->log[] = 'failed-cleanup'; }); + $container->log[] = 'failed-ran'; + }; + + $this->manager->push(new DeferringFailedJob()); + $this->runOne(); + + $this->assertSame(0, $container->leakedToOuterScope, 'nothing may accumulate in the worker command scope'); + $this->assertSame(['failed-ran', 'failed-cleanup'], $container->log); + } + + public function testDeferInsideTheProcessedHookIsIsolatedToTheJob(): void + { + $container = $this->installDeferAwareContainer(); + + $this->worker->setOnJobProcessed(function () use ($container) { + $container->defer(function () use ($container) { $container->log[] = 'hook-cleanup'; }); + }); + + $this->manager->push(new RecordingJob('a')); + $this->runOne(); + + $this->assertSame(0, $container->leakedToOuterScope); + $this->assertSame(['hook-cleanup'], $container->log); + } + + public function testContextEndsEvenWhenTheJobFails(): void + { + $this->requireExecutionContexts(); + $container = $this->installDeferAwareContainer(); + $ended = $this->recordEndedContexts($container); + + RecordingJob::$during = function () { + throw new \RuntimeException('boom'); + }; + + $this->manager->push(new RecordingJob('a')); + $this->runOne(); + + $this->assertFalse($container->hasExecutionContext()); + $this->assertCount(1, $ended); + } + public function testDeferredCallbacksRunWhenAJobFailsBeforeItIsReleased(): void { $container = $this->installDeferAwareContainer();