Pausing queues

Queues can be paused individually or all at once to temporarily prevent new jobs from being worked until they are resumed.

Pausing a single queue

The capability to pause a queue is a useful operational lever. This can be done without shutting down clients, enabling users to pause a subset of jobs but leave other queues to process jobs normally.

The Client offers QueuePause and QueueResume APIs to pause and resume individual queues. The following example demonstrates these APIs for the default queue:

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    // ...
})

if err = riverClient.QueuePause(ctx, "default"); err != nil {
    // handle error
}

if err = riverClient.QueueResume(ctx, "default"); err != nil {
    // handle error
}

Pass the queue name to queue_pause to suspend fetching, and use queue_resume to allow fetching again:

client.queue_pause :default
client.queue_resume :default

Use the client’s queue operations to pause or resume a named queue:

client.queues().pause("default").await?;
client.queues().resume("default").await?;

Call pause or resume on client.queues with the queue’s name:

await client.queues.pause("default");
await client.queues.resume("default");

Queues remain paused indefinitely until they are later resumed. However, if a queue is removed from the configs of all active workers, its record will be removed after 24 hours; if readded it will begin in an unpaused state. See the full details below for more information.

Pausing all queues

Use the special asterisk queue name * to pause or resume all known queues:

if err = riverClient.QueuePause(ctx, "*"); err != nil {
    // handle error
}

if err = riverClient.QueueResume(ctx, "*"); err != nil {
    // handle error
}

Use "*" as the queue selector to pause or resume all queues:

client.queue_pause "*"
client.queue_resume "*"

Pass QueueSelector::All when the operation should apply to every queue:

use riverqueue::QueueSelector;

client.queues().pause(QueueSelector::All).await?;
client.queues().resume(QueueSelector::All).await?;

Use the "*" selector to apply the pause or resume operation to all queues:

await client.queues.pause("*");
await client.queues.resume("*");

This only pauses or resumes queues which are currently known (tracked in the river_queue table) and has no effect on queues which are later added to clients.

Details and caveats

Active queues are tracked with records in the river_queue table. When started, clients will UPSERT (INSERT ... ON CONFLICT DO UPDATE) a row on this table for each queue configured in their queue configuration. They will also periodically bump the updated_at timestamp on these rows to indicate they are still being used. A separate queue cleaner maintenance service is responsible for periodically deleting old inactive queues from this table.

Pause functionality works as follows:

When paused, the river_queue row for the queue is updated with a paused_at timestamp to indicate that it is currently paused.
Active clients detect the change through database notifications or periodic polling, depending on the language, driver, and configuration.
Once a client detects that a queue is paused, it will cease fetching additional jobs for that queue until it is later resumed.

Even when paused, actively configured queues are still periodically updated in the river_queue table so that a queue which is paused but still in use will remain paused indefinitely. However, if a paused queue is removed from all active client configs, it will be removed by the queue cleaner after 24 hours, meaning if it is subsequently re-added to a client config it will start unpaused.

Clients which are started after a queue has been paused will detect that the queue is paused at startup and will not fetch or work any jobs on that queue until resumed.