20573383244

Committed 19 Nov 2025 03:37PM UTC coverage: 60.218% (-2.3%) from 62.5%

Build # 20573383244

Build Type

push

github

Committed by

daycry

Commit Message

- Improvements

Run Details

22 of 25 new or added lines in 12 files covered. (88.0%)

46 existing lines in 3 files now uncovered.

1105 of 1835 relevant lines covered (60.22%)

4.63 hits per line

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

67.5

/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 CodeIgniter\HTTP\Response;
use Config\Services;
use DateTimeInterface;
use Daycry\Jobs\Exceptions\JobException;
use Daycry\Jobs\Execution\ExecutionContext;
use Daycry\Jobs\Execution\JobLifecycleCoordinator;
use Daycry\Jobs\Job;
use Daycry\Jobs\Libraries\QueueManager;
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.'];
    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');

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

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

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

            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

                $ctx = new ExecutionContext(
                    source: 'queue',
                    maxRetries: $job->getMaxRetries() ?? 0,
                    notifyOnSuccess: $job->shouldNotifyOnSuccess(),
                    notifyOnFailure: $job->shouldNotifyOnFailure(),
                    singleInstance: $job->isSingleInstance(),
                    queueName: $queue,
                    queueWorker: $worker,
                    retryConfig: [
                        'strategy'   => config('Jobs')->retryBackoffStrategy,
                        'base'       => config('Jobs')->retryBackoffBase,
                        'multiplier' => config('Jobs')->retryBackoffMultiplier,
                        'jitter'     => config('Jobs')->retryBackoffJitter,
                        'max'        => config('Jobs')->retryBackoffMax,
                    ],
                    eventsEnabled: config('Jobs')->enableEvents ?? true,
                    meta: [],
                );

                $coordinator = new JobLifecycleCoordinator();
                $startExec   = microtime(true);
                $outcome     = $coordinator->run($job, $ctx);
                $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,
                ];

                // 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 CodeIgniter\HTTP\Response;
