Digging deeper

Queues

Some work is too slow to do while a user waits: sending mail, resizing an image, calling another service. Push it onto a queue instead, and a worker process runs it in the background while your app answers at once.

Connections and queues#

config/queue.ts lists your queue connections: where jobs are stored. A new app uses the database connection, whose jobs table a new app's migrations already created:

import { env } from '@orieljs/framework'export default {  default: env('QUEUE_CONNECTION', 'database'),  connections: {    sync: { driver: 'sync' },    database: { driver: 'database', table: 'jobs', queue: 'default', retry_after: 90, after_commit: false },    redis: { driver: 'redis', connection: 'default', queue: 'default', retry_after: 90, block_for: null, after_commit: false },  },  batching: { database: env('DB_CONNECTION', 'sqlite'), table: 'job_batches' },  failed: { driver: 'database-uuids', database: env('DB_CONNECTION', 'sqlite'), table: 'failed_jobs' },}

The sync driver runs each job straight away, in the same process, which is handy in tests; redis reads its server from database.redis in config/database.ts. Within a connection, jobs go onto named queues, so a worker can run the urgent ones first. Without a name, a job goes onto the connection's queue, here default.

Creating jobs#

make:job writes a job class in app/Jobs:

node oriel make:job ProcessPodcast

A job's constructor takes what the job needs, and handle() does the work. Everything the job stores in its fields travels with it to the worker:

import { Log, Queueable } from '@orieljs/framework'import type { Podcast } from '../Models/Podcast'export class ProcessPodcast extends Queueable {  constructor(public podcast: Podcast) {    super()  }  async handle(): Promise<void> {    Log.info('Processing a podcast', { file: this.podcast.audio_path })    this.podcast.published_at = new Date()    await this.podcast.save()  }}

A model in a job is stored as its class and key, not its attributes, and the worker reads it again from the database, with the relations it had loaded, before handle() runs. Dates, maps, sets, plain objects and arrays travel as they are. A model that has been deleted in the meantime fails the job; set deleteWhenMissingModels = true to drop the job quietly instead.

The worker finds your job classes in app/Jobs (and your events, listeners, mail and notifications in their folders) by their class names, so give each its own name.

Dispatching jobs#

Dispatch a job with its static dispatch(), passing what its constructor takes:

import { redirect } from '@orieljs/framework'import { ProcessPodcast } from '../../Jobs/ProcessPodcast'import { Podcast } from '../../Models/Podcast'import { Controller } from './Controller'export class PodcastController extends Controller {  async store() {    const podcast = await Podcast.create({ title: 'Episode 1', audio_path: 'podcasts/1.mp3' })    await ProcessPodcast.dispatch(podcast)    return redirect().back()  }}

dispatchIf() and dispatchUnless() take a condition first:

await ProcessPodcast.dispatchIf(accountActive, podcast)await ProcessPodcast.dispatchUnless(podcast.published_at !== null, podcast)

Delayed dispatching#

delay() keeps a job off the worker for a number of seconds, or until a date:

await ProcessPodcast.dispatch(podcast).delay(10 * 60)await ProcessPodcast.dispatch(podcast).delay(new Date('2026-12-24T09:00:00Z'))

Running a job now#

dispatchSync() runs the job in the current process and waits for it, and dispatchAfterResponse() runs it once the response has been sent to the browser:

await ProcessPodcast.dispatchSync(podcast)ProcessPodcast.dispatchAfterResponse(podcast)

Choosing the queue and the connection#

await ProcessPodcast.dispatch(podcast).onQueue('processing')await ProcessPodcast.dispatch(podcast).onConnection('redis')

A job can also name them for itself in its queue and connection fields.

Job chaining#

A chain runs its jobs one after the other, and stops when one fails. catch() takes a job to run when that happens:

await Bus.chain([new ProcessPodcast(podcast), new OptimizePodcast(podcast), new PublishPodcast(podcast)])  .onQueue('processing')  .catch(new ReportFailure(podcast.id))  .dispatch()

Retries and timeouts#

A job is attempted once unless it, or the worker, says otherwise. A job's fields set its own limits:

