Graceful shutdown

The recommended approach for stopping a River client is the use of a cancellable context passed to Start, or an explicit call to Stop. In both approaches, configure SoftStopTimeout on the client, which controls how long running jobs have to finish normally (soft stop) before their work contexts are cancelled (hard stop).

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    SoftStopTimeout: 10 * time.Second,
    // ...
})

Use client.stop to stop fetching and wait for active work. For a dedicated worker process, the river worker command handles signals, a graceful timeout, and escalation to interruption.

Keep the RunHandle returned by start. Configure a grace period before cancellation:

let client = riverqueue::Client::builder(pool)
    .workers(workers)
    .queue("default", riverqueue::QueueConfig::new(10))
    .soft_stop_timeout(std::time::Duration::from_secs(10))
    .build()?;

let mut run = client.start()?;

Keep the handle returned by start and supply a graceful timeout when stopping:

const run = await client.start();

// Later: await run.stop({ timeout: { seconds: 10 } });

Cancel the context passed to Client.Start to initiate a soft stop. The client stops fetching new jobs and waits for running jobs to finish. If they don’t finish within SoftStopTimeout, their work contexts are cancelled. Wait on Client.Stopped to know when the client has fully stopped.

if err := riverClient.Start(ctx); err != nil {
    panic(err)
}

// ... later, cancel ctx to initiate stop ...

// wait for all jobs to complete
<-riverClient.Stopped()

This is the most elegant option and pairs naturally with signal.NotifyContext for signal handling (see realistic shutdown code below).

Ruby has no Go-style start context. Use the worker command’s signal handling, or call client.stop(wait: false) to request a drain and client.stop later to wait for cleanup.

Tie startup to a shutdown future, then wait for cleanup:

let mut run = client.start_with_graceful_stop(async {
    let _ = tokio::signal::ctrl_c().await;
})?;

run.wait().await?;

For a production Unix service, also handle SIGTERM, as shown below.

Install application signal handlers that call run.stop. The client does not install process-wide signal handlers itself. See the complete example below.

In some cases it may be easier to call Client.Stop explicitly. It initiates a soft stop and blocks until the client has stopped or the provided context is cancelled. Like above, if SoftStopTimeout is configured, jobs that don’t finish in time have their contexts cancelled.

ctx, cancel := context.WithTimeout(ctx, 20 * time.Second)
defer cancel()

// blocks up to 20 seconds (see the context above) before falling through
if err := riverClient.Stop(ctx); err != nil {
    panic(err)
}

This is useful when you need to stop the client at a specific point in your program rather than tying stop to the lifecycle of a context.

Call stop at the point where your application is ready to wait for its workers to finish:

client.stop

This stops fetching and waits for active jobs. Do not make a blocking stop call from one of the client’s own workers.

Await stop on the run handle returned by client.start() to wait for shutdown:

run.stop().await?;

The call waits for active attempts and completion persistence. run.stopper().stop() requests the same stop without waiting.

Pass a timeout to run.stop to give active handlers a limited grace period to finish:

await run.stop({ timeout: { seconds: 10 } });

After the grace period, remaining handlers receive an aborted signal. Stop still needs those handlers to settle.

Client.StopAndCancel skips the soft stop phase entirely and immediately cancels the work context of all running jobs. It still waits for jobs to return and persists their results so that cancelled jobs can be picked up by another client as soon as possible.

if err := riverClient.StopAndCancel(ctx); err != nil {
    panic(err)
}

This is a more manual option for cases where you want to force an immediate hard stop without configuring SoftStopTimeout. The use of StopAndCancel was necessary before the addition of SoftStopTimeout, but now it’s generally recommended to configure SoftStopTimeout and use one of the options above instead.

Use stop_and_cancel when shutdown should interrupt current work instead of waiting for it to finish normally:

client.stop_and_cancel

This interrupts active worker threads and waits for finalization. Successfully interrupted attempts become available again without consuming an attempt.

Call stop_and_cancel on the run handle to request cancellation immediately and wait for shutdown:

run.stop_and_cancel().await?;

This immediately signals running jobs. Workers should return WorkCancelled so interrupted work becomes available without consuming an attempt.

Select cancel mode to skip the grace period and signal active handlers immediately:

await run.stop({ mode: "cancel" });

