Transactional job completion

River supports completing jobs in the same transaction as other worker operations with a transactional completion API, guaranteeing that if the transaction successfully commits, the job will never rerun, even in the presence of intermittent failure.

Completing a job atomically

Normally, jobs are marked complete out-of-band from the work functions of those jobs. This is usually good because it lets River apply optimizations during completion so that queue throughput is as high as possible, but the downside is that there’s a small chance of failure between when a job’s work function is invoked and when a River client sets it as complete. Say its process is terminated at that exact instant for example.

Should a job fail to be set complete, it’ll be rescued and rerun. This duplicates work that was already successful, but is the only way River can guarantee that the job was run.

All workers should always expect to be run at least once anyway, but those that’d like to minimize the probability of an accidental rerun can use JobCompleteTx to mark a job as complete in the same transaction as other operations. A successful commit guarantees that the job will never rerun. A failed commit discards all other changes so the next rerun starts with a fresh slate.

type TxWorker struct {
    river.WorkerDefaults[TxArgs]
    dbPool *pgxpool.Pool
}

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

    ...

    _, err := river.JobCompleteTx[*riverpgxv5.Driver](ctx, tx, job)
    if err != nil {
        return err
    }

    if err = tx.Commit(ctx); err != nil {
        return err
    }

    return nil
}

See the CompleteJobWithinTx example for complete code.

Why the generic parameter?

JobCompleteTx is a top-level river package function instead of one on the client. This is for user convenience so that a River client doesn’t need to be threaded onto every worker, but that convenience requires a concession. Because JobCompleteTx doesn’t have a client pointer to work with, it needs to take a generic parameter of the River driver in use ([*riverpgxv5.Driver] above) to give it enough information to complete a job.

The injected driver will always be the same type as the driver passed into NewClient to create a River client.

Use context.complete(connection) inside the transaction containing your application writes:

record ConfirmOrderArgs(long orderId) {}

var confirmOrder = JobType.of("confirm_order", ConfirmOrderArgs.class);
var builder = Client.builder(database).worker(confirmOrder, context -> {
    context.transaction(connection -> {
        try (var statement = connection.prepareStatement(
                "UPDATE orders SET confirmed = true WHERE id = ?")) {
            statement.setLong(1, context.args().orderId());
            statement.executeUpdate();
        }

        context.complete(connection);
        return null;
    });
});

The transaction helper commits on return and rolls back on failure. Let failures propagate from the handler so River retries the job. context.complete includes buffered output and metadata; client.complete(connection, job) is also available when no work context is needed.

Call job.client.job_complete_tx(job) inside the transaction containing your application writes:

def work(job)
  DB.transaction do
    DB[:orders].where(id: job.args.fetch("order_id")).update(confirmed: true)

    job.client.job_complete_tx(job)
  end
end

With Active Record, use ApplicationRecord.transaction. Use the driver’s connection on the same thread, and propagate errors so a rollback is retried.

Use WorkContext::job_complete_tx with the transaction that contains the application writes:

let mut tx = riverqueue::database::begin_postgres(&pool).await?;

riverqueue::sqlx::query("UPDATE orders SET confirmed = true WHERE id = $1")
    .bind(job.args.order_id)
    .execute(&mut *tx)
    .await?;

context.job_complete_tx(&mut tx).await?;
tx.commit().await?;
Ok(riverqueue::WorkOutcome::Complete)

Propagate transaction errors so River retries the attempt. Dropping an uncommitted transaction rolls it back. This protects database writes in the transaction, not external side effects such as sending email.

Use completeTx from the worker context and parameterize the registry with the driver’s transaction type:

import type { PoolClient } from "pg";

const workers = new Workers<PoolClient>().add(confirmOrder, async ({ job, completeTx }) => {
  const tx = await pool.connect();

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

    await tx.query("UPDATE orders SET confirmed = true WHERE id = $1", [job.args.orderId]);
    await completeTx(tx);

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

A rollback must fail the worker too, so River retries it. Transactional completion protects writes in that transaction, not external side effects.