Batching

Batching groups together many jobs of the same kind and executes them at the same time. A batch “leader” briefly waits for additional jobs that match the same batch key, then invokes your batch worker to process the group efficiently.

Batching is a feature of River Pro ✨. If you haven’t yet, install River Pro and run the pro migration line.

Basic usage

Enable batching for a job kind in three steps:

Implement BatchOpts() on your job args (alongside Kind()), returning a riverpro.BatchOpts to enable batching and configure how batches are formed.
Implement ManyWorker by adding a WorkMany method that processes a slice of jobs of the same kind.
In your Work method, delegate to the batching helper (riverbatch.Work) so a fetched job can prepare and execute a batch.

type MyBatchArgs struct{}

func (MyBatchArgs) Kind() string { return "my_batch" }

// Enable batching for this job kind. Customize options as needed.
func (MyBatchArgs) BatchOpts() riverpro.BatchOpts { return riverpro.BatchOpts{} }

type MyWorker struct {
    river.WorkerDefaults[MyBatchArgs]
}

func (w *MyWorker) Work(ctx context.Context, job *river.Job[MyBatchArgs]) error {
    // Invoke the batch helper so this job can gather a batch and run WorkMany.
    return riverbatch.Work[MyBatchArgs, pgx.Tx](ctx, w, job, nil)
}

func (w *MyWorker) WorkMany(ctx context.Context, jobs []*river.Job[MyBatchArgs]) error {
    // Process the entire batch at once.
    //
    // Return nil to mark the entire batch as successful, a regular error to
    // fail the entire batch with the same error, or a MultiError to return
    // individualized errors per job.
    return nil
}

Enable batching by implementing batch_opts on the argument class, then register a River::Pro::BatchWorker around a worker with a work_many(jobs) method:

class MyBatchArgs
  def kind = "my_batch"
  def to_json = JSON.generate({})
  def batch_opts = River::Pro::BatchOpts.new
end

class MyBatchWorker
  def work_many(jobs)
    # Process the entire array of jobs together.
    # Raise an exception to fail the entire batch, or return a
    # River::Pro::MultiError to fail only selected jobs.
    nil
  end
end

workers = River::Workers.new
workers.add("my_batch", River::Pro::BatchWorker.new(MyBatchWorker.new))

Pass these workers in the core configuration of a River::Pro::Client. The adapter implements the single-job work method and gathers the batch before calling work_many.

The batch Cargo feature is enabled by default. Set BatchOpts in the job type’s default insert options, then register a BatchWorker:

use riverqueue::{InsertOpts, Job, JobArgs, WorkContext, Workers};
use riverqueue_pro::batch::{BatchJobOutcome, BatchOutcomes, BatchWorker, batch_handler_fn};
use riverqueue_pro::{BatchOpts, ProInsertOptsExt as _};
use serde::{Deserialize, Serialize};
use std::convert::Infallible;

#[derive(Deserialize, JobArgs, Serialize)]
#[river(kind = "my_batch", insert_opts = Self::insert_opts)]
struct MyBatchArgs {}

impl MyBatchArgs {
    fn insert_opts() -> InsertOpts {
        InsertOpts::default().with_batch(BatchOpts::new())
    }
}

let handler = batch_handler_fn(
    |_context: WorkContext, jobs: Vec<Job<MyBatchArgs>>| async move {
        // Process the jobs together, then report an outcome for each job.
        Ok::<_, Infallible>(BatchOutcomes::map_jobs(&jobs, |_| {
            BatchJobOutcome::Complete
        }))
    },
);
let mut workers = Workers::new();
workers.add::<MyBatchArgs, _>(BatchWorker::new(handler))?;

Pass workers to the Pro client builder. The adapter gathers the batch before calling the handler. Alternatively, implement BatchHandler<MyBatchArgs>::work_batch on a worker type. Returning an error retries the whole batch; BatchOutcomes::map_jobs can return BatchJobOutcome::retry(error) for individual jobs while completing others. Every gathered job must have an outcome.

How batching works

When a batchable job is fetched, it becomes a leader and computes a batch key based on the job kind and configured options. The batch key is precomputed at insert time for efficient lookups.

