postgleam

A PostgreSQL driver for Gleam on the BEAM. The wire protocol, type codecs, authentication and connection pool are written in Gleam. The only Erlang code is a small module that sets up TLS through OTP’s ssl application (see What is not Gleam). No NIFs, no C, and no wrapping of another driver.

gleam add postgleam

A personal project. I built postgleam mainly for my own use. It’s tested and I use it, but there is no support or roadmap promise. Issues and pull requests are welcome, and so are forks if you want to take it in your own direction.

Quick start

import gleam/option.{type Option}
import postgleam
import postgleam/config
import postgleam/decode

pub type User {
  User(id: Int, name: String, email: Option(String))
}

pub fn main() {
  let assert Ok(cfg) =
    config.from_url("postgres://app:secret@localhost:5432/app_db")
  let assert Ok(conn) = postgleam.connect(cfg)

  let user = {
    use id <- decode.field("id", decode.int)
    use name <- decode.field("name", decode.text)
    use email <- decode.field("email", decode.optional(decode.text))
    decode.success(User(id:, name:, email:))
  }

  // Parameters are sent separately from the SQL, so this is injection safe.
  let assert Ok(response) =
    postgleam.query_with(
      conn,
      "SELECT id, name, email FROM users WHERE active = $1",
      [postgleam.bool(True)],
      user,
    )
  // response.rows: List(User)

  postgleam.disconnect(conn)
}

Features

• PostgreSQL wire protocol v3, binary format, with cached prepared statements.
• 35+ type codecs, including arrays (multi-dimensional too), json/jsonb, numeric, uuid, dates and times, intervals, geometric and network types. Types without a codec (ranges, hstore, enums, …) come back as text.
• TLS with certificate and host name verification, a custom CA file, and client certificates.
• SCRAM-SHA-256, MD5 and cleartext authentication, with a policy for which ones the client accepts.
• A connection pool that runs queries concurrently, with queue timeouts, health checks and supervision.
• Transactions, LISTEN/NOTIFY, COPY in/out, streaming large results, and logical replication.
• Recovery that keeps a connection correct after timeouts and errors.

Connecting

Use a URL or the builder functions.

import postgleam
import postgleam/config

// From a URL. Supported query parameters: sslmode (disable, require,
// verify-ca, verify-full), sslrootcert, connect_timeout (seconds).
// Anything else is sent to the server as a startup parameter.
let assert Ok(cfg) =
  config.from_url("postgres://app:secret@db.internal/app?sslmode=verify-full")

// Or with builders. default() is postgres@localhost:5432/postgres, no TLS.
let cfg =
  config.default()
  |> config.host("db.internal")
  |> config.database("app")
  |> config.username("app")
  |> config.password("secret")
  |> config.ssl(config.SslVerified)

let assert Ok(conn) = postgleam.connect(cfg)
postgleam.disconnect(conn)

A Connection is one PostgreSQL session, owned by an actor process that runs one request at a time. To run queries concurrently, use a pool.

Every call waits at most about config.timeout (15 s by default) for each reply. A query that runs longer is cancelled on the server and returns TimeoutError, and the connection reconnects before the next request. Calls never panic: if the connection’s process has died you get ConnectionError.

Queries

// Raw rows: List(List(Option(Value)))
let assert Ok(result) =
  postgleam.query(conn, "SELECT $1::int4 + 1", [postgleam.int(41)])

// Decoded rows
let assert Ok(response) = postgleam.query_with(conn, sql, params, decoder)

// Exactly the first row (an error if there are none)
let assert Ok(user) =
  postgleam.query_one(conn, "SELECT id, name, email FROM users WHERE id = $1",
    [postgleam.int(1)], user_decoder)

// One statement, many parameter sets, one round trip
let assert Ok(results) =
  postgleam.query_batch(conn, "INSERT INTO tags (name) VALUES ($1)", [
    [postgleam.text("gleam")],
    [postgleam.text("erlang")],
  ])

