core.database.postgres — pure-Verum PostgreSQL driver

core.database.postgres — codename spindle (postgres backend) — is a pure-Verum implementation of PostgreSQL's v3 wire protocol. Zero libpq, zero FFI: every byte that flows over the TCP socket is encoded and decoded in Verum.

The adapter implements the cross-vendor protocols defined in core.database.common.protocol (Adapter, Connection, affine Transaction, Pool, Row, Params) so handler code is portable across the SQLite / Postgres / MySQL backends — see the cross-vendor capability table.

Architectural layers

┌─────────────────────────────────────────────────────────────┐
│ L7  PUBLIC API           PgConnection, PgTransaction        │
├─────────────────────────────────────────────────────────────┤
│ L6  SESSION              cancel, listen/notify, copy        │
├─────────────────────────────────────────────────────────────┤
│ L5  EXTENDED PROTOCOL    Parse / Bind / Execute / Sync;      │
│                          per-connection prepared cache       │
├─────────────────────────────────────────────────────────────┤
│ L4  AUTH                 SCRAM-SHA-256-PLUS, channel bind    │
├─────────────────────────────────────────────────────────────┤
│ L3  WIRE FRAMES          frontend / backend / message types  │
├─────────────────────────────────────────────────────────────┤
│ L2  TLS                  via core.net.tls                    │
├─────────────────────────────────────────────────────────────┤
│ L1  TCP                  via core.net.tcp.TcpStream          │
└─────────────────────────────────────────────────────────────┘

Opening a connection

mount core.database.postgres.{PgConfig, PgConnection, connect};

let cfg = PgConfig.new()
    .with_host("localhost".into())
    .with_port(5432)
    .with_user("alice".into())
    .with_database("prod".into())
    .with_password_from_env("PGPASSWORD".into())?;

let mut conn = connect(&cfg)?;
let result = conn.simple_query(&"SELECT 1".into())?;

Affine PgTransaction

Same shape as the loom L7 Transaction (see database):

mount core.database.postgres.{
    PgTransaction, PgTxOpts,
    PgTkSerializable, PgAmReadOnly,
    begin_tx, begin_tx_with, begin_tx_serializable, begin_tx_read_only,
    commit_tx, rollback_tx,
    with_transaction, with_transaction_serializable,
    savepoint, release_savepoint, rollback_to_savepoint,
};

// Recommended — callback combinator.
with_transaction(&mut conn, |c| {
    c.execute(&"INSERT INTO orders (id, total) VALUES (1, 100)".into())?;
    c.execute(&"UPDATE accounts SET balance = balance - 100 WHERE id = 1".into())?;
    Ok(())
})?;

// Manual — affine handle the user must consume.
let tx = begin_tx_serializable(&mut conn)?;
conn.execute(&"...".into())?;
commit_tx(&mut conn, tx)?;     // or rollback_tx(...)

PgTxOpts builder selects isolation + access mode + DEFERRABLE:

Field	Values
isolation: PgTxKind	PgTkReadCommitted (default), PgTkRepeatableRead, PgTkSerializable
mode: PgAccessMode	PgAmReadWrite (default), PgAmReadOnly
deferrable: PgDeferrable	PgDfNone (default), PgDfDeferrable, PgDfNotDeferrable

Defence-in-depth: commit_tx checks conn.tx_status() (driven by every server-side ReadyForQuery) and refuses to fire if the connection is in TxFailedTransaction (Postgres requires ROLLBACK after error) or TxIdle (no tx in progress) — surfaces DbMisuse(...) rather than emitting the silent-warning NOTICE.

rollback_tx is tolerant of TxIdle — server-side may have auto-rolled, the affine consume always succeeds.

Query cancellation

Postgres protocol §6.1.7: cancellation must travel on a separate TCP connection (the original socket is held busy by the available query). The driver captures (process_id, secret_key) from the startup BackendKeyData frame and offers two entry points:

// Convenience — uses the connection's stored backend_key.
conn.cancel_running_query(&"localhost".into(), 5432)?;

// Standalone — for callers that hold the tuple outside a PgConnection
// (pool reservations, supervisor watchdogs, audit-log replay).
mount core.database.postgres.cancel_running_query_at;
cancel_running_query_at(&"localhost".into(), 5432, pid, secret_key)?;

