Download the PHP package nails/module-queue without Composer

Queue Module for Nails

A lightweight, database-backed job queue for the Nails framework. It provides a simple API to enqueue tasks, a worker to process them, retry/backoff handling, and maintenance commands.

• Storage: MySQL tables managed by migrations
• Concurrency: safe reservation using SELECT ... FOR UPDATE SKIP LOCKED
• Ordering: strict FIFO per queue (lowest id first among available_at <= NOW())
• Retries: per-task maximum with exponential backoff + jitter
• Extensibility: plug-in queues via an interface with optional setup/refresh hooks

Installation

Install via composer:

Setup required tables using migrations as normal:

Tables created:

• nails_queue_worker: active workers and their heartbeats
• nails_queue_job: queued jobs and their lifecycle

High-level Overview

Core concepts:

• Queue: a logical stream of jobs. Implemented as a class that can run setup and refresh hooks. There are built-in queues DefaultQueue and PriorityQueue.
• Job: a unit of work defined by a Task class and a Data payload. Jobs transition through PENDING → RUNNING → COMPLETE | FAILED.
• Worker: a long-running process which polls one or more queues and executes jobs. Each worker registers itself and sends periodic heartbeats.

Flow:

1. Application enqueues a job: Manager->push(Task, Data, Queue|alias|class, availableAt)
2. Worker (nails queue:work) reserves a job atomically
3. Worker runs the Task->run(Data) method
4. On success → COMPLETE; on exception → either FAILED or scheduled for retry depending on Task::getMaxRetries()
5. Maintenance (nails queue:clean) resets stuck jobs, removes stale workers, and rotates old job rows

Configuration

Set via configuration class (e.g. environment variables, Config::get()). Defaults are shown in parentheses.

• QUEUE_WORKER_WAIT_TIME (500): Initial wait time in milliseconds between polls when no work is found. Doubles up to 5000ms; a small random jitter is added each sleep.
• QUEUE_WORKER_REFRESH_INTERVAL (300): Seconds between invoking Queue::refresh() for each active queue (approx; called in the idle loop). This is effectively "~5 minutes" when jobs are running back-to-back, as the loop only checks between jobs; when idle, it can be called more frequently.
• QUEUE_WORKER_HEARTBEAT_STALE (300): Seconds after which a worker without heartbeat is considered stale and eligible for deletion by queue:clean.
• QUEUE_JOB_ROTATE_COMPLETE_DAYS (7): Days to retain COMPLETE jobs. Set 0 to disable deletion.
• QUEUE_JOB_ROTATE_FAILED_DAYS (30): Days to retain FAILED jobs. Set 0 to disable deletion.

Interfaces

Implement these in user-land to add new queues and jobs.

Queue

Built-in queues:

• Nails\Queue\Queue\Queues\DefaultQueue
• Nails\Queue\Queue\Queues\PriorityQueue

These are registered under the aliases default and priority respectively.

Task

Retry behaviour:

• On exception, if attempts < getMaxRetries(), the job is rescheduled to PENDING with an available_at delay computed by exponential backoff (+/- 20% jitter):
  • base 5s, factor 2, capped at 5 minutes
• When attempts exceed max, the job is marked FAILED and error is recorded.

Data

The module provides a simple factory to rehydrate Data from JSON when jobs are loaded.

Manager Service API (selected)

Notes:

• Passing null or 'default' as the queue argument uses the default queue alias; you can also pass a fully-qualified queue class name (FQCN) or an instance.
• FIFO is approximated by selecting the smallest available id where available_at <= NOW().
• unworkedQueues() is the list shown by the admin dashboard alert. A worker counts when its heartbeat is newer than QUEUE_WORKER_HEARTBEAT_STALE (300 seconds by default) and it was started for that queue class. isProcessed() is true when a live worker lists the queue. For a discovered queue, that is the opposite of appearing in unworkedQueues().

Creating user-land queues and jobs

1) Define a Queue (optional)

If you need custom startup or periodic maintenance, implement your own queue class and register an alias at bootstrap.

You can place your own Queues wherever you like, but for auto-discovery you should place them in the App\Queue\Queues namespace.

Optionally, register an alias:

2) Define a Task

3) Define a Data class

The supplied Data factory is usually sufficient for defining the payload for each job:

However, you are free to use your own implementation if you wish. In the following example we ensure that our payload is in the right shape and also validate that our headers match our rows:

4) Enqueue a job

Running the worker

Start a worker to process jobs from one or more queues.

Basic usage (current dir is the web root):

Options:

• --queue=<QueueFQCN-or-alias>: Process a specific queue. Repeat to listen to multiple queues. Defaults to the default queue if not provided.

Examples:

What the worker does:

• Registers itself and prints its token/ID
• Calls Queue::setup() once per queue at startup
• Event loop:
  • Attempts to reserve the next job (PENDING, available_at <= NOW()) using SKIP LOCKED
  • On success: marks RUNNING, executes Task->run(Data), then marks COMPLETE or schedules retry / marks FAILED
  • On idle: sleeps with exponential backoff (ms) and jitter up to 5000ms, then polls again
  • Periodically calls Queue::refresh() per queue (approx every QUEUE_WORKER_REFRESH_INTERVAL seconds when idle)
  • Updates its heartbeat and flushes DB cache each iteration

Graceful shutdown:

• Where possible, the worker unregisters itself on destruct. When a process is terminated the worker may be left registered. This is cleaned up using the queue:clean command, detailed below.

Maintenance command

Clean up workers and jobs:

Performs:

• Delete stale workers whose heartbeat is older than QUEUE_WORKER_HEARTBEAT_STALE seconds
• Reset stuck jobs: RUNNING with no worker_id and no finished back to PENDING
• Rotate old jobs:
  • Remove COMPLETE older than QUEUE_JOB_ROTATE_COMPLETE_DAYS
  • Remove FAILED older than QUEUE_JOB_ROTATE_FAILED_DAYS

This command auto registers itself into cron and runs every 5 minutes.

Daemonising the worker (supervisord example)

Keep the queue proccess running in the background (with auto-restart) using a process manager. Common examples:

systemd

/etc/systemd/system/nails-queue.service

Commands:

supervisord

/etc/supervisor/conf.d/queue-worker.conf:

Commands:

Tips:

• Run one program block per queue, or a single process handling multiple queues with repeated --queue flags
• Ensure your application’s environment is loaded for the worker process (env vars, config, DB access)