import { Queueable } from '@orieljs/framework'import type { Podcast } from '../Models/Podcast'export class ProcessPodcast extends Queueable {  tries = 3  backoff = [10, 60]  timeout = 120  constructor(public podcast: Podcast) {    super()  }  async handle(): Promise<void> {    if (this.attempts() > 1) {      // a retry: the first attempt failed    }  }  async failed(error: unknown): Promise<void> {    // every attempt failed: tell someone  }}

backoff is the number of seconds to wait before each retry, one number for all of them or one per retry. timeout is how many seconds an attempt may run before the worker gives up on it. Inside handle(), this.release(seconds) puts the job back on the queue for later, and this.fail(error) fails it at once, with no more retries.

Job middleware#

A job's middleware() wraps its handle(). WithoutOverlapping keeps two jobs of the same class with the same key from running at the same time (shared() makes the key count across classes); one that finds the key taken goes back on the queue:

import { Queueable, WithoutOverlapping } from '@orieljs/framework'import type { Podcast } from '../Models/Podcast'export class ProcessPodcast extends Queueable {  constructor(public podcast: Podcast) {    super()  }  middleware() {    return [new WithoutOverlapping(this.podcast.id).releaseAfterSeconds(30)]  }  async handle(): Promise<void> {}}

Unique jobs#

A job with a uniqueId() isn't queued while another job of its class with the same ID is waiting or running: the second dispatch is dropped. Uniqueness uses a cache lock, so it needs a cache store with locks, such as a new app's database store:

import { Queueable } from '@orieljs/framework'export class UpdateSearchIndex extends Queueable {  uniqueFor = 3600  constructor(public productId: number) {    super()  }  uniqueId(): string {    return String(this.productId)  }  async handle(): Promise<void> {}}

The lock is released when the job has run, or has failed for good; a job released for a retry keeps it. uniqueFor is how many seconds the lock lasts at most, in case a job never finishes; without it, the lock lasts until then. With uniqueUntilProcessing = true, the lock is released as the job starts, so the next one can be queued while it runs. uniqueVia() returns the name of the cache store that holds the lock, in place of your default store. Any of these members makes a job unique, with or without a uniqueId(): without one, the class alone is the ID.

Queued mail, notifications and listeners can be unique in the same way, with these members on the mailable, the notification or the listener. A duplicate is:

• a mailable of the same class, with the same uniqueId(), sent to the same recipients with Mail.to(), cc() and bcc() (see Unique mailables);
• a notification of the same class, with the same uniqueId(), to the same notifiable on the same channel (see Unique notifications);
• a listener of the same class, with the same uniqueId(), which receives the event: a listener has one lock for all its methods and events (see Unique listeners).

Job batching#

A batch runs many jobs side by side and tells you when they have all finished. The batch's own jobs say what happens then:

const batch = await Bus.batch(files.map((file) => new ImportCsv(file)))  .name('Import CSV')  .then(new ImportFinished())  .catch(new ImportFailed())  .allowFailures()  .dispatch()

then() runs when every job has succeeded, catch() when the first one fails, and finally() when the batch is done either way. A failure marks the batch cancelled, unless it allowFailures(), but its other jobs still run: a job that should stop checks (await this.batch())?.cancelled() first, as in Laravel. Bus.findBatch(id) reads a batch's progress later: its totalJobs, pendingJobs, failedJobs, progress() (a percentage) and finished(). Inside a job, await this.batch() returns its batch. A job in a batch keeps its own delay, except on the sync connection, which runs every job at once.

Running the queue worker#

The worker takes jobs off the queue and runs them until you stop it:

node oriel queue:work

It runs the default connection's default queue. Name another connection, and list the queues to work in order of priority:

node oriel queue:work redis --queue=high,default

A worker loads your code once, so it doesn't see your changes until it restarts. While you work, run node oriel queue:listen instead: it starts a fresh process for every job. In production, keep workers running with a process manager such as systemd or Supervisor, and after each deploy run node oriel queue:restart, which tells every worker to stop after its current job so the process manager starts it again on the new code (Deployment shows a Supervisor program).

queue:restart leaves its signal in your default cache store, which each worker reads after every job. While the cache can't be read, a worker goes on working and reports the error once per outage; it sees the next signal once the cache is back.

Failed jobs#

A job that has used up its attempts is stored in the failed_jobs table, and its failed() method runs. The queue commands manage them:

node oriel queue:failednode oriel queue:retry 5a6b8c1e-9d2f-4e3a-8b7c-1f2e3d4c5b6anode oriel queue:retry allnode oriel queue:forget 5a6b8c1e-9d2f-4e3a-8b7c-1f2e3d4c5b6anode oriel queue:flush

queue:clear deletes the jobs still waiting on a queue: node oriel queue:clear redis --queue=emails.