Resumable jobs

Resumable jobs break a worker into discrete steps so that if a job fails partway through, the next attempt can skip already-completed steps and pick up where it left off. This is useful for long-running or resource-intensive jobs with multiple stages, where re-running everything from scratch on each retry would be wasteful or cause unwanted side effects.

Basic steps

ResumableStep wraps a named unit of work. On the first attempt, all steps run in order. If a step returns an error, River records the last successfully completed step in the job’s metadata. On the next attempt, all steps up to and including the recorded step are skipped, and execution resumes at the next one.

type DataPipelineArgs struct {
    SourceURL string `json:"source_url"`
}

func (DataPipelineArgs) Kind() string { return "data_pipeline" }

type DataPipelineWorker struct {
    river.WorkerDefaults[DataPipelineArgs]
}

func (w *DataPipelineWorker) Work(ctx context.Context, job *river.Job[DataPipelineArgs]) error {
    river.ResumableStep(ctx, "download", nil, func(ctx context.Context) error {
        return downloadData(ctx, job.Args.SourceURL)
    })

    // If River was forced to stop between download and transform or midway into
    // transform, the next run of this job skips "download" and picks up here.
    river.ResumableStep(ctx, "transform", nil, func(ctx context.Context) error {
        return transformData(ctx)
    })

    river.ResumableStep(ctx, "load", nil, func(ctx context.Context) error {
        return loadData(ctx)
    })

    return nil
}

Wrap each stage in job.resumable_step with a stable name. River uses these names to track progress between attempts:

def work(job)
  job.resumable_step(:download) { download_data(job.args.fetch("source_url")) }
  job.resumable_step(:transform) { transform_data }
  job.resumable_step(:load) { load_data }
end

Call resumable_step with a name and an async operation for each stage. Propagate errors so a failed step stops the attempt:

context.resumable_step("download", || download_data(&job.args.source_url)).await?;
context.resumable_step("transform", transform_data).await?;
context.resumable_step("load", load_data).await?;

Step functions return futures resolving to Result<(), E>. Await steps sequentially and propagate their errors.

Use resumable.step from the work context to name each stage, and await the steps in order:

workers.add(dataPipeline, async ({ job, resumable }) => {
  await resumable.step("download", () => downloadData(job.args.source_url));
  await resumable.step("transform", () => transformData());
  await resumable.step("load", () => loadData());
});

If transform returns an error, download is recorded as completed. On the next attempt, download is skipped and execution begins at transform.

Step names must be unique within a worker. Code outside of steps still runs on every attempt, so generally speaking, all executable logic should be placed inside steps.

Cursor steps

ResumableStepCursor is a variant of ResumableStep that receives a cursor value from a previous attempt. This lets a step that processes items in a loop (and where each iteration of the loop is expensive enough to merit being skipped) resume from the last successfully processed item rather than starting over.

The cursor type is user-defined and can be any JSON-serializable value — a simple integer in its most basic form, or a struct with named fields for clarity or multiple cursors to handle nested loops.

type BatchProcessArgs struct {
    IDs []int `json:"ids"`
}

func (BatchProcessArgs) Kind() string { return "batch_process" }

type BatchCursor struct {
    LastProcessedID int `json:"last_processed_id"`
}

type BatchProcessWorker struct {
    river.WorkerDefaults[BatchProcessArgs]
}

func (w *BatchProcessWorker) Work(ctx context.Context, job *river.Job[BatchProcessArgs]) error {
    river.ResumableStepCursor(ctx, "process_ids", nil, func(ctx context.Context, cursor BatchCursor) error {
        for _, id := range job.Args.IDs {
            // Skip IDs that were already processed on a previous attempt.
            if id <= cursor.LastProcessedID {
                continue
            }

            if err := processID(ctx, id); err != nil {
                return err
            }

            // Record progress after each item so the next attempt can
            // resume after it.
            if err := river.ResumableSetCursor(ctx, BatchCursor{
                LastProcessedID: id,
            }); err != nil {
                return err
            }
        }

        return nil
    })

    return nil
}

On the first attempt, the cursor is the zero value of the cursor type. ResumableSetCursor records a checkpoint as progress is made. If the job fails, the cursor is saved in job metadata and passed to the step on the next attempt.

Steps and cursor steps can be freely mixed in the same worker. Steps are skipped or re-executed according to the same rules: completed steps are skipped, and a cursor step with saved cursor data is re-executed with that cursor.

See the ResumableStepCursor example for complete code.

A cursor can be any JSON value. Supply the initial value with default: and record progress after each item:

job.resumable_step_cursor :process_ids, default: 0 do |last_id|
  job.args.fetch("ids").each do |id|
    next if id <= last_id

    process_id(id)
    job.resumable_set_cursor id
  end
end

The cursor type implements Default and Deserialize; its default is used on the first attempt:

context.resumable_step_with_cursor("process_ids", |last_id: i64| async {
    for &id in &job.args.ids {
        if id <= last_id { continue; }

        process_id(id).await?;
        context.resumable_set_cursor(&id)?;
    }

    Ok::<(), riverqueue::BoxError>(())
}).await?;

The first attempt receives null. Check the JSON value’s shape before using it:

await resumable.stepWithCursor("process_ids", async (cursor) => {
  const lastId = typeof cursor === "number" ? cursor : 0;

  for (const id of job.args.ids) {
    if (id <= lastId) continue;

    await processId(id);
    resumable.setCursor(id);
  }
});

Nested loops

The cursor type is user-defined, so it can track progress across multiple levels of iteration. For a nested loop, define a struct with a field for each level:

type NestedCursor struct {
    LastProcessedID    int `json:"last_processed_id"`
    LastProcessedSubID int `json:"last_processed_sub_id"`
}

func (w *NestedWorker) Work(ctx context.Context, job *river.Job[NestedArgs]) error {
    river.ResumableStepCursor(ctx, "process_ids", nil, func(ctx context.Context, cursor NestedCursor) error {
        for _, id := range job.Args.IDs {
            if id < cursor.LastProcessedID {
                continue
            }

            for _, subID := range job.Args.SubIDs[id] {
                if id == cursor.LastProcessedID && subID <= cursor.LastProcessedSubID {
                    continue
                }

                if err := processSubID(ctx, id, subID); err != nil {
                    return err
                }

                if err := river.ResumableSetCursor(ctx, NestedCursor{
                    LastProcessedID:    id,
                    LastProcessedSubID: subID,
                }); err != nil {
                    return err
                }
            }
        }

        return nil
    })

    return nil
}

On resume, the outer loop skips fully completed IDs and the inner loop skips sub-IDs already processed within the current ID. Each call to ResumableSetCursor records both levels so no work is repeated.

Use a JSON object to track both levels of a loop. Keep the IDs sorted and skip only items at or before the recorded pair:

job.resumable_step_cursor :process_ids, default: {"id" => 0, "sub_id" => 0} do |cursor|
  ids.each do |id|
    next if id < cursor.fetch("id")

    sub_ids.fetch(id).each do |sub_id|
      next if id == cursor.fetch("id") && sub_id <= cursor.fetch("sub_id")

      process_sub_id(id, sub_id)
      job.resumable_set_cursor({id: id, sub_id: sub_id})
    end
  end
end

Use a serializable cursor struct with both positions:

#[derive(Default, serde::Deserialize, serde::Serialize)]
struct NestedCursor { id: i64, sub_id: i64 }

context.resumable_step_with_cursor("process_ids", |cursor: NestedCursor| async {
    for &id in &ids {
        if id < cursor.id { continue; }

        for &sub_id in &sub_ids[&id] {
            if id == cursor.id && sub_id <= cursor.sub_id { continue; }

            process_sub_id(id, sub_id).await?;
            context.resumable_set_cursor(&NestedCursor { id, sub_id })?;
        }
    }

    Ok::<(), riverqueue::BoxError>(())
}).await?;

Validate an object cursor to track both positions:

const cursorSchema = z.object({ id: z.number(), sub_id: z.number() });

await resumable.stepWithCursor("process_ids", async (value) => {
  const cursor = value === null ? { id: 0, sub_id: 0 } : cursorSchema.parse(value);

  for (const id of ids) {
    if (id < cursor.id) continue;

    for (const subId of subIds.get(id) ?? []) {
      if (id === cursor.id && subId <= cursor.sub_id) continue;

      await processSubId(id, subId);
      resumable.setCursor({ id, sub_id: subId });
    }
  }
});

Durable checkpoints

By default, checkpoints persist when an attempt ends, so a process crash can lose progress. Transactional checkpoints persist alongside application writes: either both commit, or both roll back. External side effects still need to be idempotent.

By default, step progress and cursors are saved to job metadata after the worker returns. This is efficient, but means that if the process crashes or is forcefully killed mid-step, the progress recorded by ResumableSetCursor may be lost.

ResumableSetStepTx and ResumableSetStepCursorTx provide a stronger guarantee by persisting step progress immediately as part of a database transaction. If the transaction commits, the checkpoint is durable. If it rolls back, the checkpoint is discarded along with the other work.

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

        // Do work within the transaction...
        if err := doWork(ctx, tx); err != nil {
            return err
        }

        // Persist step completion as part of the same transaction.
        if _, err := river.ResumableSetStepTx[*riverpgxv5.Driver](
            ctx, tx, job,
        ); err != nil {
            return err
        }

        return tx.Commit(ctx)
    })

    return nil
}

Use the transactional variants when a step performs database writes that must be atomic with the step’s completion record. For steps that are idempotent or where the default post-execution persistence is sufficient, ResumableStep and ResumableSetCursor are simpler and don’t require an extra database operation.

Start the application transaction inside the step and use the same connection for the checkpoint. Let rollback errors propagate:

