postgleam/pool

A pool of PostgreSQL connections for running queries concurrently.

Each connection is its own actor, so queries on different connections run in parallel. A caller checks a connection out, uses it, and checks it back in; the pool then rolls back anything the caller left open before handing the connection to the next caller. If no connection is free the caller waits up to config.queue_timeout and then gets PoolTimeout.

import postgleam
import postgleam/config
import postgleam/decode
import postgleam/pool

pub fn main() {
  let assert Ok(cfg) = config.from_url("postgres://app:secret@localhost/app")
  let assert Ok(db) = pool.start(cfg, size: 10)
  let assert Ok(response) =
    pool.query_with(db, "SELECT $1::int4 + 1", [postgleam.int(41)], {
      use n <- decode.element(0, decode.int)
      decode.success(n)
    })
  pool.shutdown(db)
}

The pool also:

pings connections that have been idle for config.idle_interval ms, reconnecting broken ones before a caller needs them;
when it finds a broken connection, checks every idle connection at once, so sockets that died together (for example while the host was suspended) are replaced in one sweep instead of one failed query at a time;
replaces a connection whose actor crashed;
takes a connection back if the caller holding it exits.

An application that knows its sockets are dead, for example right after the machine resumed from a suspend, can call reconnect_all.

Types

Pool

opaque </>

A handle to a running pool.

pub opaque type Pool

PoolMessage

opaque </>

Messages the pool process handles. Only needed to name a pool (see supervised).

pub opaque type PoolMessage

Stats

</>

A snapshot of the pool, for monitoring.

pub type Stats {
  Stats(size: Int, idle: Int, busy: Int, waiting: Int)
}

Constructors

```
Stats(size: Int, idle: Int, busy: Int, waiting: Int)
```
Arguments

size

Configured number of connections.

idle

Connections free right now.

busy

Connections checked out, being reset after check-in, or being pinged or reconnected.

waiting

Callers waiting for a connection.

Values

named

</>

pub fn named(name: process.Name(PoolMessage)) -> Pool

A handle to the pool registered under name by supervised.

query

</>

pub fn query(
  pool: Pool,
  sql: String,
  params: List(option.Option(value.Value)),
) -> Result(connection.ExtendedQueryResult, error.Error)

Run a parameterized query on a pooled connection.

query_batch

</>

pub fn query_batch(
  pool: Pool,
  sql: String,
  param_sets: List(List(option.Option(value.Value))),
) -> Result(List(connection.ExtendedQueryResult), error.Error)

Run one statement with several parameter sets in one round trip.

query_one

</>

pub fn query_one(
  pool: Pool,
  sql: String,
  params: List(option.Option(value.Value)),
  decoder: decode.RowDecoder(a),
) -> Result(a, error.Error)

Run a query and decode the first row. Errors if there are no rows.

query_with

</>

pub fn query_with(
  pool: Pool,
  sql: String,
  params: List(option.Option(value.Value)),
  decoder: decode.RowDecoder(a),
) -> Result(postgleam.Response(a), error.Error)

Run a parameterized query on a pooled connection and decode each row.

reconnect_all

</>

pub fn reconnect_all(pool: Pool) -> Result(Nil, error.Error)

Reconnect every connection in the pool: idle ones now, and ones in use as soon as they are checked back in. The old sockets are closed without waiting for the server, so this is quick even when they are dead.

Use it when the application knows its connections cannot have survived, for example right after the machine resumed from a suspend. Returns once the pool has started the reconnects; callers that check out a connection meanwhile get one that has been reconnected.

shutdown

</>

pub fn shutdown(pool: Pool) -> Nil

Stop the pool and close all its connections. Callers still holding a connection get errors on their next call.

simple_query

</>

pub fn simple_query(
  pool: Pool,
  sql: String,
) -> Result(List(connection.SimpleQueryResult), error.Error)

Run a simple (text protocol) query on a pooled connection.

start

</>

pub fn start(
  config: config.Config,
  size size: Int,
) -> Result(Pool, error.Error)

Start a pool of size connections. All connections are opened before this returns, so a bad config or an unreachable server fails here. The pool is linked to the calling process.

stats

</>

pub fn stats(pool: Pool) -> Result(Stats, error.Error)

Get a snapshot of the pool’s state.

supervised

</>

pub fn supervised(
  config: config.Config,
  size: Int,
  name: process.Name(PoolMessage),
) -> supervision.ChildSpecification(Pool)

A child specification for running the pool under a supervisor, registered under name. Get a handle with named(name).

let name = process.new_name("db")
let assert Ok(_) =
  static_supervisor.new(static_supervisor.OneForOne)
  |> static_supervisor.add(pool.supervised(cfg, 10, name))
  |> static_supervisor.start
let db = pool.named(name)

transaction

</>

pub fn transaction(
  pool: Pool,
  f: fn(postgleam.Connection) -> Result(a, e),
) -> Result(a, postgleam.TransactionError(e))

Run f in a transaction on one pooled connection. See postgleam.transaction.

with_connection

</>

pub fn with_connection(
  pool: Pool,
  f: fn(postgleam.Connection) -> Result(a, error.Error),
) -> Result(a, error.Error)

Check out a connection, run f with it, and check it back in. The connection must not be used after f returns. Anything f leaves open (such as a transaction) is rolled back before the next caller gets the connection. If the calling process exits while holding the connection, the pool takes it back.

Constructors

Arguments