Queue

Reference for the Queue module: register queues and process their items. Two queue types are supported. Lock-based queues, registered with QueueConfig::AddQueue, are drained in-process using row locking. Task-queues, registered with QueueTaskConfig::AddQueue, lease work to external, anonymous-but-authenticated workers and reconcile a lease mirror against the application's own pending set.

This page is a pure reference. For how the framework fits together, see the Guide.

Classes

QueueConfig - register lock-based queues for processing.
QueueConfigQueueData - the registration record for one lock-based queue.
QueueDatabase - count, list and process lock-based queue items.
QueueInfo - convert queue info to and from JSON.
QueueInfoData - a snapshot of one queue's config and item count.
QueueRateLimitError - raised when a queue's rate limit is exceeded.
QueueResultDatabase - record the outcome of processing a queue item.
QueueTaskConfig - register task-queues for external workers.
QueueTaskConfigData - the registration record for one task-queue.
QueueTaskDatabase - the task-queue engine: claim, progress, complete, fail and reconcile.
QueueTaskStats - log per-queue health counts and measures for task-queues.

QueueConfig

Register a lock-based queue for processing and set how often the queue list is re-polled. Registering a queue only records it; processing is exposed as API endpoints by the separate QueueApi module. Each queue supplies two callbacks: one that locks and returns a batch of item ids, and one that processes a single item within a transaction.

Properties

Property	Type	Description
`QueueConfig::$listPollSeconds`	`int`	How often the driver should re-poll the queue list, in seconds; defaults to 30.
`QueueConfig::$queues`	`array`	The registered queues, keyed by 'moduleName_queueName'.

Functions

AddQueue SetListPollSeconds

AddQueue

QueueConfig::AddQueue(string $moduleName, string $queueName, int $maxProcessesPerSecond, int $maxProcessesInFlight, callable $lockNextCallback, callable $processItemCallback, int $idleRecheckSeconds = 5): void

Name	Type	Description
`$moduleName`	`string`	Module name, used for error logging.
`$queueName`	`string`	The queue name identifier.
`$maxProcessesPerSecond`	`int`	Maximum number of processes per second.
`$maxProcessesInFlight`	`int`	Maximum number of processes in flight.
`$lockNextCallback`	`callable`	Locks and returns a batch of item ids; takes an int limit, returns an array of ints.
`$processItemCallback`	`callable`	Processes the item with the given id, within the transaction.
`$idleRecheckSeconds`	`int`	When drained and idle, how long before re-probing; defaults to 5.

Returns void

SetListPollSeconds

QueueConfig::SetListPollSeconds(int $seconds): void

Override how often the driver re-polls the queue list.

Name	Type	Description
`$seconds`	`int`	The new list poll interval, in seconds.

Returns void

QueueConfigQueueData

The registration record for one lock-based queue, created by QueueConfig::AddQueue. Its fields are readonly. This class has no functions beyond its constructor.

Property	Type	Description
`$queue->moduleName`	`string`	Module name, used for error logging.
`$queue->queueName`	`string`	The queue name identifier.
`$queue->maxProcessesPerSecond`	`int`	Maximum number of processes per second.
`$queue->maxProcessesInFlight`	`int`	Maximum number of processes in flight.
`$queue->idleRecheckSeconds`	`int`	When drained and idle, how long before re-probing.
`$queue->lockNextCallback`	`mixed`	The callback that locks and returns a batch of item ids.
`$queue->processItemCallback`	`mixed`	The callback that processes a single item.

QueueDatabase

Count, list and process the items of registered lock-based queues. Counting and processing both go through the queue's own lock callback inside a transaction; processing captures console output and records a result on success or failure.

Functions

CountQueueItems CountResults GetConfig GetQueueOptional GetQueues ListQueues ProcessQueueItem

CountQueueItems

QueueDatabase::CountQueueItems(QueueConfigQueueData $queue): int

Count the items in a queue using its lock callback, capped at 100 for performance.

Name	Type	Description
`$queue`	`QueueConfigQueueData`	The queue configuration.

Returns int The number of items, capped at 100.

CountResults

QueueDatabase::CountResults(?object $where): int

