Stuck jobs

A job is considered stuck when it exceeds Config.JobTimeout, River cancels its context, and it still hasn't returned after Config.JobStuckThreshold (default 10 seconds). This is generally caused by a job waiting on a channel or mutex and not respecting context cancellations.

In Go, one goroutine can't forcibly terminate another, so River has limited options for resolving stuck jobs. See use of JobStuckHandler below.

How jobs become stuck

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.

Handling stuck jobs with JobStuckHandler

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.

Replacing a worker slot

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.

Capping replacement slots and restarting

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.