21479393233

Committed 28 Jan 2026 05:28PM UTC coverage: 53.81% (-2.2%) from 55.964%

Build # 21479393233

Build Type

push

github

Committed by

daycry

Commit Message

Add background worker support to jobs:queue:run

Implemented cross-platform background execution for the jobs:queue:run command using detached processes (nohup/& for POSIX, start /B for Windows). Updated documentation to explain usage, PHP binary selection, and troubleshooting. Refactored code to select the PHP binary appropriately and improved process spawning logic.

Run Details

0 of 12 new or added lines in 1 file covered. (0.0%)

44 existing lines in 2 files now uncovered.

1130 of 2100 relevant lines covered (53.81%)

4.17 hits per line

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

47.06

/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\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;

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

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

                if ($oneTime) {
                    return;
                }

                sleep(config('Jobs')->defaultTimeout ?? 5);
            }
        }
    }

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

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

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

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

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

daycry / jobs / 21479393233

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