24458365411

Committed 05 Apr 2026 08:48AM UTC coverage: 53.938% (-2.2%) from 56.164%

Build # 24458365411

Build Type

push

github

Committed by

daycry

Commit Message

Optimize

Coverage Stats

50 of 192 new or added lines in 14 files covered. (26.04%)

3 existing lines in 3 files now uncovered.

1219 of 2260 relevant lines covered (53.94%)

4.42 hits per line

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

43.64

/src/Commands/QueueRunCommand.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\Commands;

use CodeIgniter\CLI\CLI;
use CodeIgniter\Exceptions\ExceptionInterface;
use Config\Services;
use DateTimeInterface;
use Daycry\Jobs\Exceptions\JobException;
use Daycry\Jobs\Execution\JobLifecycleCoordinator;
use Daycry\Jobs\Job;
use Daycry\Jobs\Libraries\CircuitBreaker;
use Daycry\Jobs\Libraries\QueueManager;
use Daycry\Jobs\Libraries\RateLimiter;
use Daycry\Jobs\Metrics\Metrics;
use Daycry\Jobs\Queues\JobEnvelope;
use Daycry\Jobs\Queues\RequeueHelper;

/**
 * Long-running (or one-shot) queue worker command.
 * Pulls messages from the configured queue backend, executes them via lifecycle coordinator,
 * and applies basic retry (requeue) fallback when failures occur.
 */
class QueueRunCommand extends BaseJobsCommand
{
    protected $name                       = 'jobs:queue:run';
    protected $description                = 'Start queue worker.';
    protected $usage                      = 'queue:run <queue> [Options]';
    protected $arguments                  = ['queue' => 'The queue name.'];
    protected $options                    = ['--oneTime' => 'Only executes one time.', '--background' => 'Run the worker in background.'];
    protected bool $locked                = false;
    private ?RequeueHelper $requeueHelper = null;
    private bool $shouldStop              = false;

    protected function earlyChecks(Job $job): void
    {
    }

    protected function lateChecks(Job $job): void
    {
    }

    protected function earlyCallbackChecks(Job $job): void
    {
    }

    protected function lateCallbackChecks(Job $job): void
    {
    }

    protected function conditionalChecks(): bool
    {
        return true;
    }

    public function run(array $params): void
    {
        $queue      = $params['queue'] ?? $params[0] ?? CLI::getOption('queue');
        $oneTime    = array_key_exists('oneTime', $params) ? true : CLI::getOption('oneTime');
        $background = array_key_exists('background', $params) ? true : CLI::getOption('background');

        // Spawn background child and exit parent if requested (avoid respawn with --noBackground)
        if ($background) {
            $phpBin    = $this->getPhpBinary();
            $sparkPath = ROOTPATH . 'spark';

            if (str_starts_with(strtolower(PHP_OS), strtolower('WIN'))) {
                // Windows: use start /B and redirect to NUL
                $cmd    = sprintf('%s %s %s', escapeshellarg($phpBin), escapeshellarg($sparkPath), $this->name . ' --queue ' . escapeshellarg($queue) . ($oneTime ? ' --oneTime' : ''));
                $winCmd = 'start "" /B ' . $cmd . ' > NUL 2>&1';
                pclose(popen($winCmd, 'r'));
            } else {
                // POSIX: use nohup and redirect to /dev/null, detach with &
                $cmd      = sprintf('%s %s %s', escapeshellarg($phpBin), escapeshellarg($sparkPath), $this->name . ' --queue ' . escapeshellarg($queue) . ($oneTime ? ' --oneTime' : ''));
                $posixCmd = 'nohup ' . $cmd . ' > /dev/null 2>&1 &';
                exec($posixCmd);
            }

            return;
        }

        if (empty($queue)) {
            $queue = CLI::prompt(lang('Queue.insertQueue'), config('Jobs')->queues, 'required');
        }

        // Register signal handlers for graceful shutdown (POSIX only)
        if (function_exists('pcntl_signal')) {
            pcntl_signal(SIGTERM, function (): void {
                $this->shouldStop = true;
                CLI::write('[Worker] SIGTERM received, finishing current job...', 'yellow');
            });
            pcntl_signal(SIGINT, function (): void {
                $this->shouldStop = true;
                CLI::write('[Worker] SIGINT received, finishing current job...', 'yellow');
            });
        }

        while (true) {
            if (function_exists('pcntl_signal_dispatch')) {
                pcntl_signal_dispatch();
            }

            if ($this->shouldStop) {
                CLI::write('[Worker] Graceful shutdown complete.', 'yellow');
                break;
            }

            if ($this->conditionalChecks()) {
                $this->processQueue($queue);

                if ($oneTime) {
                    return;
                }

                sleep(config('Jobs')->pollInterval);
            }
        }
    }