Race semantics match Postgres: the server may complete the query before the cancel arrives, in which case the cancel is silently ignored. Don't treat cancel as guaranteed delivery.

LISTEN / NOTIFY

// Subscribe.
conn.listen(&"orders_channel".into())?;

// Publish (single-quote-escapes the payload per SQL standard).
conn.notify(&"orders_channel".into(), &"order_id=42".into())?;

// Drain all pending notifications.
let pending = conn.take_notifications();
for n in pending.iter() {
    let _channel = &n.channel;
    let _payload = &n.payload;
    let _sender_pid = n.process_id;
}

// Block until at least one arrives.
let batch = conn.wait_for_notification()?;

Implementation: read_message auto-demuxes BmNotification(...) frames into conn.notifications whenever the server pushes one between query rounds. Higher-level callers see only synchronous response streams; notifications accumulate transparently.

COPY FROM / TO — bulk-streaming

5–10× faster than even pipelined INSERT for bulk ingest because the server skips per-row Parse+Bind round-trips:

mount core.database.postgres.copy.{copy_in, copy_out};

// Bulk ingest — one row per List<Text> entry, tab-separated.
let n_rows = copy_in(&"events".into())
    .with_columns(["ts".into(), "user_id".into(), "kind".into()])
    .run(&mut conn, &rows)?;

// Bulk export.
let chunks = copy_out(&"users".into())
    .with_columns(["id".into(), "email".into()])
    .run(&mut conn)?;

Driver methods on PgConnection (lower-level):

• copy_in_bytes(sql, rows: &List<List<Byte>>) -> Result<Int, DbError>
• copy_in_text_lines(sql, rows: &List<Text>) -> Result<Int, DbError> — auto-newline-terminates
• copy_fail_in(reason: &Text) -> Result<(), DbError> — abort mid-stream
• copy_out_bytes(sql) -> Result<List<List<Byte>>, DbError>

SCRAM-SHA-256-PLUS authentication

core.database.postgres.auth.scram — pure-Verum SCRAM-SHA-256 client. Channel-binding (-PLUS) is preferred when a TLS layer is present. md5, password, trust flows are compile-time-rejected unless @allow_legacy_auth is opted in.

Extended protocol

core.database.postgres.extended covers Parse / Bind / Describe / Execute / Sync. Per-connection LRU prepared-statement cache (size default 256) sits in core.database.postgres.stmt_cache so repeated queries with the same fingerprint = hash(normalised_sql_ast) skip the Parse round-trip.

Wire-protocol surface

Frame builders + parsers live under core.database.postgres.wire:

Module	Surface
wire.frame	Frame { tag: Byte, payload: List<Byte> }; length-prefixed framing
wire.frontend	startup, simple_query, parse, bind, execute, sync, flush, terminate, cancel_request, copy_data, copy_done, copy_fail, SCRAM helpers
wire.backend	parse_backend(frame, formats) → BackendMessage; BmAuth / BmReadyForQuery / BmRowDescription / BmDataRow / BmCommandComplete / BmError / BmNotice / BmNotification / BmCopyInResponse / BmCopyOutResponse / BmCopyData / BmCopyDone / BmParseComplete / BmBindComplete / BmCloseComplete / etc.
wire.types	BE_* / FE_* byte constants, TX_* status bytes, AUTH_* SASL sub-codes

All decoded &arena references live in connection-arena; recv_buf / last_formats reset at connection drop.

Binary type codecs

core.database.postgres.codec carries the wire-level binary codec for every PostgreSQL built-in scalar plus arrays plus composites. The codec splits the encode and decode paths so parameter binding (Verum value → PG bytes) and row decoding (PG bytes → Verum value) can evolve independently per type.

The dispatch table supports_binary returns true for every type the binary path can carry; types not in the table fall back to the textual FmtText representation so the connection never gets stuck on an exotic OID.

Coverage matrix