This aborts active handlers immediately. Propagate the signal’s reason instead of converting shutdown into a permanent job cancellation.

The Go programming language is designed in such a way that no goroutine can kill another. Instead, concurrency constructs are used to pass messages to other goroutines that instruct them to terminate. One of those concurrency constructs are contexts, which are inherited across all components in a Go app in a tree structure, and can be used to pass information or a cancellation signal. If a context high up in the tree is cancelled, all inherited contexts are cancelled as well, which gives a Go program a way of stopping as all goroutines throughout the process respond to cancellation by exiting cleanly.

River is built entirely around the idea of context cancellation. Each worker’s Work function receives a context as its first argument, and is expected to needle that context into all subsequent invocations that it makes. In the event of a hard stop via StopAndCancel, the context is cancelled and active jobs are expected to notice and return.

Many of the low-level components in Go already respect context cancellation and will return an error naturally, so as long as user code is respecting returned errors it doesn’t need to do any additional work. For example, an HTTP request through net/http will return context.Canceled as long as the worker’s context was threaded into the request (be careful to use NewRequestWithContext instead of NewRequest):

resp, err := http.DefaultClient.Do(req)
if err != nil {
    return err // will return context.Canceled
}

The same generally goes for database drivers, SDKs, and other types of network communication. Context cancellation is respected at a low level, and will bubble back through user code with minimal effort.

However, there are cases where user code needs to be careful to respect context cancellation in its own right, especially around sends and receives on channels. Take the simplest example, a channel receive:

item := <-myChan // WRONG

A send on myChan might eventually be received by this code, but in the interim if the job’s context is cancelled, it won’t stop the job. This can be corrected by rewriting the code with a select to handle both conditions:

select { // RIGHT
    case item := <-myChan:
    case <-ctx.Done():
        return ctx.Err()
}

To ensure jobs can be cancelled quickly, all channel receives or sends on blocking channels should be in a select statement alongside a receive on ctx.Done().

In the event of a cancelled context, the code block above returns context.Canceled. This is to ensure that in the case of job cancellation, an error is written to the database and the job isn’t accidentally lost (returning a nil counts as a success). The job will be picked up by another client or the next time one is available.

Let interruption exceptions propagate, and release application resources with ensure. Do not rescue all exceptions and then return success:

def work(job)
  resource = open_resource

  begin
    perform_work(resource, job.args)
  ensure
    resource.close
  end
end

Set finite timeouts on blocking I/O. A native call that cannot be interrupted may prevent a timely stop.

Use tokio::select! to wait for either the work or the cancellation token. Return WorkCancelled if cancellation wins:

tokio::select! {
    () = context.cancellation_token().cancelled() => Err(riverqueue::WorkCancelled.into()),
    result = perform_work(&job.args) => {
        result?;
        Ok(riverqueue::WorkOutcome::Complete)
    }
}

Do not block a Tokio runtime thread indefinitely. Design cancelled futures so dropping them does not leave application state inconsistent; use transactions or idempotent operations for partially completed work.

Pass the worker’s AbortSignal to operations that accept it, and check it between operations so shutdown can interrupt the handler:

workers.add(download, async ({ job, signal }) => {
  const response = await fetch(job.args.url, { signal });
  if (!response.ok) throw new Error(`Download failed: ${response.status}`);

  const body = await response.text();
  signal.throwIfAborted();
  await saveDownload(body, { signal });
});

Pass the signal to APIs that support it. Check it between units of CPU work and yield so cancellation can be delivered. Do not swallow an abort and return success.

A goroutine can’t terminate another goroutine, so in the event of a job that doesn’t respect context cancellation, calls to Stop or StopAndCancel may hang forever.

River can detect stuck jobs with JobStuckHandler, temporarily replace their worker slots, and initiate a restart after a configured limit.

Robustly designed programs should either have a supervisor terminate a process stuck on Stop or StopAndCancel after an appropriate timeout, or stop waiting on them.

Care should be taken to try and prevent this from happening because failing to wait on stop runs the risk of River exiting uncleanly, meaning that it may not have been able to persist the result of running jobs as it’s shutting down, leaving them in running state. These jobs will eventually be rescued so they can be reworked, but not for an hour (see Config.RescueStuckJobsAfter), and their work will be considerably delayed.