19	use Config\Services;
20	use DateTimeInterface;
21	use Daycry\Jobs\Exceptions\JobException;
22	use Daycry\Jobs\Execution\ExecutionContext;
23	use Daycry\Jobs\Execution\JobLifecycleCoordinator;
24	use Daycry\Jobs\Job;
25	use Daycry\Jobs\Libraries\QueueManager;
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.'];
42	protected bool $locked = false;
43	private ?RequeueHelper $requeueHelper = null;
44
45	protected function earlyChecks(Job $job): void
46	{
47	}	1✔
48
49	protected function lateChecks(Job $job): void
50	{
51	}	1✔
52
53	protected function earlyCallbackChecks(Job $job): void
54	{
55	}	×
56
57	protected function lateCallbackChecks(Job $job): void
58	{
59	}	×
60
61	protected function conditionalChecks(): bool
62	{
63	return true;	×
64	}
65
66	public function run(array $params): void
67	{
68	$queue = $params[0] ?? CLI::getOption('queue');	×
69	$oneTime = array_key_exists('oneTime', $params) \|\| CLI::getOption('oneTime');	×
70
71	if (empty($queue)) {	×
72	$queue = CLI::prompt(lang('Queue.insertQueue'));	×
73	}
74
75	while (true) {	×
NEW 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	Services::resetSingle('request');	1✔
95	Services::resetSingle('response');	1✔
96
97	try {
98	$worker = $this->getWorker();	1✔
99	$queueEntity = $worker->watch($queue);	1✔
100
101	if ($queueEntity !== null) {	1✔
102	$metrics->increment('jobs_fetched', 1, ['queue' => $queue]);	1✔
103	$this->locked = true;	1✔
104	if (! ($queueEntity instanceof JobEnvelope)) {	1✔
105	throw JobException::validationError('Legacy queue entity unsupported (expecting JobEnvelope).');	×
106	}
107	$decoded = $queueEntity->payload;	1✔
108	if (! is_object($decoded)) {	1✔
109	throw JobException::validationError('Invalid envelope payload format.');	×
110	}
111	$job = Job::fromQueueRecord($decoded);	1✔
112
113	$this->earlyChecks($job);	1✔
114
115	$this->lateChecks($job); // todavía antes de ejecutar? mantener orden original	1✔
116
117	$ctx = new ExecutionContext(	1✔
118	source: 'queue',	1✔
119	maxRetries: $job->getMaxRetries() ?? 0,	1✔
120	notifyOnSuccess: $job->shouldNotifyOnSuccess(),	1✔
121	notifyOnFailure: $job->shouldNotifyOnFailure(),	1✔
122	singleInstance: $job->isSingleInstance(),	1✔
123	queueName: $queue,	1✔
124	queueWorker: $worker,	1✔
125	retryConfig: [	1✔
126	'strategy' => config('Jobs')->retryBackoffStrategy,	1✔
127	'base' => config('Jobs')->retryBackoffBase,	1✔
128	'multiplier' => config('Jobs')->retryBackoffMultiplier,	1✔
129	'jitter' => config('Jobs')->retryBackoffJitter,	1✔
130	'max' => config('Jobs')->retryBackoffMax,	1✔
131	],	1✔
132	eventsEnabled: config('Jobs')->enableEvents ?? true,	1✔
133	meta: [],	1✔
134	);	1✔
135
136	$coordinator = new JobLifecycleCoordinator();	1✔
137	$startExec = microtime(true);	1✔
138	$outcome = $coordinator->run($job, $ctx);	1✔
139	$latency = microtime(true) - $startExec;	1✔
140	$exec = $outcome->finalResult;	1✔
141
142	$response = [	1✔
143	'status' => $exec->success,	1✔
144	'statusCode' => $exec->success ? 200 : 500,	1✔
145	'data' => $exec->output,	1✔
146	'error' => $exec->success ? null : $exec->error,	1✔
147	];	1✔
148
149	// Finalización: usar completion strategy ya ejecutada dentro del coordinator.
150	// Remoción/requeue ya la maneja la estrategia QueueCompletionStrategy (si lo configuramos). Si aún no, aplicamos fallback:
151	$this->requeueHelper->finalize($job, $queueEntity, static fn ($j, $r) => $worker->removeJob($j, $r), $exec->success);	1✔
152	if ($queueEntity instanceof JobEnvelope && $queueEntity->createdAt instanceof DateTimeInterface) {	1✔
153	$age = microtime(true) - $queueEntity->createdAt->getTimestamp();	1✔
154	$metrics->observe('jobs_age_seconds', $age, ['queue' => $queue]);	1✔
155	}
156	$metrics->observe('jobs_exec_seconds', $latency, ['queue' => $queue]);	1✔
157	}
158	} catch (ExceptionInterface $e) {	×
159	$response = $this->handleException($e, $worker ?? null, $job ?? null);	×
160	}
161
162	$this->locked = false;	1✔
163	unset($job, $queueEntity);	1✔
164	}
165
166	protected function getWorker()
167	{
168	return QueueManager::instance()->getDefault();	×
169	}
170
171	/*protected function prepareResponse($result): array
172	{
173	$response['status'] = true;
174
175	if (! $result instanceof Response) {
176	$result = (Services::response(null, true))->setStatusCode(200)->setBody($result);
177	}
178
179	$response['statusCode'] = $result->getStatusCode();
180	$response['data'] = $result->getBody();
181
182	return $response;
183	}*/
184
185	protected function handleException($e, $worker, $job): array
186	{
187	$response['statusCode'] = $e->getCode();	×
188	$response['error'] = $e->getMessage();	×
189	$response['status'] = false;	×
190
191	if ($worker && $job) {	×
192	$worker->removeJob($job, true);	×
193	}
194
195	$this->showError($e);	×
196
197	return $response;	×
198	}
199
200	protected function finalizeJob(array $response, $worker, Job $job): void
201	{
202	// Ya no se usa: la lógica de finalize se hace inline tras ejecutar el JobExecutor.
203	}	×
204
205	// Metrics retrieval now centralized in Metrics::get(); no local logic needed.
206	}

daycry / jobs / 20573383244

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