Testing

Job inserts are verified with the RequireInserted* family of helpers provided by river/rivertest. They’re designed to be symmetrical with the Client.Insert functions, so there’s variants to verify a single insert or many, and on a database pool or a transaction.

Check the insertion of a single job with RequireInsertedTx:

import (
    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riverpgxv5"
    "github.com/riverqueue/river/rivertest"
)

type RequiredArgs struct {
    Message string `json:"message"`
}

func (RequiredArgs) Kind() string { return "required" }

func TestInsert(t *testing.T) {
    ...

    tx, err := dbPool.Begin(ctx)
    if err != nil {
        // handle error
    }
    defer tx.Rollback(ctx)

    _, err = riverClient.InsertTx(ctx, tx, &RequiredArgs{ Message: "Hello."}, nil)
    if err != nil {
        // handle error
    }

    job := rivertest.RequireInsertedTx[*riverpgxv5.Driver](ctx, t, tx, &RequiredArgs{}, nil)
    fmt.Printf("Test passed with message: %s\n", job.Args.Message)
}

See the RequiredInserted example for complete code.

RequiredInsertedTx takes a testing.TB why is provided by the argument to any Go test or benchmark as its first parameter like t *testing.T. If the job was inserted, it returns the inserted job. If not, the test fails:

--- FAIL: TestRequireInsertedTx (0.00s)
    --- FAIL: TestRequireInsertedTx/FailsWithoutInsert (0.12s)
        rivertest.go:352:
                River assertion failure:
                No jobs found with kind: required

Load the optional RSpec integration and assert on jobs inserted by a block:

require "riverqueue/testing/rspec"

RSpec.configure { |config| config.include River::Testing::RSpec }

expect { enqueue_order(42) }.to insert_job(
  client, kind: :fulfill_order, args: {"order_id" => 42}
)

These helpers read real database rows. They do not migrate or clean up for you. Minitest users can load riverqueue/testing/minitest and include River::Testing::Minitest.

Add riverqueue-test as a development dependency, then assert against a client connected to an isolated test database:

enqueue_order(&client, 42).await?;

let job = riverqueue_test::require_inserted::<FulfillOrderArgs>(&client).await;
assert_eq!(job.args.order_id, 42);

The _tx variants inspect jobs inside an open transaction before it commits.

Install @riverqueue/test as a development dependency. Producer code can accept the InsertClient interface, which is implemented by the real client and this in-memory recorder:

import { createTestClient, requireInserted } from "@riverqueue/test";

const { client, insertions } = createTestClient();

await enqueueOrder(client, 42);

const job = requireInserted(insertions, fulfillOrder, { args: { order_id: 42 } });
expect(job.state).toBe("available");

This records insertions without a database. Use requireInsertedInDatabase when testing real persistence.

The RequireInserted* functions use arguments sent to them like &RequiredArgs{} only to extract an expected job kind that’s used to query for matching jobs. Any properties set in these job args are ignored. To check specific properties of an inserted job, assertions should be made against helper return values using built-in Go comparisons, or an assertion library of choice like testify/require.

import   "github.com/stretchr/testify/require"

...

job := rivertest.RequireInsertedTx[*riverpgxv5.Driver](ctx, t, tx, &RequiredArgs{}, nil)
require.Equal(t, "Hello.", job.Args.Message)

Pass argument expectations through args: using string keys to match the job’s JSON. A literal hash requires exact equality; a_hash_including checks only the fields that matter to the test:

expect { enqueue_order(42) }.to insert_job(
  client, args: a_hash_including("order_id" => 42)
)

The insertion helper returns a job with typed arguments. Use ordinary Rust assertions on job.args to check their values:

let job = riverqueue_test::require_inserted::<FulfillOrderArgs>(&client).await;
assert_eq!(job.args.order_id, 42);

Pass expected argument fields in the args option to requireInserted. It compares the supplied fields and leaves omitted fields unchecked:

const job = requireInserted(insertions, fulfillOrder, { args: { order_id: 42 } });

RequireInsertedOptions can be used to assert various insertion options like maximum number of attempts, priority, queue, and scheduled time. RequireInsert* functions takes them as an optional last parameter:

_ = rivertest.RequireInsertedTx[*riverpgxv5.Driver](ctx, t, tx, &RequiredArgs{}, &rivertest.RequireInsertedOpts{
    Priority: 1,
    Queue:    river.QueueDefault,
})

Add insertion options to the matcher to verify how the job was enqueued. This expectation checks both its destination queue and priority:

expect { enqueue_order(42) }.to insert_job(client, queue: :orders, priority: 1)

Build RequireInsertedOpts with the properties to check, then pass it to require_inserted_with:

let options = riverqueue_test::RequireInsertedOpts::new().with_queue("orders").with_priority(1);
let job = riverqueue_test::require_inserted_with::<FulfillOrderArgs>(&client, &options).await;

Include options such as queue and priority in the match object passed to requireInserted:

requireInserted(insertions, fulfillOrder, { queue: "orders", priority: 1 });

Similar to requiring a single insert, many insertions can be checked simultaneously for a nominal performance benefit:

jobs := rivertest.RequireManyInsertedTx[*riverpgxv5.Driver](ctx, t, tx, []rivertest.ExpectedJob{
    {Args: &FirstRequiredArgs{}},
    {Args: &SecondRequiredArgs{}},
    {Args: &FirstRequiredArgs{}},
})

for i, job := range jobs {
    fmt.Printf("Job %d args: %s\n", i, string(job.EncodedArgs))
}

See the RequiredManyInserted example for complete code.

RequireManyInserted* functions take a mixed set of job args of different kinds, so unlike the the single job check, they return JobRow instead of Job[T], so arguments will have to be unmarshaled from job.EncodedArgs to be inspected.

The slice of jobs returned is ordered identically to the []rivertest.ExpectedJob input slice.

When checking many jobs at once, RequireManyInserted* expects all jobs of any included kinds to be included in the expectation list. The snippet above passes if exactly two FirstRequiredArgs and one SecondRequiredArgs were inserted, but if a third FirstRequiredArgs was inserted in addition to the first two, it’d fail.

Use insert_jobs with count: to verify that the block enqueues the expected number of jobs matching a kind:

expect { enqueue_orders }.to insert_jobs(client, count: 3, kind: :fulfill_order)

Build an expectation list with ExpectedJob::of for each job kind. The list can include multiple jobs of the same kind:

use riverqueue_test::{ExpectedJob, require_many_inserted};

let jobs = require_many_inserted(&client, &[
    ExpectedJob::of::<FirstArgs>(),
    ExpectedJob::of::<SecondArgs>(),
    ExpectedJob::of::<FirstArgs>(),
]).await;

Pass an array of job expectations to requireManyInserted to check a batch of recorded insertions:

import { requireManyInserted } from "@riverqueue/test";

const jobs = requireManyInserted(insertions, [
  { job: firstJob }, { job: secondJob }, { job: firstJob },
]);

The examples above show requiring insertions on a transaction, but non-Tx variants are provided to check against a database pool instead:

_ = rivertest.RequireInserted(ctx, t, riverpgxv5.New(dbPool), &RequiredArgs{}, nil)

Because a driver is passed to the function directly, the [*riverpgx5.Driver] type parameter can be omitted as Go can infer it.

Insertion assertions query through the supplied client’s driver. To inspect rows already present, use:

expect(client).to have_job(kind: :fulfill_order, queue: :orders)

require_inserted queries through the client’s pool. require_inserted_tx queries through a supplied transaction, so uncommitted insertions can be tested.

Use requireInsertedInDatabase with a real client when the assertion should inspect persisted jobs instead of the in-memory insertion log:

import { requireInsertedInDatabase } from "@riverqueue/test";

const job = await requireInsertedInDatabase(client, fulfillOrder, { queue: "orders" });

Sophisticated users may want to make use of test transactions in their tests which are rolled back after each test case, insulating tests running in parallel from each other, and avoiding tainting global state. In such cases, it may not be convenient to inject a full database pool to River’s client. To support test transactions, River clients can be initialized without database pool by passing nil to their driver:

// a nil database pool is sent to the driver
riverClient, err := river.NewClient(riverpgxv5.New(nil), &river.Config{})

The initialized client is more limited, supporting only inserts on transactions with InsertTx and InsertManyTx. Calls to the non-transactional variants Insert and InsertMany, or trying to start the client with Start, will fail.

Run insertion and synchronous worker tests inside the test framework’s transaction, using the same database connection as River’s driver. River::Testing.perform_job stays on the calling thread. Sharing only a database URL is not enough to share a transaction.

Begin a test transaction with riverqueue::database::begin_postgres(&pool).await?, insert using .tx(&mut tx), and query it with the _tx test helpers. Roll it back after the test. Workers on other connections cannot see uncommitted jobs.

Pass the same pg transaction connection to application writes, { tx } insertions, and requireInsertedInDatabase(..., { tx }). Roll back and release the connection in cleanup. Worker processes cannot see those jobs until the transaction commits.

Test transactions aren’t always suitable, like when the code under test needs multiple connections, or where a test suite is trying to test River end-to-end in inserting a job and verifying that it’s worked all the way through by a listening client on the other side.

For these cases we’d recommend pgtestdb, which takes advantage of Postgres template databases to quickly clone a pristine archetype (~100 ms per clone). Tests using pgtestdb can use real connection pools, start River clients, and still run independently in parallel.

