Writing reliable workers

One of the biggest benefits of River is that jobs can be enqueued transactionally and atomically along with other database changes. Once a job is committed to the database, River will not lose track of it and will ensure that it is executed.

However, even the most robust and reliable applications will inevitably encounter job errors or other failures, particularly at scale. Any River job may be retried multiple times due to scenarios including:

An error returned, raised, or thrown by the worker
An error preventing the completed job’s state from being saved to the database
A process crash or runtime failure

River is designed to retry jobs in the event of an error. This at-least-once execution pattern ensures that programming errors, crashes, or transient network failures do not result in a job being lost or forgotten. It is therefore important to write workers with the expectation that jobs can and will be retried.

Jobs are automatically completed or errored

When River fetches a job, it passes the job to its registered worker. Successful completion marks the job completed asynchronously. A worker error records the failure and schedules a retry, or discards the job when its attempt budget is exhausted. Each language also has explicit outcomes for cancelling or snoozing.

Timeouts and cancellation

Go does not provide a way for a goroutine to cancel or terminate another goroutine. This means that the only practical way for River to timeout or cancel jobs is with context cancellation.

Respect context cancellation

Workers should respect context cancellation for timeouts, job cancellation, and safe rapid shutdown. If a worker blocks and does not respect ctx.Done(), these features will not function and River will be stuck waiting for the job to complete.

To ensure that workers will always respect a cancelled or timed out context, workers should always inherit from the context provided in Work().

River interrupts worker threads on timeout or cancellation. Let the exception propagate and put cleanup in ensure blocks. Avoid swallowing interruption exceptions or blocking indefinitely in native extensions, where interruption can be delayed.

Workers receive a cancellation token through WorkContext. Await it alongside long-running operations and return WorkCancelled when interrupted. Tokio can abort an asynchronous task that keeps yielding, but cannot reliably interrupt CPU-bound or blocking native code.

Workers receive an AbortSignal. Pass it to cancellable APIs and check it in loops. In-process JavaScript cannot forcibly stop a handler; CPU-bound handlers can use the optional worker-thread runtime. See graceful shutdown.

Client job timeouts

The default job timeout is one minute. Disabling it also prevents ordinary timeout-based rescue of abandoned jobs, so do so only for jobs that need it.

By default, clients will time out all jobs after 1 minute (DefaultJobTimeout). This can be customized for all jobs at the client level with the JobTimeout config:

client, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    JobTimeout: 10*time.Minute,
})

River can also run jobs with no timeout, though it is not recommended to do this globally:

client, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    JobTimeout: -1,
})

Set job_timeout in seconds to change the client’s default time limit for an attempt:

config = River::Config.new(job_timeout: 10 * 60)
# Disable with job_timeout: nil.

Configure the default attempt timeout with job_timeout on the client builder:

let client = riverqueue::Client::builder(pool)
    .job_timeout(std::time::Duration::from_secs(10 * 60))
    .build()?;
// Use .without_job_timeout() to disable timeouts.

Set jobTimeout in the client configuration to change the default time limit for worker attempts:

const client = new Client(driver, { jobTimeout: { minutes: 10 } });
// Use jobTimeout: null to disable timeouts.

Worker-level job timeouts

Timeouts can also be customized at the level of an individual worker:

func (w *MyWorker) Timeout(*Job[T]) time.Duration { return 30 * time.Second }

Some use cases require a job to run for an indefinite amount of time. To disable timeouts on a worker, return -1:

func (w *MyWorker) Timeout(*Job[T]) time.Duration { return -1 }

Jobs with no timeout are still subject to being manually cancelled or an abrupt graceful shutdown via StopAndCancel, but this will only function if the worker respects the context’s cancellation.

Define timeout on the worker to override the client default for its attempts. The return value is a duration in seconds:

def timeout(_job) = 30
# Return nil to disable the timeout, or 0 to use the client default.

Override timeout in the worker’s trait implementation:

fn timeout(&self, _job: &riverqueue::Job<MyArgs>) -> riverqueue::WorkerTimeout {
    riverqueue::WorkerTimeout::After(std::time::Duration::from_secs(30))
}
// WorkerTimeout::Disabled disables the timeout for this kind.