Once a leader is fetched, it polls at the configured interval to find more jobs matching the same batch key until the maximum count, timeout, or deadline is reached. Once a full batch is fetched or either time limit is reached, the batch worker is invoked once with the collected jobs.

Multiple leaders with the same key

River Pro’s batching design is fully decentralized with no central coordinator. Multiple leaders with the same key may exist concurrently.

Operational considerations

The batch leader occupies a worker slot from the moment it is fetched until the batch finishes. This can make jobs run longer than expected while a leader waits to collect a full batch. Account for this when choosing the queue’s worker count and the maximum delay.

If partitioning by arguments, choose partitions that reflect isolation boundaries (e.g., per customer) to avoid over-aggregation and long waits.
Tune batch size, timeout, and poll interval to balance throughput and latency. Larger batches improve efficiency but increase tail latency for early jobs in the group.
A batch worker can report individual job failures while allowing the other jobs to succeed.

Configuration

Batches are always formed from jobs of a single kind, however their formation and partitioning can be further configured with riverpro.BatchOpts returned from BatchOpts() on your job args type.

Once a leader job is fetched, it will begin polling for more jobs that match the same batch key at a configured interval until either a maximum batch size is reached or a maximum wait time elapses. To customize the batching behavior, you can override the default options with riverbatch.WorkerOpts when calling the riverbatch.Work helper:

func (w *MyWorker) Work(ctx context.Context, job *river.Job[MyBatchArgs]) error {
    // Override the default batch options so that we wait up to 10 seconds
    // for a full batch before executing:
    return riverbatch.Work[MyBatchArgs, pgx.Tx](ctx, w, job, &riverbatch.WorkerOpts{
        MaxCount:     1000,             // 100 by default
        MaxDelay:     10 * time.Second, // 5 seconds by default
        PollInterval: 10 * time.Second, // 1 second by default
    })
}

Batches are gathered by a leader job for a short window: the leader polls for more jobs at a configured interval until either a maximum batch size is reached or a timeout elapses.

Common considerations when choosing options (see WorkerOpts):

Maximum batch size (MaxCount): upper bound on jobs collected per batch.
Batch timeout (MaxDelay): how long a leader waits to gather more jobs before executing.
Batch deadline (Deadline): an absolute time to stop gathering, as an alternative to a relative MaxDelay.
Poll interval (PollInterval): how frequently the leader checks for additional matching jobs while waiting.

Set Deadline to a time.Time, computed for each call to riverbatch.Work. For example, stop gathering 30 seconds after the leader job’s scheduled time, or after five seconds of gathering, whichever comes first:

func (w *MyWorker) Work(ctx context.Context, job *river.Job[MyBatchArgs]) error {
    return riverbatch.Work[MyBatchArgs, pgx.Tx](ctx, w, job, &riverbatch.WorkerOpts{
        Deadline: job.ScheduledAt.Add(30 * time.Second),
        MaxDelay: 5 * time.Second,
    })
}

Omit MaxDelay or leave it at zero to use only Deadline. A zero Deadline is unset; when both are unset, the default five-second MaxDelay applies.

Configure grouping with River::Pro::BatchOpts on the argument class, and collection timing on River::Pro::BatchWorker:

workers.add("my_batch", River::Pro::BatchWorker.new(
  MyBatchWorker.new,
  max_count: 1000,    # 100 by default, including the leader
  max_delay: 10,     # 5 seconds by default
  poll_interval: 10  # 1 second by default
))

max_count caps the number of jobs, max_delay bounds the time spent filling the batch, and poll_interval controls how often the leader looks for more jobs. Durations are in seconds. Set max_delay: 0 to work immediately with the available jobs.

For an absolute gathering cutoff, pass deadline: as a Time or a callable evaluated once per batch with the leader job. For example, stop gathering 30 seconds after the leader’s scheduled time, or after five seconds of gathering, whichever comes first:

workers.add("my_batch", River::Pro::BatchWorker.new(
  MyBatchWorker.new,
  deadline: ->(job) { job.scheduled_at + 30 },
  max_delay: 5
))

Omit max_delay to use only the deadline. A callable may return nil to leave the deadline unset for that batch; when neither limit is supplied, the default five-second delay applies. A fixed Time is reused across batches, so use a callable when the deadline should vary by leader.