Blocked native calls can delay Ruby thread interruption. Keep an external supervisor’s kill deadline as a backstop and let River finish persisting results whenever possible. If the process dies first, jobs left running depend on the rescuer.

River can abort a yielding Tokio task that ignores cancellation beyond the stuck threshold, but cannot force arbitrary blocking code to stop. Wait for the run handle to finish; use an external supervisor as a final deadline. See stuck jobs.

In-process handlers can ignore their abort signal indefinitely. run.stop cannot force them to settle. Use cooperative I/O, the optional worker-thread executor for CPU-bound work, and an external supervisor’s termination deadline. See stuck jobs.

A realistic shutdown waits for an external signal like SIGINT/SIGTERM and initiates stop, giving up if exiting cleanly is taking too long. SoftStopTimeout is configured on the client and a cancellable context (in this case from signal.NotifyContext) is sent to Start:

• SIGINT/SIGTERM cancels the start context, initiating a soft stop. Producers stop locking new jobs. Running jobs are given SoftStopTimeout to finish normally.
• If jobs don’t finish within SoftStopTimeout, the client cancels their work contexts (hard stop).
• A second SIGINT/SIGTERM exits the process immediately.

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Queues: map[string]river.QueueConfig{
        river.QueueDefault: {MaxWorkers: 100},
    },
    SoftStopTimeout: 10 * time.Second,
    Workers:         workers,
})
if err != nil {
    panic(err)
}

// Cancel the start context on SIGINT/SIGTERM. This initiates a soft stop,
// giving running jobs SoftStopTimeout to finish before their contexts are
// cancelled.
signalCtx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
defer stop()

if err := riverClient.Start(signalCtx); err != nil {
    panic(err)
}

<-signalCtx.Done()
stop()

// Initiated inside river.Client when signalCtx is cancelled.
fmt.Printf("Received SIGINT/SIGTERM; initiating soft stop (waiting for jobs to finish)\n")

// Wait for all jobs to complete.
<-riverClient.Stopped()

See also the graceful shutdown executable example, or a variant of it using StopAndCancel.

Have config/river.rb boot the application and return an unstarted client. Then run:

bundle exec river worker --config config/river.rb --stop-timeout 30

SIGINT or SIGTERM begins draining. A second signal or expiration of the grace period interrupts attempts. The runner allows a finalization window before forcing exit. See the worker command guide.

On Unix, handle both Ctrl-C and the termination signal sent by deployment systems:

let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;

let mut run = client.start_with_graceful_stop(async move {
    tokio::select! {
        _ = tokio::signal::ctrl_c() => {},
        _ = terminate.recv() => {},
    }
})?;

run.wait().await?;

Configure soft_stop_timeout on the client and give the supervisor additional time for finalization.

Handle process signals by stopping the run, then wait for run.completed before closing the database pool. Remove the signal listeners during cleanup:

const run = await client.start();

const stop = () => {
  void run.stop({ timeout: { seconds: 30 } }).catch((error) => {
    console.error("River stop failed", error);
    process.exitCode = 1;
  });
};

process.once("SIGINT", stop);
process.once("SIGTERM", stop);

try {
  await run.completed;
} finally {
  process.removeListener("SIGINT", stop);
  process.removeListener("SIGTERM", stop);

  await pool.end();
}

The application owns the pool. Close it after River stops, with a supervisor deadline longer than the grace period.

If there are jobs that don’t respect context cancellation (see stuck programs), the program blocks on Stopped() indefinitely. Most operating systems (on a host stopping) and deployment orchestrators (on a deploy/restart) will eventually time out and kill the process, but you can preempt that by providing your own timeout with a select block:

select {
case <-riverClient.Stopped():
    // client stopped successfully
case <-time.After(10 * time.Second):
    fmt.Fprintf(os.Stderr, "Jobs didn't respond to cancellation in time; exiting anyway\n")
}

The worker command applies a five-second finalization window after interrupting attempts. Give the process supervisor longer than the configured grace period plus that window.

A process supervisor should allow the grace period and completion persistence before killing the process. Dropping a run handle requests cancellation but does not wait for persistence; retain it and await shutdown whenever possible.

Use a process supervisor to terminate handlers that cannot settle after cancellation. Returning from a timeout race does not terminate those handlers, and closing the database pool while they are still finalizing can leave jobs running until rescue.