26886467550

Committed 03 Jun 2026 01:01PM UTC coverage: 88.948% (+14.0%) from 74.974%

Build # 26886467550

Build Type

push

github

Committed by

web-flow

Commit Message

v3.0: single clean architecture (remove V1, lease-based queues, secure-by-default)

Complete v3.0 rewrite into a single, clean architecture. The v1 API and the V2\ scaffolding
are removed (no facade, no dual code); the package passes PHPStan level 6 + strict-rules +
codeigniter with NO baseline.

- Definition: Jobs::define()->...->dispatch() fluent builder -> immutable JobDefinition.
- Handlers decoupled from the god-object (JobHandlerInterface / AbstractJobHandler / TypedJobHandler + JobContext).
- One QueueBackend contract (enqueue/fetch(lease)/ack/nack(delay)/abandon/reapExpired) with 5 backends:
  Sync, Database, Redis, Beanstalk, ServiceBus.
- Runtime: one attempt per fetch; real interrupting Timeout; opt-in idempotency; single-instance lock.
- Worker/Cron: jobs:queue:work, jobs:queue:reap, jobs:cronjob:run, jobs:queue:purge.
- Secure-by-default: HMAC-signed envelopes, per-queue handler allowlist, ShellHandler deny-by-default,
  EventHandler allowlist, UrlHandler anti-SSRF.

Resolves audit findings #1,#2,#3,#4,#5,#6,#7,#8,#10,#12,#13,#17,#18,#19,#20,#22.
Tests: 359 (Beanstalk live); line coverage 88.9%; PHPStan/Psalm/Rector/cs green on PHP 8.2-8.5.

BREAKING CHANGE: v1 API removed. See docs/MIGRATION-v1-to-v3.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Coverage Stats

983 of 1103 new or added lines in 43 files covered. (89.12%)

15 existing lines in 3 files now uncovered.

1497 of 1683 relevant lines covered (88.95%)

7.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.91

/src/Execution/JobRuntime.php

<?php

declare(strict_types=1);

/**
 * This file is part of Daycry Queues.
 *
 * (c) Daycry <daycry9@proton.me>
 *
 * For the full copyright and license information, please view
 * the LICENSE file that was distributed with this source code.
 */

namespace Daycry\Jobs\Execution;

use Daycry\Jobs\Config\Jobs;
use Daycry\Jobs\Definition\JobDefinition;
use Daycry\Jobs\Handlers\HandlerRegistry;
use Daycry\Jobs\Handlers\JobHandlerInterface;
use Throwable;

/**
 * Executes a job ONE time and returns an {@see ExecutionResult}.
 *
 * This is the v3 replacement for {@see JobLifecycleCoordinator}. Crucially it does NOT
 * loop or sleep for retries: a single attempt runs here, and the worker decides whether to
 * requeue with backoff (via QueueBackend::nack) — which removes the legacy double-retry
 * (coordinator loop + RequeueHelper) that consumed maxRetries twice and blocked the worker.
 *
 * Responsibilities: resolve the handler (with per-queue allowlist), capture output, apply a
 * real timeout that interrupts the work, and turn any Throwable into a failed result.
 */