Install pgtestdb:

go get github.com/peterldowns/pgtestdb@latest

Then implement a pgtestdb migrator that runs River’s and your own migrations against the template database:

import (
    "context"
    "database/sql"
    "fmt"
    "testing"

    "github.com/jackc/pgx/v5/pgxpool"
    _ "github.com/jackc/pgx/v5/stdlib"
    "github.com/peterldowns/pgtestdb"
    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riverpgxv5"
    "github.com/riverqueue/river/rivermigrate"
)

var testDatabaseConfig = pgtestdb.Config{
    DriverName: "pgx",
    Host:       "localhost",
    Port:       "5432",
    User:       "postgres",
    Password:   "postgres",
    Database:   "postgres",
    Options:    "sslmode=disable",
}

type testDatabaseMigrator struct{}

func (testDatabaseMigrator) Hash() (string, error) {
    // pgtestdb rebuilds its template when this value changes. Bump it for any
    // new River or other internal migrations.
    return "river-migrations-v1", nil
}

func (testDatabaseMigrator) Migrate(
    ctx context.Context,
    _ *sql.DB,
    config pgtestdb.Config,
) error {
    dbPool, err := pgxpool.New(ctx, config.URL())
    if err != nil {
        return fmt.Errorf("opening migration pool: %w", err)
    }
    defer dbPool.Close()

    migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
    if err != nil {
        return fmt.Errorf("creating River migrator: %w", err)
    }

    if _, err := migrator.Migrate(ctx, rivermigrate.DirectionUp, nil); err != nil {
        return fmt.Errorf("running River migrations: %w", err)
    }

    //
    // You will also probably want to add your own non-River schema here!
    //

    return nil
}

func newTestPool(ctx context.Context, t *testing.T) *pgxpool.Pool {
    t.Helper()

    config := pgtestdb.Custom(t, testDatabaseConfig, testDatabaseMigrator{})

    dbPool, err := pgxpool.New(ctx, config.URL())
    if err != nil {
        t.Fatalf("opening test pool: %v", err)
    }
    t.Cleanup(dbPool.Close)

    return dbPool
}

func TestInsertOnFullDatabase(t *testing.T) {
    t.Parallel()

    dbPool := newTestPool(t.Context(), t)

    riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{})
    if err != nil {
        t.Fatal(err)
    }

    _, err = riverClient.Insert(
        t.Context(),
        &RequiredArgs{Message: "Hello."},
        nil,
    )
    if err != nil {
        t.Fatal(err)
    }
}

A few notes on use:

• Configure testDatabaseConfig for a local or CI Postgres server, keeping in mind that pgtestdb’s database user must be able to create databases and roles.
• Make sure to add your application’s migrations to testDatabaseMigrator.Migrate (only River’s are shown in this sample code).
• The value returned by testDatabaseMigrator.Hash controls reuse of the migrated template; update it whenever River is upgraded or application migrations are changed.

Use a disposable Postgres database/schema or an in-memory SQLite database. Migrate it before the test, configure a stopped client with registered workers, then use River::Testing.perform_job or River::Testing.drain. Start a real client only when testing consumer threads or maintenance; stop it before cleaning up.

Use a disposable Postgres database for SQLx integration tests, or a single-connection in-memory SQLite pool for SQLite tests. Migrate first and give parallel tests separate databases or schemas. Use work_with_client when the worker needs a real client or transactional completion; start the runtime for tests of polling and maintenance.

Use a real in-memory SQLite runtime for integration tests:

import { SqliteDriver } from "@riverqueue/driver-sqlite";
import { createMigrator } from "@riverqueue/migrate";

using driver = SqliteDriver.memory();
await createMigrator(driver).migrateUp();

const client = new Client(driver, {
  workers,
  leaderElectionDisabled: true,
  queues: { default: { maxWorkers: 1 } },
});

using events = client.subscribe({ kinds: ["job_completed"] });
await using run = await client.start();

const { job } = await client.insert(myJob, {});

const { value: event } = await events.next();
expect(event?.kind === "job_completed" && event.job.id).toBe(job.id);

await run.stop();

Use a disposable Postgres database when testing notifications, custom schemas, or other Postgres-specific behavior. Give asynchronous tests an overall timeout.

River defaults to producing informational logging, and some logging may be emitted while tests run. Although not hugely harmful, logging output won’t be indented to match other test output, and in the presence of t.Parallel() will be interleaved so as to become unusable for debugging.

To avoid these problems, we recommend bridging slog and testing using a package like slogt, which will send River’s log output to t.Log so it can be cleanly collated by Go’s test framework:

import "github.com/neilotoole/slogt"