PG type	Decode	Encode	Notes
BOOL / BYTEA	✓	✓	Pass-through.
INT2 / INT4 / INT8	✓	✓	Network-byte-order.
FLOAT4 / FLOAT8	✓	✓	IEEE 754.
TEXT / VARCHAR	✓	✓	UTF-8.
UUID	✓	✓	16 bytes.
DATE / TIME / TIMESTAMP[TZ]	✓	✓	µs-precision Postgres epoch.
INTERVAL	✓	✓	(months, days, µs) triple.
JSON / JSONB	✓	✓	JSONB version byte preserved.
INET / CIDR	✓	✓	Family-byte + prefix.
MACADDR / MACADDR8	✓	✓	6 / 8 bytes.
BIT / VARBIT	✓	✓	Length-prefixed.
NUMERIC	✓	✓ V0/V1	V0 integer fast-path; V1 text → binary via Decimal.
tsvector	✓	—	Decode-only V0; encode via to_tsvector server-side.
composite / record	✓	✓	(type_oid, ArenaSlice) per field; auto-encode from Verum record values is V1.
array<T> for every T above	✓	✓	Header + per-element dispatch.
range<T>	✓	✓	(lower, upper, flags).

The NUMERIC, tsvector, and composite codecs landed in the 2026-05-04 batch (changelog entries NUMERIC). With them every PG built-in scalar plus array plus composite has a wire codec; parameter binding no longer hits FmtText for any built-in type.

NUMERIC parameter binding

Three encoders cover the full surface:

// Integer fast-path — no Decimal allocation.
public fn encode_numeric_from_int(
    buf:   &mut WireBuf,
    arena: &Arena,
    n:     Int,
) -> Bool;

// V1 — non-integer NUMERIC values.
public fn encode_numeric_from_decimal(
    buf:   &mut WireBuf,
    arena: &Arena,
    d:     &Decimal,
) -> Bool;

// Convenience: Text → Decimal → encode.
public fn encode_numeric_from_text(
    buf:   &mut WireBuf,
    arena: &Arena,
    s:     &Text,
) -> Bool;

The decimal path bridges through core.text.numeric.decimal, so non-integer binding is full-fidelity (no rounding) for any value within the V0 i64-coefficient range.

tsvector decode

public type PgTsVector  is { lexemes: List<PgTsLexeme> };
public type PgTsLexeme  is { word: Text, positions: List<PgTsPosition> };
public type PgTsPosition is { position: Int, weight: Int };

public fn decode_tsvector(slice: &ArenaSlice) -> Result<PgTsVector, DbError>;

Constants PG_TS_WEIGHT_{D,C,B,A} (0=D, 1=C, 2=B, 3=A) match the PG source. Encode is a V1 follow-up — most callers ship Text and let PG run to_tsvector server-side.

composite codec

public type PgComposite      is { fields: List<PgCompositeField> };
public type PgCompositeField is { type_oid: Int, value: Maybe<ArenaSlice> };

public fn decode_composite(
    slice:    &ArenaSlice,
) -> Result<PgComposite, DbError>;

public fn encode_composite(
    buf:    &mut WireBuf,
    arena:  &Arena,
    fields: &List<PgCompositeField>,
) -> Bool;

Decoded fields surface as (type_oid, ArenaSlice) so callers dispatch via the existing per-OID decoder table sql# uses for top-level columns. Auto-encoding from a Verum record value (deriving type_oid and per-field encoder) is V1 and depends on the row-resolver.

Error model

Every wire-decode failure becomes BackendDecodeError; every high-level operation surfaces a unified core.database.common.error.DbError. See core.database.postgres.error.dberr_from_pg for the server-side ErrorResponse → DbError translation (preserves SQLSTATE class

• severity + message).

Typed row façade — core.database.postgres.row.Row

Row is a column-name-keyed view over the wire-level (RowDescription, List<WireValue>) pair returned by PgConnection.simple_query / execute_prepared_typed. Eight typed getters cover the registered postgres scalar / array families with per-OID guards, NULL discipline, and TEXT + BINARY wire-format coverage on every path.

mount core.database.postgres.row.Row;

let row: Row = pool.acquire().await?
    .query_one(&"SELECT id, name, active, created_at FROM users WHERE id = $1".into(),
               vec![Some(TvInt4(42))]).await?;