final readonly class JobRuntime
{
    private Jobs $config;
    private Timeout $timeout;
    private HandlerRegistry $registry;
    private SingleInstanceLock $lock;

    public function __construct(
        ?Timeout $timeout = null,
        ?HandlerRegistry $registry = null,
        ?Jobs $config = null,
        ?SingleInstanceLock $lock = null,
    ) {
        $this->config   = $config ?? config('Jobs');
        $this->timeout  = $timeout ?? new Timeout();
        $this->registry = $registry ?? new HandlerRegistry($this->config);
        $this->lock     = $lock ?? new SingleInstanceLock();
    }

    public function run(JobDefinition $definition, JobContext $context): ExecutionResult
    {
        $start = microtime(true);

        try {
            $handler = $this->registry->resolveForQueue($definition->handler, $context->queue ?? 'default');
        } catch (Throwable $e) {
            return new ExecutionResult(false, null, $e->getMessage(), $start, microtime(true));
        }

        $timeoutSeconds = $this->resolveTimeout($definition);
        $handlerClass   = $handler::class;
        $bufferActive   = false;

        // Single-instance guard: prevent concurrent runs of the same named job. The lock
        // carries an ownership token (only this run releases it). Contention surfaces as a
        // failed result, so the worker requeues it to run once the holder finishes.
        $lockName  = $context->name ?? $definition->name ?? $definition->handler;
        $lockOwner = '';
        $locked    = false;
        if ($definition->singleInstance) {
            $lockOwner = bin2hex(random_bytes(16));
            if (! $this->lock->acquire($lockName, $lockOwner, max(120, $timeoutSeconds + 60))) {
                return new ExecutionResult(false, null, "single-instance job '{$lockName}' is already running", $start, microtime(true), $handlerClass);
            }
            $locked = true;
        }

        try {
            $handler->beforeRun($context);

            ob_start();
            $bufferActive = true;

            $returned = $this->timeout->run(
                $timeoutSeconds,
                static fn (): mixed => $handler->handle($context),
                $context->name ?? $definition->handler,
            );

            $buffer       = ob_get_clean();
            $bufferActive = false;

            $result = new ExecutionResult(
                success: true,
                output: $this->normalizeOutput($returned, $buffer === false ? null : $buffer),
                error: null,
                startedAt: $start,
                endedAt: microtime(true),
                handlerClass: $handlerClass,
            );

            $this->safeAfterRun($handler, $context, $result);

            return $result;
        } catch (Throwable $e) {
            if ($bufferActive && ob_get_level() > 0) {
                ob_end_clean();
            }

            $result = new ExecutionResult(false, null, $e->getMessage(), $start, microtime(true), $handlerClass);
            $this->safeAfterRun($handler, $context, $result);

            return $result;
        } finally {
            if ($locked) {
                $this->lock->release($lockName, $lockOwner);
            }
        }
    }

    private function resolveTimeout(JobDefinition $definition): int
    {
        if ($definition->timeout !== null) {
            return max(0, $definition->timeout);
        }

        $default = $this->config->defaultTimeout;

        return $default !== null ? max(0, $default) : 0;
    }

    private function safeAfterRun(JobHandlerInterface $handler, JobContext $context, ExecutionResult $result): void
    {
        try {
            $handler->afterRun($context, $result);
        } catch (Throwable) {
            // afterRun is a best-effort hook; never let it change the recorded outcome.
        }
    }

    private function normalizeOutput(mixed $returned, ?string $buffer): ?string
    {
        $data = $returned;

        if ($buffer !== null && $buffer !== '') {
            if ($data === null) {
                $data = $buffer;
            } elseif (is_string($data)) {
                $separator = str_starts_with($buffer, "\n") ? '' : "\n";
                $data .= $separator . $buffer;
            }
        }

        if ($data === null) {
            return null;
        }
        if (is_scalar($data)) {
            return (string) $data;
        }

        $encoded = json_encode($data);

        return $encoded === false ? null : $encoded;
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	/**
6	* This file is part of Daycry Queues.
7	*
8	* (c) Daycry <daycry9@proton.me>
9	*
10	* For the full copyright and license information, please view
11	* the LICENSE file that was distributed with this source code.
12	*/
13
14	namespace Daycry\Jobs\Execution;
15
16	use Daycry\Jobs\Config\Jobs;
17	use Daycry\Jobs\Definition\JobDefinition;
18	use Daycry\Jobs\Handlers\HandlerRegistry;
19	use Daycry\Jobs\Handlers\JobHandlerInterface;
20	use Throwable;
21
22	/**
23	* Executes a job ONE time and returns an {@see ExecutionResult}.
24	*
25	* This is the v3 replacement for {@see JobLifecycleCoordinator}. Crucially it does NOT
26	* loop or sleep for retries: a single attempt runs here, and the worker decides whether to
27	* requeue with backoff (via QueueBackend::nack) — which removes the legacy double-retry
28	* (coordinator loop + RequeueHelper) that consumed maxRetries twice and blocked the worker.
29	*
30	* Responsibilities: resolve the handler (with per-queue allowlist), capture output, apply a
31	* real timeout that interrupts the work, and turn any Throwable into a failed result.
32	*/
33	final readonly class JobRuntime
34	{
35	private Jobs $config;
36	private Timeout $timeout;
37	private HandlerRegistry $registry;
38	private SingleInstanceLock $lock;
39
40	public function __construct(
41	?Timeout $timeout = null,
42	?HandlerRegistry $registry = null,
43	?Jobs $config = null,
44	?SingleInstanceLock $lock = null,
45	) {
46	$this->config = $config ?? config('Jobs');	35✔
47	$this->timeout = $timeout ?? new Timeout();	35✔
48	$this->registry = $registry ?? new HandlerRegistry($this->config);	35✔
49	$this->lock = $lock ?? new SingleInstanceLock();	35✔
50	}
51
52	public function run(JobDefinition $definition, JobContext $context): ExecutionResult
53	{
54	$start = microtime(true);	28✔
55
56	try {
57	$handler = $this->registry->resolveForQueue($definition->handler, $context->queue ?? 'default');	28✔
58	} catch (Throwable $e) {	2✔
59	return new ExecutionResult(false, null, $e->getMessage(), $start, microtime(true));	2✔
60	}
61
62	$timeoutSeconds = $this->resolveTimeout($definition);	26✔
63	$handlerClass = $handler::class;	26✔
64	$bufferActive = false;	26✔
65
66	// Single-instance guard: prevent concurrent runs of the same named job. The lock
67	// carries an ownership token (only this run releases it). Contention surfaces as a
68	// failed result, so the worker requeues it to run once the holder finishes.
69	$lockName = $context->name ?? $definition->name ?? $definition->handler;	26✔
70	$lockOwner = '';	26✔
71	$locked = false;	26✔
72	if ($definition->singleInstance) {	26✔
73	$lockOwner = bin2hex(random_bytes(16));	2✔
74	if (! $this->lock->acquire($lockName, $lockOwner, max(120, $timeoutSeconds + 60))) {	2✔
75	return new ExecutionResult(false, null, "single-instance job '{$lockName}' is already running", $start, microtime(true), $handlerClass);	1✔
76	}
77	$locked = true;	1✔
78	}
79
80	try {
81	$handler->beforeRun($context);	25✔
82
83	ob_start();	25✔
84	$bufferActive = true;	25✔
85
86	$returned = $this->timeout->run(	25✔
87	$timeoutSeconds,	25✔
88	static fn (): mixed => $handler->handle($context),	25✔
89	$context->name ?? $definition->handler,	25✔
90	);	25✔
91
92	$buffer = ob_get_clean();	18✔
93	$bufferActive = false;	18✔
94
95	$result = new ExecutionResult(	18✔
96	success: true,	18✔
97	output: $this->normalizeOutput($returned, $buffer === false ? null : $buffer),	18✔
98	error: null,	18✔
99	startedAt: $start,	18✔
100	endedAt: microtime(true),	18✔
101	handlerClass: $handlerClass,	18✔
102	);	18✔
103
104	$this->safeAfterRun($handler, $context, $result);	18✔
105
106	return $result;	18✔
107	} catch (Throwable $e) {	7✔
108	if ($bufferActive && ob_get_level() > 0) {	7✔
109	ob_end_clean();	7✔
110	}
111
112	$result = new ExecutionResult(false, null, $e->getMessage(), $start, microtime(true), $handlerClass);	7✔
113	$this->safeAfterRun($handler, $context, $result);	7✔
114
115	return $result;	7✔
116	} finally {
117	if ($locked) {	25✔
118	$this->lock->release($lockName, $lockOwner);	25✔
119	}
120	}
121	}
122
123	private function resolveTimeout(JobDefinition $definition): int
124	{
125	if ($definition->timeout !== null) {	26✔
126	return max(0, $definition->timeout);	1✔
127	}
128
129	$default = $this->config->defaultTimeout;	25✔
130
131	return $default !== null ? max(0, $default) : 0;	25✔
132	}
133
134	private function safeAfterRun(JobHandlerInterface $handler, JobContext $context, ExecutionResult $result): void
135	{
136	try {
137	$handler->afterRun($context, $result);	25✔
NEW 138	} catch (Throwable) {	×
139	// afterRun is a best-effort hook; never let it change the recorded outcome.
140	}
141	}
142
143	private function normalizeOutput(mixed $returned, ?string $buffer): ?string
144	{
145	$data = $returned;	18✔
146
147	if ($buffer !== null && $buffer !== '') {	18✔
148	if ($data === null) {	1✔
149	$data = $buffer;	1✔
NEW 150	} elseif (is_string($data)) {	×
NEW 151	$separator = str_starts_with($buffer, "\n") ? '' : "\n";	×
NEW 152	$data .= $separator . $buffer;	×
153	}
154	}
155
156	if ($data === null) {	18✔
157	return null;	4✔
158	}
159	if (is_scalar($data)) {	14✔
160	return (string) $data;	14✔
161	}
162
NEW 163	$encoded = json_encode($data);	×
164
NEW 165	return $encoded === false ? null : $encoded;	×
166	}
167	}

daycry / jobs / 26886467550

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous