Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions src/QueueWorker.php
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
4 changes: 4 additions & 0 deletions tests/Support/claim_worker.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
4 changes: 4 additions & 0 deletions tests/Support/slow_job_worker.php
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
88 changes: 88 additions & 0 deletions tests/Unit/QueueWorkerTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
}
Loading