// Large results without loading them all: rows are fetched 500 at a time
let amount = {
  use n <- decode.element(0, decode.int)
  decode.success(n)
}
let assert Ok(total) =
  postgleam.query_fold(conn, "SELECT amount FROM payments", [], 500, amount,
    0, fn(sum, n) { sum + n })

// Text protocol: DDL, several statements at once, values come back as text
let assert Ok(_) =
  postgleam.simple_query(conn, "CREATE TABLE t (id serial); INSERT INTO t DEFAULT VALUES")

Decoders

Decode columns by name with decode.field or by position with decode.element. Errors name the column that failed.

let decoder = {
  use id <- decode.field("id", decode.int)
  use tags <- decode.field("tags", decode.list(decode.text))
  use deleted_at <- decode.field("deleted_at", decode.optional(decode.timestamp))
  decode.success(#(id, tags, deleted_at))
}

Parameters

postgleam.query(conn, "INSERT INTO users (name, age, email) VALUES ($1, $2, $3)", [
  postgleam.text("alice"),
  postgleam.int(30),
  postgleam.nullable(maybe_email, postgleam.text),  // None becomes NULL
])

Constructors include int, float, text, bool, null, nullable, bytea, uuid, uuid_string, json, jsonb, numeric, date, timestamp, timestamptz, time, interval, array and the geometric and network types. text works for any column type PostgreSQL can parse from text (enums, uuid, ranges, …). uuid_string returns a Result, so an invalid UUID is an error rather than a NULL.

Transactions

let result =
  postgleam.transaction(conn, fn(conn) {
    use _ <- result.try(
      postgleam.query(conn, "UPDATE accounts SET balance = balance - 10 WHERE id = 1", [])
      |> result.map_error(DbError),
    )
    use _ <- result.try(check_limits(conn))  // returns Result(_, MyError)
    Ok(Nil)
  })

case result {
  Ok(Nil) -> // committed
  Error(postgleam.RolledBack(my_error)) -> // your function returned Error; rolled back
  Error(postgleam.TransactionFailed(db_error)) -> // BEGIN, COMMIT or ROLLBACK failed
}

If the connection is lost in the middle of a transaction, the rest of the transaction fails with ConnectionError instead of running outside it, and transaction returns TransactionFailed.

Pool

import postgleam/pool

let assert Ok(db) = pool.start(cfg, size: 10)

let assert Ok(response) = pool.query_with(db, sql, params, decoder)

// Several statements on one connection
let assert Ok(n) =
  pool.with_connection(db, fn(conn) {
    use _ <- result.try(postgleam.query(conn, "SET LOCAL work_mem = '64MB'", []))
    postgleam.query_one(conn, report_sql, [], decoder)
  })

let assert Ok(_) = pool.transaction(db, fn(conn) { ... })

pool.shutdown(db)

Each pooled connection is its own actor, so queries from different processes run in parallel. When no connection is free a caller waits up to config.queue_timeout (5 s) and then gets PoolTimeout. When a connection comes back, anything left open (such as a transaction) is rolled back first; this also happens if the process holding it exits. Idle connections are pinged every config.idle_interval, and a crashed connection is replaced.

When the pool finds a broken connection (a ping or query timed out or the socket closed), it checks every idle connection at once and reconnects the ones that don’t answer. Sockets that died together, for example while the machine was suspended, are replaced in one sweep instead of one failed query at a time. Health-check pings wait at most config.connect_timeout. If your application knows its sockets are dead, such as right after a resume, call pool.reconnect_all(db). It reconnects idle connections right away, and ones in use as soon as they are checked back in.

Under a supervisor:

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)

pool.stats(db) reports idle, busy and waiting counts.

TLS and authentication

config.default()
|> config.ssl(config.SslVerified)          // verify certificate and host name
|> config.ssl_ca_file("/etc/ssl/db-ca.pem") // default: the system trust store
|> config.ssl_client_cert(certfile: "client.pem", keyfile: "client-key.pem")
|> config.allowed_auth([config.AuthScram])  // refuse anything weaker