    protected function processQueue(string $queue): void
    {
        $response = [];
        $metrics  = Metrics::get();
        $this->requeueHelper ??= new RequeueHelper($metrics);

        // Rate limiting check
        $config    = config('Jobs');
        $rateLimit = $config->queueRateLimits[$queue] ?? 0;

        if ($rateLimit > 0) {
            $rateLimiter = new RateLimiter();
            if (! $rateLimiter->allow($queue, $rateLimit)) {
                // Rate limit exceeded, skip processing this cycle
                CLI::write("[Rate Limited] Queue '{$queue}' has reached limit of {$rateLimit} jobs/minute", 'yellow');

                return;
            }
        }

        // Circuit breaker check
        $breaker = new CircuitBreaker(
            'queue_' . $queue,
            $config->circuitBreakerThreshold,
            $config->circuitBreakerCooldown,
        );

        if (! $breaker->isAvailable()) {
            CLI::write("[Circuit Open] Backend for '{$queue}' is temporarily unavailable, skipping.", 'red');

            return;
        }

        Services::resetSingle('request');
        Services::resetSingle('response');

        try {
            $worker      = $this->getWorker();
            $queueEntity = $worker->watch($queue);
            $breaker->recordSuccess();

            if ($queueEntity === null) {
                // No available job for this queue at this time.
                return;
            }

            if ($queueEntity !== null) {
                $metrics->increment('jobs_fetched', 1, ['queue' => $queue]);
                $this->locked = true;
                if (! ($queueEntity instanceof JobEnvelope)) {
                    throw JobException::validationError('Legacy queue entity unsupported (expecting JobEnvelope).');
                }
                $decoded = $queueEntity->payload;
                if (! is_object($decoded)) {
                    throw JobException::validationError('Invalid envelope payload format.');
                }
                $job = Job::fromQueueRecord($decoded);
                // Inject backend ID into job instance for context availability
                $job->setJobId($queueEntity->id);

                $this->earlyChecks($job);

                $this->lateChecks($job); // todavía antes de ejecutar? mantener orden original

                $coordinator = new JobLifecycleCoordinator();
                $startExec   = microtime(true);
                $outcome     = $coordinator->run($job, 'queue');
                $latency     = microtime(true) - $startExec;
                $exec        = $outcome->finalResult;
                $response    = [
                    'status'     => $exec->success,
                    'statusCode' => $exec->success ? 200 : 500,
                    'data'       => $exec->output,
                    'error'      => $exec->success ? null : $exec->error,
                ];

                // Execution completed; outcome handled below.

                // Finalización: usar completion strategy ya ejecutada dentro del coordinator.
                // Remoción/requeue ya la maneja la estrategia QueueCompletionStrategy (si lo configuramos). Si aún no, aplicamos fallback:
                $this->requeueHelper->finalize($job, $queueEntity, static fn ($j, $r) => $worker->removeJob($j, $r), $exec->success);
                if ($queueEntity instanceof JobEnvelope && $queueEntity->createdAt instanceof DateTimeInterface) {
                    $age = microtime(true) - $queueEntity->createdAt->getTimestamp();
                    $metrics->observe('jobs_age_seconds', $age, ['queue' => $queue]);
                }
                $metrics->observe('jobs_exec_seconds', $latency, ['queue' => $queue]);
            }
        } catch (ExceptionInterface $e) {
            $breaker->recordFailure();
            $response = $this->handleException($e, $worker ?? null, $job ?? null);
        }

        $this->locked = false;
        unset($job, $queueEntity);
    }

    protected function getWorker()
    {
        return QueueManager::instance()->getDefault();
    }

    protected function handleException($e, $worker, $job): array
    {
        $response['statusCode'] = $e->getCode();
        $response['error']      = $e->getMessage();
        $response['status']     = false;

        if ($worker && $job) {
            $worker->removeJob($job, true);
        }

        $this->showError($e);

        return $response;
    }

