Skip to content
River Docs

PgBouncer

River supports PgBouncer. Insert-only clients can use transaction pooling. For running clients, a direct Postgres or session-pooled connection is the simplest configuration across languages because some runtime services use session-scoped database features.

Application queries inside workers can use a separate transaction-pooled connection pool, as shown below.

PgBouncer modes

As background, PgBouncer supports three modes:

  • Session pooling: Pooling occurs at the session level. When a client requests a session, it’s assigned a “real” Postgres connection, and is allowed to keep it until explicitly released.

  • Transaction pooling: Pooling occurs at the transaction level. When a client starts a transaction, it’s affixed to a single Postgres connection until the transaction commits or is rolled back.

  • Statement pooling: Pooling occurs at the statement level. Clients only stay affixed to a Postgres connection for as long as it takes to execute a single statement. Multi-statement transactions aren’t allowed.

Ways to use River client

An insert-only client is never started and needs transactions for features such as unique insertion. Statement pooling is unsuitable.

A running client claims jobs, handles queue control, and may become the maintenance leader. Go, Rust, and TypeScript use Postgres notifications unless configured for polling; Ruby currently polls. Disabling notifications does not by itself establish that every other session-scoped service will work through transaction pooling. Use direct or session-pooled connections for the runtime unless you have verified the full configuration.

Workers may use a separate pool for their own application queries. When an application transaction also inserts jobs, use an insert client with the matching driver and transaction connection.

A realistic scenario

A plausible scenario where River and PgBouncer are used together is to configure PgBouncer with transaction pooling, and use it for inserting all jobs and performing all work, but giving a raw Postgres connection pool to work coordinators.

Each running client can work many jobs concurrently, so a relatively small number of worker processes can handle substantial throughput. Size pools for the runtime’s reserved connections and the application queries the workers issue; increasing concurrency does not create additional database capacity.

Insert clients

To make this concrete, here’s an insert-only client like might be used in a web or API process, configured to use PgBouncer as its pool:

dbPool, err := pgxpool.New(ctx, os.Getenv("PGBOUNCER_DATABASE_URL"))
if err != nil {
    // handle error
}


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


// client NOT started
_, err = riverClient.Insert(ctx, DatabaseArgs{SQL: "SELECT 1"}, nil)
if err != nil {
    // handle error
}

Workers

Individual workers also use PgBouncer. They’re worked by a work coordinator, but don’t need to use the same database pool configuration as their progenitor:

type DatabaseArgs struct {
    SQL string `json:"sql"`
}


func (DatabaseArgs) Kind() string { return "database" }


type DatabaseWorker struct {
    river.WorkerDefaults[DatabaseArgs]
    dbPool *pgxpool.Pool
}


func (w *DatabaseWorker) Work(ctx context.Context, job *river.Job[DatabaseArgs]) error {
    _, err := w.dbPool.Exec(ctx, job.Args.SQL)
    if err != nil {
        return err
    }


    return nil
}
dbPool, err := pgxpool.New(ctx, os.Getenv("PGBOUNCER_DATABASE_URL"))
if err != nil {
    // handle error
}


workers := river.NewWorkers()
river.AddWorker(workers, &DatabaseWorker{dbPool: dbPool})

Work coordinator

The work coordinator needs a real Postgres database pool so it can use LISTEN/NOTIFY, but is configured with high concurrency so relatively few total coordinators are required:

dbPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL")) // NOT PgBouncer
if err != nil {
    // handle error
}


riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Queues: map[string]river.QueueConfig{
        river.QueueDefault: {MaxWorkers: 1_000},
    },
    Workers: workers,
})
if err != nil {
    // handle error
}


if err := riverClient.Start(ctx); err != nil {
    // handle error
}