Database drivers

River’s database adapters let job insertion share application connections and transactions. Choose the adapter for your language and database.

Drivers wrap third party packages

The River Client takes a generic TTx type parameter representing the type of the transaction in use for functions like InsertTx and InsertManyTx. TTx is derived from the client’s driver, an agnostic interface to a third party package that provides access to the target database.

Most of the time, the only time code references a database driver is when it’s initializing a River client. NewClient takes a driver as its first parameter, and the driver wraps a database pool:

import "github.com/riverqueue/river"
import "github.com/riverqueue/river/riverdriver/riverpgxv5"

...

dbPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
    panic(err)
}
defer dbPool.Close()

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

See the InsertAndWork example for complete code.

Use riverqueue-sequel or riverqueue-activerecord, plus the application’s pg or sqlite3 gem:

require "riverqueue-sequel"

DB = Sequel.connect(ENV.fetch("DATABASE_URL"))
client = River::Client.new(River::Driver::Sequel.new(DB))

The Active Record equivalent is River::Driver::ActiveRecord.new. Pass connection_class: when using a particular Active Record connection pool.

River uses SQLx. Postgres is enabled by default; SQLite uses the sqlite Cargo feature. Use River’s SQLx re-export to keep transaction types compatible:

let pool = riverqueue::sqlx::PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
let client = riverqueue::Client::builder(pool).build()?;

Use @riverqueue/driver-pg for Postgres, @riverqueue/driver-sqlite for SQLite, or @riverqueue/driver-prisma for producer-only Prisma integration:

import { PgDriver } from "@riverqueue/driver-pg";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const client = new Client(new PgDriver(pool));

The application owns the pg pool. End it after stopping every client that uses it.

Using `riverdatabasesql`

The riverdatabasesql driver adapts a standard *sql.DB for use with River. This lets River share a database pool and transactions with other packages built around database/sql, including Bun and GORM. Clients using this driver use *sql.Tx as their transaction type.

Create a client by passing its *sql.DB to riverdatabasesql.New:

riverClient, err := river.NewClient(
    riverdatabasesql.New(sqlDB),
    &river.Config{
        Workers: workers,
    },
)
if err != nil {
    panic(err)
}

This constructor uses only the supplied database/sql pool. It’s appropriate for clients that only insert jobs, migration tools, or installations where LISTEN isn’t available. Clients that work jobs should generally also configure a Pgx listener.

In general, riverpgxv5 should be considered River’s main supported driver. Pgx is performant, feature complete, production hardened, and well maintained, while Go’s database/sql is broadly considered misdesigned, and too generic to provide access to important Postgres features.

Using `LISTEN`/`NOTIFY`

Go’s database/sql package doesn’t expose a way to implement Postgres’ LISTEN feature. A client created with riverdatabasesql.New therefore runs in “poll only mode”, polling periodically for newly available jobs.

Clients that work jobs can receive notifications by supplying a separate Pgx pool to riverdatabasesql.NewWithPgxListener:

listenerPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
    panic(err)
}
defer listenerPool.Close()

riverClient, err := river.NewClient(
    riverdatabasesql.NewWithPgxListener(sqlDB, listenerPool),
    &river.Config{
        Workers: workers,
    },
)
if err != nil {
    panic(err)
}

The database/sql pool continues to execute all database operations and transactions. The Pgx pool is used only to acquire dedicated connections for LISTEN, and both pools must connect to the same database and resolve the same schema. The caller owns and must close both pools.

River Pro provides the equivalent riverprodatabasesql.NewWithPgxListener constructor.

Ruby currently polls for jobs and queue control changes. It does not consume Postgres LISTEN notifications. Choose fetch polling intervals appropriate to your latency requirements.

The SQLx Postgres runtime uses notifications and polling. Use .without_notifications() on the client builder when notifications are unavailable. See PgBouncer for connection-pooling constraints.

The Postgres driver uses a dedicated connection for LISTEN, with polling as a fallback. Configure pollOnly: true on the client to disable the listener. See PgBouncer for pooling constraints.

Database drivers

Drivers wrap third party packages

Using riverdatabasesql

Using LISTEN/NOTIFY

Using `riverdatabasesql`

Using `LISTEN`/`NOTIFY`