func TestRiverInsertions(t *testing.T) {
  riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
      Logger: slogt.New(t),
      ...
  })
  ...
}

Pass a test logger through River::Config.new(logger: logger). For job-persisted logging, install the plugin on the client used by River::Testing.perform_job so the worker receives its normal logger.

River emits tracing events. Configure your test’s tracing subscriber to capture them; avoid installing a new global subscriber in each parallel test.

workOnce captures worker logs in its result. For a real client, supply a test logger or logger: false to silence client logs. Keep failures visible through outcome and event assertions.

Workers in River are plain old Go structs and can often be tested by directly invoking the Work function. They’re testable by initializing a worker, invoking its Work function, and checking against the results:

type SortArgs struct {
    // Strings is a slice of strings to sort.
    Strings []string `json:"strings"`
}

func (SortArgs) Kind() string { return "sort" }

type SortWorker struct {
    river.WorkerDefaults[SortArgs]
}

func (w *SortWorker) Work(ctx context.Context, job *river.Job[SortArgs]) error {
    sort.Strings(job.Args.Strings)
    fmt.Printf("Sorted strings: %+v\n", job.Args.Strings)
    return nil
}

func TestSortWorker(t *testing.T) {
    err := (&SortWorker{}).Work(context.Background(), &river.Job[SortArgs]{
        Args: SortArgs{Strings: []string{"whale", "tiger", "bear"}},
    })
    if err != nil {
        t.Fatal(err)
    }
}

A worker is an ordinary Ruby object. Extract business logic into methods that can be tested independently. Use River::Testing.perform_job for behavior that needs River’s job wrapper, timeouts, or persistence.

Build a test job and pass it to work_once with the worker under test. Inspect the returned result to check the attempt’s outcome:

let job = riverqueue_test::TestJobBuilder::new(MyArgs {}).build()?;

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

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

This captures output and resumable metadata but does not execute hooks, middleware, retries, or database persistence.

Create a test job with testJob, then run it once through the registered workers with workOnce:

import { testJob, workOnce } from "@riverqueue/test";

const job = await testJob(myJob, {});

const result = await workOnce(job, workers);

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

This runs the handler and captures output, metadata, logs, and follow-up insertions. It does not run the runtime’s middleware, hooks, scheduling, or persistence.

Some River features depend on the worker being run in a real worker context, or which depend on the database, and can’t be fully tested in isolation. Examples include river.JobCompleteTx and recorded output. To comprehensively test workers, River provides powerful test helpers to exercise workers using a real work context and River’s internal execution logic.

These features are exposed through the rivertest.Worker type, which must be initialized for the specific worker type being tested:

import (
    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riverpgxv5"
    "github.com/riverqueue/river/rivertest"
)

var (
    config = &river.Config{}
    driver = riverpgxv5.New(nil)
    worker = &MyWorker{}
)
testWorker := rivertest.NewWorker(t, driver, config, worker)

The testWorker can then be used to execute as many jobs as desired:

result, err := testWorker.Work(ctx, t, tx, MyJobArgs{CustomerID: 123}, nil)
require.NoError(t, err)
require.Equal(t, river.EventKindJobCompleted, result.EventKind)
require.Equal(t, rivertype.JobStateCompleted, result.Job.State)

The Work function inserts a real job within the provided transaction, executes it, and records its result in the same transaction. The error return value includes any “real” errors the worker returned (excluding intentional snooze and cancel errors), as well as any recovered panics. The WorkResult struct includes the overall execution result (completed, failed, etc) as well as the final job row recorded in the database following execution.

To test execution of an existing job, use WorkJob:

job := client.InsertTx(ctx, tx, args, nil)
// ...
result, err := testWorker.WorkJob(ctx, t, tx, job.JobRow)

Load riverqueue/testing, insert a job, and execute it synchronously with perform_job. The result exposes the outcome, error, and updated job row:

require "riverqueue/testing"

row = client.insert(MyArgs.new).job

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

expect(result.outcome).to eq(:completed)
expect(result.error).to be_nil
expect(result.job.state).to eq("completed")

The client must be stopped and have the worker registered. This executes a real synchronous attempt with plugins, retries, output, and finalization. Worker exceptions are returned in result.error, so always assert the outcome. Your test owns database cleanup.

riverqueue_test::work_with_client inserts and claims a job and runs its worker with a real client in WorkContext. It does not persist ordinary completion for you; the row stays running unless the worker completes itself transactionally. Use a started client and subscriptions to test full completion persistence. See the test crate guide.

Pass a real client to workOnce when a handler needs to insert follow-up jobs into a database. For hooks, middleware, timeouts, retries, and completion persistence, use a started client as in the integration example above. The lightweight helper intentionally does not reproduce the entire runtime.