Merge pull request #32 from techmahedy/techmahedy-4.x

(feat) Worker: run every job in its own `defer()` scope

Author: techmahedy

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

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;

tests/Support/slow_job_worker.php

@@ -7,6 +7,10 @@
  * Usage: php slow_job_worker.php <base64 json spec>
  */
 
+// 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;

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<int, string>}
+     */
+    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());
+    }
 }