21442227581

Committed 28 Jan 2026 02:30PM UTC coverage: 56.301% (-3.1%) from 59.413%

Build # 21442227581

Build Type

push

github

Committed by

daycry

Commit Message

Improve queue handling and job execution logic

Refactor queue worker to support background execution and improve job fetching logic. Update JobLifecycleCoordinator to prioritize job-specific and default timeouts without a global cap. Replace custom UUID generation with service-based UUID v7 in JobLogger. Ensure queue scheduling uses application timezone for consistency.

Run Details

13 of 24 new or added lines in 5 files covered. (54.17%)

63 existing lines in 4 files now uncovered.

1175 of 2087 relevant lines covered (56.3%)

4.26 hits per line

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

55.56

/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[0] ?? CLI::getOption('queue');
        $oneTime    = array_key_exists('oneTime', $params)    || CLI::getOption('oneTime');
        $background = array_key_exists('background', $params) || CLI::getOption('background');

        if (empty($queue)) {
            $queue = CLI::prompt(lang('Queue.insertQueue'));
        }

        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 prepareResponse($result): array
    {
        $response['status'] = true;

        if (! $result instanceof Response) {
            $result = (Services::response(null, true))->setStatusCode(200)->setBody($result);
        }

        $response['statusCode'] = $result->getStatusCode();
        $response['data']       = $result->getBody();

        return $response;
    }*/

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

    protected function finalizeJob(array $response, $worker, Job $job): void
    {
        // Ya no se usa: la lógica de finalize se hace inline tras ejecutar el JobExecutor.
    }

    // Metrics retrieval now centralized in Metrics::get(); no local logic needed.
}

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	{
NEW 67	$queue = $params[0] ?? CLI::getOption('queue');	×
NEW 68	$oneTime = array_key_exists('oneTime', $params) \|\| CLI::getOption('oneTime');	×
NEW 69	$background = array_key_exists('background', $params) \|\| CLI::getOption('background');	×
70
71	if (empty($queue)) {	×
72	$queue = CLI::prompt(lang('Queue.insertQueue'));	×
73	}
74
75	while (true) {	×
76	if ($this->conditionalChecks()) {	×
77	$this->processQueue($queue);	×
78
79	if ($oneTime) {	×
80	return;	×
81	}
82
83	sleep(config('Jobs')->defaultTimeout ?? 5);	×
84	}
85	}
86	}
87
88	protected function processQueue(string $queue): void
89	{
90	$response = [];	1✔
91	$metrics = Metrics::get();	1✔
92	$this->requeueHelper ??= new RequeueHelper($metrics);	1✔
93
94	// Rate limiting check
95	$config = config('Jobs');	1✔
96	$rateLimit = $config->queueRateLimits[$queue] ?? 0;	1✔
97
98	if ($rateLimit > 0) {	1✔
99	$rateLimiter = new RateLimiter();	×
100	if (! $rateLimiter->allow($queue, $rateLimit)) {	×
101	// Rate limit exceeded, skip processing this cycle
102	CLI::write("[Rate Limited] Queue '{$queue}' has reached limit of {$rateLimit} jobs/minute", 'yellow');	×
103
104	return;	×
105	}
106	}
107
108	Services::resetSingle('request');	1✔
109	Services::resetSingle('response');	1✔
110
111	try {
112	$worker = $this->getWorker();	1✔
113	$queueEntity = $worker->watch($queue);	1✔
114
115	if ($queueEntity === null) {	1✔
116	// No available job for this queue at this time.
NEW 117	return;	×
118	}
119
120	if ($queueEntity !== null) {	1✔
121	$metrics->increment('jobs_fetched', 1, ['queue' => $queue]);	1✔
122	$this->locked = true;	1✔
123	if (! ($queueEntity instanceof JobEnvelope)) {	1✔
124	throw JobException::validationError('Legacy queue entity unsupported (expecting JobEnvelope).');	×
125	}
126	$decoded = $queueEntity->payload;	1✔
127	if (! is_object($decoded)) {	1✔
128	throw JobException::validationError('Invalid envelope payload format.');	×
129	}
130	$job = Job::fromQueueRecord($decoded);	1✔
131
132	$this->earlyChecks($job);	1✔
133
134	$this->lateChecks($job); // todavía antes de ejecutar? mantener orden original	1✔
135
136	$coordinator = new JobLifecycleCoordinator();	1✔
137	$startExec = microtime(true);	1✔
138	$outcome = $coordinator->run($job, 'queue');	1✔
139	$latency = microtime(true) - $startExec;	1✔
140	$exec = $outcome->finalResult;	1✔
141	$response = [	1✔
142	'status' => $exec->success,	1✔
143	'statusCode' => $exec->success ? 200 : 500,	1✔
144	'data' => $exec->output,	1✔
145	'error' => $exec->success ? null : $exec->error,	1✔
146	];	1✔
147
148	// Execution completed; outcome handled below.
149
150	// Finalización: usar completion strategy ya ejecutada dentro del coordinator.
151	// Remoción/requeue ya la maneja la estrategia QueueCompletionStrategy (si lo configuramos). Si aún no, aplicamos fallback:
152	$this->requeueHelper->finalize($job, $queueEntity, static fn ($j, $r) => $worker->removeJob($j, $r), $exec->success);	1✔
153	if ($queueEntity instanceof JobEnvelope && $queueEntity->createdAt instanceof DateTimeInterface) {	1✔
154	$age = microtime(true) - $queueEntity->createdAt->getTimestamp();	1✔
155	$metrics->observe('jobs_age_seconds', $age, ['queue' => $queue]);	1✔
156	}
157	$metrics->observe('jobs_exec_seconds', $latency, ['queue' => $queue]);	1✔
158	}
159	} catch (ExceptionInterface $e) {	×
160	$response = $this->handleException($e, $worker ?? null, $job ?? null);	×
161	}
162
163	$this->locked = false;	1✔
164	unset($job, $queueEntity);	1✔
165	}
166
167	protected function getWorker()
168	{
169	return QueueManager::instance()->getDefault();	×
170	}
171
172	/*protected function prepareResponse($result): array
173	{
174	$response['status'] = true;
175
176	if (! $result instanceof Response) {
177	$result = (Services::response(null, true))->setStatusCode(200)->setBody($result);
178	}
179
180	$response['statusCode'] = $result->getStatusCode();
181	$response['data'] = $result->getBody();
182
183	return $response;
184	}*/
185
186	protected function handleException($e, $worker, $job): array
187	{
188	$response['statusCode'] = $e->getCode();	×
189	$response['error'] = $e->getMessage();	×
190	$response['status'] = false;	×
191
192	if ($worker && $job) {	×
193	$worker->removeJob($job, true);	×
194	}
195
196	$this->showError($e);	×
197
198	return $response;	×
199	}
200
201	protected function finalizeJob(array $response, $worker, Job $job): void
202	{
203	// Ya no se usa: la lógica de finalize se hace inline tras ejecutar el JobExecutor.
204	}	×
205
206	// Metrics retrieval now centralized in Metrics::get(); no local logic needed.
207	}

daycry / jobs / 21442227581

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