    private function getPhpBinary(): string
    {
        if (PHP_SAPI === 'cli') {
            return PHP_BINARY;
        }

        return env('PHP_BINARY_PATH', 'php');
    }
}

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\Commands;
15
16	use CodeIgniter\CLI\CLI;
17	use CodeIgniter\Exceptions\ExceptionInterface;
18	use Config\Services;
19	use DateTimeInterface;
20	use Daycry\Jobs\Exceptions\JobException;
21	use Daycry\Jobs\Execution\JobLifecycleCoordinator;
22	use Daycry\Jobs\Job;
23	use Daycry\Jobs\Libraries\CircuitBreaker;
24	use Daycry\Jobs\Libraries\QueueManager;
25	use Daycry\Jobs\Libraries\RateLimiter;
26	use Daycry\Jobs\Metrics\Metrics;
27	use Daycry\Jobs\Queues\JobEnvelope;
28	use Daycry\Jobs\Queues\RequeueHelper;
29
30	/**
31	* Long-running (or one-shot) queue worker command.
32	* Pulls messages from the configured queue backend, executes them via lifecycle coordinator,
33	* and applies basic retry (requeue) fallback when failures occur.
34	*/
35	class QueueRunCommand extends BaseJobsCommand
36	{
37	protected $name = 'jobs:queue:run';
38	protected $description = 'Start queue worker.';
39	protected $usage = 'queue:run <queue> [Options]';
40	protected $arguments = ['queue' => 'The queue name.'];
41	protected $options = ['--oneTime' => 'Only executes one time.', '--background' => 'Run the worker in background.'];
42	protected bool $locked = false;
43	private ?RequeueHelper $requeueHelper = null;
44	private bool $shouldStop = false;
45
46	protected function earlyChecks(Job $job): void
47	{
48	}	1✔
49
50	protected function lateChecks(Job $job): void
51	{
52	}	1✔
53
54	protected function earlyCallbackChecks(Job $job): void
55	{
56	}	×
57
58	protected function lateCallbackChecks(Job $job): void
59	{
60	}	×
61
62	protected function conditionalChecks(): bool
63	{
64	return true;	×
65	}
66
67	public function run(array $params): void
68	{
69	$queue = $params['queue'] ?? $params[0] ?? CLI::getOption('queue');	×
70	$oneTime = array_key_exists('oneTime', $params) ? true : CLI::getOption('oneTime');	×
71	$background = array_key_exists('background', $params) ? true : CLI::getOption('background');	×
72
73	// Spawn background child and exit parent if requested (avoid respawn with --noBackground)
74	if ($background) {	×
75	$phpBin = $this->getPhpBinary();	×
76	$sparkPath = ROOTPATH . 'spark';	×
77
78	if (str_starts_with(strtolower(PHP_OS), strtolower('WIN'))) {	×
79	// Windows: use start /B and redirect to NUL
80	$cmd = sprintf('%s %s %s', escapeshellarg($phpBin), escapeshellarg($sparkPath), $this->name . ' --queue ' . escapeshellarg($queue) . ($oneTime ? ' --oneTime' : ''));	×
81	$winCmd = 'start "" /B ' . $cmd . ' > NUL 2>&1';	×
82	pclose(popen($winCmd, 'r'));	×
83	} else {
84	// POSIX: use nohup and redirect to /dev/null, detach with &
85	$cmd = sprintf('%s %s %s', escapeshellarg($phpBin), escapeshellarg($sparkPath), $this->name . ' --queue ' . escapeshellarg($queue) . ($oneTime ? ' --oneTime' : ''));	×
86	$posixCmd = 'nohup ' . $cmd . ' > /dev/null 2>&1 &';	×
87	exec($posixCmd);	×
88	}
89
90	return;	×
91	}
92
93	if (empty($queue)) {	×
94	$queue = CLI::prompt(lang('Queue.insertQueue'), config('Jobs')->queues, 'required');	×
95	}
96
97	// Register signal handlers for graceful shutdown (POSIX only)
NEW 98	if (function_exists('pcntl_signal')) {	×
NEW 99	pcntl_signal(SIGTERM, function (): void {	×
NEW 100	$this->shouldStop = true;	×
NEW 101	CLI::write('[Worker] SIGTERM received, finishing current job...', 'yellow');	×
NEW 102	});	×
NEW 103	pcntl_signal(SIGINT, function (): void {	×
NEW 104	$this->shouldStop = true;	×
NEW 105	CLI::write('[Worker] SIGINT received, finishing current job...', 'yellow');	×
NEW 106	});	×
107	}
108
109	while (true) {	×
NEW 110	if (function_exists('pcntl_signal_dispatch')) {	×
NEW 111	pcntl_signal_dispatch();	×
112	}
113
NEW 114	if ($this->shouldStop) {	×
NEW 115	CLI::write('[Worker] Graceful shutdown complete.', 'yellow');	×
NEW 116	break;	×
117	}
118
119	if ($this->conditionalChecks()) {	×
120	$this->processQueue($queue);	×
121
122	if ($oneTime) {	×
123	return;	×
124	}
125
NEW 126	sleep(config('Jobs')->pollInterval);	×
127	}
128	}
129	}
130
131	protected function processQueue(string $queue): void
132	{
133	$response = [];	1✔
134	$metrics = Metrics::get();	1✔
135	$this->requeueHelper ??= new RequeueHelper($metrics);	1✔
136
137	// Rate limiting check
138	$config = config('Jobs');	1✔
139	$rateLimit = $config->queueRateLimits[$queue] ?? 0;	1✔
140
141	if ($rateLimit > 0) {	1✔
142	$rateLimiter = new RateLimiter();	×
143	if (! $rateLimiter->allow($queue, $rateLimit)) {	×
144	// Rate limit exceeded, skip processing this cycle
145	CLI::write("[Rate Limited] Queue '{$queue}' has reached limit of {$rateLimit} jobs/minute", 'yellow');	×
146
147	return;	×
148	}
149	}
150
151	// Circuit breaker check
152	$breaker = new CircuitBreaker(	1✔
153	'queue_' . $queue,	1✔
154	$config->circuitBreakerThreshold,	1✔
155	$config->circuitBreakerCooldown,	1✔
156	);	1✔
157
158	if (! $breaker->isAvailable()) {	1✔
NEW 159	CLI::write("[Circuit Open] Backend for '{$queue}' is temporarily unavailable, skipping.", 'red');	×
160
NEW 161	return;	×
162	}
163
164	Services::resetSingle('request');	1✔
165	Services::resetSingle('response');	1✔
166
167	try {
168	$worker = $this->getWorker();	1✔
169	$queueEntity = $worker->watch($queue);	1✔
170	$breaker->recordSuccess();	1✔
171
172	if ($queueEntity === null) {	1✔
173	// No available job for this queue at this time.
174	return;	×
175	}
176
177	if ($queueEntity !== null) {	1✔
178	$metrics->increment('jobs_fetched', 1, ['queue' => $queue]);	1✔
179	$this->locked = true;	1✔
180	if (! ($queueEntity instanceof JobEnvelope)) {	1✔
181	throw JobException::validationError('Legacy queue entity unsupported (expecting JobEnvelope).');	×
182	}
183	$decoded = $queueEntity->payload;	1✔
184	if (! is_object($decoded)) {	1✔
185	throw JobException::validationError('Invalid envelope payload format.');	×
186	}
187	$job = Job::fromQueueRecord($decoded);	1✔
188	// Inject backend ID into job instance for context availability
189	$job->setJobId($queueEntity->id);	1✔
190
191	$this->earlyChecks($job);	1✔
192
193	$this->lateChecks($job); // todavía antes de ejecutar? mantener orden original	1✔
194
195	$coordinator = new JobLifecycleCoordinator();	1✔
196	$startExec = microtime(true);	1✔
197	$outcome = $coordinator->run($job, 'queue');	1✔
198	$latency = microtime(true) - $startExec;	1✔
199	$exec = $outcome->finalResult;	1✔
200	$response = [	1✔
201	'status' => $exec->success,	1✔
202	'statusCode' => $exec->success ? 200 : 500,	1✔
203	'data' => $exec->output,	1✔
204	'error' => $exec->success ? null : $exec->error,	1✔
205	];	1✔
206
207	// Execution completed; outcome handled below.
208
209	// Finalización: usar completion strategy ya ejecutada dentro del coordinator.
210	// Remoción/requeue ya la maneja la estrategia QueueCompletionStrategy (si lo configuramos). Si aún no, aplicamos fallback:
211	$this->requeueHelper->finalize($job, $queueEntity, static fn ($j, $r) => $worker->removeJob($j, $r), $exec->success);	1✔
212	if ($queueEntity instanceof JobEnvelope && $queueEntity->createdAt instanceof DateTimeInterface) {	1✔
213	$age = microtime(true) - $queueEntity->createdAt->getTimestamp();	1✔
214	$metrics->observe('jobs_age_seconds', $age, ['queue' => $queue]);	1✔
215	}
216	$metrics->observe('jobs_exec_seconds', $latency, ['queue' => $queue]);	1✔
217	}
218	} catch (ExceptionInterface $e) {	×
NEW 219	$breaker->recordFailure();	×
UNCOV 220	$response = $this->handleException($e, $worker ?? null, $job ?? null);	×
221	}
222
223	$this->locked = false;	1✔
224	unset($job, $queueEntity);	1✔
225	}
226
227	protected function getWorker()
228	{
229	return QueueManager::instance()->getDefault();	×
230	}
231
232	protected function handleException($e, $worker, $job): array
233	{
234	$response['statusCode'] = $e->getCode();	×
235	$response['error'] = $e->getMessage();	×
236	$response['status'] = false;	×
237
238	if ($worker && $job) {	×
239	$worker->removeJob($job, true);	×
240	}
241
242	$this->showError($e);	×
243
244	return $response;	×
245	}
246
247	private function getPhpBinary(): string
248	{
249	if (PHP_SAPI === 'cli') {	×
250	return PHP_BINARY;	×
251	}
252
253	return env('PHP_BINARY_PATH', 'php');	×
254	}
255	}

daycry / jobs / 24458365411

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