Replace BatchWorker::new(handler) in the basic example with a configured worker:

use riverqueue_pro::batch::BatchWorker;
use std::{num::NonZeroUsize, time::Duration};

let worker = BatchWorker::builder(handler)
    .max_count(NonZeroUsize::new(1000).expect("nonzero batch size"))
    .max_delay(Duration::from_secs(10))
    .poll_interval(Duration::from_secs(10))
    .build()?;
workers.add::<MyBatchArgs, _>(worker)?;

Defaults are 100 jobs including the leader, a five-second gathering delay when no deadline or explicit delay is set, and a one-second poll interval. max_count must be nonzero, and the poll interval must be positive. A zero maximum delay checks for available peers once and then works the batch immediately.

Use .deadline(timestamp) for a fixed jiff::Timestamp, or compute a deadline once per batch from its leader job:

use riverqueue_pro::batch::BatchWorker;
use std::time::Duration;

let worker = BatchWorker::builder(handler)
    .deadline_func(|leader| Some(leader.row.scheduled_at + Duration::from_secs(30)))
    .max_delay(Duration::from_secs(10))
    .build()?;
workers.add::<MyBatchArgs, _>(worker)?;

The earlier of the deadline and an explicit max_delay applies. Returning None from the function leaves the deadline unset for that batch. An expired deadline gathers available jobs without waiting; a fixed timestamp is reused for every batch. The leader’s deadline applies to all peers, so group jobs with compatible cutoffs. A final fetch may finish after the cutoff; the deadline does not cancel work or guarantee when execution starts or finishes.

An expired deadline gathers available jobs without waiting for more. The batch may fill before the deadline, and a final database fetch may finish after it. The deadline only limits gathering: it does not cancel execution or guarantee when the batch starts or finishes. The leader’s deadline applies to every job in the batch, so group jobs with compatible deadlines.

Batching by arguments

Batches are always formed per job kind, meaning that when any batchable job is fetched, it will attempt to fill a batch with other available jobs of the same kind.

To achieve more granular batching, you can further partition by job arguments (via BatchOpts.ByArgs) to form independent batches per argument set (e.g., per customer or tenant).

When BatchOpts.ByArgs is enabled, the batch key includes all of the job’s encoded args (sorted prior to hashing). You can opt to include only a subset of args by marking fields on your JobArgs struct with the river:"batch" tag:

type MyBatchArgs struct {
    CustomerID string `json:"customer_id" river:"batch"`
    TraceID    string `json:"trace_id"`
}

func (MyBatchArgs) Kind() string { return "my_batch" }

func (MyBatchArgs) BatchOpts() riverpro.BatchOpts {
    return riverpro.BatchOpts{ByArgs: true}
}

Only customer_id will contribute to the batch key; jobs with different trace_ids but the same customer_id will be batched together.

Set by_args: true to group by all encoded arguments, or select specific fields with by_args: [:customer_id]:

MyBatchArgs = Struct.new(:customer_id, :trace_id, keyword_init: true) do
  def kind = "my_batch"
  def to_json = JSON.generate(to_h)

  def batch_opts
    River::Pro::BatchOpts.new(by_args: [:customer_id])
  end
end

Only customer_id contributes to the batch key, so jobs with different trace IDs can share a batch. Strings and symbols select JSON keys; nested fields use path arrays such as by_args: [[:customer, :id]].

Select encoded JSON fields with BatchOpts::with_arguments. ArgumentSelection::All includes every argument; components selects only the fields that define the batch:

use riverqueue::{InsertOpts, JobArgs};
use riverqueue_pro::{ArgumentSelection, BatchOpts, ProInsertOptsExt as _};
use serde::{Deserialize, Serialize};

#[derive(Deserialize, JobArgs, Serialize)]
#[river(kind = "my_batch", insert_opts = Self::insert_opts)]
struct MyBatchArgs {
    customer_id: String,
    trace_id: String,
}

impl MyBatchArgs {
    fn insert_opts() -> InsertOpts {
        InsertOpts::default().with_batch(BatchOpts::new().with_arguments(
            ArgumentSelection::components(&[&["customer_id"]]).expect("valid argument paths"),
        ))
    }
}