Count result rows, all rows when the where object is null or empty, otherwise filtered.

Name	Type	Description
`$where`	`?object`	Fields to filter by, or null to count all rows.

Returns int The number of matching result rows.

GetConfig

QueueDatabase::GetConfig(string $moduleName, string $queueName): QueueConfigQueueData

Get a registered queue's configuration by module name and queue name. Throws when not found.

Name	Type	Description
`$moduleName`	`string`	The module name.
`$queueName`	`string`	The queue name.

Returns QueueConfigQueueData The queue configuration.

GetQueueOptional

QueueDatabase::GetQueueOptional(string $moduleName, string $queueName): ?QueueConfigQueueData

Get a registered queue by module name and queue name, or null when there is no match.

Name	Type	Description
`$moduleName`	`string`	The module name.
`$queueName`	`string`	The queue name.

Returns ?QueueConfigQueueData The queue, or null.

GetQueues

QueueDatabase::GetQueues(): array

All registered lock-based queues.

Returns array A list of QueueConfigQueueData objects.

ListQueues

QueueDatabase::ListQueues(): array

List every registered queue with its process interval and item count calculated. Queues that error are included with fallback values.

Returns array A list of QueueInfoData objects.

ProcessQueueItem

QueueDatabase::ProcessQueueItem(QueueConfigQueueData $queue): void

Lock and process the next item in a queue within a transaction, recording a success or failure result. Returns early when there is nothing to do; throws and rolls back when processing fails.

Name	Type	Description
`$queue`	`QueueConfigQueueData`	The queue configuration.

Returns void

QueueInfo

Convert queue information between JSON and QueueInfoData objects, for one entry or a list.

Functions

FromJson ListFromJson ListToJson

FromJson

QueueInfo::FromJson(string $json): QueueInfoData

Parse a single QueueInfoData object from a JSON string.

Name	Type	Description
`$json`	`string`	The JSON string to parse.

Returns QueueInfoData The parsed queue info.

ListFromJson

QueueInfo::ListFromJson(string $json): array

Parse an array of QueueInfoData objects from a JSON string, calling FromJson on each entry.

Name	Type	Description
`$json`	`string`	The JSON string to parse.

Returns array A list of QueueInfoData objects.

ListToJson

QueueInfo::ListToJson(array $queueInfoDataList): array

Convert an array of QueueInfoData objects to stdClass objects ready for JSON encoding.

Name	Type	Description
`$queueInfoDataList`	`array`	The QueueInfoData objects to convert.

Returns array A list of stdClass objects.

QueueInfoData

A snapshot of one queue's configuration and current item count, used when listing queues and moving that information over JSON. Its fields are readonly.

Property	Type	Description
`$info->moduleName`	`string`	The owning module name.
`$info->queueName`	`string`	The queue name.
`$info->processIntervalMs`	`int`	The interval between processes, in milliseconds.
`$info->count`	`int`	The item count, capped at 100.
`$info->maxInFlight`	`int`	Maximum number of processes in flight; defaults to 10.
`$info->maxRatePerSecond`	`int`	Maximum number of processes per second; defaults to 1.
`$info->idleRecheckSeconds`	`int`	When drained and idle, how long before re-probing; defaults to 5.
`$info->countFetchTimeMs`	`float`	How long the count fetch took, in milliseconds; defaults to 0.0.

Functions

ToJson

QueueInfoData::ToJson(): stdClass

Convert this queue info to a stdClass object ready for JSON encoding.

Returns stdClass The queue info as an object.

QueueRateLimitError

Raised when a queue's rate limit is exceeded. Extends Exception. It adds no members of its own.

QueueResultDatabase

Record the outcome of processing a lock-based queue item, whether it succeeded or failed.

Functions

Add

QueueResultDatabase::Add(string $moduleName, string $queueName, int $rowId, bool $success, string $description, ?string $consoleOutput): void

Insert a queue result entry recording a success or error for one processed row.

Name	Type	Description
`$moduleName`	`string`	The owning module name.
`$queueName`	`string`	The queue name.
`$rowId`	`int`	The id of the processed row.
`$success`	`bool`	Whether processing succeeded.
`$description`	`string`	A description of the outcome.
`$consoleOutput`	`?string`	Raw captured console output, or null.

