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

70.0

/src/Models/QueueModel.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\Models;

use CodeIgniter\Database\ConnectionInterface;
use CodeIgniter\Model;
use CodeIgniter\Validation\ValidationInterface;
use Config\Database;
use DateTime;
use DateTimeZone;
use Daycry\Jobs\Entities\Queue;

/**
 * Model for interacting with queued job records (insertion, fetching next pending job).
 */
class QueueModel extends Model
{
    protected $primaryKey     = 'id';
    protected $returnType     = Queue::class;
    protected $useSoftDeletes = false;
    protected $allowedFields  = [
        'identifier',
        'queue',
        'payload',
        'priority',
        'schedule',
        'status',
        'max_retries',
        'attempts',
    ];
    protected $useTimestamps = true;
    protected $createdField  = 'created_at';
    protected $updatedField  = 'updated_at';
    protected $deletedField  = 'deleted_at';

    public function __construct(?ConnectionInterface &$db = null, ?ValidationInterface $validation = null)
    {
        if ($db === null) {
            $db            = Database::connect(config('Jobs')->database['group']);
            $this->DBGroup = config('Jobs')->database['group'];
        }

        parent::__construct($db, $validation);
    }

    protected function initialize(): void
    {
        parent::initialize();

        $this->table = config('Jobs')->database['table'];
    }

    /**
     * Fetch next pending job ready for execution ordered by priority then schedule.
     */
    public function getJob(): ?Queue
    {
        $now = (new DateTime('now', new DateTimeZone(config('App')->appTimezone)))->format('Y-m-d H:i:s');

        return $this->where('status', 'pending')
            ->where('schedule <=', $now)
            ->orderBy('priority ASC, schedule ASC')
            ->first();
    }

    /**
     * Reserve a job from the queue safely (Atomic operation).
     */
    public function reserveJob(string $queue): ?Queue
    {
        $table       = $this->db->prefixTable($this->table);
        $attempts    = 0;
        $maxAttempts = 3;

        while ($attempts < $maxAttempts) {
            // 1. Find a candidate ID
            $now = (new DateTime('now', new DateTimeZone(config('App')->appTimezone)))->format('Y-m-d H:i:s');

            $sql = "SELECT id FROM {$table}
                    WHERE queue = ? AND status = 'pending' AND schedule <= ?
                    ORDER BY priority ASC, schedule ASC LIMIT 1";

            $query = $this->db->query($sql, [$queue, $now]);
            $row   = $query->getRow();

            if (! $row) {
                return null; // Queue empty
            }

            // 2. Try to lock it
            $updateSql = "UPDATE {$table}
                          SET status = 'in_progress', updated_at = ?
                          WHERE id = ? AND status = 'pending'";

            $this->db->query($updateSql, [$now, $row->id]);

            if ($this->db->affectedRows() > 0) {
                return $this->find($row->id);
            }

            // If we failed, someone else took it. Retry.
            $attempts++;
            usleep(10000); // 10ms wait
        }

        return null; // Could not lock any job after retries
    }
}

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\Models;
15
16	use CodeIgniter\Database\ConnectionInterface;
17	use CodeIgniter\Model;
18	use CodeIgniter\Validation\ValidationInterface;
19	use Config\Database;
20	use DateTime;
21	use DateTimeZone;
22	use Daycry\Jobs\Entities\Queue;
23
24	/**
25	* Model for interacting with queued job records (insertion, fetching next pending job).
26	*/
27	class QueueModel extends Model
28	{
29	protected $primaryKey = 'id';
30	protected $returnType = Queue::class;
31	protected $useSoftDeletes = false;
32	protected $allowedFields = [
33	'identifier',
34	'queue',
35	'payload',
36	'priority',
37	'schedule',
38	'status',
39	'max_retries',
40	'attempts',
41	];
42	protected $useTimestamps = true;
43	protected $createdField = 'created_at';
44	protected $updatedField = 'updated_at';
45	protected $deletedField = 'deleted_at';
46
47	public function __construct(?ConnectionInterface &$db = null, ?ValidationInterface $validation = null)
48	{
49	if ($db === null) {	6✔
50	$db = Database::connect(config('Jobs')->database['group']);	6✔
51	$this->DBGroup = config('Jobs')->database['group'];	6✔
52	}
53
54	parent::__construct($db, $validation);	6✔
55	}
56
57	protected function initialize(): void
58	{
59	parent::initialize();	6✔
60
61	$this->table = config('Jobs')->database['table'];	6✔
62	}
63
64	/**
65	* Fetch next pending job ready for execution ordered by priority then schedule.
66	*/
67	public function getJob(): ?Queue
68	{
NEW 69	$now = (new DateTime('now', new DateTimeZone(config('App')->appTimezone)))->format('Y-m-d H:i:s');	×
70
NEW 71	return $this->where('status', 'pending')	×
NEW 72	->where('schedule <=', $now)	×
NEW 73	->orderBy('priority ASC, schedule ASC')	×
NEW 74	->first();	×
75	}
76
77	/**
78	* Reserve a job from the queue safely (Atomic operation).
79	*/
80	public function reserveJob(string $queue): ?Queue
81	{
82	$table = $this->db->prefixTable($this->table);	3✔
83	$attempts = 0;	3✔
84	$maxAttempts = 3;	3✔
85
86	while ($attempts < $maxAttempts) {	3✔
87	// 1. Find a candidate ID
88	$now = (new DateTime('now', new DateTimeZone(config('App')->appTimezone)))->format('Y-m-d H:i:s');	3✔
89
90	$sql = "SELECT id FROM {$table}	3✔
91	WHERE queue = ? AND status = 'pending' AND schedule <= ?
92	ORDER BY priority ASC, schedule ASC LIMIT 1";	3✔
93
94	$query = $this->db->query($sql, [$queue, $now]);	3✔
95	$row = $query->getRow();	3✔
96
97	if (! $row) {	3✔
98	return null; // Queue empty	×
99	}
100
101	// 2. Try to lock it
102	$updateSql = "UPDATE {$table}	3✔
103	SET status = 'in_progress', updated_at = ?
104	WHERE id = ? AND status = 'pending'";	3✔
105
106	$this->db->query($updateSql, [$now, $row->id]);	3✔
107
108	if ($this->db->affectedRows() > 0) {	3✔
109	return $this->find($row->id);	3✔
110	}
111
112	// If we failed, someone else took it. Retry.
113	$attempts++;	×
114	usleep(10000); // 10ms wait	×
115	}
116
117	return null; // Could not lock any job after retries	×
118	}
119	}

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