riverbatch

Package riverbatch provides functionality for batching multiple River jobs together for execution as a single group. Batched jobs must all have the same kind, and are enabled with 3 steps:

1. The job args type must implement JobArgsWithBatchOpts with a `BatchOpts` method alongside the usual `Kind` method. Batching can be partitioned by job args (or a subset) if riverpro.BatchOpts.ByArgs is true.

2. The worker must implement ManyWorker with a `WorkMany` method for processing a batch of slice of jobs of the same kind.

3. In the worker's `Work` method, call the riverbatch.Work helper to perform batching.

For example:

import (
	"riverqueue.com/riverpro"
	"riverqueue.com/riverpro/driver/riverpropgxv5"
	"riverqueue.com/riverpro/riverbatch"
)

type MyBatchArgs struct {}

func (MyBatchArgs) Kind() string { return "my_batch" }
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 {
	return riverbatch.Work[MyBatchArgs, pgx.Tx](ctx, job, w, &w.opts)
}

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

See riverpro for a complete executable example.

Implementation notes

For efficient querying of partitioned batch jobs, batches utilize a key that is precomputed at insert time based on the job args and batch options.

Batching is performed lazily without any centralized coordination. When a batch job starts and the Work helper is invoked, the job acts as a leader and fetches available jobs with the same batch key, polling for more jobs at the specified interval until either the max count or batch timeout is reached. Once all jobs are fetched, the worker executes the batch with [ManyWorker.WorkMany].

With this architecture it is possible for multiple batches with the same key to be active at the same time. To avoid this, batching can be combined with a global concurrency limit of 1 and `ByKind: true` to ensure there are no redundant batches.

The batch leader job will occupy a worker slot from the moment it is fetched until the batch is completed. This may result in jobs running for longer than expected and should be considered when choosing `MaxWorkers`.