From cb509b80ce02aa8eefd4d87da7fee51f17b109d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20de=20la=20Pen=CC=83a?= Date: Mon, 2 Nov 2026 20:50:08 +0000 Subject: [PATCH] docs: Describe the queue module Adds a Queue section covering tasks and payloads, queues, running workers, the admin overview, and keeping workers alive with systemd or Supervisor. Notes that the priority queue needs its own worker process. Links to it from the Find & Replace queue section. Co-Authored-By: Claude Sonnet 5.5 --- SUMMARY.md | 6 + modules/find-replace.md | 2 +- modules/queue.md | 85 ++++++++++++ modules/queue/admin.md | 87 ++++++++++++ modules/queue/process-managers.md | 213 ++++++++++++++++++++++++++++++ modules/queue/queues.md | 134 +++++++++++++++++++ modules/queue/tasks.md | 176 ++++++++++++++++++++++++ modules/queue/workers.md | 113 ++++++++++++++++ 8 files changed, 815 insertions(+), 1 deletion(-) create mode 100644 modules/queue.md create mode 100644 modules/queue/admin.md create mode 100644 modules/queue/process-managers.md create mode 100644 modules/queue/queues.md create mode 100644 modules/queue/tasks.md create mode 100644 modules/queue/workers.md diff --git a/SUMMARY.md b/SUMMARY.md index 3b0651b..93507c6 100644 --- a/SUMMARY.md +++ b/SUMMARY.md @@ -153,6 +153,12 @@ * [Drivers](modules/multi-factor-auth/drivers/README.md) * [Email](modules/multi-factor-auth/drivers/email.md) * [Authenticator](modules/multi-factor-auth/drivers/authenticator.md) +* [Queue](modules/queue.md) + * [Tasks](modules/queue/tasks.md) + * [Queues](modules/queue/queues.md) + * [Workers](modules/queue/workers.md) + * [Process Managers](modules/queue/process-managers.md) + * [Admin](modules/queue/admin.md) * [Subscription](modules/subscription.md) * [Webhooks](modules/webhooks.md) * [Handlers](modules/webhooks/handlers.md) diff --git a/modules/find-replace.md b/modules/find-replace.md index 1c2e934..cd706be 100644 --- a/modules/find-replace.md +++ b/modules/find-replace.md @@ -64,7 +64,7 @@ A revert writes the previous value only when the live cell is still exactly the Admin searches run as a queue job. The queue is `default` unless `FIND_REPLACE_QUEUE` names another alias or queue class, such as `priority` or `App\Queue\Queues\Reports`. A blank value also uses `default`. -`nails queue:work` listens to `default`. Pass `--queue` when the search uses another queue. +`nails queue:work` listens to `default`. Pass `--queue` when the search uses another queue. See [Queue](queue.md) for [running workers](queue/workers.md) and keeping them alive with a [process manager](queue/process-managers.md). The Find & Replace index shows a warning when that queue has no live worker. A worker is live when its heartbeat is within `QUEUE_WORKER_HEARTBEAT_STALE` seconds (300 by default) and it was started for that queue. This is the same check as the queue dashboard alert. A stale worker, or a worker listening to a different queue, leaves the warning in place. The form can still be submitted; the search stays pending until a worker picks it up. If `FIND_REPLACE_QUEUE` does not name a queue, the index says so and the search is not saved. diff --git a/modules/queue.md b/modules/queue.md new file mode 100644 index 0000000..48ae72a --- /dev/null +++ b/modules/queue.md @@ -0,0 +1,85 @@ +--- +description: >- + A database-backed job queue. Push work from a request, run it in a + long-lived worker, and retry it when it fails. +--- + +# Queue + +The queue module moves slow or fragile work out of the request. The app pushes a job, a worker process picks it up and runs it, and a failed job is retried with a growing delay. Jobs live in MySQL, so there is no broker to install. + +```bash +composer require nails/module-queue +nails db:migrate +``` + +The module needs `nails/module-console` for the worker commands and [`nails/module-cron`](cron.md) to schedule its cleanup. Reserving a job uses `SELECT … FOR UPDATE SKIP LOCKED`, so the database must be MySQL 8.0 or newer, or MariaDB 10.6 or newer. + +The migration creates two tables, prefixed with `NAILS_DB_PREFIX` as usual: + +| Table | Holds | +|---|---| +| `nails_queue_job` | Every job, from the moment it is pushed until it is rotated out | +| `nails_queue_worker` | One row per running worker, with its last heartbeat | + +Both tables are hidden from [Find & Replace](find-replace/exclusions.md). + +## How it fits together + +| Piece | What it is | +|---|---| +| [Task](queue/tasks.md) | A class with a `run()` method. It is the work. | +| Data | The JSON payload stored with a job and handed to `run()`. | +| [Queue](queue/queues.md) | A named stream of jobs. A worker listens to one or more queues. | +| [Worker](queue/workers.md) | The `nails queue:work` process. It runs one job at a time. | +| [Manager](queue/tasks.md#pushing-a-job) | The service the app uses to push jobs. Workers use it to reserve them. | + +A job moves through four statuses: + +| Status | Meaning | +|---|---| +| `PENDING` | Waiting for a worker. A job with a future `available_at` is _scheduled_ and is still `PENDING`. | +| `RUNNING` | A worker has reserved it and is running the task. | +| `COMPLETE` | The task returned without throwing. | +| `FAILED` | The task threw and has no retries left. The errors are kept on the job. | + +A failed attempt that still has retries left goes back to `PENDING` with a later `available_at`. + +1. The app calls `push()` with a task, a payload, and optionally a queue and a start time. +2. A [worker](queue/workers.md) reserves the oldest job that is `PENDING` and due, in the queues it was started for. +3. The worker calls `run()` on the task with the payload. +4. The job becomes `COMPLETE`, is rescheduled for a retry, or becomes `FAILED`. +5. [`queue:clean`](queue/workers.md#maintenance), run by cron, removes dead workers, recovers stuck jobs, and deletes old rows. + +{% hint style="warning" %} +Nothing runs a job until a worker is running. Jobs sit in `PENDING` and the admin dashboard shows a warning. Keep a worker alive with a [process manager](queue/process-managers.md). Each [queue](queue/queues.md) a worker should serve is named when it starts, and the built-in `priority` queue needs a [worker of its own](queue/queues.md#priority). +{% endhint %} + +## Pages + +* [Tasks](queue/tasks.md): write a task, push a job, and how retries work +* [Queues](queue/queues.md): the built-in queues, your own queues, and aliases +* [Workers](queue/workers.md): run workers, read their log, stop them, and the maintenance command +* [Process managers](queue/process-managers.md): keep workers running with systemd or Supervisor +* [Admin](queue/admin.md): the overview screen, the dashboard alert, and the API behind them + +## Configuration + +Set these as [config values](../getting-started/configuration.md), usually environment variables. All are optional. + +| Config | Default | Effect | +|---|---|---| +| `QUEUE_WORKER_WAIT_TIME` | `500` | Milliseconds an idle worker first sleeps between polls. The wait doubles on each empty poll, up to 5,000 ms, and starts again after a job. A random 0 to 200 ms is added each time. | +| `QUEUE_WORKER_REFRESH_INTERVAL` | `300` | Seconds between calls to each queue's `refresh()` hook. | +| `QUEUE_WORKER_HEARTBEAT_DEBOUNCE` | `5` | Minimum seconds between heartbeat writes from one worker. | +| `QUEUE_WORKER_HEARTBEAT_STALE` | `300` | Seconds without a heartbeat before a worker counts as dead. | +| `QUEUE_JOB_ROTATE_COMPLETE_DAYS` | `7` | Days to keep `COMPLETE` jobs. `0` keeps them forever. | +| `QUEUE_JOB_ROTATE_FAILED_DAYS` | `30` | Days to keep `FAILED` jobs. `0` keeps them forever. | + +{% hint style="danger" %} +A worker records its heartbeat between jobs, not during one. If a single job can run for longer than `QUEUE_WORKER_HEARTBEAT_STALE`, the worker looks dead, [`queue:clean`](queue/workers.md#maintenance) deletes it, and the job is put back in `PENDING` while it is still running. Another worker then runs it a second time. Set `QUEUE_WORKER_HEARTBEAT_STALE` higher than your longest job. +{% endhint %} + +## Overriding + +The Manager, both models, both resources, and the `Data` factory can each be replaced by defining a class of the same name under `App\Queue`: `App\Queue\Service\Manager`, `App\Queue\Model\Job`, `App\Queue\Factory\Data`, and so on. See [Overloading](../key-concepts/factory/overloading.md). diff --git a/modules/queue/admin.md b/modules/queue/admin.md new file mode 100644 index 0000000..eb784ca --- /dev/null +++ b/modules/queue/admin.md @@ -0,0 +1,87 @@ +--- +description: >- + The queue overview screen, the dashboard warning for unworked queues, and + the API behind them. +--- + +# Admin + +**Queue → Overview** shows what the queue is doing now: how many jobs are in each state, which workers are alive, and which jobs failed. It needs `nails/module-admin`. + +The screen loads its data from the [API](#api) and asks again every ten seconds, so it can stay open on a second monitor. + +## Permission + +| Permission | Class | Allows | +|---|---|---| +| Can view queue overview | `Nails\Queue\Admin\Permission\Overview\View` | See the screen and its data, and retry failed jobs | + +Grant it on a user group, as described in [User Permissions](../admin/user-permissions.md). The one permission covers retrying a failed job as well as looking at it. + +## Overview + +### Totals + +A row of tiles across the top: + +| Tile | Meaning | +|---|---| +| Workers (total, active, stale) | Worker rows. A worker is stale when its heartbeat is older than `QUEUE_WORKER_HEARTBEAT_STALE`. | +| Jobs (total) | Every job in every status. | +| Jobs pending | `PENDING` jobs that are due and waiting for a worker. | +| Jobs scheduled | `PENDING` jobs whose start time is still in the future. | +| Jobs running, complete, failed | Jobs in that status. Complete jobs disappear after [rotation](workers.md#maintenance). | +| Oldest queued age | How long the oldest due `PENDING` job has been waiting. A figure that keeps growing means the workers are not keeping up, or are not running. | +| Avg. latency (24h) | The average time between a job becoming available and a worker starting it. | +| Avg. duration (24h) | The average time a job takes to run, for jobs that finished. | +| Throughput (1h, 24h) | Jobs finished in that period, both `COMPLETE` and `FAILED`. | + +### Queues + +One row per [discovered queue](queues.md#discovery), with its job counts by status, its average latency and duration over the last hour, and its throughput over the last hour and the last 24 hours. A queue you have not [put in the discovery namespace](queues.md#discovery) is missing here. + +### Workers + +Every row in the worker table, with its ID, token, queues, how long it has been running, and when it last reported. A stale worker is highlighted. It is either dead or in the middle of a job longer than `QUEUE_WORKER_HEARTBEAT_STALE`; see [the warning](../queue.md#configuration) on that setting. `queue:clean` removes it. + +### Failed jobs + +The latest 100 `FAILED` jobs, newest first. Each shows its queue, task, payload, when it failed, the number of attempts, and the message from the last error. **Error Trace** opens the stack trace of the last error. + +**Retry** puts the job back in `PENDING`. It waits the usual [backoff](tasks.md#retries) for its next attempt number before it runs, so it is not instant. A retried job gets one more attempt: its retries are already used up, so if it fails again it is `FAILED` straight away. To give a task more tries, raise its `getMaxRetries()`. + +The old errors stay on the job. + +## Dashboard alert + +The admin dashboard shows the warning **Some queues are not being processed** when a [discovered queue](queues.md#discovery) has no live worker. It lists the queues without one. A worker is live when its heartbeat is newer than `QUEUE_WORKER_HEARTBEAT_STALE` and it was started for that queue's class. + +It is the same check [Find & Replace](../find-replace.md#queue) uses to warn about its own queue. The check is also available to your code: + +```php +/** @var \Nails\Queue\Service\Manager $manager */ +$manager = Factory::service('Manager', \Nails\Queue\Constants::MODULE_SLUG); + +$manager->unworkedQueues(); // Queue[]: discovered queues with no live worker +$manager->isProcessed('priority'); // bool: does a live worker cover this queue? +``` + +## API + +The screen reads and writes through two routes. Both need a logged-in user with the permission above. See [API](../api/README.md) for how routes and responses work. + +| Method | Path | Purpose | +|---|---|---| +| `GET` | `/api/queue/overview` | Everything on the screen | +| `POST` | `/api/queue/overview/retry` | Retry a failed job | + +`GET` returns four keys in `data`: + +| Key | Holds | +|---|---| +| `kpis` | The tiles: a `label`, `value`, `hint`, and `type` (`number`, `age`, or `duration`) each | +| `queues` | One entry per discovered queue, with `label`, `jobs`, `latency`, `duration`, and `throughput` | +| `workers` | One entry per worker, with `id`, `token`, `queues`, `created`, `heartbeat`, and `stale` | +| `failed` | Up to 100 failed jobs, with `id`, `queue`, `task`, `payload`, `finished`, `attempts`, and `errors` | + +`POST` takes `job_id`. It answers `400` if `job_id` is missing or the job is not `FAILED`, and `404` if there is no such job. diff --git a/modules/queue/process-managers.md b/modules/queue/process-managers.md new file mode 100644 index 0000000..de2626b --- /dev/null +++ b/modules/queue/process-managers.md @@ -0,0 +1,213 @@ +--- +description: >- + Keep queue workers running in production with systemd or Supervisor: start + on boot, restart on exit, stop cleanly, and restart after a deploy. +--- + +# Process managers + +`nails queue:work` runs until it is stopped. Something has to start it when the server boots, start it again if it crashes or is killed, and stop it cleanly. That is the job of a process manager. This page covers [systemd](#systemd) and [Supervisor](#supervisor). Use one of them, not both. + +Whichever you use, set these up the same way: + +| Setting | Why | +|---|---| +| Run it as the web server's user | The worker writes logs and files the web app reads. A worker run as `root` creates files the web app can't overwrite. | +| Start in the project root | `nails` finds the application from the working directory. | +| Pass the same configuration as the web app | The worker is a separate process and does not inherit what PHP-FPM or Apache has. Anything set as an environment variable for the web app must be set for the worker too. In particular `ENVIRONMENT` defaults to `DEVELOPMENT`, so a worker without it runs as a development environment. | +| Restart on any exit | A worker that exits, for any reason, should come back. | +| Stop with `SIGTERM` and wait | The worker finishes its current job on `SIGTERM`. The manager's stop timeout has to be longer than your longest job, or the job is killed part-way and run again. This needs the `pcntl` extension: see [Stopping](workers.md#stopping). | +| Use absolute paths to `nails` | Both managers start the process with a minimal environment. Find the path with `command -v nails`. | + +{% hint style="info" %} +You do not need a process manager to run `queue:clean`. It is a [cron](../cron.md) task. If cron is running, it is already scheduled. +{% endhint %} + +## systemd + +Create a template unit. The `@` makes it a template: the text after it in the name is the instance, available as `%i`, so one file runs as many workers as you like. + +`/etc/systemd/system/nails-queue@.service` + +```ini +[Unit] +Description=Nails queue worker %i +After=network.target + +[Service] +Type=simple +User=www-data +Group=www-data +WorkingDirectory=/var/www/app +Environment=ENVIRONMENT=PRODUCTION +ExecStart=/usr/local/bin/nails queue:work --queue=default +Restart=always +RestartSec=5 +KillSignal=SIGTERM +TimeoutStopSec=300 +SyslogIdentifier=nails-queue-%i + +[Install] +WantedBy=multi-user.target +``` + +| Line | Effect | +|---|---| +| `WorkingDirectory` | The project root. | +| `Environment` | Config for the worker. For several values use `EnvironmentFile=/etc/nails/app.env`, a file of `KEY=value` lines. | +| `ExecStart` | The worker command. `--queue` names the [queue](queues.md) this worker serves. It is `default` here, and every other queue needs [a worker of its own](#a-worker-per-queue). | +| `Restart=always`, `RestartSec` | Restart five seconds after any exit, except a `systemctl stop`. | +| `TimeoutStopSec` | How long to wait after `SIGTERM` before `SIGKILL`. The default is 90 seconds. Set it above your longest job. | + +Load the unit and start two workers: + +```bash +sudo systemctl daemon-reload +sudo systemctl enable --now nails-queue@1 nails-queue@2 +``` + +`enable` starts them at boot, `--now` also starts them now. The `1` and `2` are only labels. + +Check on them and read their output: + +```bash +systemctl status 'nails-queue@*' +journalctl -u 'nails-queue@*' -f +``` + +### Run it as one service + +If a single worker is enough, drop the `@` from the file name, delete `%i` from the two lines that use it, and use the plain name: `systemctl enable --now nails-queue`. + +## Supervisor + +Supervisor reads one file per program. On Debian and Ubuntu that is in `/etc/supervisor/conf.d/`; other distributions put it in `/etc/supervisord.d/` with an `.ini` extension. + +`/etc/supervisor/conf.d/nails-queue.conf` + +```ini +[program:nails-queue] +command=/usr/local/bin/nails queue:work --queue=default +directory=/var/www/app +user=www-data +environment=ENVIRONMENT="PRODUCTION" +process_name=%(program_name)s_%(process_num)02d +numprocs=2 +autostart=true +autorestart=true +startsecs=5 +startretries=10 +stopsignal=TERM +stopwaitsecs=300 +redirect_stderr=true +stdout_logfile=/var/log/nails-queue_%(process_num)02d.log +stdout_logfile_maxbytes=20MB +stdout_logfile_backups=5 +``` + +| Line | Effect | +|---|---| +| `directory` | The project root. | +| `environment` | Config for the worker, as a comma-separated list of `KEY="value"` pairs. | +| `numprocs`, `process_name` | How many workers to run. With more than one, `process_name` has to include `%(process_num)02d`. | +| `autorestart=true` | Restart on any exit. The default, `unexpected`, does not restart a worker that exits with status 0. | +| `startsecs`, `startretries` | A worker that exits before `startsecs` counts as a failed start. After `startretries` failed starts Supervisor stops trying and marks it `FATAL`. A high number lets the worker ride out a database that isn't up yet at boot. | +| `stopsignal`, `stopwaitsecs` | Send `SIGTERM`, then `SIGKILL` after this many seconds. The default is 10, which would kill most jobs part-way. Set it above your longest job. | +| `stdout_logfile` | Console output. The worker also writes its own [log](workers.md#log). | + +Tell Supervisor about the new file and start the workers: + +```bash +sudo supervisorctl reread +sudo supervisorctl update +sudo supervisorctl status 'nails-queue:*' +``` + +`update` starts a program that is new, because of `autostart=true`. + +## A worker per queue + +A worker serves only the queues it was started with. Each queue gets its own program, with its own `--queue`, so that a slow queue never holds up a fast one. + +The built-in `priority` queue is the case that matters. It is a queue in name only: nothing starts a worker for it, and a single worker given both `default` and `priority` runs jobs oldest first across the two, so priority jobs gain nothing. The app has to configure a separate process for `priority`, or those jobs are never run. See [Priority](queues.md#priority). The same applies to any queue you [define](queues.md#defining-a-queue). + +{% tabs %} +{% tab title="systemd" %} +Copy the unit to `/etc/systemd/system/nails-queue-priority@.service`. Change two lines and leave the rest as it was: + +```ini +ExecStart=/usr/local/bin/nails queue:work --queue=priority +SyslogIdentifier=nails-queue-priority-%i +``` + +Then load it and start a worker: + +```bash +sudo systemctl daemon-reload +sudo systemctl enable --now nails-queue-priority@1 +``` +{% endtab %} + +{% tab title="Supervisor" %} +Add a second program to the same file, or to a file of its own: + +```ini +[program:nails-queue-priority] +command=/usr/local/bin/nails queue:work --queue=priority +directory=/var/www/app +user=www-data +environment=ENVIRONMENT="PRODUCTION" +process_name=%(program_name)s_%(process_num)02d +numprocs=1 +autostart=true +autorestart=true +startsecs=5 +startretries=10 +stopsignal=TERM +stopwaitsecs=300 +redirect_stderr=true +stdout_logfile=/var/log/nails-queue-priority_%(process_num)02d.log +stdout_logfile_maxbytes=20MB +stdout_logfile_backups=5 +``` + +Then load it: + +```bash +sudo supervisorctl reread +sudo supervisorctl update +``` +{% endtab %} +{% endtabs %} + +Priority jobs are usually short, so `TimeoutStopSec` and `stopwaitsecs` can be lower here than for `default`. Set them from the longest job that queue runs. + +## Checking it works + +1. Open **Queue → Overview** in [admin](admin.md). Each running worker is in the **Workers** table with a recent heartbeat, listing the queues it serves. The dashboard warning for queues with no worker should be gone, which means every [discovered queue](queues.md#discovery), `priority` included, has a worker. +2. Push a `DoNothing` job, as in [Trying it out](tasks.md#trying-it-out), and watch it complete. +3. Restart the worker (below) and check a replacement appears in **Workers**. If you stop a worker without `SIGTERM`, its old row stays until `queue:clean` removes it. + +## Deploying + +A worker loads your code once, when it starts. After a deploy, restart the workers so they run the new code. + +{% tabs %} +{% tab title="systemd" %} +```bash +sudo systemctl restart 'nails-queue*' +``` +{% endtab %} + +{% tab title="Supervisor" %} +```bash +sudo supervisorctl restart 'nails-queue:*' 'nails-queue-priority:*' +``` +{% endtab %} +{% endtabs %} + +Both managers send `SIGTERM` and wait for the job in progress, up to the timeout you set, so a restart does not cut a job off. A restart is only as quick as the longest job running when it starts. + +## Other environments + +The same rules apply anywhere else a worker runs, such as a Docker container or a Kubernetes pod. Run `nails queue:work` as the container's main process, restart the container if it exits, forward `SIGTERM` to the process, and make the grace period before `SIGKILL` longer than your longest job. Give the container the same environment as the web app. diff --git a/modules/queue/queues.md b/modules/queue/queues.md new file mode 100644 index 0000000..8e5b645 --- /dev/null +++ b/modules/queue/queues.md @@ -0,0 +1,134 @@ +--- +description: >- + The built-in queues, how to add your own, and how aliases and discovery + work. +--- + +# Queues + +A queue is a name that jobs are filed under and workers listen to. It is a class that implements `Nails\Queue\Interface\Queue`. The job stores the class name. + +Most apps only need the built-in `default` queue. Add a queue when a kind of work needs its own workers: a slow report that should not hold up password-reset emails, or work that needs setup before the first job. + +## Built-in queues + +| Class | Alias | +|---|---| +| `Nails\Queue\Queue\Queues\DefaultQueue` | `default` | +| `Nails\Queue\Queue\Queues\PriorityQueue` | `priority` | + +`push()` uses `default` when no queue is given, and `nails queue:work` listens to `default` when no `--queue` is given. + +### Priority + +`priority` is a separate queue in name only. The module gives it no extra weight, no head start, and no worker. Nothing starts a worker for it, and a job pushed to it waits until the app runs one. + +{% hint style="warning" %} +**The app has to run a separate worker process for `priority`.** One worker that listens to both `default` and `priority` takes the oldest job across the two, exactly as if they were one queue, so a priority job still waits behind everything pushed to `default` before it. +{% endhint %} + +Give each queue its own process, so priority work never waits behind a slow `default` job: + +```bash +nails queue:work --queue=default +nails queue:work --queue=priority +``` + +Keep both running with a [process manager](process-managers.md#a-worker-per-queue). + +## Defining a queue + +A queue has two static hooks. Both may do nothing. + +```php +namespace App\Queue\Queues; + +use Nails\Queue\Interface\Queue; +use Nails\Queue\Resource\Worker; + +class Reports implements Queue +{ + public static function setup(Worker $worker): void + { + // Runs once, when a worker for this queue starts + } + + public static function refresh(Worker $worker): void + { + // Runs again every QUEUE_WORKER_REFRESH_INTERVAL seconds + } +} +``` + +| Hook | When | +|---|---| +| `setup()` | Once as the worker starts, before it takes any job. | +| `refresh()` | Periodically while the worker is running. Use it for work that goes stale, such as refreshing a token or clearing temporary files. | + +Both receive the `Nails\Queue\Resource\Worker` that is running. An exception from either is logged and the worker carries on. + +`refresh()` is checked once per pass through the worker loop, so a long job delays it until the job finishes. An idle worker passes through the loop at least every five seconds (see [Workers](workers.md#what-a-worker-does)), so the hook runs close to on time. + +The hooks are static because the job records only the class name. A queue is built with `new` and no arguments. + +## Using a queue + +Pass the class name or an instance when pushing a job, and give the class name to the worker: + +```php +$manager->push( + task: new Tasks\Reports\Generate(), + data: Data\Reports\Generate::make(123), + queue: \App\Queue\Queues\Reports::class, +); +``` + +```bash +nails queue:work --queue='App\Queue\Queues\Reports' +``` + +Quote the class name in the shell so the backslashes survive. + +## Aliases + +An alias is a short name for a queue: `default` and `priority` are aliases. `push()` and `--queue` accept an alias anywhere they accept a class name, and aliases are not case sensitive. A name resolves in this order: + +1. A queue instance is used as it is. +2. A registered alias. +3. A class name. + +Anything else throws an `InvalidArgumentException`. + +Aliases are not stored. Each PHP process starts with just `default` and `priority`, so a custom alias has to be registered in every process that uses it: the web request that pushes the job, and the worker. The place that runs for both is the Manager. Override it with `App\Queue\Service\Manager`: + +```php +namespace App\Queue\Service; + +use App\Queue\Queues; +use Nails\Common\Service\Database; +use Nails\Queue\Model; +use Nails\Queue\Service\Manager as BaseManager; + +class Manager extends BaseManager +{ + public function __construct(Database $database, Model\Worker $workerModel, Model\Job $jobModel) + { + parent::__construct($database, $workerModel, $jobModel); + $this->addAlias('reports', new Queues\Reports()); + } +} +``` + +Then `queue: 'reports'` and `--queue=reports` both work. If that is more setup than you want, use the class name. + +## Discovery + +The Manager finds queues by looking in the `Queue\Queues` namespace of every installed component: the app (`App\Queue\Queues`), `Nails\Queue\Queue\Queues` in this module, and any other module that ships one. A class is discovered when it implements `Queue` and can be instantiated. + +Discovery is what makes a queue visible to [admin](admin.md). The overview lists discovered queues, and the dashboard alert warns when a discovered queue has no worker. A queue outside that namespace still works for pushing and running jobs, but admin does not list it. + +{% hint style="warning" %} +`DefaultQueue` and `PriorityQueue` are discovered, so the dashboard warns while `priority` has no worker, even if you never push to it. Run a [separate `priority` worker](#priority) to clear the warning. +{% endhint %} + +A worker counts for a queue when its heartbeat is recent and it was started with that queue's class. A worker started with `--queue=priority` counts for `PriorityQueue` whichever way the queue was spelled. diff --git a/modules/queue/tasks.md b/modules/queue/tasks.md new file mode 100644 index 0000000..016fc09 --- /dev/null +++ b/modules/queue/tasks.md @@ -0,0 +1,176 @@ +--- +description: >- + Write a task, push a job with a payload, and how failed jobs are retried. +--- + +# Tasks + +A task is the work. It implements `Nails\Queue\Interface\Task` and has two methods: how many times a failure may be retried, and `run()`. + +```php +namespace App\Queue\Tasks\Reports; + +use App\Model; +use Nails\Factory; +use Nails\Queue\Interface\Data; +use Nails\Queue\Interface\Task; + +class Generate implements Task +{ + public static function getMaxRetries(): int + { + return 3; + } + + public function run(Data $data): void + { + $payload = $data->get(); + + /** @var Model\Report $model */ + $model = Factory::model('Report', 'app'); + $report = $model->getById($payload->report_id); + + // Do the work. Throw to fail the attempt. + } +} +``` + +`run()` returns nothing. Returning normally completes the job. Throwing anything, including an `Error`, fails the attempt. + +A worker builds the task with `new`, so a task cannot have required constructor arguments. Everything the task needs comes from the payload. + +## Payload + +The payload is wrapped in a `Data` object. Build one with the `Data` factory: + +```php +use Nails\Factory; +use Nails\Queue\Constants; + +$data = Factory::factory('Data', Constants::MODULE_SLUG, (object) [ + 'report_id' => 123, +]); +``` + +It accepts an array, string, int, float, bool, `stdClass`, or `null`. It is stored as JSON, so everything in it must survive `json_encode()`. + +The worker reads the JSON back with `json_decode()`, so what `get()` returns in `run()` is not always what went in: + +| Pushed | `get()` returns | +|---|---| +| `stdClass` | `stdClass` | +| Associative array | `stdClass` | +| List array | List array | +| Scalar or `null` | The same scalar or `null` | + +Push an object and read properties with `->`, as in the example above. + +Keep payloads small. Pass an ID and load the row in `run()`, not the row itself. The job may wait, and the row may change before it runs. + +### Your own Data class + +A subclass of `Nails\Queue\Factory\Data` is a good place to validate a payload before it is stored: + +```php +namespace App\Queue\Data\Reports; + +use InvalidArgumentException; +use Nails\Queue\Factory\Data; + +class Generate extends Data +{ + public static function make(int $reportId): self + { + if ($reportId <= 0) { + throw new InvalidArgumentException('A report ID is required'); + } + + return new self((object) ['report_id' => $reportId]); + } +} +``` + +The class is only used to build the payload. A worker always reads a job back through the `Data` factory, so the task receives the plain `Data` class and not your subclass. To change what the worker builds, define `App\Queue\Factory\Data`. + +## Pushing a job + +Resolve the Manager and call `push()`: + +```php +use App\Queue\Data; +use App\Queue\Tasks; +use Nails\Factory; +use Nails\Queue\Constants; +use Nails\Queue\Service\Manager; + +/** @var Manager $manager */ +$manager = Factory::service('Manager', Constants::MODULE_SLUG); + +$job = $manager->push( + task: new Tasks\Reports\Generate(), + data: Data\Reports\Generate::make(123), +); +``` + +| Argument | Default | Effect | +|---|---|---| +| `task` | Required | The task to run. Only its class name is stored. | +| `data` | Required | The payload. | +| `queue` | `null` | A [queue](queues.md) instance, alias, or class name. `null` means `default`. | +| `availableAt` | Now | A `DateTimeInterface`. No worker runs the job before then. | + +`push()` returns the stored `Nails\Queue\Resource\Job`. A job with a future `availableAt` stays `PENDING` and is counted as _scheduled_ in [admin](admin.md). + +```php +$manager->push( + task: new Tasks\Reports\Generate(), + data: Data\Reports\Generate::make(123), + queue: 'priority', + availableAt: new DateTimeImmutable('+10 minutes'), +); +``` + +{% hint style="info" %} +Pushing does not start anything. If no [worker](workers.md) is running for the queue the job waits. A job pushed inside a database [transaction](../../core-services/database/transactions.md) is invisible to workers until the transaction commits, so a job that depends on the transaction's changes never runs ahead of them. +{% endhint %} + +### Order + +A worker takes the lowest job `id` among the jobs that are `PENDING` and due. Jobs run first in, first out. A retried job keeps its `id`, so it goes back to its place once its delay has passed. A worker listening to several queues takes the lowest `id` across all of them. See [Queues](queues.md#priority). + +## Retries + +`getMaxRetries()` is the number of times a failed task is tried again. A task with `3` runs up to four times: the first run and three retries. `0` means the first failure is final. + +After a failure with retries left the job goes back to `PENDING`, its `attempts` count goes up by one, and `available_at` moves into the future. The delay doubles each time, starting at five seconds, is capped at five minutes, and has up to 20% random jitter either way. + +| Retry | Wait before it, roughly | +|---|---| +| 1 | 5 s | +| 2 | 10 s | +| 3 | 20 s | +| 4 | 40 s | +| 5 | 80 s | +| 6 | 160 s | +| 7 and later | 300 s | + +When the retries are used up the job is `FAILED`. A failed job is never run again unless someone [retries it from admin](admin.md#failed-jobs) or calls `retryJob()`. It is deleted after `QUEUE_JOB_ROTATE_FAILED_DAYS`. + +Every failure is appended to the job's `errors`, so a failed job carries the whole history. Each entry has the exception class, message, code, file, line, a stack trace cut to 2,000 characters, and when it happened. + +### Write tasks that can run twice + +A job can run more than once even when it succeeded the first time. If a worker is killed mid-job, or a job outlasts `QUEUE_WORKER_HEARTBEAT_STALE`, [`queue:clean`](workers.md#maintenance) puts it back in `PENDING` and another worker runs it from the start. Make `run()` safe to repeat. Check whether the work is already done before doing it, and prefer writes that set a value to ones that add to it. + +## Trying it out + +`Nails\Queue\Tasks\DoNothing` is a task that does nothing and never retries. Use it to check that workers are picking up jobs: + +```php +$manager->push( + task: new \Nails\Queue\Tasks\DoNothing(), + data: Factory::factory('Data', Constants::MODULE_SLUG, null), +); +``` + +The job should go `PENDING`, `RUNNING`, `COMPLETE` within a second or so of a worker starting. Watch the worker output or the [overview](admin.md). diff --git a/modules/queue/workers.md b/modules/queue/workers.md new file mode 100644 index 0000000..a2b62c1 --- /dev/null +++ b/modules/queue/workers.md @@ -0,0 +1,113 @@ +--- +description: >- + Run workers, read their log, stop them cleanly, run one job by hand, and + keep the tables tidy with queue:clean. +--- + +# Workers + +A worker is the `nails queue:work` command. It is a long-running process that polls for jobs and runs them, one at a time. + +```bash +nails queue:work +``` + +Run it from the project root. In production, don't start it by hand: use a [process manager](process-managers.md) so it starts on boot and comes back if it exits. + +## Choosing queues + +| Option | Effect | +|---|---| +| `--queue` | A [queue](queues.md) alias or class name to work. Repeat it to work several. Defaults to `default`. | + +```bash +# The default queue +nails queue:work + +# By alias, two queues in one worker +nails queue:work --queue=default --queue=reports + +# By class name +nails queue:work --queue='App\Queue\Queues\Reports' +``` + +A worker only takes jobs from the queues it was started with. A job on a queue that no worker was started for is never run. That includes the built-in `priority` queue: it is not served by the default worker, so [start one for it](queues.md#priority). + +## What a worker does + +1. Registers itself in `nails_queue_worker` and prints its ID and token. +2. Calls `setup()` on each of its queues once. +3. Loops until told to stop: + 1. Reserves the oldest `PENDING` job that is due, from any of its queues. Two workers never get the same job. + 2. If there is one, marks it `RUNNING`, runs the task, then marks it `COMPLETE`, schedules a [retry](tasks.md#retries), or marks it `FAILED`. + 3. If there isn't one, sleeps for `QUEUE_WORKER_WAIT_TIME` milliseconds plus up to 200 ms of random jitter, then doubles the wait for the next empty poll, up to 5,000 ms. The wait starts again after the next job. A job pushed to an idle worker can wait up to about five seconds. + 4. Calls `refresh()` on each queue when `QUEUE_WORKER_REFRESH_INTERVAL` seconds have passed. + 5. Writes its heartbeat, at most once every `QUEUE_WORKER_HEARTBEAT_DEBOUNCE` seconds. +4. Removes its own row from `nails_queue_worker` on the way out. + +A worker runs one job at a time. To run more jobs at once, run more workers. They can share a queue, and they can share a machine. + +{% hint style="info" %} +A worker reads your code once, when it starts. After a deploy, restart the workers or they keep running the old code. See [Deploying](process-managers.md#deploying). +{% endhint %} + +### Log + +The worker prints to the terminal and also writes the same lines to `queue-worker-YYYY-MM-DD.php` in the application log directory, next to the [cron](../cron.md) log. Each line starts with `[worker:#ID]`. The file switches to the new day's name the next time `refresh()` is due. + +``` +[worker:#12] Worker ID: +[worker:#12] #12:web-01:48213:9f3a1c2e +[worker:#12] Working queues: +[worker:#12] Nails\Queue\Queue\Queues\DefaultQueue +[worker:#12] Waiting for jobs: +[worker:#12] [App\Queue\Tasks\Reports\Generate:#481] ......... RUNNING +[worker:#12] [App\Queue\Tasks\Reports\Generate:#481] ......... COMPLETE [0.2314s] +[worker:#12] [App\Queue\Tasks\Reports\Generate:#482] ......... RETRY 1/3 (next at 14:02:07) [0.0412s] +``` + +A job line ends in `COMPLETE`, `RETRY n/max` with the time the retry becomes due, or `FAILED`, followed by how long the attempt took. The error itself is not in the log. It is stored on the job: see [Failed jobs](admin.md#failed-jobs). + +## Stopping + +Send the worker `SIGTERM` or `SIGINT` (`Ctrl+C` in a terminal). It finishes the job it is running, takes no more, removes its row, and exits with status 0. An idle worker stops within about a tenth of a second. + +This needs the `pcntl` PHP extension. Without it the signal kills the process at once, in the middle of a job, and the worker's row stays behind until [`queue:clean`](#maintenance) removes it. + +A process manager that stops a worker has to wait for the job to end. If it gives up and sends `SIGKILL` first, the job is cut off and later run again. Set the stop timeout above your longest job. The [examples](process-managers.md) do. + +## Running one job by hand + +`queue:run` takes a job ID and runs its task in the current process. Use it to reproduce a failure with a debugger or `var_dump()`. + +```bash +nails queue:run 481 +``` + +It prints the job's ID, queue, task, and payload, registers a short-lived worker for the job's queue, calls `setup()`, runs the task, and removes the worker. + +It does not change the job. The status, attempts, and errors are left as they were, a task that throws prints `ERROR: …` and the command still exits 0, and the job can be in any status, including `COMPLETE`. Running a job that is `PENDING` can run it twice if a worker takes it at the same time. + +## Maintenance + +`queue:clean` repairs what a crashed or killed worker leaves behind and removes old rows. + +```bash +nails queue:clean +``` + +It does three things, in this order: + +| Step | What it does | +|---|---| +| Stale workers | Deletes workers whose heartbeat is more than `QUEUE_WORKER_HEARTBEAT_STALE` seconds old. | +| Stuck jobs | Puts `RUNNING` jobs that no longer belong to a worker back in `PENDING`. A job loses its worker when the worker's row is deleted, so the previous step feeds this one. `attempts` is not changed. | +| Old jobs | Deletes `COMPLETE` jobs older than `QUEUE_JOB_ROTATE_COMPLETE_DAYS` and `FAILED` jobs older than `QUEUE_JOB_ROTATE_FAILED_DAYS`, counted from when they finished. `0` turns that deletion off. | + +The module registers this as a [cron](../cron.md) task that runs every five minutes, so there is nothing to schedule as long as the cron runner is already running on the server. + +### When a worker dies + +If a worker is killed with `SIGKILL`, loses its machine, or runs out of memory, the job it was running stays `RUNNING`. It is recovered by `queue:clean`: the worker is deleted once its heartbeat is stale, and the job is reset and run again by another worker. That takes at least `QUEUE_WORKER_HEARTBEAT_STALE` seconds and then up to five minutes more, for the next `queue:clean`, so budget five to ten minutes with the defaults. + +Nothing is lost, but the job runs again from the start. See [Write tasks that can run twice](tasks.md#write-tasks-that-can-run-twice).