• SslVerified checks the certificate chain and the host name (or, for an IP address, the certificate’s IP entries). Use it for anything that crosses a network.
• SslUnverified encrypts but accepts any certificate, so it only protects against passive eavesdropping.
• SslDisabled (the default) is plain TCP, for local development.
• Cleartext passwords are never sent over a connection without TLS, even if AuthCleartext is allowed.
• The password is stored as an opaque Secret, so it does not show up when a config is printed or included in a crash report.

LISTEN/NOTIFY

A listener owns a connection and sends every notification to a Subject. If its connection drops it reconnects and listens again (notifications sent while it was disconnected are lost, as PostgreSQL does not keep them).

import postgleam/listener

let inbox = process.new_subject()
let assert Ok(l) = listener.start(cfg, inbox)
let assert Ok(Nil) = listener.listen(l, "jobs")

let assert Ok(Nil) = postgleam.notify(conn, "jobs", "42")

let assert Ok(notification) = process.receive(inbox, 1000)
// notification.channel == "jobs", notification.payload == "42"
listener.stop(l)

COPY

let assert Ok("COPY 2") =
  postgleam.copy_in(conn, "COPY users (id, name) FROM STDIN", [
    <<"1\tAlice\n":utf8>>,
    <<"2\tBob\n":utf8>>,
  ])

let assert Ok(rows) = postgleam.copy_out(conn, "COPY users TO STDOUT")

PgBouncer

With config.pgbouncer(True) the driver uses only unnamed statements, which work with PgBouncer in transaction pooling mode. It costs one extra round trip’s worth of work per query (see the benchmarks).

Low-level API

postgleam/connection exposes the protocol directly on a ConnectionState value with no actor, for when you want full control. postgleam/stream, postgleam/copy, postgleam/notifications and postgleam/replication (logical replication) work on it too.

Benchmarks

Numbers from make bench-throughput on a 4-core Linux machine with PostgreSQL 16 on localhost, OTP 27. Every query is SELECT $1::int4, so they measure the driver and the round trip, not PostgreSQL’s work.

On one connection postgleam reaches about 85% of libpq’s speed (pgbench). Through the pool, throughput grows with the number of callers until the pool size or the server’s CPUs run out. On a real network the round trip dominates: expect each sequential query to take roughly one network round trip, and use the pool, query_batch or copy_in to do more per trip.

Version 0.9 fixed two large slowdowns found in review. Without TCP_NODELAY, Linux delayed ACKs held every extended query for about 40 ms (22 queries/s), and the old pool ran every query inside the pool process, one at a time, so ten callers got no more throughput than one.

The other benchmark suites:

make bench-pure          # codecs, message parsing, decoders (no database)
make bench-integration   # common operations against a database
make bench-realworld     # insert/query workloads modelled on go-sqlite-bench
make bench-throughput    # the table above

What is not Gleam

src/postgleam_ffi.erl upgrades the socket to TLS with OTP’s ssl module. Gleam has no TLS library for the BEAM, and building the ssl options (including the host name check for IP addresses) is clearer in Erlang. A few other calls go straight to the Erlang standard library with @external: crypto:pbkdf2_hmac and unicode:characters_to_nfkc_binary for SCRAM, and inet:setopts to turn on TCP_NODELAY.

Development

docker compose up -d            # PostgreSQL 16
./scripts/setup_test_db.sh      # create the test databases and roles
gleam test                      # 550+ tests (README examples are compiled too)
make check                      # format check, build, tests, docs

The TLS tests run when build/tls/ca.pem exists and the server uses the matching certificate (scripts/make_test_certs.sh creates them). The PgBouncer tests run when PgBouncer in transaction mode answers on localhost:6432. CI runs both.

Target

Erlang only. The JavaScript target is not supported, since the driver uses TCP sockets and OTP processes.