Returns void

QueueTaskConfig

Register task-queues, the second queue type, and look them up by key. A task-queue is processed by external, anonymous-but-authenticated workers that pull tasks, report progress and return a result; the application supplies the callbacks. The queue key is 'Module.queueName'.

Properties

Property	Type	Description
`QueueTaskConfig::$queues`	`array`	The registered task-queues, keyed by 'Module.queueName'.

Functions

AddQueue All Get GetOptional

AddQueue

QueueTaskConfig::AddQueue(string $moduleName, string $queueName, int $reassignSeconds, ?int $overallTimeoutSeconds, callable $candidateCallback, callable $payloadCallback, callable $applyResultCallback, callable $applyFailureCallback, ?callable $applyProgressCallback = null): void

Name	Type	Description
`$moduleName`	`string`	The owning application module.
`$queueName`	`string`	The queue name.
`$reassignSeconds`	`int`	The no-contact window before a new worker may be assigned.
`$overallTimeoutSeconds`	`?int`	Seconds from entry into the queue to permanent failure, or null for never.
`$candidateCallback`	`callable`	Returns pending record ids; takes a limit, does no locking.
`$payloadCallback`	`callable`	Returns the data a worker needs for a record.
`$applyResultCallback`	`callable`	Idempotent result write; returns true only when this call set it.
`$applyFailureCallback`	`callable`	Writes the permanent-failure sentinel for a record.
`$applyProgressCallback`	`?callable`	Optional; persists domain-relevant progress bits.

Returns void

All

QueueTaskConfig::All(): array

All registered task-queues.

Returns array A list of QueueTaskConfigData objects.

Get

QueueTaskConfig::Get(string $key): QueueTaskConfigData

Get a registered task-queue by its key. Throws when the key is unknown.

Name	Type	Description
`$key`	`string`	The queue key, 'Module.queueName'.

Returns QueueTaskConfigData The task-queue registration.

GetOptional

QueueTaskConfig::GetOptional(string $key): ?QueueTaskConfigData

Get a registered task-queue by its key, or null when the key is unknown.

Name	Type	Description
`$key`	`string`	The queue key, 'Module.queueName'.

Returns ?QueueTaskConfigData The task-queue registration, or null.

QueueTaskConfigData

The registration record for one task-queue, created by QueueTaskConfig::AddQueue. Its fields are readonly.

Property	Type	Description
`$queue->moduleName`	`string`	The owning application module.
`$queue->queueName`	`string`	The queue name.
`$queue->reassignSeconds`	`int`	The rolling no-contact window before a new worker may be assigned.
`$queue->overallTimeoutSeconds`	`?int`	Seconds from entry into the queue to permanent failure, or null for never.
`$queue->candidateCallback`	`mixed`	The callback returning pending record ids.
`$queue->payloadCallback`	`mixed`	The callback returning a worker's payload for a record.
`$queue->applyResultCallback`	`mixed`	The idempotent result-write callback.
`$queue->applyFailureCallback`	`mixed`	The permanent-failure callback.
`$queue->applyProgressCallback`	`mixed`	The optional progress callback, or null.

Functions

Key

QueueTaskConfigData::Key(): string

The queue key used everywhere, formed as 'Module.queueName'.

Returns string The queue key.

QueueTaskDatabase

The task-queue engine. Workers claim, heartbeat, complete and fail tasks; the scheduler runs a maintenance pass that reconciles the lease mirror against the application's pending set and samples gauges; admin helpers list and count raw task rows. Every operation is a single autocommit statement or a short independent sequence, so no transaction is held.

Constants

Constant	Value	Description
`QueueTaskDatabase::CANDIDATE_LIMIT`	`10000`	Upper bound on candidate ids pulled per Process pass.

Functions

Claim Complete CountTasks DistinctQueueKeysAssoc Fail ListTasks Process Progress Stats

Claim

QueueTaskDatabase::Claim(QueueTaskConfigData $queue): ?object

Atomically lease the next claimable task to a new worker reference.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.

