Error and panic handling

Failure in a job queue is inevitable. River provides an interface to handle errors and panics for custom application telemetry and to provide execution feedback.

Implementing an error handler

At sufficient scale, it’s inevitable that jobs will occasionally return errors, and even panic given unforeseen conditions. River has a comprehensive retry system to ensure that even in the presence of errors, no jobs are lost, and have a chance to be reworked in case of an intermittent failure or while a bug fix is deployed.

River rescues panics during work automatically, and will log information on any errors or panics that occur, but sophisticated applications will often want to hook up their own telemetry to handle these events. River provides the ErrorHandler interface to make this possible.

type ErrorHandler interface {
    // HandleError is invoked in case of an error occurring in a job. It's
    // used to add custom telemetry or provide feedback on an errored job.
    //
    // Context is descended from the one used to start the River client that
    // worked the job.
    HandleError(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult

    // HandleError is invoked in case of a panic occurring in a job. It's
    // used to add custom telemetry or provide feedback on a panicked job.
    //
    // Context is descended from the one used to start the River client that
    // worked the job.
    HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *ErrorHandlerResult
}

type ErrorHandlerResult struct {
    // SetCancelled can be set to true to fail the job immediately and
    // permanently. By default it'll continue to follow the configured retry
    // schedule.
    SetCancelled bool
}

Applications can provide an implementation for ErrorHandler and configure it when creating a client:

type CustomErrorHandler struct{}

func (*CustomErrorHandler) HandleError(ctx context.Context, job *rivertype.JobRow, err error) *river.ErrorHandlerResult {
    fmt.Printf("Job errored with: %s\n", err)
    return nil
}

func (*CustomErrorHandler) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *river.ErrorHandlerResult {
    fmt.Printf("Job panicked with: %v\n", panicVal)
    fmt.Printf("Stack trace: %s\n", trace)
    return nil
}

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

See the ErrorHandler example for complete code.

Pass an error handler to the client configuration. Exceptions also follow the normal retry policy:

config = River::Config.new(
  error_handler: ->(error, job) {
    warn "Job #{job.id} failed: #{error.message}"
    nil
  }
)

Implement ErrorHandler and attach it to the client builder. Returning Continue leaves River’s normal retry handling in place after logging the failure:

use riverqueue::{BoxError, ErrorHandler, ErrorHandlerDecision, JobRow, WorkContext, WorkResult};

struct CustomErrorHandler;

impl ErrorHandler for CustomErrorHandler {
    async fn handle_error(
        &self, _context: &WorkContext, job: &JobRow, result: &WorkResult,
    ) -> Result<ErrorHandlerDecision, BoxError> {
        eprintln!("Job {} failed: {:?}", job.id, result);
        Ok(ErrorHandlerDecision::Continue)
    }
}

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

Provide an errorHandler callback in the client configuration. It receives the job’s work context and the error, which you can send to your logger or error tracker:

const client = new Client(driver, {
  errorHandler: ({ job }, error) => {
    console.error(`Job ${job.id} failed`, error);
  },
});

Reacting to errors and panics

ErrorHandler also lets an implementation provide feedback to job execution. ErrorHandlerResult.SetCancelled can be set to permanently cancel the job (preventing any future retries):

func (*CustomErrorHandler) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *river.ErrorHandlerResult {
    fmt.Printf("Job panicked with: %v\n", panicVal)
    fmt.Printf("Stack trace: %s\n", trace)

    // Cancel the job to prevent it from being retried:
    return &river.ErrorHandlerResult{
        SetCancelled: true,
    }
}

Return :cancel from an error handler to stop retries for a permanent failure:

config = River::Config.new(
  error_handler: ->(error, _job) { :cancel if error.is_a?(PermanentError) }
)

Return Ok(ErrorHandlerDecision::Cancel) from handle_error when the failure should permanently cancel the job. Return Continue to retain the ordinary retry policy. The handler receives worker errors and panics through WorkResult.

Return { cancel: true } from the error handler for failures that should not be retried. This example distinguishes permanent errors from ordinary failures:

class PermanentError extends Error {}

const client = new Client(driver, {
  errorHandler: (_context, error) => {
    if (error instanceof PermanentError) return { cancel: true };
    return undefined;
  },
});