Stuck jobs

Common causes include:

• Sending to or receiving from a channel when the corresponding receiver or sender has exited.
• Sending to or receiving from a nil channel.
• Ranging over a channel that is never closed.
• Waiting for a mutex that is never unlocked.
• Calling application or third-party code that blocks without observing the job context.

For example, this worker can wait forever if no value is sent to resultChan. The receive doesn’t observe context, so cancelling ctx won’t interrupt it:

func (w *ReportWorker) Work(ctx context.Context, job *river.Job[ReportArgs]) error {
    result := <-w.resultChan
    return saveReport(ctx, result)
}

A corrected form combines a select block with a receive on ctx.Done():

func (w *ReportWorker) Work(ctx context.Context, job *river.Job[ReportArgs]) error {
    select {
    case result := <-w.resultChan:
        return saveReport(ctx, result)
    case <-ctx.Done():
        return ctx.Err()
    }
}

See designing cancellable jobs for more details.

Ruby interrupts worker threads, but a native extension can block delivery of the interruption. Avoid swallowing interruption exceptions, put cleanup in ensure, and set timeouts on application I/O.

Await cancellation alongside long-running work. A yielding Tokio task can be aborted after the cancellation grace period; blocking native calls and CPU loops that never yield cannot be reliably interrupted. See cancellable jobs.

A handler that ignores its AbortSignal may continue indefinitely. Pass it to fetch and other cancellable operations, check it between work units, and yield to the event loop. CPU-bound handlers can use @riverqueue/worker-threads.

Configure JobStuckHandler on the client to report stuck jobs to logs or telemetry:

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    JobStuckHandler: func(ctx context.Context, params river.JobStuckHandlerParams) river.JobStuckHandlerResult {
        slog.ErrorContext(ctx, "River job is stuck",
            "job_id", params.ID,
            "job_kind", params.Kind,
            "queue", params.Queue,
            "total_stuck_jobs", params.TotalStuckJobs,
        )

        return river.JobStuckHandlerResult{}
    },
})

TotalStuckJobs is the number of jobs currently considered stuck across every queue in the client. A job that later returns is removed from the count.

Ruby does not expose Go’s stuck-job callback or replacement-slot policy. Use error handling, logging, and a supervised worker process with a shutdown deadline.

Implement ErrorHandler::handle_stuck to report attempts that outlive cancellation:

struct StuckHandler;

impl riverqueue::ErrorHandler for StuckHandler {
    async fn handle_stuck(&self, job: &riverqueue::JobRow) -> Result<(), riverqueue::BoxError> {
        eprintln!("Stuck job {} of kind {}", job.id, job.kind);
        Ok(())
    }
}

let client = riverqueue::Client::builder(pool).error_handler(StuckHandler).build()?;

Provide a stuckHandler to report jobs that have not returned after cancellation. Its context includes the job’s identity and the total number of stuck jobs:

const client = new Client(driver, {
  stuckHandler: ({ id, kind, queue, totalStuckJobs }) => {
    console.error("River job is stuck", { id, kind, queue, totalStuckJobs });
  },
});

Setting JobStuckHandlerResult.AddWorkerSlot to true in JobStuckHandler’s return value tells River to add one more worker slot for the queue where the stuck job is located, keeping execution capacity the same as before:

JobStuckHandler: func(_ context.Context, _ river.JobStuckHandlerParams) river.JobStuckHandlerResult {
    return river.JobStuckHandlerResult{
        AddWorkerSlot: true,
    }
},

Use AddWorkerSlot with care. River can open a replacement slot, but it can’t terminate the original goroutine. The stuck execution may continue retaining memory, locks, database connections, or other resources, and the queue may temporarily have more active job goroutines than its configured MaxWorkers. Replacing every stuck slot without a limit can eventually exhaust the process.

There is no public replacement-slot callback in Ruby. Diagnose the blocked operation and use supervised restarts when it cannot be interrupted.

Rust’s stuck handler observes the job; it does not return a replacement-slot decision. The runtime can abort yielding tasks after the configured threshold and finalize the failed attempt.

Return addWorkerSlot from the stuck handler to request replacement capacity. This example stops requesting replacements once the stuck-job count exceeds ten:

const client = new Client(driver, {
  stuckHandler: ({ totalStuckJobs }) => ({ addWorkerSlot: totalStuckJobs <= 10 }),
});

A replacement slot does not terminate the original handler. Bound replacements so stuck jobs cannot consume unlimited resources.

To preserve worker capacity without allowing resource use to grow indefinitely, pair AddWorkerSlot with a maximum number of stuck jobs. The following handler adds replacement slots up to a limit. Once the limit is exceeded, it asks the client to stop so the application can exit and its process supervisor can restart it. A supervisor like systemd or Kubernetes is required for this to work:

const maxStuckJobsBeforeRestart = 10

errTooManyStuckJobs := errors.New("too many stuck River jobs")

// Normal SIGINT/SIGTERM shutdown and stuck-job restarts both cancel clientCtx.
signalCtx, stopSignals := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)
defer stopSignals()

clientCtx, cancelClient := context.WithCancelCause(signalCtx)
defer cancelClient(nil)

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    JobStuckHandler: func(ctx context.Context, params river.JobStuckHandlerParams) river.JobStuckHandlerResult {
        if params.TotalStuckJobs <= maxStuckJobsBeforeRestart {
            return river.JobStuckHandlerResult{AddWorkerSlot: true}
        }

        slog.ErrorContext(ctx, "Too many stuck River jobs; restarting process",
            "total_stuck_jobs", params.TotalStuckJobs,
        )
        cancelClient(errTooManyStuckJobs)

        return river.JobStuckHandlerResult{}
    },
    SoftStopTimeout: 10 * time.Second,
})
if err != nil {
    return err
}

if err := riverClient.Start(clientCtx); err != nil {
    return err
}

<-clientCtx.Done()
stopSignals() // Restore default signal handling so a second signal exits immediately.

// Give River a chance to stop cleanly, but don't wait forever for stuck jobs.
select {
case <-riverClient.Stopped():
case <-time.After(30 * time.Second):
    slog.Warn("River client did not stop before restart deadline")
}

if cause := context.Cause(clientCtx); errors.Is(cause, errTooManyStuckJobs) {
    return cause // Produce a non-zero exit so the supervisor restarts the process.
}

return nil // Normal SIGINT/SIGTERM shutdown.

Cancelling the context passed to Start initiates River’s normal graceful stop, whether cancellation came from SIGINT/SIGTERM or JobStuckHandler. context.Cause distinguishes the stuck-job case so only it produces a non-zero exit for the supervisor to restart.

If permanently stuck jobs prevent the client from stopping, the application stops waiting after the deadline and exits anyway. Jobs left in running state are later recovered by River’s rescuer.

Run river worker under a supervisor and give it a finite stop timeout. See graceful shutdown. Jobs that cannot finalize before process exit are recovered later by the rescuer.

Use handle_stuck to alert or signal the application’s shutdown coordinator. Keep blocking work outside the Tokio runtime and supervise the process so a blocked native call cannot prevent a restart indefinitely.

Combine a bounded replacement policy with a request to the application’s shutdown coordinator:

const client = new Client(driver, {
  stuckHandler: ({ totalStuckJobs }) => {
    if (totalStuckJobs <= 10) return { addWorkerSlot: true };

    requestRestart(); // Application callback: request stop, then let the supervisor restart.
    return undefined;
  },
});

Do not wait for run.stop inside this callback. Signal an external coordinator, allow a bounded finalization period, and let the supervisor enforce the final deadline.