Queues

• Introduction
• Requirements
• Configuration
• Registering the service
• Using the queue in a service
• Queue operations
• Tracking failed attempts
• Consistency guarantees
• Queue logging
  • Overriding the logger

Introduction

Sukarix\Queue\QueueService is a Redis-backed FIFO queue for passing work between the stages of a pipeline — typically CLI workers scheduled by TaskScheduler.

It provides three things on top of a plain Redis list:

• FIFO delivery — items are pushed onto the head and popped from the tail.
• Deduplication — a companion Redis set tracks membership, so pushing a payload that is already queued is a no-op instead of a duplicate.
• Attempt tracking — a per-queue hash counts failed processing attempts per item, so a stage can decide when an item has exhausted its retries.

Note

Retry policy stays in your application

The queue counts attempts but never decides what to do with them. Whether a failing item is retried, parked or dead-lettered is a business decision and belongs in your own stage code.

Requirements

The queue requires the redis PHP extension and a reachable Redis server.

Configuration

Connection settings are read from the hive:

[globals]
redis.host = 127.0.0.1
redis.port = 6379
; connection timeout in seconds, keep it bounded so workers fail fast
redis.timeout = 2

The constructor throws \RedisException when the connection cannot be established, so a misconfigured worker fails immediately instead of part-way through a run.

Registering the service

Register the queue once during boot so it can be resolved through the Injector:

protected function setupQueueService(): void
{
    \Registry::set('queue', \Sukarix\Queue\QueueService::instance());
}

Using the queue in a service

Add the HasQueue behaviour to get a $queueService property. See Behaviours for how init<TraitName> methods are invoked.

use Sukarix\Behaviours\HasQueue;
use Sukarix\Core\Processor;

class MyWorker
{
    use HasQueue;

    public function __construct()
    {
        Processor::instance()->initialize($this);
    }

    public function run(): void
    {
        while (null !== $item = $this->queueService->popFromQueue('queue.work.pending')) {
            // process $item
        }
    }
}

Warning

Register before constructing

HasQueue throws a \LogicException if no queue service is registered when the consuming class is constructed. This is deliberate: it surfaces the misconfiguration at boot instead of failing later with a null member access.

Queue operations

Payloads are JSON-encoded, so any JSON-serialisable value works — scalars, arrays and nested structures all survive the round trip:

$queue = \Sukarix\Queue\QueueService::instance();

$queue->pushToQueue('queue.work.pending', ['order_id' => 'A-1', 'total' => 42]);
$queue->pushToQueue('queue.work.pending', ['order_id' => 'A-1', 'total' => 42]); // skipped

$queue->getQueueSize('queue.work.pending'); // 1
$item = $queue->popFromQueue('queue.work.pending'); // ['order_id' => 'A-1', 'total' => 42]

Note

getQueueItems is a snapshot

getQueueItems() is an observational read for reporting and administration. Concurrent producers and consumers may change the queue before you act on the result, so never treat it as a lock.

Tracking failed attempts

Attempt counters are keyed by the payload, independently of the queue contents:

try {
    $this->process($item);
    // Success clears the history so a future failure starts from a clean count
    $this->queueService->clearAttempts('queue.work.failed', $item);
} catch (\Throwable $e) {
    $attempts = $this->queueService->incrementAttempts('queue.work.failed', $item);

    if ($attempts >= $this->f3->get('pipeline.max_retries')) {
        // Your policy: park it, alert, or leave it for manual analysis
        $this->logger->critical('Item exhausted its retries', ['attempts' => $attempts]);
    }
}

getAttempts() returns 0 for an item that has never failed, so a fresh item is always below any positive limit.

Consistency guarantees

Enqueue and dequeue are executed as Redis Lua scripts, which Redis runs atomically. A worker killed mid-operation can therefore never leave the list and the membership set disagreeing — an item is either fully queued and tracked, or neither.

Warning

clearQueue is an administrative operation

clearQueue() removes three keys and is not atomic with respect to a running pipeline. Do not call it while producers or consumers are active.

Queue logging

Sukarix\Queue\QueueProgressLogger emits structured queue events with a distinctive prefix and a context array, so the output is parsable by log aggregators rather than only human-readable.

$logger = \Sukarix\Queue\QueueProgressLogger::instance();

$logger->logQueueStart('queue.work.pending', 'PROCESS', ['max_items' => 100]);
$logger->logQueueProcess('queue.work.pending', 40, 2, 58, 100);
$logger->logQueueComplete('queue.work.pending', 'PROCESS', 98, 2);

Available methods: logQueueStart(), logQueueComplete(), logQueueStatus(), logQueueFill(), logQueueProcess(), logBatchProgress(), logQueueThroughput(), logItemAdded(), logItemRemoved(), logQueueWarning() and logQueueError().

The queue service logs enqueue and dequeue events through this logger. Item previews are deliberately generic — an array is reported as array[3], never with its values — so payload contents never leak into logs.

Overriding the logger

Register your own subclass under the queue.progress_logger alias to route queue telemetry to a different sink:

\Registry::set('queue.progress_logger', new MetricsQueueProgressLogger());

The queue validates the alias and throws \UnexpectedValueException if the registered service does not extend QueueProgressLogger. When the alias is absent, the framework default is used.