job.resumable_step :durable_step do
  job.client.driver.transaction do
    do_work
    job.resumable_checkpoint
  end
end

For a cursor step, pass cursor: value. Without an explicit transaction the checkpoint is durable immediately, but is not atomic with separate application writes.

Pass the same transaction to the application writes and resumable_set_step_tx, then commit them together so the checkpoint survives a process crash:

context.resumable_step("durable_step", || async {
    let mut tx = riverqueue::database::begin_postgres(&pool).await?;

    do_work(&mut tx).await?;
    context.resumable_set_step_tx(&mut tx).await?;

    tx.commit().await?;
    Ok::<(), riverqueue::BoxError>(())
}).await?;

Inside a cursor step, use context.resumable_set_step_cursor_tx(&mut tx, &cursor).await?. Roll back the transaction if any operation fails or is cancelled.

Call resumable.checkpoint({ tx }) inside the transaction that performs the step’s database writes. The work and its checkpoint then commit together:

await resumable.step("durable_step", async () => {
  const tx = await pool.connect();

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

    await doWork(tx);
    await resumable.checkpoint({ tx });

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

Inside a cursor step, pass { tx, cursor: value }. Register workers as Workers<PoolClient> for a pg transaction.

Testing resumable workers

Worker test helpers can seed a checkpoint to exercise a worker partway through its steps. See testing for each language’s test setup.

This is particularly useful for making sure that jobs can run successfully even if they picked up midway. It’s relatively easy, for example, to accidentally hit a nil pointer exception because a step was skipped that would’ve initialized the pointer had the job run all steps from the beginning. Jobs with skippable steps need to have a fallback route to avoid that, and tests make sure that the fallback route works.

Testing from a specific step

ResumableStepAfter configures a job to skip all steps up to and including the named step, simulating a retry after that step completed successfully:

func TestWorker_ResumableStepAfter(t *testing.T) {
    result, err := rivertest.NewWorker(t, driver, config, &MyWorker{}).
        Work(ctx, t, tx, MyArgs{},
            rivertest.ResumableStepAfter(&river.InsertOpts{}, "download"))
    require.NoError(t, err)
    require.Equal(t, river.EventKindJobCompleted, result.EventKind)
}

This runs only the steps that come after download, letting you test later steps in isolation.

Seed river:resumable_step when inserting the test job, then use perform_job to exercise the worker with that checkpoint:

row = client.insert(MyArgs.new, metadata: {"river:resumable_step" => "download"}).job

result = River::Testing.perform_job(client, row.id)

expect(result.outcome).to eq(:completed)

Set checkpoint metadata on TestJobBuilder before passing the job to work_once:

let job = riverqueue_test::TestJobBuilder::new(MyArgs {})
    .metadata(riverqueue::serde_json::from_value(
        riverqueue::serde_json::json!({"river:resumable_step": "download"})
    )?)
    .build()?;

let worked = riverqueue_test::work_once(&MyWorker, job).await;

assert!(worked.result.is_ok());

Pass checkpoint metadata to testJob to exercise an existing resume point with workOnce:

const job = await testJob(myJob, {}, {
  metadata: { "river:resumable_step": "download" },
});

const result = await workOnce(job, workers);

expect(result.status).toBe("succeeded");

Testing with a cursor

ResumableStepAtCursor configures a job to resume at a cursor step with specific cursor data, simulating a retry after partial progress:

func TestWorker_ResumeWithCursor(t *testing.T) {
    result, err := rivertest.NewWorker(t, driver, config, &BatchProcessWorker{}).
        Work(ctx, t, tx, BatchProcessArgs{IDs: []int{1, 2, 3}},
            rivertest.ResumableStepAtCursor(&river.InsertOpts{}, "process_ids",
                BatchCursor{LastProcessedID: 1}))
    require.NoError(t, err)
    require.Equal(t, river.EventKindJobCompleted, result.EventKind)
}

This runs the process_ids step with a cursor indicating that ID 1 was already processed, so only IDs 2 and 3 are processed. All steps before process_ids are skipped.

Both helpers modify InsertOpts in place and return the same pointer, so they can be used inline as shown above, or applied to existing options:

opts := &river.InsertOpts{Queue: "custom"}
rivertest.ResumableStepAfter(opts, "step1")
result, err := tw.Work(ctx, t, tx, args, opts)

For a realistic resume test, run an attempt that fails after recording a cursor, then use River::Testing.perform_job on the same row again. Assert that already processed IDs are skipped. Use allow_scheduled: true to run a retry before its scheduled time.

Pass a failed attempt’s metadata_updates into the next TestJobBuilder with .metadata(...), then call work_once again. Assert that the worker resumes after the recorded item. This exercises the persisted cursor format rather than constructing it by hand.

Pass a failed workOnce result’s metadata into the next testJob call, then run workOnce again. Assert that the worker skips the items before the saved cursor.