let id:         Int          = row.get_int("id")?;
let name:       Text         = row.get_text("name")?;
let active:     Bool         = row.get_bool("active")?;
let nickname:   Maybe<Text>  = row.get_text_opt("nickname")?;  // nullable column
let created_at: Rfc3339Time  = row.get_timestamp("created_at")?;

Getter coverage:

Method	OID family	Wire formats	NULL handling
get_text / get_text_opt	TEXT, VARCHAR, BPCHAR	UTF-8 (TEXT + BINARY converge)	_opt returns Ok(None)
get_bool	BOOL	t/f (TEXT) or 0x00/0x01 (BINARY)	error
get_int / get_int_opt	INT2, INT4, INT8	parse_int (TEXT) or sign-extending big-endian (BINARY)	_opt returns Ok(None)
get_bytes	BYTEA	\x<hex> per bytea_output=hex (TEXT) or raw (BINARY)	error
get_timestamp / get_timestamp_opt	TIMESTAMP, TIMESTAMPTZ	postgres YYYY-MM-DD HH:MM:SS+00 → RFC 3339	_opt returns Ok(None)
get_text_array	_text (1009), _varchar (1015), _bpchar (1014)	{a,b,"c"} literal (TEXT) or length-prefix (BINARY)	NULL elements collapse to ""

Plus introspection: column_count(), column_index(name) -> Maybe<Int>, is_null(name) -> Bool, Row.new(description, values) for adapter wiring. Errors via DbError.DecodeFailed { fault: DecodeFault { column_index, column_name, expected_type, found_oid, detail } } for type-family mismatches; DbError.Adapter("PG_ROW_COL_NOT_FOUND") for unknown column names.

NULL discipline. get_X errors when the column is SQL-NULL; get_X_opt returns Ok(None). Callers that query nullable columns MUST use the _opt form — using get_text on a NULL column is a programmer error.

Parameterised query API — AsyncPgPoolGuard

Four ergonomic methods on AsyncPgPoolGuard that route through prepare → execute_prepared_typed → close_prepared internally and project rows into the Row façade for downstream use:

let pool: AsyncPgPool = ...;
let conn = pool.acquire().await?;

// Full result set as Vec of Row.
let rows: List<Row> = conn.query(
    &"SELECT * FROM users WHERE active = $1".into(),
    vec![Some(TvBool(true))],
).await?;

// Exactly one row — errors with SQLSTATE 02000 (no_data) on zero.
let row: Row = conn.query_one(
    &"SELECT email FROM users WHERE id = $1".into(),
    vec![Some(TvInt4(user_id))],
).await?;

// Optional single row — Ok(None) on empty.
let row_opt: Maybe<Row> = conn.query_one_opt(
    &"SELECT email FROM users WHERE id = $1".into(),
    vec![Some(TvInt4(user_id))],
).await?;

// DML returning command_tag (e.g. "UPDATE 3").
let tag: Text = conn.execute_with_params(
    &"UPDATE users SET active = $1 WHERE id = $2".into(),
    vec![Some(TvBool(false)), Some(TvInt4(user_id))],
).await?;

Plus four parameter-less delegations forwarding to the inner AsyncPgConnection: execute(sql), simple_query(sql), ping(), prepare(sql).

Param encoding. Caller hands a List<Maybe<TypedValue>> — typically constructed via the Tv* constructors from core.database.postgres.wire.typed. None encodes as SQL NULL; Some(value) runs through the per-OID binary encoder and ends up in the Bind frame's parameter section.

Result projection. Every cell from execute_prepared_typed's List<List<TypedValue>> is mapped to a TEXT-format WireValue via the typed_row_to_wire_values helper, so the Row façade's TEXT-decode path handles both simple-query and prepared-statement-typed result sets uniformly.

What's NOT in this module (yet)

• TLS verify-full — TLS handshake works (PtmDisable mode), but verify-full mode currently rejects with a "TLS_PENDING" adapter error; depends on core.net.tls channel-binding wiring.
• Logical replication consumer (core.database.postgres.replication) — pgoutput-format CDC stream. Spec §6.1.8.
• Connection-pool resilience patterns (auto-reconnect + retry).
• pgvector typed codec.

See also: stdlib → database for the cross-vendor surface, and stdlib → database (mysql) for the sister implementation.