Pass a timeout when registering the handler to give that worker its own attempt limit:

workers.add(myJob, handler, { timeout: { seconds: 30 } });
// Use timeout: null to disable it for this worker.

Cancellation and shutdown

A running job can be cancelled remotely or interrupted during shutdown. Cooperate with the runtime’s cancellation mechanism and propagate interruption instead of reporting unfinished work as successful.

Job idempotency

A database’s ACID properties will ensure that any particular insert operation occurs exactly once or errors. This guarantee can be augmented with unique jobs so that only one job will be inserted based on a set of uniquely defining properties.

Once a job is inserted, it will be worked with at least once semantics. A successful job in a normal system will run exactly once in the overwhelming majority number of cases, but there are exceptions. If the job errors, it’ll be reworked until it succeeds, and there may be side effects between these runs. Rarer cases are possible too, like if a job finishes successfully, but fails to be marked as completed, in which case it’ll be rescued and run again.

Jobs should be written so that they can still succeed even if run multiple times, which is generally accomplished by making every operation in the job idempotent.

Idempotency with transactions

River workers have a shortcut to easier idempotency in that they can open a transaction in their work body, and in case of an error, roll changes back so that the next attempt will start with a clean slate.

Begin a transaction inside Work and defer rollback so early returns discard partial database changes. Commit only after the work succeeds:

func (w *TxWorker) Work(ctx context.Context, job *river.Job[TxArgs]) error {
    tx, err := w.dbPool.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    ...

    if err != nil {
        // Rollback occurs, reverting database changes made during the run.
        return err
    }

    return tx.Commit(ctx)
}

Wrap the worker’s database operations in a Sequel transaction so they commit or roll back together:

def work(job)
  DB.transaction do
    perform_database_work(job)
  end
end

Let errors escape the transaction so it rolls back.

Pass a transaction through the worker’s database operations and commit it only after they succeed:

let mut tx = riverqueue::database::begin_postgres(&pool).await?;
perform_database_work(&mut tx, &job).await?;
tx.commit().await?;

Use one checked-out connection for the transaction. Roll back on failure and release the connection in finally:

const tx = await pool.connect();

try {
  await tx.query("BEGIN");

  await performDatabaseWork(tx, job);

  await tx.query("COMMIT");
} catch (error) {
  await tx.query("ROLLBACK");
  throw error;
} finally {
  tx.release();
}

But it’s important to understand that even a transaction is not a full idempotency guarantee. Consider:

A work function may have called out to external systems while running like HTTP APIs or other non-transactional data stores (e.g. ElasticSearch). These changes will not roll back with a transaction.
To maximize work throughput, jobs are normally completed out-of-band of a work function, and in rare cases that completion could fail even if the worker succeeded and committed its transaction. See below for a possible remedy.

Transactions may also have performance and operational side effects. A long-lived transaction will tie up a Postgres backend while it’s open, and will contribute to database bloat as any rows still technically visible to the transaction have to be retained until it’s closed, even if they’ve since been deleted or updated to a new state. Workers may want to avoid opening a transaction when one isn’t needed, or when they’re expected to be long running.

Transactionally completing jobs alongside other changes

A way to hedge against possible failures while completing a job and get a little closer to an exactly once guarantee is to wrap operations in a transaction and complete the job as part of the same transaction using JobCompleteTx. A successful commit guarantees that the job will never rerun. A failed commit discards all changes so the next run starts fresh.

func (w *TxWorker) Work(ctx context.Context, job *river.Job[TxArgs]) error {
    tx, err := w.dbPool.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    ...

    _, err := river.JobCompleteTx[*riverpgxv5.Driver](ctx, tx, job)
    if err != nil {
        return err
    }

    return tx.Commit(ctx)
}

River normally completes jobs in large, efficient batches, so there is a throughput trade off to completing jobs in this way. Opening a transaction also has a small marginal cost, so that’ll add some overhead to the worker unless one was in use already.

See transactional job completion for more details.

Call context.job_complete_tx(&mut tx).await? after application writes and before committing the transaction. Propagate errors so a rollback is retried. See transactional job completion for the complete pattern.

Call await completeTx(tx) from the worker context after application writes and before COMMIT. Propagate errors after a rollback. See transactional job completion for the complete pattern.