Returns ?object An object with workerRef, recordId and leaseSeconds, or null when nothing is claimable.

Complete

QueueTaskDatabase::Complete(QueueTaskConfigData $queue, int $recordId, string $workerRef, object $result): object

Record a terminal success. Not ref-gated: any worker's result is accepted and the idempotent result callback makes the first writer win. Removes the task.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.
`$recordId`	`int`	The application record id.
`$workerRef`	`string`	The reporting worker's reference.
`$result`	`object`	The result to apply.

Returns object An object with accepted (always true) and first (whether this call set the result).

CountTasks

QueueTaskDatabase::CountTasks(string $queueKey): int

Count task rows, all rows when the key is empty, otherwise filtered to one queue key.

Name	Type	Description
`$queueKey`	`string`	The queue key to filter by, or empty for all.

Returns int The number of task rows.

DistinctQueueKeysAssoc

QueueTaskDatabase::DistinctQueueKeysAssoc(): array

The distinct queue keys currently present in the task table, as a value-to-value map for filtering.

Returns array A map of queue key to queue key.

Fail

QueueTaskDatabase::Fail(QueueTaskConfigData $queue, int $recordId, string $workerRef, bool $permanent, string $message): object

Record a terminal failure. A permanent failure is ref-gated and writes the sentinel; a transient failure releases the lease early for immediate re-claim.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.
`$recordId`	`int`	The application record id.
`$workerRef`	`string`	The reporting worker's reference.
`$permanent`	`bool`	Whether the failure is permanent.
`$message`	`string`	The failure message, for logging.

Returns object An object with accepted, true only when the current holder's failure counted.

ListTasks

QueueTaskDatabase::ListTasks(string $queueKey, int $page, int $limit): array

A page of raw task rows, optionally filtered by queue key, newest first.

Name	Type	Description
`$queueKey`	`string`	The queue key to filter by, or empty for all.
`$page`	`int`	The page number, from 1.
`$limit`	`int`	The number of rows per page.

Returns array A list of raw task row objects.

Process

QueueTaskDatabase::Process(QueueTaskConfigData $queue): object

Run one maintenance pass: fail tasks past their overall timeout, reconcile the mirror against the application's candidate list, then sample and log gauges.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.

Returns object A summary with queue, added, removed, failed, pending, inFlight and oldestPendingSeconds.

Progress

QueueTaskDatabase::Progress(QueueTaskConfigData $queue, int $recordId, string $workerRef, ?object $progress): string

Heartbeat, progress and cancellation in one call. Extends the lease and records progress for the current holder; a superseded worker is told to continue but has no authority.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.
`$recordId`	`int`	The application record id.
`$workerRef`	`string`	The reporting worker's reference.
`$progress`	`?object`	Progress bits to record, or null.

Returns string 'continue' while the task still exists, or 'stop' once it is gone.

Stats

QueueTaskDatabase::Stats(QueueTaskConfigData $queue): object

Live counts for a task-queue, read straight from the task table with no logging.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.

Returns object An object with queue, moduleName, queueName, pending, inFlight and oldestPendingSeconds.

QueueTaskStats

Log per-queue health statistics for task-queues via LogCounter. Counts are recorded as summed values; measures record a duration or gauge. Stats must never break queue operations, so all failures are swallowed.

Constants

Constant	Value	Description
`QueueTaskStats::PREFIX`	`'Queue.task.'`	The prefix for every logged metric identifier.

Functions

Count Measure

Count

QueueTaskStats::Count(QueueTaskConfigData $queue, string $event, float $n = 1.0): void

Log an event count for a queue. Does nothing when the count is not positive.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.
`$event`	`string`	The event name, appended to the prefix.
`$n`	`float`	The count to add; defaults to 1.0.

Returns void

Measure

QueueTaskStats::Measure(QueueTaskConfigData $queue, string $metric, float $value): void

Log a measured value for a queue, such as a duration in seconds or a sampled gauge.

Name	Type	Description
`$queue`	`QueueTaskConfigData`	The task-queue registration.
`$metric`	`string`	The metric name, appended to the prefix.
`$value`	`float`	The measured value.

Returns void