Only customer_id contributes to the batch key. Use &[&["customer", "id"]] to select a nested field. Names refer to serialized JSON keys, including any Serde renames.

Prefer specific args for batching

Choose stable identifiers such as a customer ID. Including timestamps or other unique values can create a separate batch for every job, preventing effective batching.

Avoiding redundant batches

Because leaders are decentralized, more than one leader may begin forming batches for the same key at the same time. If you want to avoid redundant overlapping batches, combine batching with a global concurrency limit of 1, partitioned by kind:

&riverpro.Config{
    ProQueues: map[string]riverpro.QueueConfig{
        "my_queue": {
            Concurrency: riverpro.ConcurrencyConfig{
                GlobalLimit: 1,
                Partition:   riverpro.PartitionConfig{ByKind: true},
            },
            MaxWorkers: 100,
        },
    },
}

River::Pro::Config.new(
  core: River::Config.new(workers: workers),
  pro_queues: {
    "my_queue" => River::Pro::QueueConfig.new(
      concurrency: River::Pro::ConcurrencyConfig.new(
        global_limit: 1,
        partition: River::Pro::PartitionConfig.new(by_kind: true)
      ),
      max_workers: 100
    )
  }
)

use riverqueue::{QueueConfig, database::PostgresDatabase};
use riverqueue_pro::{
    ConcurrencyConfig, ConcurrencyLimit, PartitionConfig, ProClient, QueueProConfig,
};

let concurrency = ConcurrencyConfig::global(ConcurrencyLimit::new(1)?)
    .with_partition(PartitionConfig::new().with_by_kind(true));
let client = ProClient::builder(PostgresDatabase::new(pool))
    .workers(workers)
    .queue(
        "my_queue",
        QueueProConfig::new(QueueConfig::new(100)).with_concurrency(concurrency),
    )
    .build()?;

If you’re also batching by arguments, your concurrency limits should also be partitioned by the same arguments.

See Concurrency limits for details.

Function-worker helpers

Instead of implementing a full worker type, you can use the riverbatch.WorkFunc or riverbatch.WorkFuncSafely helpers to define a batch worker with just a function for working the batch:

workers := river.NewWorkers()
river.AddWorker(workers, riverbatch.WorkFunc[MyBatchArgs, pgx.Tx](
    func(ctx context.Context, jobs []*river.Job[MyBatchArgs]) error {
        return nil // success
    },
    &riverbatch.WorkerOpts{MaxCount: 4, MaxDelay: time.Millisecond, PollInterval: 10 * time.Millisecond},
))

These are similar to work functions for regular workers.

A Deadline configured through these helpers remains fixed across calls. To compute a new deadline for each leader, use a Work method as shown in Configuration.

Pass a callable or block to River::Pro::BatchWorker to define the batch’s work without a separate worker class:

workers = River::Workers.new
workers.add("my_batch", River::Pro::BatchWorker.new(
  max_count: 4,
  max_delay: 0.001,
  poll_interval: 0.01
) do |jobs|
  # Process jobs together. Raise to fail the batch.
  nil
end)

The argument class still needs batch_opts to enable batching.

Use batch_handler_fn to adapt an async closure into a handler:

use riverqueue::{Job, WorkContext, Workers};
use riverqueue_pro::batch::{BatchJobOutcome, BatchOutcomes, BatchWorker, batch_handler_fn};
use std::{convert::Infallible, num::NonZeroUsize, time::Duration};

let handler = batch_handler_fn(
    |_context: WorkContext, jobs: Vec<Job<MyBatchArgs>>| async move {
        Ok::<_, Infallible>(BatchOutcomes::map_jobs(&jobs, |_| {
            BatchJobOutcome::Complete
        }))
    },
);
let worker = BatchWorker::builder(handler)
    .max_count(NonZeroUsize::new(4).expect("nonzero batch size"))
    .max_delay(Duration::from_millis(1))
    .poll_interval(Duration::from_millis(10))
    .build()?;
let mut workers = Workers::new();
workers.add::<MyBatchArgs, _>(worker)?;

Keep BatchOpts in MyBatchArgs’ default insert options as in the basic example.

Compatibility

Batching is compatible with all other River and River Pro features, including concurrency limits, workflows, and sequences.