21138471384

Committed 19 Jan 2026 12:16PM UTC coverage: 62.662% (+0.1%) from 62.561%

Build # 21138471384

Build Type

push

github

Committed by

daycry

Commit Message

Improve queue atomicity and performance, add index

Introduces atomic job reservation in QueueModel via reserveJob to prevent race conditions in DatabaseQueue. Adds a composite index (status, schedule, priority) for efficient job fetching. Updates ShellJob to sanitize command input. Documentation updated to reflect security and performance improvements.

Run Details

20 of 25 new or added lines in 4 files covered. (80.0%)

2 existing lines in 2 files now uncovered.

1163 of 1856 relevant lines covered (62.66%)

4.74 hits per line

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

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

daycry / jobs / 21138471384

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