From 6db50516b7c073526590a4b41ee5fa60880fceea Mon Sep 17 00:00:00 2001 From: ekklesia Date: Wed, 7 Oct 2026 16:01:46 -0500 Subject: [PATCH 1/2] feat: Core's Module Database ABI v1 executor, proven against PostgreSQL (ADR 0050) Implements the first ADR 0050 slice: Core's ORM-neutral executor and its PostgreSQL evidence. Nothing is wired to modules yet; manifest v8, SDK adapters, schema metadata, packaging, and Signage are later slices. The legacy ModuleDatabase facade is unchanged. Contracts (@ekklesia/contracts, no imports at all): - ABI version 1; query { text, values, rowMode } returning { command, rowCount, rows, fields: [{ name, typeOid }] }; - executor, transaction, and savepoint types with exactly the ABI's members; parameter and result value types; - ModuleDatabaseError { code, sqlState? }: eight stable codes, and a SQLSTATE only when PostgreSQL supplied one; static messages, no cause, frozen; isModuleDatabaseError recognises it by shape. Core (createModuleDatabaseExecutor over the API's existing pool): - one statement per query, enforced by forcing the extended query protocol so PostgreSQL itself refuses a second statement (42601); no SQL parsing; - parameters: null, boolean, number, bigint, string, Uint8Array, and rectangular arrays of them; everything else refused before sending; - a Core-owned decoding table that never consults pg's global parser registry: int8, numeric, temporal, and JSON as exact text; bytea as a plain Uint8Array; built-in arrays element by element; - every operation pins TimeZone, DateStyle, IntervalStyle, bytea_output, and extra_float_digits, then resets with DISCARD ALL before returning the connection, or destroys it when it cannot be reset cleanly; - callback-scoped transactions with explicit isolation and access mode; a COMMIT that PostgreSQL answers with ROLLBACK is reported as failed; - savepoints named by Core; only the innermost scope is active; scopes are revoked when their callback settles; - a callback that settles with work outstanding is rolled back, never committed or released; - transaction-control statements sent through query are refused by command tag and transaction status, and fail the transaction; - the module ID is carried for Core's attribution only. Tests: - 83 integration tests on a database with hostile defaults: queries, the one-statement rule, parameters, decoding, errors, transactions, savepoints, forgotten awaits, pool safety after every failure class, and session state; - unit tests for encoding, decoding, options, sanitization, and scope bookkeeping; - type-level and source checks that no ORM or driver object crosses the boundary. ADR 0050 is Accepted. Co-Authored-By: Claude Opus 5.5 --- core/src/modules/database-abi-values.ts | 349 +++++ core/src/modules/database-abi.ts | 563 +++++++ core/src/modules/index.ts | 1 + core/tests/module-database-abi.integration.ts | 1326 +++++++++++++++++ core/tests/module-database-abi.test.ts | 833 +++++++++++ .../decisions/0050-module-database-abi-v1.md | 2 +- docs/architecture/decisions/README.md | 2 +- docs/architecture/platform-architecture.md | 6 +- docs/development/testing.md | 25 + packages/contracts/src/index.ts | 1 + packages/contracts/src/module-database-abi.ts | 217 +++ .../tests/module-database-abi.test.ts | 176 +++ vitest.migrations.integration.config.ts | 1 + 13 files changed, 3498 insertions(+), 4 deletions(-) create mode 100644 core/src/modules/database-abi-values.ts create mode 100644 core/src/modules/database-abi.ts create mode 100644 core/tests/module-database-abi.integration.ts create mode 100644 core/tests/module-database-abi.test.ts create mode 100644 packages/contracts/src/module-database-abi.ts create mode 100644 packages/contracts/tests/module-database-abi.test.ts diff --git a/core/src/modules/database-abi-values.ts b/core/src/modules/database-abi-values.ts new file mode 100644 index 0000000..3259f0f --- /dev/null +++ b/core/src/modules/database-abi-values.ts @@ -0,0 +1,349 @@ +import { + type ModuleDatabaseField, + ModuleDatabaseError, + type ModuleDatabaseValue, +} from "@ekklesia/contracts"; + +/** + * The value contract of Module Database ABI v1 (ADR 0050): how parameters are encoded before they + * reach PostgreSQL, and how result text is decoded after. Both are Core's, and neither consults + * `pg`'s process-global type registry or any ORM, which module code could change or differ from. + * + * Every value travels in PostgreSQL's text format. The session settings the decoding depends on + * (`DateStyle`, `IntervalStyle`, `TimeZone`, `bytea_output`, `extra_float_digits`) are pinned by the + * executor for every ABI operation (`./database-abi.ts`). + */ + +/** PostgreSQL's limit on bound parameters in one statement (the protocol's 16-bit count). */ +export const MODULE_DATABASE_MAX_PARAMETERS = 65_535; +/** PostgreSQL's limit on array dimensions (`MAXDIM`). */ +const maximumArrayDimensions = 6; + +function parameterInvalid(): ModuleDatabaseError { + return new ModuleDatabaseError("MODULE_DATABASE_PARAMETER_INVALID"); +} + +function isScalar(value: unknown): boolean { + return ( + value === null || + typeof value === "boolean" || + typeof value === "number" || + typeof value === "bigint" || + typeof value === "string" || + value instanceof Uint8Array + ); +} + +function scalarText(value: boolean | number | bigint | string): string { + // `String` spells NaN, Infinity, and -Infinity exactly as PostgreSQL's float and numeric input + // accepts them. + return typeof value === "string" ? value : String(value); +} + +function byteaText(value: Uint8Array): string { + return `\\x${Buffer.from(value.buffer, value.byteOffset, value.byteLength).toString("hex")}`; +} + +function quoteArrayElement(text: string): string { + return `"${text.replaceAll("\\", "\\\\").replaceAll('"', '\\"')}"`; +} + +/** + * Validates one array level and returns its PostgreSQL array literal. Every element of a level is + * either a sub-array or a scalar, and sibling sub-arrays have equal lengths, because PostgreSQL + * arrays are rectangular. Every non-null element is quoted, which PostgreSQL accepts for any element + * type. + */ +function arrayLiteral(values: readonly unknown[], depth: number): string { + if (depth > maximumArrayDimensions) { + throw parameterInvalid(); + } + const elements: string[] = []; + let nestedLength: number | undefined; + let nested: boolean | undefined; + for (let index = 0; index < values.length; index += 1) { + if (!Object.hasOwn(values, index)) { + throw parameterInvalid(); + } + const element = values[index]; + const isNested = Array.isArray(element); + if (nested === undefined) { + nested = isNested; + } else if (nested !== isNested) { + throw parameterInvalid(); + } + if (isNested) { + if (nestedLength === undefined) { + nestedLength = element.length; + } else if (nestedLength !== element.length) { + throw parameterInvalid(); + } + elements.push(arrayLiteral(element, depth + 1)); + } else if (element === null) { + elements.push("NULL"); + } else if (element instanceof Uint8Array) { + elements.push(quoteArrayElement(byteaText(element))); + } else if (isScalar(element)) { + elements.push(quoteArrayElement(scalarText(element as boolean | number | bigint | string))); + } else { + throw parameterInvalid(); + } + } + return `{${elements.join(",")}}`; +} + +/** + * Validates `values` against the ABI's parameter set and encodes each one into the only forms the + * driver then passes through unchanged: `null`, a string, or a `Buffer` copy of a `Uint8Array`. + * Anything else, including `undefined`, `Date`, plain objects, other class instances, functions, + * symbols, sparse or ragged arrays, and arrays deeper than PostgreSQL allows, is refused before + * anything reaches PostgreSQL. The error never names the value. + */ +export function encodeModuleDatabaseParameters(values: unknown): (string | Buffer | null)[] { + if (values === undefined) { + return []; + } + if (!Array.isArray(values) || values.length > MODULE_DATABASE_MAX_PARAMETERS) { + throw parameterInvalid(); + } + const encoded: (string | Buffer | null)[] = []; + for (let index = 0; index < values.length; index += 1) { + if (!Object.hasOwn(values, index)) { + throw parameterInvalid(); + } + const value: unknown = values[index]; + if (value === null) { + encoded.push(null); + } else if (value instanceof Uint8Array) { + // A copy: the driver serializes when the statement is sent, which may be after the caller + // has changed its own bytes. + encoded.push(Buffer.from(value)); + } else if (Array.isArray(value)) { + encoded.push(arrayLiteral(value, 1)); + } else if (isScalar(value)) { + encoded.push(scalarText(value as boolean | number | bigint | string)); + } else { + throw parameterInvalid(); + } + } + return encoded; +} + +// --- Results ----------------------------------------------------------------------------------- + +type ScalarDecoder = (text: string) => ModuleDatabaseValue; + +const asText: ScalarDecoder = (text) => text; +const asBoolean: ScalarDecoder = (text) => text === "t"; +const asInteger: ScalarDecoder = (text) => Number.parseInt(text, 10); +// `Number` reads PostgreSQL's NaN, Infinity, -Infinity, and exponent forms. With +// `extra_float_digits` pinned to 1 the text is the shortest form that round-trips exactly. +const asFloat: ScalarDecoder = (text) => Number(text); + +function hexDigit(code: number): number { + if (code >= 48 && code <= 57) { + return code - 48; + } + if (code >= 97 && code <= 102) { + return code - 87; + } + if (code >= 65 && code <= 70) { + return code - 55; + } + return -1; +} + +/** + * `bytea` text to bytes. The executor pins `bytea_output` to `hex`; the older `escape` form is read + * too, so a transaction that changed the setting still decodes faithfully rather than failing. + * Always a plain `Uint8Array`, never a `Buffer`. + */ +const asBytes: ScalarDecoder = (text) => { + if (text.startsWith("\\x")) { + const length = (text.length - 2) / 2; + const bytes = new Uint8Array(length); + for (let index = 0; index < length; index += 1) { + const high = hexDigit(text.charCodeAt(2 + index * 2)); + const low = hexDigit(text.charCodeAt(3 + index * 2)); + bytes[index] = high * 16 + low; + } + return bytes; + } + const bytes: number[] = []; + for (let index = 0; index < text.length; index += 1) { + const code = text.charCodeAt(index); + if (code === 92 && text.charCodeAt(index + 1) === 92) { + bytes.push(92); + index += 1; + } else if (code === 92) { + bytes.push(Number.parseInt(text.slice(index + 1, index + 4), 8)); + index += 3; + } else { + bytes.push(code); + } + } + return Uint8Array.from(bytes); +}; + +/** Built-in scalar type OIDs with a decoding other than "PostgreSQL's text output, unchanged". Every + * type not listed here, `int8`, `numeric`, date and time types, `json`, and `jsonb` included, + * arrives as its exact text. */ +const scalarDecoders: ReadonlyMap = new Map([ + [16, asBoolean], // bool + [17, asBytes], // bytea + [21, asInteger], // int2 + [23, asInteger], // int4 + [26, asInteger], // oid + [700, asFloat], // float4 + [701, asFloat], // float8 +]); + +/** Built-in array type OIDs, each with its element type OID. An array of a type not listed here + * (an enum, a domain, an extension type) arrives as its text, like any other type. */ +const arrayElementTypes: ReadonlyMap = new Map([ + [1000, 16], // bool[] + [1001, 17], // bytea[] + [1002, 18], // "char"[] + [1003, 19], // name[] + [1005, 21], // int2[] + [1007, 23], // int4[] + [1009, 25], // text[] + [1014, 1042], // bpchar[] + [1015, 1043], // varchar[] + [1016, 20], // int8[] + [1021, 700], // float4[] + [1022, 701], // float8[] + [1028, 26], // oid[] + [1115, 1114], // timestamp[] + [1182, 1082], // date[] + [1183, 1083], // time[] + [1185, 1184], // timestamptz[] + [1187, 1186], // interval[] + [1231, 1700], // numeric[] + [1270, 1266], // timetz[] + [2951, 2950], // uuid[] + [199, 114], // json[] + [3807, 3802], // jsonb[] +]); + +class ArrayTextError extends Error {} + +/** + * Parses PostgreSQL's array text output into nested arrays, each element decoded by `decode`. + * Unquoted `NULL` is null; a quoted `"NULL"` is the string. A non-default lower bound + * (`[0:1]={…}`) is not represented: a JavaScript array is a zero-based list. Every element type this + * decodes uses `,` as its delimiter. + */ +export function parsePostgresArrayText(text: string, decode: ScalarDecoder): ModuleDatabaseValue[] { + let position = 0; + if (text.startsWith("[")) { + const equals = text.indexOf("="); + if (equals === -1) { + throw new ArrayTextError(); + } + position = equals + 1; + } + + function parseArray(): ModuleDatabaseValue[] { + if (text[position] !== "{") { + throw new ArrayTextError(); + } + position += 1; + const items: ModuleDatabaseValue[] = []; + if (text[position] === "}") { + position += 1; + return items; + } + for (;;) { + const character = text[position]; + if (character === "{") { + items.push(parseArray()); + } else if (character === '"') { + position += 1; + let element = ""; + for (;;) { + const next = text[position]; + if (next === undefined) { + throw new ArrayTextError(); + } + if (next === "\\") { + element += text[position + 1] ?? ""; + position += 2; + } else if (next === '"') { + position += 1; + break; + } else { + element += next; + position += 1; + } + } + items.push(decode(element)); + } else { + const start = position; + while (position < text.length && text[position] !== "," && text[position] !== "}") { + position += 1; + } + const element = text.slice(start, position); + items.push(element.toUpperCase() === "NULL" ? null : decode(element)); + } + if (text[position] === ",") { + position += 1; + } else if (text[position] === "}") { + position += 1; + return items; + } else { + throw new ArrayTextError(); + } + } + } + + const parsed = parseArray(); + if (position !== text.length) { + throw new ArrayTextError(); + } + return parsed; +} + +function scalarDecoder(typeOid: number): ScalarDecoder { + return scalarDecoders.get(typeOid) ?? asText; +} + +/** Decodes one non-null column value from its PostgreSQL text by the column's type OID. */ +export function decodeModuleDatabaseValue(typeOid: number, text: string): ModuleDatabaseValue { + const elementType = arrayElementTypes.get(typeOid); + if (elementType === undefined) { + return scalarDecoder(typeOid)(text); + } + try { + return parsePostgresArrayText(text, scalarDecoder(elementType)); + } catch { + // PostgreSQL produced this text, so it cannot be malformed; should it ever be, the column still + // arrives faithfully, as text, rather than as a guess. + return text; + } +} + +/** + * The `types` object handed to `pg` for every ABI statement. Its parser returns the raw text for + * every type, so the driver's global registry, which module code could change, is never consulted + * and the ABI decodes from text itself. It is an empty array as well because the driver also reads + * the same option as the Parse message's parameter-type list: empty, so PostgreSQL infers every + * parameter type from the statement. + */ +export const rawTextTypes: { getTypeParser: () => (text: string) => string } & never[] = + Object.freeze( + Object.assign([] as never[], { + getTypeParser: () => (text: string) => text, + }), + ); + +/** Decodes one positional row of raw driver text. */ +export function decodeModuleDatabaseRow( + fields: readonly ModuleDatabaseField[], + raw: readonly unknown[], +): ModuleDatabaseValue[] { + return fields.map((field, index) => { + const value = raw[index]; + return typeof value === "string" ? decodeModuleDatabaseValue(field.typeOid, value) : null; + }); +} diff --git a/core/src/modules/database-abi.ts b/core/src/modules/database-abi.ts new file mode 100644 index 0000000..9019110 --- /dev/null +++ b/core/src/modules/database-abi.ts @@ -0,0 +1,563 @@ +import { + MODULE_DATABASE_ABI_VERSION, + type ModuleDatabaseArrayRow, + ModuleDatabaseError, + type ModuleDatabaseErrorCode, + type ModuleDatabaseExecutor, + type ModuleDatabaseField, + type ModuleDatabaseObjectRow, + type ModuleDatabaseQueryable, + type ModuleDatabaseQueryResult, + type ModuleDatabaseTransaction, + type ModuleId, + moduleIdSchema, +} from "@ekklesia/contracts"; +import { DatabaseError, type Pool, type PoolClient, type QueryArrayConfig } from "pg"; +import { + decodeModuleDatabaseRow, + encodeModuleDatabaseParameters, + rawTextTypes, +} from "./database-abi-values.js"; + +/** + * Core's Module Database ABI v1 executor (ADR 0050). + * + * A module hands Core SQL text, ordered parameter values, and a row mode; Core hands back decoded + * rows, column metadata, and stable errors. No ORM, driver, pool, client, or connection object ever + * crosses in either direction. The executor is built over the one pool the API already owns + * (ADR 0016) and never closes it. + * + * This is a correctness and compatibility boundary, **not a security sandbox**: a trusted module can + * still send any SQL its database role permits. Nothing here inspects table names. + * + * ### Connections and session state + * + * Every ABI operation, a top-level statement or a whole transaction, borrows a connection from the + * pool, pins the session settings the ABI's decoding depends on, does its work, and resets the + * session with `DISCARD ALL` before returning the connection. So no setting, temporary table, + * listener, advisory lock, prepared statement, or role change a module makes survives the operation + * that made it, and nothing the ABI pinned reaches the pool's other users. A connection that cannot + * be reset cleanly, or that is not idle when the operation ends, is destroyed rather than returned. + * `DISCARD ALL` also deallocates named prepared statements; Core uses none on the shared pool. + * + * ### One statement per `query` + * + * Every ABI statement is sent with PostgreSQL's extended query protocol, forced with the driver's + * `queryMode: "extended"` even when there are no parameters. PostgreSQL itself refuses a Parse + * message holding more than one statement (SQLSTATE 42601), using its own SQL grammar, so comments, + * quoted literals, and semicolons inside them are never mistaken for a second statement. Core does + * not parse SQL. A statement text that holds no statement at all is refused after PostgreSQL reports + * it empty. + * + * ### Transaction control belongs to Core + * + * A statement whose command tag is transaction control (`BEGIN`, `START`, `COMMIT`, `ROLLBACK`, + * `SAVEPOINT`, `RELEASE`, `PREPARE`), or that leaves the connection's transaction status other than + * expected, is refused. PostgreSQL has already run it by then, so inside a transaction it also marks + * the transaction failed: Core rolls back and never commits it. The command tag and the protocol's + * transaction status come from PostgreSQL, not from parsing the SQL. + */ + +/** The settings the ABI's decoding and determinism depend on, pinned for every ABI operation. */ +const pinSessionStatement = + "SET TimeZone TO 'UTC'; SET DateStyle TO 'ISO, MDY'; SET IntervalStyle TO 'postgres'; " + + "SET bytea_output TO 'hex'; SET extra_float_digits TO 1"; +const resetSessionStatement = "DISCARD ALL"; + +const transactionControlCommands: ReadonlySet = new Set([ + "BEGIN", + "COMMIT", + "PREPARE", + "RELEASE", + "ROLLBACK", + "SAVEPOINT", + "START", +]); + +const isolationStatements: ReadonlyMap = new Map([ + ["read committed", "READ COMMITTED"], + ["repeatable read", "REPEATABLE READ"], + ["serializable", "SERIALIZABLE"], +]); + +/** `pg` 8.22 reports the protocol's ReadyForQuery transaction status through this method, which + * its type declarations omit. `acquire` refuses a client without it rather than guessing. */ +type AbiClient = PoolClient & { getTransactionStatus(): string | null }; + +interface PreparedQuery { + readonly config: QueryArrayConfig; + readonly rowMode: "array" | "object"; +} + +interface RawResult { + readonly command: string | null; + readonly fields: readonly { readonly dataTypeID: number; readonly name: string }[]; + readonly rowCount: number | null; + readonly rows: readonly unknown[][]; +} + +function databaseError(code: ModuleDatabaseErrorCode, cause?: unknown): ModuleDatabaseError { + // Only PostgreSQL's own SQLSTATE survives. The driver's message, detail, hint, position, and + // object names, and every non-PostgreSQL error, are dropped here. + if (cause instanceof DatabaseError && typeof cause.code === "string") { + return new ModuleDatabaseError(code, { sqlState: cause.code }); + } + return new ModuleDatabaseError(code); +} + +/** A PostgreSQL error leaves the connection usable; anything else (a dropped connection, a + * client-side timeout) may leave it mid-protocol, so it must be destroyed. */ +function connectionSurvives(error: unknown): boolean { + return error instanceof DatabaseError; +} + +function prepareQuery(request: unknown): PreparedQuery { + if (typeof request !== "object" || request === null || Array.isArray(request)) { + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + for (const key of Object.keys(request)) { + if (key !== "text" && key !== "values" && key !== "rowMode") { + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + } + const { rowMode, text, values } = request as Record; + if (typeof text !== "string" || text.length === 0) { + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + if (rowMode !== undefined && rowMode !== "object" && rowMode !== "array") { + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + const config = { + queryMode: "extended", + rowMode: "array", + text, + types: rawTextTypes, + values: encodeModuleDatabaseParameters(values), + }; + return { config: config as unknown as QueryArrayConfig, rowMode: rowMode ?? "object" }; +} + +function beginStatement(options: unknown): string { + if (options === undefined) { + return "BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE"; + } + if (typeof options !== "object" || options === null || Array.isArray(options)) { + throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_OPTION_INVALID"); + } + for (const key of Object.keys(options)) { + if (key !== "isolation" && key !== "readOnly") { + throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_OPTION_INVALID"); + } + } + const { isolation, readOnly } = options as Record; + const level = isolation === undefined ? "READ COMMITTED" : isolationStatements.get(isolation); + if (level === undefined || (readOnly !== undefined && typeof readOnly !== "boolean")) { + throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_OPTION_INVALID"); + } + return `BEGIN ISOLATION LEVEL ${level} ${readOnly === true ? "READ ONLY" : "READ WRITE"}`; +} + +function buildResult( + raw: RawResult, + rowMode: "array" | "object", +): ModuleDatabaseQueryResult { + if (typeof raw.command !== "string" || raw.command === "") { + // PostgreSQL answered with EmptyQueryResponse: the text held no statement. + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + const fields: readonly ModuleDatabaseField[] = Object.freeze( + raw.fields.map((field) => Object.freeze({ name: field.name, typeOid: field.dataTypeID })), + ); + const decoded = raw.rows.map((row) => decodeModuleDatabaseRow(fields, row)); + let rows: readonly (ModuleDatabaseArrayRow | ModuleDatabaseObjectRow)[]; + if (rowMode === "array") { + rows = decoded; + } else { + if (new Set(fields.map((field) => field.name)).size !== fields.length) { + // Object rows cannot hold two columns with one name; refusing beats dropping one. + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + // `Object.fromEntries` defines properties, so a column named `__proto__` stays a column. + rows = decoded.map((row) => + Object.fromEntries(fields.map((field, index) => [field.name, row[index] ?? null])), + ); + } + return Object.freeze({ + command: raw.command, + fields, + rowCount: raw.rowCount ?? null, + rows: Object.freeze(rows), + }); +} + +/** A borrowed client's `error` events are already reported through the statement they interrupt. + * The pool only listens to idle clients, so without this a connection the server terminates while + * borrowed would surface as an unhandled `error` event. */ +function ignoreClientError(): void {} + +function destroy(client: AbiClient): void { + client.removeListener("error", ignoreClientError); + client.release(true); +} + +async function acquire(pool: Pool, failure: ModuleDatabaseErrorCode): Promise { + let client: AbiClient; + try { + client = (await pool.connect()) as AbiClient; + } catch (error) { + throw databaseError(failure, error); + } + client.on("error", ignoreClientError); + if (typeof client.getTransactionStatus !== "function") { + destroy(client); + throw new ModuleDatabaseError(failure); + } + try { + await client.query(pinSessionStatement); + } catch (error) { + destroy(client); + throw databaseError(failure, error); + } + return client; +} + +/** Returns a connection to the pool only when it is idle and resets cleanly; destroys it + * otherwise. Never throws. */ +async function relinquish(client: AbiClient, healthy: boolean): Promise { + if (!healthy || client.getTransactionStatus() !== "I") { + destroy(client); + return; + } + try { + await client.query(resetSessionStatement); + } catch { + destroy(client); + return; + } + client.removeListener("error", ignoreClientError); + client.release(); +} + +/** One open scope: a transaction or a savepoint inside it. */ +interface Scope { + activeChild: Scope | undefined; + readonly children: Set; + closed: boolean; + /** Operations started through this scope's executor that have not settled yet. */ + pending: number; +} + +interface TransactionState { + readonly client: AbiClient; + healthy: boolean; + /** Set when the transaction can no longer be committed: transaction control sent through + * `query`, or a savepoint that could not be undone. */ + failure: ModuleDatabaseError | undefined; + finished: boolean; + /** Settled-never-rejecting shadows of every statement sent on the connection. */ + readonly inflight: Set>; + savepoints: number; +} + +function newScope(parent?: Scope): Scope { + const scope: Scope = { activeChild: undefined, children: new Set(), closed: false, pending: 0 }; + parent?.children.add(scope); + return scope; +} + +function closeScope(scope: Scope): void { + scope.closed = true; + for (const child of scope.children) { + closeScope(child); + } +} + +function assertActive(scope: Scope): void { + if (scope.closed) { + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); + } + if (scope.activeChild !== undefined) { + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_INACTIVE"); + } +} + +function track(state: TransactionState, promise: Promise): Promise { + const shadow = promise.then( + () => undefined, + () => undefined, + ); + state.inflight.add(shadow); + void shadow.then(() => state.inflight.delete(shadow)); + return promise; +} + +/** Waits until every statement already sent has settled, including any sent while waiting. */ +async function drain(state: TransactionState): Promise { + while (state.inflight.size > 0) { + await Promise.all([...state.inflight]); + } +} + +/** Runs one of Core's own transaction-control statements. Refuses, without sending anything, once + * the transaction has finished. */ +async function control(state: TransactionState, statement: string): Promise { + if (state.finished) { + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); + } + try { + await track(state, state.client.query(statement)); + } catch (error) { + if (!connectionSurvives(error)) { + state.healthy = false; + } + throw databaseError("MODULE_DATABASE_TRANSACTION_FAILED", error); + } +} + +async function rollback(state: TransactionState): Promise { + try { + await state.client.query("ROLLBACK"); + } catch { + state.healthy = false; + } +} + +type Outcome = { readonly value: T } | { readonly error: unknown }; + +async function settle(run: () => Promise | T): Promise> { + try { + return { value: await run() }; + } catch (error) { + return { error }; + } +} + +const moduleByExecutor = new WeakMap(); + +/** Which module an ABI executor, or a transaction or savepoint executor derived from it, belongs + * to. For Core's own logging, diagnostics, and attribution; never exposed to modules. */ +export function moduleDatabaseExecutorModuleId(executor: object): ModuleId | undefined { + return moduleByExecutor.get(executor); +} + +export interface ModuleDatabaseExecutorOptions { + /** The module this executor serves. Carried for attribution only: the ABI enforces no table + * ownership at run time (ADR 0050). */ + readonly moduleId: ModuleId; +} + +/** + * Creates one module's ABI v1 executor over the API's existing pool. Synchronous and free of I/O: + * nothing touches the database until the module calls it. The returned object, and every + * transaction and savepoint executor it hands out, is frozen and holds exactly the ABI's members. + */ +export function createModuleDatabaseExecutor( + pool: Pool, + options: ModuleDatabaseExecutorOptions, +): ModuleDatabaseExecutor { + const parsedModuleId = moduleIdSchema.safeParse(options.moduleId); + if (!parsedModuleId.success) { + throw new TypeError("A module database executor needs a valid module ID."); + } + const moduleId = parsedModuleId.data; + + async function query( + request: unknown, + ): Promise> { + const prepared = prepareQuery(request); + const client = await acquire(pool, "MODULE_DATABASE_QUERY_FAILED"); + let healthy = true; + try { + let raw: RawResult; + try { + raw = (await client.query(prepared.config)) as unknown as RawResult; + } catch (error) { + healthy = connectionSurvives(error); + throw databaseError("MODULE_DATABASE_QUERY_FAILED", error); + } + if ( + (raw.command !== null && transactionControlCommands.has(raw.command)) || + client.getTransactionStatus() !== "I" + ) { + // The connection is destroyed below if the statement left a transaction open. + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + return buildResult(raw, prepared.rowMode); + } finally { + await relinquish(client, healthy); + } + } + + function scopeExecutor(state: TransactionState, scope: Scope): ModuleDatabaseTransaction { + async function scopeQuery( + request: unknown, + ): Promise> { + assertActive(scope); + const prepared = prepareQuery(request); + scope.pending += 1; + try { + let raw: RawResult; + try { + raw = (await track(state, state.client.query(prepared.config))) as unknown as RawResult; + } catch (error) { + if (!connectionSurvives(error)) { + state.healthy = false; + } + throw databaseError("MODULE_DATABASE_QUERY_FAILED", error); + } + if ( + (raw.command !== null && transactionControlCommands.has(raw.command)) || + state.client.getTransactionStatus() === "I" + ) { + state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + return buildResult(raw, prepared.rowMode); + } finally { + scope.pending -= 1; + } + } + + async function savepoint( + callback: (savepoint: ModuleDatabaseTransaction) => Promise | T, + ): Promise { + assertActive(scope); + if (typeof callback !== "function") { + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + scope.pending += 1; + const child = newScope(scope); + scope.activeChild = child; + state.savepoints += 1; + // Core's own identifier: a fixed prefix and a counter, never anything a module supplied. + const name = `ekklesia_savepoint_${state.savepoints}`; + try { + await control(state, `SAVEPOINT ${name}`); + if (child.closed) { + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); + } + const outcome = await settle(() => callback(scopeExecutor(state, child))); + const outstanding = child.pending > 0; + closeScope(child); + await drain(state); + if (scope.closed) { + // The enclosing transaction finished first; it rolls back, and nothing more is sent. + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); + } + if (outstanding || "error" in outcome) { + try { + await control(state, `ROLLBACK TO SAVEPOINT ${name}`); + if (!outstanding) { + await control(state, `RELEASE SAVEPOINT ${name}`); + } + } catch (error) { + state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); + throw error; + } + if (outstanding) { + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + } + throw (outcome as { readonly error: unknown }).error; + } + try { + await control(state, `RELEASE SAVEPOINT ${name}`); + } catch (releaseError) { + // Typically a failed statement left the transaction aborted. Undo the savepoint so the + // transaction stays usable, and report that this savepoint did not complete. + try { + await control(state, `ROLLBACK TO SAVEPOINT ${name}`); + await control(state, `RELEASE SAVEPOINT ${name}`); + } catch { + state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); + } + throw releaseError; + } + return (outcome as { readonly value: T }).value; + } finally { + closeScope(child); + if (scope.activeChild === child) { + scope.activeChild = undefined; + } + scope.pending -= 1; + } + } + + const executor = Object.freeze({ + abiVersion: MODULE_DATABASE_ABI_VERSION, + query: scopeQuery as ModuleDatabaseQueryable["query"], + savepoint, + }); + moduleByExecutor.set(executor, moduleId); + return executor; + } + + async function transaction( + callback: (transaction: ModuleDatabaseTransaction) => Promise | T, + options?: unknown, + ): Promise { + if (typeof callback !== "function") { + throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + } + const begin = beginStatement(options); + const client = await acquire(pool, "MODULE_DATABASE_TRANSACTION_FAILED"); + const state: TransactionState = { + client, + failure: undefined, + finished: false, + healthy: true, + inflight: new Set(), + savepoints: 0, + }; + try { + try { + await client.query(begin); + } catch (error) { + state.healthy = connectionSurvives(error); + throw databaseError("MODULE_DATABASE_TRANSACTION_FAILED", error); + } + const scope = newScope(); + const outcome = await settle(() => callback(scopeExecutor(state, scope))); + const outstanding = scope.pending > 0; + closeScope(scope); + await drain(state); + state.finished = true; + + if (outstanding) { + await rollback(state); + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + } + if ("error" in outcome) { + await rollback(state); + throw outcome.error; + } + if (state.failure !== undefined) { + await rollback(state); + throw state.failure; + } + let committed: { readonly command: string | null }; + try { + committed = await client.query("COMMIT"); + } catch (error) { + state.healthy = connectionSurvives(error); + throw databaseError("MODULE_DATABASE_TRANSACTION_FAILED", error); + } + if (committed.command !== "COMMIT") { + // PostgreSQL answers COMMIT with ROLLBACK when a failed statement had aborted the + // transaction. Nothing was committed, and PostgreSQL supplied no SQLSTATE to report. + throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); + } + return outcome.value; + } finally { + state.finished = true; + await relinquish(client, state.healthy); + } + } + + const executor = Object.freeze({ + abiVersion: MODULE_DATABASE_ABI_VERSION, + query: query as ModuleDatabaseQueryable["query"], + transaction: transaction as ModuleDatabaseExecutor["transaction"], + }); + moduleByExecutor.set(executor, moduleId); + return executor; +} diff --git a/core/src/modules/index.ts b/core/src/modules/index.ts index 020f131..8c7a318 100644 --- a/core/src/modules/index.ts +++ b/core/src/modules/index.ts @@ -2,6 +2,7 @@ export * from "./configuration.js"; export * from "./configuration-schema.js"; export * from "./configuration-store.js"; export * from "./database.js"; +export * from "./database-abi.js"; export * from "./discovery.js"; export * from "./lifecycle.js"; export * from "./lifecycle-schema.js"; diff --git a/core/tests/module-database-abi.integration.ts b/core/tests/module-database-abi.integration.ts new file mode 100644 index 0000000..d3476a2 --- /dev/null +++ b/core/tests/module-database-abi.integration.ts @@ -0,0 +1,1326 @@ +import { randomBytes } from "node:crypto"; +import { + isModuleDatabaseError, + type ModuleDatabaseErrorCode, + type ModuleDatabaseExecutor, + type ModuleDatabaseIsolationLevel, + type ModuleDatabaseTransaction, +} from "@ekklesia/contracts"; +import { Client, Pool, type PoolClient } from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createValidatedPostgresPool } from "../src/database/index.js"; +import { createModuleDatabaseExecutor } from "../src/modules/index.js"; + +/** + * Module Database ABI v1 (ADR 0050) against real PostgreSQL. + * + * Everything the ABI promises that depends on PostgreSQL itself is proven here, not mocked: type + * OIDs and decoding, the one-statement rule, SQLSTATEs, transactions, isolation, savepoints, + * session pinning and reset, and that every failure leaves the pool's next borrower a clean + * connection. + * + * The disposable database is created with deliberately hostile defaults (a non-UTC `TimeZone`, a + * non-ISO `DateStyle`, `sql_standard` intervals, `escape` bytea output, and reduced float digits), + * so every decoding assertion also proves the executor pins its own session settings, and every + * pool-safety assertion proves it gives the database's defaults back afterwards. The fixture table + * is created directly in that disposable database; no migration runs. + */ +const integrationDatabaseUrlEnvironment = "EKKLESIA_MIGRATION_TEST_DATABASE_URL"; +const trackedDatabases = new Set(); + +class IntegrationHarnessError extends Error { + constructor(message: string) { + super(message); + this.name = "IntegrationHarnessError"; + this.stack = `${this.name}: ${this.message}`; + } +} + +function baseDatabaseUrl(): string { + const value = process.env[integrationDatabaseUrlEnvironment]; + if (typeof value !== "string" || value.trim() === "") { + throw new IntegrationHarnessError( + "The module database ABI integration database is not configured.", + ); + } + try { + const parsed = new URL(value); + if (parsed.protocol !== "postgres:" && parsed.protocol !== "postgresql:") { + throw new TypeError("Unsupported protocol."); + } + } catch { + throw new IntegrationHarnessError( + "The module database ABI integration database configuration is invalid.", + ); + } + return value; +} + +function quoteDatabaseName(databaseName: string): string { + if (!/^[a-z][a-z0-9_]{0,62}$/u.test(databaseName)) { + throw new IntegrationHarnessError("The generated integration database name is invalid."); + } + return `"${databaseName}"`; +} + +async function withAdministrativeClient( + connectionString: string, + run: (client: Client) => Promise, +): Promise { + const client = new Client({ connectionString }); + client.on("error", () => undefined); + try { + await client.connect(); + } catch { + await client.end().catch(() => undefined); + throw new IntegrationHarnessError("The integration database connection failed."); + } + try { + return await run(client); + } finally { + await client.end().catch(() => undefined); + } +} + +async function createDisposableDatabase(): Promise { + const name = `ekk_db_abi_${process.pid}_${randomBytes(4).toString("hex")}`; + const quoted = quoteDatabaseName(name); + await withAdministrativeClient(baseDatabaseUrl(), async (admin) => { + await admin.query(`CREATE DATABASE ${quoted} TEMPLATE template0`); + trackedDatabases.add(name); + // Hostile defaults: everything the ABI pins is set to something else here. + await admin.query(`ALTER DATABASE ${quoted} SET TimeZone TO 'America/Chicago'`); + await admin.query(`ALTER DATABASE ${quoted} SET DateStyle TO 'SQL, DMY'`); + await admin.query(`ALTER DATABASE ${quoted} SET IntervalStyle TO 'sql_standard'`); + await admin.query(`ALTER DATABASE ${quoted} SET bytea_output TO 'escape'`); + await admin.query(`ALTER DATABASE ${quoted} SET extra_float_digits TO -3`); + }); + const parsed = new URL(baseDatabaseUrl()); + parsed.pathname = `/${name}`; + parsed.searchParams.delete("application_name"); + return parsed.toString(); +} + +async function dropDatabase(name: string): Promise { + await withAdministrativeClient(baseDatabaseUrl(), async (admin) => { + await admin.query(`DROP DATABASE IF EXISTS ${quoteDatabaseName(name)} WITH (FORCE)`); + }); + trackedDatabases.delete(name); +} + +let databaseUrl: string; +let pool: Pool; +let closePool: () => Promise; +let database: ModuleDatabaseExecutor; + +beforeAll(async () => { + databaseUrl = await createDisposableDatabase(); + await withAdministrativeClient(databaseUrl, async (admin) => { + await admin.query( + "CREATE TABLE example_records (id integer PRIMARY KEY, value text NOT NULL, secret text UNIQUE)", + ); + await admin.query( + "CREATE TABLE example_deferred (id integer, CONSTRAINT example_deferred_id_key UNIQUE (id) DEFERRABLE INITIALLY DEFERRED)", + ); + }); + // The production pool: validated, with Core's own per-statement bounds. + ({ close: closePool, pool } = createValidatedPostgresPool(databaseUrl, { + connectionTimeoutMs: 10_000, + operationTimeoutMs: 20_000, + })); + database = createModuleDatabaseExecutor(pool, { moduleId: "example" }); +}); + +afterAll(async () => { + await closePool?.(); + for (const name of [...trackedDatabases]) { + await dropDatabase(name).catch(() => undefined); + } +}); + +async function rejection(promise: Promise): Promise { + try { + await promise; + } catch (error) { + return error; + } + throw new Error("Expected the operation to reject."); +} + +/** Asserts a sanitized ABI error: the code, PostgreSQL's SQLSTATE or none, and nothing else. */ +async function expectAbiError( + promise: Promise, + code: ModuleDatabaseErrorCode, + sqlState?: string, + secrets: readonly string[] = [], +): Promise { + const error = await rejection(promise); + expect(isModuleDatabaseError(error)).toBe(true); + expect((error as { code: string }).code).toBe(code); + expect((error as { sqlState?: string }).sqlState).toBe(sqlState); + expect(Object.keys(error as object).sort()).toEqual( + sqlState === undefined ? ["code", "name"] : ["code", "name", "sqlState"], + ); + expect((error as Error).cause).toBeUndefined(); + expect((error as Error).stack).toBe(`ModuleDatabaseError: ${(error as Error).message}`); + const serialized = `${JSON.stringify(error)} ${String(error)}`; + for (const secret of [...secrets, databaseUrl, "example_records", "postgres://"]) { + expect(serialized).not.toContain(secret); + } +} + +async function reset(): Promise { + await withAdministrativeClient(databaseUrl, async (admin) => { + await admin.query("TRUNCATE example_records, example_deferred"); + }); +} + +async function storedIds(): Promise { + return withAdministrativeClient(databaseUrl, async (admin) => { + const result = await admin.query<{ id: number }>("SELECT id FROM example_records ORDER BY id"); + return result.rows.map((row) => row.id); + }); +} + +// --- Queries ------------------------------------------------------------------------------------ + +describe("query", () => { + it("runs a simple select in object mode by default", async () => { + const result = await database.query({ text: "SELECT 1 AS one, 'two' AS two" }); + expect(result).toEqual({ + command: "SELECT", + fields: [ + { name: "one", typeOid: 23 }, + { name: "two", typeOid: 25 }, + ], + rowCount: 1, + rows: [{ one: 1, two: "two" }], + }); + }); + + it("binds parameters and never interpolates them", async () => { + const hostile = "x'); DROP TABLE example_records; --"; + const result = await database.query({ text: "SELECT $1::text AS v", values: [hostile] }); + expect(result.rows).toEqual([{ v: hostile }]); + expect(await storedIds()).toEqual([]); + }); + + it("keeps array-mode columns positional, and returns metadata even with no rows", async () => { + const result = await database.query({ + rowMode: "array", + text: "SELECT 1 AS a, 'b' AS b WHERE false", + }); + expect(result.rows).toEqual([]); + expect(result.rowCount).toBe(0); + expect(result.fields).toEqual([ + { name: "a", typeOid: 23 }, + { name: "b", typeOid: 25 }, + ]); + }); + + it("returns duplicate column names positionally in array mode, and refuses them in object mode", async () => { + const arrayResult = await database.query({ rowMode: "array", text: "SELECT 1 AS a, 2 AS a" }); + expect(arrayResult.rows).toEqual([[1, 2]]); + expect(arrayResult.fields.map((field) => field.name)).toEqual(["a", "a"]); + await expectAbiError( + database.query({ text: "SELECT 1 AS a, 2 AS a" }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + }); + + it("keeps a column named __proto__ a column", async () => { + const result = await database.query({ text: 'SELECT 1 AS "__proto__"' }); + const row = result.rows[0] as Record; + expect(Object.hasOwn(row, "__proto__")).toBe(true); + expect(Object.getPrototypeOf(row)).toBe(Object.prototype); + }); + + it("reports command tags and row counts for writes, and RETURNING rows as ordinary rows", async () => { + await reset(); + const inserted = await database.query({ + text: "INSERT INTO example_records (id, value) VALUES ($1, $2), ($3, $4) RETURNING id, value", + values: [1, "a", 2, "b"], + }); + expect(inserted.command).toBe("INSERT"); + expect(inserted.rowCount).toBe(2); + expect(inserted.rows).toEqual([ + { id: 1, value: "a" }, + { id: 2, value: "b" }, + ]); + const updated = await database.query({ + text: "UPDATE example_records SET value = 'c' WHERE id > $1", + values: [0], + }); + expect(updated).toMatchObject({ command: "UPDATE", rowCount: 2, rows: [] }); + const deleted = await database.query({ text: "DELETE FROM example_records" }); + expect(deleted).toMatchObject({ command: "DELETE", rowCount: 2 }); + const created = await database.query({ text: "CREATE TEMPORARY TABLE scratch (x int)" }); + expect(created).toMatchObject({ command: "CREATE", rowCount: null }); + }); +}); + +describe("one statement per query", () => { + it("runs exactly one statement, a trailing semicolon included", async () => { + expect((await database.query({ text: "SELECT 1 AS n" })).rows).toEqual([{ n: 1 }]); + expect((await database.query({ text: "SELECT 1 AS n;" })).rows).toEqual([{ n: 1 }]); + }); + + it("refuses two statements, with PostgreSQL's own SQLSTATE, and runs neither", async () => { + await reset(); + await expectAbiError( + database.query({ text: "SELECT 1; SELECT 2" }), + "MODULE_DATABASE_QUERY_FAILED", + "42601", + ); + await expectAbiError( + database.query({ + text: "INSERT INTO example_records (id, value) VALUES (1, 'a'); INSERT INTO example_records (id, value) VALUES (2, 'b')", + }), + "MODULE_DATABASE_QUERY_FAILED", + "42601", + ); + await expectAbiError( + database.query({ text: "SELECT $1::int; SELECT 2", values: [1] }), + "MODULE_DATABASE_QUERY_FAILED", + "42601", + ); + expect(await storedIds()).toEqual([]); + }); + + it("never mistakes semicolons in literals, identifiers, or comments for a second statement", async () => { + const result = await database.query({ + rowMode: "array", + text: [ + "SELECT ';' AS a,", + " $$;SELECT 2;$$ AS b,", + " $tag$;$tag$ AS c,", + " E'\\x3B' AS d,", + ' 1 AS "semi;colon" -- ; SELECT 3', + "/* ; SELECT 4; */", + ].join("\n"), + }); + expect(result.rows).toEqual([[";", ";SELECT 2;", ";", ";", 1]]); + expect(result.fields.map((field) => field.name)).toEqual(["a", "b", "c", "d", "semi;colon"]); + }); + + it("refuses statement text that holds no statement", async () => { + for (const text of [" ", "-- only a comment", "/* nothing */", ";"]) { + await expectAbiError(database.query({ text }), "MODULE_DATABASE_REQUEST_INVALID"); + } + }); +}); + +// --- Parameters --------------------------------------------------------------------------------- + +describe("parameters", () => { + it("binds every allowed category with its exact value", async () => { + const result = await database.query({ + rowMode: "array", + text: [ + "SELECT $1::text IS NULL, $2::bool, $3::float8, $4::float8, $5::float8, $6::float8,", + " $7::int8, $8::numeric, $9::text, $10::bytea, $11::bytea, $12::int4", + ].join(" "), + values: [ + null, + true, + 0.1, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.NEGATIVE_INFINITY, + 9_223_372_036_854_775_807n, + 12_345_678_901_234_567_890n, + 'it\'s "quoted" \\ and ünïcödé 🙂', + new Uint8Array([0, 1, 254, 255]), + Buffer.from([7]), + -2_147_483_648, + ], + }); + const row = result.rows[0] as readonly unknown[]; + expect(row.slice(0, 3)).toEqual([true, true, 0.1]); + expect(row[3]).toBeNaN(); + expect(row.slice(4, 10)).toEqual([ + Number.POSITIVE_INFINITY, + Number.NEGATIVE_INFINITY, + "9223372036854775807", + "12345678901234567890", + 'it\'s "quoted" \\ and ünïcödé 🙂', + new Uint8Array([0, 1, 254, 255]), + ]); + expect(row.slice(10)).toEqual([new Uint8Array([7]), -2_147_483_648]); + }); + + it("binds arrays, nulls inside them, and more dimensions", async () => { + const result = await database.query({ + rowMode: "array", + text: "SELECT $1::int[], $2::text[], $3::bytea[], $4::int[], $5::int8[], $6::text[], $7::bool[]", + values: [ + [1, null, 3], + ['a"b', "c,d", "e\\f", "", "NULL", null, "{x}"], + [new Uint8Array([0, 255]), null], + [ + [1, 2], + [3, 4], + ], + [9_223_372_036_854_775_807n], + [], + [true, false], + ], + }); + expect(result.rows[0]).toEqual([ + [1, null, 3], + ['a"b', "c,d", "e\\f", "", "NULL", null, "{x}"], + [new Uint8Array([0, 255]), null], + [ + [1, 2], + [3, 4], + ], + ["9223372036854775807"], + [], + [true, false], + ]); + }); + + it("finds rows with an array parameter", async () => { + await reset(); + await database.query({ + text: "INSERT INTO example_records (id, value) VALUES (1, 'a'), (2, 'b'), (3, 'c')", + }); + const result = await database.query({ + text: "SELECT id FROM example_records WHERE id = ANY($1) ORDER BY id", + values: [[1, 3]], + }); + expect(result.rows).toEqual([{ id: 1 }, { id: 3 }]); + }); + + const refused: readonly [string, unknown][] = [ + ["a Date", new Date("2026-10-07T00:00:00Z")], + ["a plain object", { secret: "hunter2" }], + ["undefined", undefined], + ["a function", () => "hunter2"], + ["a symbol", Symbol("hunter2")], + ["a class instance", new URL("https://hunter2.invalid")], + ["an array holding a Date", [new Date("2026-10-07T00:00:00Z")]], + ["an array holding an object", [{ secret: "hunter2" }]], + ["an array holding undefined", ["hunter2", undefined]], + ["ragged sub-arrays", [["hunter2"], ["a", "b"]]], + ]; + + for (const [label, value] of refused) { + it(`refuses ${label} before PostgreSQL sees anything`, async () => { + await reset(); + await expectAbiError( + database.query({ + text: "INSERT INTO example_records (id, value) VALUES (1, $1::text)", + values: [value as never], + }), + "MODULE_DATABASE_PARAMETER_INVALID", + undefined, + ["hunter2", "2026-10-07", "INSERT"], + ); + expect(await storedIds()).toEqual([]); + }); + } +}); + +// --- Results ------------------------------------------------------------------------------------ + +describe("result decoding", () => { + it("decodes every type in ADR 0050's table deterministically, despite hostile database defaults", async () => { + const result = await database.query({ + text: [ + "SELECT", + " NULL::text AS null_value,", + " true AS bool, false AS bool_false,", + " (-32768)::int2 AS int2, 2147483647::int4 AS int4,", + " 9223372036854775807::int8 AS int8,", + " 1.5::float4 AS float4, 0.1::float8 AS float8, 'NaN'::float8 AS nan,", + " 'Infinity'::float8 AS inf, '-Infinity'::float4 AS neg_inf,", + " 12345678901234567890.123456789::numeric AS numeric, 'NaN'::numeric AS numeric_nan,", + " 1.10::decimal(5,2) AS decimal,", + " 'text'::text AS text, 'var'::varchar(10) AS varchar, 'ch'::char(3) AS bpchar,", + " 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid AS uuid,", + " '2026-10-07 12:34:56.789'::timestamp AS timestamp,", + " '2026-10-07 12:34:56.789+02'::timestamptz AS timestamptz,", + " '2026-10-07'::date AS date,", + " '12:34:56.5'::time AS time, '12:34:56+05:30'::timetz AS timetz,", + " '1 year 2 mons 3 days 04:05:06.7'::interval AS interval,", + " '-1 day'::interval AS negative_interval,", + ` '{"big": 12345678901234567890, "f": 0.10000000000000000001}'::json AS json,`, + ` '{"big": 12345678901234567890}'::jsonb AS jsonb,`, + " '\\x00ff10'::bytea AS bytea, ''::bytea AS empty_bytea", + ].join("\n"), + }); + const row = result.rows[0] as Record; + + expect(row).toMatchObject({ + bool: true, + bool_false: false, + bpchar: "ch ", + date: "2026-10-07", + decimal: "1.10", + float4: 1.5, + float8: 0.1, + inf: Number.POSITIVE_INFINITY, + int2: -32_768, + int4: 2_147_483_647, + int8: "9223372036854775807", + interval: "1 year 2 mons 3 days 04:05:06.7", + json: '{"big": 12345678901234567890, "f": 0.10000000000000000001}', + jsonb: '{"big": 12345678901234567890}', + neg_inf: Number.NEGATIVE_INFINITY, + negative_interval: "-1 days", + null_value: null, + numeric: "12345678901234567890.123456789", + numeric_nan: "NaN", + text: "text", + time: "12:34:56.5", + timestamp: "2026-10-07 12:34:56.789", + timestamptz: "2026-10-07 10:34:56.789+00", + timetz: "12:34:56+05:30", + uuid: "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11", + varchar: "var", + }); + expect(row.nan).toBeNaN(); + expect(row.bytea).toEqual(new Uint8Array([0, 255, 16])); + expect(Object.getPrototypeOf(row.bytea)).toBe(Uint8Array.prototype); + expect(row.empty_bytea).toEqual(new Uint8Array([])); + expect(result.fields.map((field) => [field.name, field.typeOid])).toEqual([ + ["null_value", 25], + ["bool", 16], + ["bool_false", 16], + ["int2", 21], + ["int4", 23], + ["int8", 20], + ["float4", 700], + ["float8", 701], + ["nan", 701], + ["inf", 701], + ["neg_inf", 700], + ["numeric", 1700], + ["numeric_nan", 1700], + ["decimal", 1700], + ["text", 25], + ["varchar", 1043], + ["bpchar", 1042], + ["uuid", 2950], + ["timestamp", 1114], + ["timestamptz", 1184], + ["date", 1082], + ["time", 1083], + ["timetz", 1266], + ["interval", 1186], + ["negative_interval", 1186], + ["json", 114], + ["jsonb", 3802], + ["bytea", 17], + ["empty_bytea", 17], + ]); + }); + + it("keeps float precision exact: the shortest text that round-trips", async () => { + const values = [0.1, 1 / 3, 2 ** -1074, Number.MAX_VALUE, 123_456_789.123_456_78]; + const result = await database.query({ + rowMode: "array", + text: "SELECT unnest($1::float8[])", + values: [values], + }); + expect(result.rows.map((row) => row[0])).toEqual(values); + }); + + it("decodes representative arrays element by element", async () => { + const result = await database.query({ + text: [ + "SELECT", + " ARRAY[1, NULL, 9223372036854775807]::int8[] AS int8s,", + " ARRAY[1.10, 2]::numeric[] AS numerics,", + " ARRAY[true, false, NULL] AS bools,", + " ARRAY[[1, 2], [3, 4]]::int4[] AS matrix,", + " ARRAY[0.5, 'NaN']::float8[] AS floats,", + ` ARRAY['a"b', 'c,d', '', 'NULL', NULL, '{}', 'back\\slash']::text[] AS texts,`, + " ARRAY['\\x00ff'::bytea, NULL] AS byteas,", + " ARRAY['2026-10-07 12:00:00+02'::timestamptz] AS instants,", + " ARRAY['2026-10-07'::date] AS dates,", + " ARRAY['1 day'::interval] AS intervals,", + ` ARRAY['{"n": 12345678901234567890}'::jsonb] AS documents,`, + " ARRAY['a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid] AS uuids,", + " '{}'::int[] AS empty,", + " '[0:1]={5,6}'::int[] AS lower_bound", + ].join("\n"), + }); + const row = result.rows[0] as Record; + expect(row).toEqual({ + bools: [true, false, null], + byteas: [new Uint8Array([0, 255]), null], + dates: ["2026-10-07"], + documents: ['{"n": 12345678901234567890}'], + empty: [], + floats: [0.5, Number.NaN], + instants: ["2026-10-07 10:00:00+00"], + int8s: ["1", null, "9223372036854775807"], + intervals: ["1 day"], + lower_bound: [5, 6], + matrix: [ + [1, 2], + [3, 4], + ], + numerics: ["1.10", "2"], + texts: ['a"b', "c,d", "", "NULL", null, "{}", "back\\slash"], + uuids: ["a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"], + }); + }); + + it("decodes the same instant to the same text whatever the session started as", async () => { + const instant = "2026-03-08 08:30:00+00"; + const first = await database.query({ text: "SELECT $1::timestamptz AS t", values: [instant] }); + const second = await database.transaction((tx) => + tx.query({ text: "SELECT $1::timestamptz AS t", values: [instant] }), + ); + expect(first.rows).toEqual([{ t: instant }]); + expect(second.rows).toEqual([{ t: instant }]); + }); +}); + +// --- Errors ------------------------------------------------------------------------------------- + +describe("errors", () => { + it("reports a syntax error as a failed query with PostgreSQL's SQLSTATE and nothing more", async () => { + await expectAbiError( + database.query({ text: "SELEC secret_column FROM example_records" }), + "MODULE_DATABASE_QUERY_FAILED", + "42601", + ["SELEC", "secret_column", "syntax"], + ); + await expectAbiError( + database.query({ text: "SELECT missing_column FROM example_records" }), + "MODULE_DATABASE_QUERY_FAILED", + "42703", + ["missing_column"], + ); + }); + + it("keeps a unique violation's SQLSTATE and hides its message, detail, constraint, and value", async () => { + await reset(); + await database.query({ + text: "INSERT INTO example_records (id, value, secret) VALUES (1, 'a', $1)", + values: ["hunter2"], + }); + await expectAbiError( + database.query({ + text: "INSERT INTO example_records (id, value, secret) VALUES (2, 'b', $1)", + values: ["hunter2"], + }), + "MODULE_DATABASE_QUERY_FAILED", + "23505", + ["hunter2", "example_records_secret_key", "duplicate", "Key ("], + ); + }); + + it("reports a primary-key conflict inside a transaction the same way", async () => { + await reset(); + await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await expectAbiError( + database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'b')" }); + }), + "MODULE_DATABASE_QUERY_FAILED", + "23505", + ["example_records_pkey"], + ); + }); +}); + +// --- Transactions ------------------------------------------------------------------------------- + +describe("transactions", () => { + it("commits a resolved callback and returns its value", async () => { + await reset(); + const value = await database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + return "committed"; + }); + expect(value).toBe("committed"); + expect(await storedIds()).toEqual([1, 2]); + }); + + it("rolls back a callback that throws, and rejects with the callback's own error", async () => { + await reset(); + const own = new Error("module failure"); + const error = await rejection( + database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + throw own; + }), + ); + expect(error).toBe(own); + expect(await storedIds()).toEqual([]); + }); + + it("sees its own writes, and other connections do not until it commits", async () => { + await reset(); + await database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + expect( + (await tx.query({ text: "SELECT count(*)::int AS n FROM example_records" })).rows, + ).toEqual([{ n: 1 }]); + expect(await storedIds()).toEqual([]); + }); + expect(await storedIds()).toEqual([1]); + }); + + it("never retries: a serialization failure reaches the callback once and rejects the transaction", async () => { + await reset(); + await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + let attempts = 0; + await expectAbiError( + database.transaction( + async (tx) => { + attempts += 1; + await tx.query({ text: "SELECT value FROM example_records WHERE id = 1" }); + // Another connection commits a change to the same row after this snapshot was taken. + await database.query({ text: "UPDATE example_records SET value = 'b' WHERE id = 1" }); + await tx.query({ text: "UPDATE example_records SET value = 'c' WHERE id = 1" }); + }, + { isolation: "repeatable read" }, + ), + "MODULE_DATABASE_QUERY_FAILED", + "40001", + ); + expect(attempts).toBe(1); + const rows = await database.query({ text: "SELECT value FROM example_records WHERE id = 1" }); + expect(rows.rows).toEqual([{ value: "b" }]); + }); + + it("reports a failure at COMMIT, such as a deferred constraint, with its SQLSTATE", async () => { + await reset(); + let attempts = 0; + await expectAbiError( + database.transaction(async (tx) => { + attempts += 1; + await tx.query({ text: "INSERT INTO example_deferred (id) VALUES (1), (1)" }); + }), + "MODULE_DATABASE_TRANSACTION_FAILED", + "23505", + ["example_deferred_id_key"], + ); + expect(attempts).toBe(1); + const count = await database.query({ text: "SELECT count(*)::int AS n FROM example_deferred" }); + expect(count.rows).toEqual([{ n: 0 }]); + }); + + it("never commits a transaction a failed statement aborted, even if the callback carries on", async () => { + await reset(); + await expectAbiError( + database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await expectAbiError( + tx.query({ text: "SELECT 1/0" }), + "MODULE_DATABASE_QUERY_FAILED", + "22012", + ); + await expectAbiError( + tx.query({ text: "SELECT 1" }), + "MODULE_DATABASE_QUERY_FAILED", + "25P02", + ); + return "carried on"; + }), + "MODULE_DATABASE_TRANSACTION_FAILED", + ); + expect(await storedIds()).toEqual([]); + }); + + it("revokes an escaped executor, which then reaches nothing", async () => { + await reset(); + let escaped: ModuleDatabaseTransaction | undefined; + await database.transaction((tx) => { + escaped = tx; + }); + await expectAbiError( + (escaped as ModuleDatabaseTransaction).query({ + text: "INSERT INTO example_records (id, value) VALUES (1, 'a')", + }), + "MODULE_DATABASE_SCOPE_CLOSED", + ); + await expectAbiError( + (escaped as ModuleDatabaseTransaction).savepoint(async () => undefined), + "MODULE_DATABASE_SCOPE_CLOSED", + ); + expect(await storedIds()).toEqual([]); + }); + + it("refuses writes in a read-only transaction", async () => { + await reset(); + await expectAbiError( + database.transaction( + async (tx) => { + expect((await tx.query({ text: "SHOW transaction_read_only" })).rows).toEqual([ + { transaction_read_only: "on" }, + ]); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + }, + { readOnly: true }, + ), + "MODULE_DATABASE_QUERY_FAILED", + "25006", + ); + expect(await storedIds()).toEqual([]); + }); + + const levels: readonly [ModuleDatabaseIsolationLevel | undefined, string][] = [ + [undefined, "read committed"], + ["read committed", "read committed"], + ["repeatable read", "repeatable read"], + ["serializable", "serializable"], + ]; + for (const [isolation, expected] of levels) { + it(`runs ${isolation ?? "the default"} isolation as ${expected}, read-write`, async () => { + const rows = await database.transaction( + async (tx) => + ( + await tx.query({ + text: "SELECT current_setting('transaction_isolation') AS level, current_setting('transaction_read_only') AS ro", + }) + ).rows, + isolation === undefined ? undefined : { isolation }, + ); + expect(rows).toEqual([{ level: expected, ro: "off" }]); + }); + } + + it("overrides a database whose default isolation is different", async () => { + const name = new URL(databaseUrl).pathname.slice(1); + await withAdministrativeClient(databaseUrl, async (admin) => { + await admin.query( + `ALTER DATABASE ${quoteDatabaseName(name)} SET default_transaction_isolation TO 'serializable'`, + ); + }); + // A new pool, so its connections start with the new database default. + const freshPool = new Pool({ connectionString: databaseUrl, max: 1 }); + freshPool.on("error", () => undefined); + try { + const raw = await freshPool.query<{ level: string }>( + "SELECT current_setting('default_transaction_isolation') AS level", + ); + expect(raw.rows).toEqual([{ level: "serializable" }]); + const fresh = createModuleDatabaseExecutor(freshPool, { moduleId: "example" }); + const rows = await fresh.transaction( + async (tx) => + (await tx.query({ text: "SELECT current_setting('transaction_isolation') AS level" })) + .rows, + ); + expect(rows).toEqual([{ level: "read committed" }]); + } finally { + await freshPool.end(); + await withAdministrativeClient(databaseUrl, async (admin) => { + await admin.query( + `ALTER DATABASE ${quoteDatabaseName(name)} RESET default_transaction_isolation`, + ); + }); + } + }); + + it("refuses invalid options before it starts anything", async () => { + let called = false; + for (const options of [ + { isolation: "read uncommitted" }, + { deferrable: true }, + { readOnly: "yes" }, + { isolation: "serializable", timeout: 5 }, + ]) { + await expectAbiError( + database.transaction(async () => { + called = true; + }, options as never), + "MODULE_DATABASE_TRANSACTION_OPTION_INVALID", + ); + } + expect(called).toBe(false); + }); + + it("refuses transaction control sent through query, and never commits the transaction", async () => { + await reset(); + for (const statements of [ + ["SAVEPOINT mine"], + ["SAVEPOINT mine", "RELEASE SAVEPOINT mine"], + ["SAVEPOINT mine", "ROLLBACK TO SAVEPOINT mine"], + ["BEGIN"], + ["START TRANSACTION"], + ]) { + await expectAbiError( + database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + for (const statement of statements) { + // PostgreSQL has run each statement by the time it is refused. + await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID"); + } + return "carried on"; + }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + expect(await storedIds()).toEqual([]); + } + }); + + it("refuses a COMMIT sent through query and fails the transaction", async () => { + await reset(); + await expectAbiError( + database.transaction(async (tx) => { + await expectAbiError(tx.query({ text: "COMMIT" }), "MODULE_DATABASE_REQUEST_INVALID"); + return "carried on"; + }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + }); + + it("applies Core's statement bound inside a transaction, and transaction-local settings work", async () => { + await expectAbiError( + database.transaction(async (tx) => { + await tx.query({ text: "SET LOCAL statement_timeout = 50" }); + await tx.query({ text: "SELECT pg_sleep(2)" }); + }), + "MODULE_DATABASE_QUERY_FAILED", + "57014", + ); + }); +}); + +// --- Forgotten await ---------------------------------------------------------------------------- + +describe("outstanding work", () => { + it("never commits a transaction whose callback forgot to await a write", async () => { + await reset(); + let forgotten: Promise | undefined; + await expectAbiError( + database.transaction((tx) => { + forgotten = tx.query({ + text: "INSERT INTO example_records (id, value) SELECT 1, 'a' FROM pg_sleep(0.2)", + }); + }), + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", + ); + // The forgotten statement itself ran (Core let it settle before rolling back) but its write + // was rolled back with the transaction. + await expect(forgotten).resolves.toMatchObject({ command: "INSERT", rowCount: 1 }); + expect(await storedIds()).toEqual([]); + }); + + it("never commits when the callback awaited some writes and forgot one", async () => { + await reset(); + await expectAbiError( + database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + void tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + }), + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", + ); + expect(await storedIds()).toEqual([]); + }); + + it("treats an unawaited savepoint as outstanding work", async () => { + await reset(); + let forgotten: Promise | undefined; + await expectAbiError( + database.transaction((tx) => { + forgotten = tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + }); + // Observed below; handled now so its rejection is never briefly unhandled. + forgotten.catch(() => undefined); + }), + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", + ); + await expect(forgotten).rejects.toMatchObject({ code: "MODULE_DATABASE_SCOPE_CLOSED" }); + expect(await storedIds()).toEqual([]); + }); +}); + +// --- Savepoints --------------------------------------------------------------------------------- + +describe("savepoints", () => { + it("releases a resolved savepoint, keeping its work", async () => { + await reset(); + const value = await database.transaction(async (tx) => + tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + return "kept"; + }), + ); + expect(value).toBe("kept"); + expect(await storedIds()).toEqual([1]); + }); + + it("rolls back a rejected savepoint and lets the transaction carry on and commit", async () => { + await reset(); + const own = new Error("savepoint failure"); + await database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + const error = await rejection( + tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + throw own; + }), + ); + expect(error).toBe(own); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" }); + }); + expect(await storedIds()).toEqual([1, 3]); + }); + + it("recovers from a failed statement inside a savepoint, the usual way to continue after one", async () => { + await reset(); + await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await database.transaction(async (tx) => { + await expectAbiError( + tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')" }); + }), + "MODULE_DATABASE_QUERY_FAILED", + "23505", + ); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + }); + expect(await storedIds()).toEqual([1, 2]); + }); + + it("reports a savepoint whose callback swallowed a failed statement, and keeps the transaction usable", async () => { + await reset(); + await database.transaction(async (tx) => { + await expectAbiError( + tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await rejection(sp.query({ text: "SELECT 1/0" })); + return "swallowed"; + }), + "MODULE_DATABASE_TRANSACTION_FAILED", + "25P02", + ); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + }); + expect(await storedIds()).toEqual([2]); + }); + + it("nests and sequences savepoints, keeping exactly the work that was released", async () => { + await reset(); + await database.transaction(async (tx) => { + await tx.savepoint(async (outer) => { + await outer.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await rejection( + outer.savepoint(async (inner) => { + await inner.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + throw new Error("inner failure"); + }), + ); + await outer.savepoint(async (inner) => { + await inner.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" }); + }); + }); + await tx.savepoint(async (second) => { + await second.query({ text: "INSERT INTO example_records (id, value) VALUES (4, 'd')" }); + }); + }); + expect(await storedIds()).toEqual([1, 3, 4]); + }); + + it("revokes an escaped savepoint executor", async () => { + await reset(); + await database.transaction(async (tx) => { + let escaped: ModuleDatabaseTransaction | undefined; + await tx.savepoint((sp) => { + escaped = sp; + }); + await expectAbiError( + (escaped as ModuleDatabaseTransaction).query({ + text: "INSERT INTO example_records (id, value) VALUES (1, 'a')", + }), + "MODULE_DATABASE_SCOPE_CLOSED", + ); + }); + expect(await storedIds()).toEqual([]); + }); + + it("refuses overlapping scopes: the parent is inactive while a savepoint is open", async () => { + await reset(); + await database.transaction(async (tx) => { + await tx.savepoint(async (sp) => { + await expectAbiError( + tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }), + "MODULE_DATABASE_SCOPE_INACTIVE", + ); + await expectAbiError( + tx.savepoint(async () => undefined), + "MODULE_DATABASE_SCOPE_INACTIVE", + ); + await sp.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + }); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" }); + }); + expect(await storedIds()).toEqual([2, 3]); + }); + + it("rolls back, never releases, a savepoint whose callback forgot to await", async () => { + await reset(); + await database.transaction(async (tx) => { + await expectAbiError( + tx.savepoint((sp) => { + void sp.query({ + text: "INSERT INTO example_records (id, value) SELECT 1, 'a' FROM pg_sleep(0.2)", + }); + }), + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", + ); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }); + }); + expect(await storedIds()).toEqual([2]); + }); +}); + +// --- Pool safety and session state -------------------------------------------------------------- + +interface BorrowerState { + readonly bytea: string; + readonly datestyle: string; + readonly listening: number; + readonly locks: number; + readonly status: string | null; + readonly temporaryTables: number; + readonly timezone: string; + readonly workMem: string; +} + +/** What the next borrower of a one-connection pool finds, through the raw driver, as Core's own + * stores would. */ +async function nextBorrower(onePool: Pool): Promise { + const client = (await onePool.connect()) as PoolClient & { + getTransactionStatus(): string | null; + }; + try { + const status = client.getTransactionStatus(); + const result = await client.query<{ + bytea: string; + datestyle: string; + listening: number; + locks: number; + temporary_tables: number; + timezone: string; + work_mem: string; + }>( + [ + "SELECT current_setting('TimeZone') AS timezone, current_setting('DateStyle') AS datestyle,", + " current_setting('bytea_output') AS bytea, current_setting('work_mem') AS work_mem,", + " (SELECT count(*)::int FROM pg_listening_channels()) AS listening,", + " (SELECT count(*)::int FROM pg_locks WHERE locktype = 'advisory' AND pid = pg_backend_pid()) AS locks,", + " (SELECT count(*)::int FROM pg_class WHERE relnamespace = pg_my_temp_schema()) AS temporary_tables", + ].join(" "), + ); + const row = result.rows[0]; + if (row === undefined) { + throw new IntegrationHarnessError("The borrower probe returned no row."); + } + return { + bytea: row.bytea, + datestyle: row.datestyle, + listening: row.listening, + locks: row.locks, + status, + temporaryTables: row.temporary_tables, + timezone: row.timezone, + workMem: row.work_mem, + }; + } finally { + client.release(); + } +} + +/** Exactly what a fresh connection to the hostile database has. */ +const clean: BorrowerState = { + bytea: "escape", + datestyle: "SQL, DMY", + listening: 0, + locks: 0, + status: "I", + temporaryTables: 0, + timezone: "America/Chicago", + workMem: "4MB", +}; + +describe("pool safety", () => { + let onePool: Pool; + let single: ModuleDatabaseExecutor; + + beforeAll(() => { + onePool = new Pool({ connectionString: databaseUrl, max: 1 }); + onePool.on("error", () => undefined); + single = createModuleDatabaseExecutor(onePool, { moduleId: "example" }); + }); + + afterAll(async () => { + await onePool?.end(); + }); + + const failures: readonly [string, (executor: ModuleDatabaseExecutor) => Promise][] = [ + ["a successful query", (executor) => executor.query({ text: "SELECT 1" })], + [ + "a successful transaction", + (executor) => executor.transaction(async (tx) => tx.query({ text: "SELECT 1" })), + ], + [ + "a callback throw", + (executor) => executor.transaction(async () => Promise.reject(new Error("x"))), + ], + ["a SQL error", (executor) => executor.query({ text: "SELECT 1/0" })], + [ + "a SQL error inside a transaction", + (executor) => executor.transaction((tx) => tx.query({ text: "SELECT 1/0" })), + ], + [ + "an aborted transaction the callback ignored", + (executor) => + executor.transaction(async (tx) => { + await rejection(tx.query({ text: "SELECT 1/0" })); + }), + ], + [ + "escaped executor use", + async (executor) => { + let escaped: ModuleDatabaseTransaction | undefined; + await executor.transaction((tx) => { + escaped = tx; + }); + return (escaped as ModuleDatabaseTransaction).query({ text: "SELECT 1" }); + }, + ], + [ + "outstanding work", + (executor) => + executor.transaction((tx) => { + void tx.query({ text: "SELECT pg_sleep(0.1)" }); + }), + ], + [ + "a savepoint rollback", + (executor) => + executor.transaction(async (tx) => { + await rejection(tx.savepoint(async () => Promise.reject(new Error("x")))); + }), + ], + [ + "an outstanding savepoint", + (executor) => + executor.transaction((tx) => { + // The module ignores the savepoint's own rejection; it is handled here only so the test + // run has no unhandled rejection. + tx.savepoint((sp) => sp.query({ text: "SELECT pg_sleep(0.1)" })).catch(() => undefined); + }), + ], + ["a refused top-level BEGIN", (executor) => executor.query({ text: "BEGIN" })], + [ + "a refused COMMIT inside a transaction", + (executor) => executor.transaction((tx) => tx.query({ text: "COMMIT" })), + ], + ["session-level settings", (executor) => executor.query({ text: "SET work_mem = '64MB'" })], + [ + "session-level settings inside a transaction", + (executor) => + executor.transaction(async (tx) => { + await tx.query({ text: "SET work_mem = '64MB'" }); + await tx.query({ text: "SET TimeZone = 'Asia/Tokyo'" }); + }), + ], + [ + "a temporary table", + (executor) => executor.query({ text: "CREATE TEMPORARY TABLE scratch (x int)" }), + ], + ["LISTEN", (executor) => executor.query({ text: "LISTEN example_channel" })], + [ + "LISTEN inside a transaction", + (executor) => executor.transaction((tx) => tx.query({ text: "LISTEN example_channel" })), + ], + [ + "a session advisory lock", + (executor) => executor.query({ text: "SELECT pg_advisory_lock(42)" }), + ], + ["a prepared statement", (executor) => executor.query({ text: "PREPARE mine AS SELECT 1" })], + [ + "a terminated connection", + (executor) => + executor.transaction(async (tx) => { + await tx.query({ text: "SELECT pg_terminate_backend(pg_backend_pid())" }); + }), + ], + ]; + + for (const [label, run] of failures) { + it(`hands the next borrower a clean connection after ${label}`, async () => { + await run(single).catch(() => undefined); + expect(await nextBorrower(onePool)).toEqual(clean); + // And the ABI keeps working on it. + expect((await single.query({ text: "SELECT 1 AS n" })).rows).toEqual([{ n: 1 }]); + }); + } + + it("reports a terminated connection without a SQLSTATE of its own making", async () => { + const error = await rejection( + single.transaction(async (tx) => { + await tx.query({ text: "SELECT pg_terminate_backend(pg_backend_pid())" }); + }), + ); + expect(isModuleDatabaseError(error)).toBe(true); + // PostgreSQL may report the termination itself (57P01) or the driver may only see the socket + // close; either way the code is Core's, and any SQLSTATE is PostgreSQL's. + expect((error as { code: string }).code).toMatch( + /^MODULE_DATABASE_(QUERY_FAILED|TRANSACTION_FAILED)$/u, + ); + expect([undefined, "57P01"]).toContain((error as { sqlState?: string }).sqlState); + expect(await nextBorrower(onePool)).toEqual(clean); + }); +}); + +describe("session state", () => { + it("lets no session setting, temporary table, or listener carry from one query to the next", async () => { + await database.query({ text: "SET work_mem = '64MB'" }); + await database.query({ text: "CREATE TEMPORARY TABLE scratch (x int)" }); + await database.query({ text: "LISTEN example_channel" }); + const result = await database.query({ + text: [ + "SELECT current_setting('work_mem') AS work_mem,", + " to_regclass('pg_temp.scratch') IS NULL AS no_scratch,", + " (SELECT count(*)::int FROM pg_listening_channels()) AS listening", + ].join(" "), + }); + expect(result.rows).toEqual([{ listening: 0, no_scratch: true, work_mem: "4MB" }]); + }); + + it("keeps transaction-local settings for the rest of the transaction, and no longer", async () => { + const inside = await database.transaction(async (tx) => { + await tx.query({ text: "SET LOCAL work_mem = '64MB'" }); + return (await tx.query({ text: "SELECT current_setting('work_mem') AS work_mem" })).rows; + }); + expect(inside).toEqual([{ work_mem: "64MB" }]); + const after = await database.query({ text: "SELECT current_setting('work_mem') AS work_mem" }); + expect(after.rows).toEqual([{ work_mem: "4MB" }]); + }); + + it("pins the ABI's settings for every operation, and never leaves them for Core's own borrowers", async () => { + const pinned = await database.query({ + text: [ + "SELECT current_setting('TimeZone') AS tz, current_setting('DateStyle') AS ds,", + " current_setting('IntervalStyle') AS iv, current_setting('bytea_output') AS bytea,", + " current_setting('extra_float_digits') AS digits", + ].join(" "), + }); + expect(pinned.rows).toEqual([ + { bytea: "hex", digits: "1", ds: "ISO, MDY", iv: "postgres", tz: "UTC" }, + ]); + const core = await pool.query<{ tz: string }>("SELECT current_setting('TimeZone') AS tz"); + expect(core.rows).toEqual([{ tz: "America/Chicago" }]); + }); + + it("keeps Core's own connection-level statement bound after resetting a session", async () => { + await database.query({ text: "SET statement_timeout = 0" }); + const bound = await pool.query<{ timeout: string }>( + "SELECT current_setting('statement_timeout') AS timeout", + ); + expect(bound.rows).toEqual([{ timeout: "20s" }]); + }); +}); diff --git a/core/tests/module-database-abi.test.ts b/core/tests/module-database-abi.test.ts new file mode 100644 index 0000000..40ead6f --- /dev/null +++ b/core/tests/module-database-abi.test.ts @@ -0,0 +1,833 @@ +import { readFile } from "node:fs/promises"; +import { + isModuleDatabaseError, + type ModuleDatabaseErrorCode, + type ModuleDatabaseTransaction, +} from "@ekklesia/contracts"; +import { DatabaseError, type Pool } from "pg"; +import { describe, expect, it, vi } from "vitest"; +import { + createModuleDatabaseExecutor, + moduleDatabaseExecutorModuleId, +} from "../src/modules/database-abi.js"; +import { + decodeModuleDatabaseValue, + encodeModuleDatabaseParameters, + MODULE_DATABASE_MAX_PARAMETERS, + parsePostgresArrayText, +} from "../src/modules/database-abi-values.js"; + +/** + * Module Database ABI v1 (ADR 0050) without a database: parameter validation and encoding, result + * decoding from PostgreSQL text, option validation, error sanitization, and scope bookkeeping. + * + * Everything that depends on PostgreSQL's own behaviour (type OIDs, the protocol, SQLSTATEs, + * transactions, savepoints, session settings) is proven against a real database in + * `module-database-abi.integration.ts`. The scripted client here only replays command tags and + * transaction status, so the executor's own bookkeeping can be observed statement by statement. + */ + +async function rejection(promise: Promise): Promise { + try { + await promise; + } catch (error) { + return error; + } + throw new Error("Expected the operation to reject."); +} + +async function expectAbiError( + promise: Promise, + code: ModuleDatabaseErrorCode, + sqlState?: string, +): Promise { + const error = await rejection(promise); + expect(isModuleDatabaseError(error)).toBe(true); + expect(error).toMatchObject({ code }); + expect((error as { sqlState?: string }).sqlState).toBe(sqlState); +} + +function syncThrows(run: () => unknown): unknown { + try { + run(); + } catch (error) { + return error; + } + throw new Error("Expected the call to throw."); +} + +// --- Parameters -------------------------------------------------------------------------------- + +describe("encodeModuleDatabaseParameters", () => { + it("encodes every scalar category as text, or bytes, and nothing else", () => { + const bytes = new Uint8Array([0, 1, 255]); + const encoded = encodeModuleDatabaseParameters([ + null, + true, + false, + 1.5, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.NEGATIVE_INFINITY, + 9_007_199_254_740_993n, + 'it\'s \\ "quoted"', + bytes, + ]); + expect(encoded.slice(0, 9)).toEqual([ + null, + "true", + "false", + "1.5", + "NaN", + "Infinity", + "-Infinity", + "9007199254740993", + 'it\'s \\ "quoted"', + ]); + expect(Buffer.isBuffer(encoded[9])).toBe(true); + expect([...(encoded[9] as Buffer)]).toEqual([0, 1, 255]); + }); + + it("copies bytes, so a later change by the caller cannot reach the statement", () => { + const bytes = new Uint8Array([1, 2, 3]); + const view = bytes.subarray(1); + const [encoded] = encodeModuleDatabaseParameters([view]); + bytes[1] = 9; + expect([...(encoded as Buffer)]).toEqual([2, 3]); + }); + + it("encodes arrays as quoted PostgreSQL array literals, nested for more dimensions", () => { + expect( + encodeModuleDatabaseParameters([ + [1, null, 3], + ['a"b', "c\\d", "", "NULL", "{,}"], + [ + [1, 2], + [3, 4], + ], + [new Uint8Array([0, 255])], + [true, 2n], + [], + ]), + ).toEqual([ + '{"1",NULL,"3"}', + '{"a\\"b","c\\\\d","","NULL","{,}"}', + '{{"1","2"},{"3","4"}}', + '{"\\\\x00ff"}', + '{"true","2"}', + "{}", + ]); + }); + + it("treats omitted values as no parameters", () => { + expect(encodeModuleDatabaseParameters(undefined)).toEqual([]); + }); + + class Point { + readonly x = 1; + } + + const refused: readonly [string, unknown][] = [ + ["a Date", new Date(0)], + ["a plain object", { a: 1 }], + ["a null-prototype object", Object.create(null)], + ["undefined", undefined], + ["a function", () => 1], + ["a symbol", Symbol("s")], + ["a class instance", new Point()], + ["a Map", new Map()], + ["another typed array", new Int16Array([1])], + ["an ArrayBuffer", new ArrayBuffer(1)], + ["a boxed string", new String("x")], + ["an array holding a Date", [new Date(0)]], + ["an array holding undefined", [1, undefined]], + ["an array holding an object", [{ a: 1 }]], + // biome-ignore lint/suspicious/noSparseArray: the hole is the case under test. + ["a sparse array", [1, , 3]], + ["scalars mixed with sub-arrays", [1, [2]]], + ["a null where a sub-array belongs", [[1], null]], + ["ragged sub-arrays", [[1], [1, 2]]], + ["more dimensions than PostgreSQL allows", [[[[[[[1]]]]]]]], + ]; + + for (const [label, value] of refused) { + it(`refuses ${label}, without naming the value`, () => { + const error = syncThrows(() => encodeModuleDatabaseParameters(["fine", value])); + expect(isModuleDatabaseError(error)).toBe(true); + expect(error).toMatchObject({ code: "MODULE_DATABASE_PARAMETER_INVALID" }); + expect((error as Error).message).not.toMatch(/fine|1970|Point|Symbol/u); + }); + } + + it("accepts exactly PostgreSQL's six array dimensions", () => { + expect(encodeModuleDatabaseParameters([[[[[[[1]]]]]]])).toEqual(['{{{{{{"1"}}}}}}']); + }); + + it("refuses a values list that is not an array, or longer than the protocol allows", () => { + for (const values of [ + "x", + { 0: "x", length: 1 }, + new Array(MODULE_DATABASE_MAX_PARAMETERS + 1).fill(1), + ]) { + const error = syncThrows(() => encodeModuleDatabaseParameters(values)); + expect(error).toMatchObject({ code: "MODULE_DATABASE_PARAMETER_INVALID" }); + } + expect( + encodeModuleDatabaseParameters(new Array(MODULE_DATABASE_MAX_PARAMETERS).fill(1)), + ).toHaveLength(MODULE_DATABASE_MAX_PARAMETERS); + }); + + it("refuses a self-referencing array instead of recursing forever", () => { + const cyclic: unknown[] = []; + cyclic.push(cyclic); + expect(syncThrows(() => encodeModuleDatabaseParameters([cyclic]))).toMatchObject({ + code: "MODULE_DATABASE_PARAMETER_INVALID", + }); + }); +}); + +// --- Results ----------------------------------------------------------------------------------- + +describe("decodeModuleDatabaseValue", () => { + it("decodes the scalar types the table names, and keeps everything else as exact text", () => { + expect(decodeModuleDatabaseValue(16, "t")).toBe(true); + expect(decodeModuleDatabaseValue(16, "f")).toBe(false); + expect(decodeModuleDatabaseValue(21, "-32768")).toBe(-32_768); + expect(decodeModuleDatabaseValue(23, "2147483647")).toBe(2_147_483_647); + expect(decodeModuleDatabaseValue(26, "4294967295")).toBe(4_294_967_295); + expect(decodeModuleDatabaseValue(700, "1.5")).toBe(1.5); + expect(decodeModuleDatabaseValue(701, "NaN")).toBeNaN(); + expect(decodeModuleDatabaseValue(701, "Infinity")).toBe(Number.POSITIVE_INFINITY); + expect(decodeModuleDatabaseValue(701, "-Infinity")).toBe(Number.NEGATIVE_INFINITY); + expect(decodeModuleDatabaseValue(701, "1e+300")).toBe(1e300); + // Precision-sensitive and zone-sensitive types stay exactly as PostgreSQL wrote them. + expect(decodeModuleDatabaseValue(20, "9223372036854775807")).toBe("9223372036854775807"); + expect(decodeModuleDatabaseValue(1700, "0.1000000000000000000001")).toBe( + "0.1000000000000000000001", + ); + expect(decodeModuleDatabaseValue(3802, '{"a": 12345678901234567890}')).toBe( + '{"a": 12345678901234567890}', + ); + expect(decodeModuleDatabaseValue(1184, "2026-10-07 12:00:00+00")).toBe( + "2026-10-07 12:00:00+00", + ); + // An unknown OID (an enum, a domain, an extension type) is text. + expect(decodeModuleDatabaseValue(987_654, "anything")).toBe("anything"); + }); + + it("decodes bytea to a plain Uint8Array, from hex or the older escape form", () => { + const hex = decodeModuleDatabaseValue(17, "\\x00ff10"); + expect(Object.getPrototypeOf(hex)).toBe(Uint8Array.prototype); + expect([...(hex as Uint8Array)]).toEqual([0, 255, 16]); + expect([...(decodeModuleDatabaseValue(17, "\\x") as Uint8Array)]).toEqual([]); + const escaped = decodeModuleDatabaseValue(17, "a\\000\\\\\\377"); + expect([...(escaped as Uint8Array)]).toEqual([97, 0, 92, 255]); + }); + + it("decodes built-in arrays element by element", () => { + expect(decodeModuleDatabaseValue(1016, "{1,NULL,9223372036854775807}")).toEqual([ + "1", + null, + "9223372036854775807", + ]); + expect(decodeModuleDatabaseValue(1231, "{1.10,2}")).toEqual(["1.10", "2"]); + expect(decodeModuleDatabaseValue(1000, "{t,f}")).toEqual([true, false]); + expect(decodeModuleDatabaseValue(1007, "{{1,2},{3,4}}")).toEqual([ + [1, 2], + [3, 4], + ]); + const bytes = decodeModuleDatabaseValue(1001, '{"\\\\x00ff"}') as Uint8Array[]; + expect([...(bytes[0] as Uint8Array)]).toEqual([0, 255]); + expect(decodeModuleDatabaseValue(1009, "{}")).toEqual([]); + // An array of a type the table does not know arrives as text, like the type itself. + expect(decodeModuleDatabaseValue(987_655, "{a,b}")).toBe("{a,b}"); + }); +}); + +describe("parsePostgresArrayText", () => { + const text = (value: string) => value; + + it('honours quoting, escapes, and the difference between NULL and "NULL"', () => { + expect(parsePostgresArrayText('{"a\\"b","c,d",NULL,"NULL","","{}",null}', text)).toEqual([ + 'a"b', + "c,d", + null, + "NULL", + "", + "{}", + null, + ]); + expect(parsePostgresArrayText('{"back\\\\slash"}', text)).toEqual(["back\\slash"]); + }); + + it("drops a non-default lower bound: a JavaScript array is a zero-based list", () => { + expect(parsePostgresArrayText("[0:1]={5,6}", text)).toEqual(["5", "6"]); + expect(parsePostgresArrayText("[1:1][2:3]={{7,8}}", text)).toEqual([["7", "8"]]); + }); + + it("refuses text that is not an array literal", () => { + for (const malformed of ["", "{", "{1", '{"a', "{1}x", "[1:2]{1,2}", "1"]) { + expect(() => parsePostgresArrayText(malformed, text)).toThrow(); + } + }); +}); + +// --- The executor, over a scripted client ------------------------------------------------------- + +interface ScriptedStatement { + readonly text: string; + readonly values?: readonly unknown[]; +} + +interface Script { + readonly pool: Pool; + readonly statements: ScriptedStatement[]; + readonly released: (boolean | undefined)[]; + /** Resolve a statement only when `release` is called, to model in-flight work. */ + holdNext(): () => void; + failNext(error: unknown): void; + /** Fail the next statement whose text is exactly `text`. */ + failWhen(text: string, error: unknown): void; + connectCount(): number; +} + +function commandOf(text: string): string { + const word = /^\s*([A-Za-z]+)/u.exec(text)?.[1]?.toUpperCase() ?? ""; + return word === "END" ? "COMMIT" : word; +} + +/** + * A pool whose one client replays PostgreSQL's command tags and transaction status for simple SQL. + * It is not PostgreSQL: it exists only so the executor's bookkeeping can be observed. + */ +function scriptedPool(): Script { + const statements: ScriptedStatement[] = []; + const released: (boolean | undefined)[] = []; + let status = "I"; + let connects = 0; + let hold: Promise | undefined; + let failure: unknown; + const failures = new Map(); + + const client = { + getTransactionStatus: () => status, + on: () => client, + removeListener: () => client, + async query(config: string | { text: string; values?: unknown[] }) { + const text = typeof config === "string" ? config : config.text; + statements.push( + typeof config === "string" ? { text } : { text, values: config.values ?? [] }, + ); + const held = hold; + hold = undefined; + if (held !== undefined) { + await held; + } + if (failure !== undefined || failures.has(text)) { + const error = failure ?? failures.get(text); + failure = undefined; + failures.delete(text); + if (status === "T") { + status = "E"; + } + throw error; + } + const command = commandOf(text); + if (command === "BEGIN" || command === "START") { + status = "T"; + } else if (command === "COMMIT" || (command === "ROLLBACK" && !/\bTO\b/iu.test(text))) { + const aborted = status === "E"; + status = "I"; + return { command: aborted ? "ROLLBACK" : command, fields: [], rowCount: null, rows: [] }; + } + return { + command, + fields: [{ dataTypeID: 23, name: "n" }], + rowCount: 1, + rows: [["1"]], + }; + }, + release: (destroy?: boolean) => { + released.push(destroy); + }, + }; + + return { + connectCount: () => connects, + failNext(error) { + failure = error; + }, + failWhen(text, error) { + failures.set(text, error); + }, + holdNext() { + let release!: () => void; + hold = new Promise((resolve) => { + release = resolve; + }); + return release; + }, + pool: { + connect: vi.fn(async () => { + connects += 1; + return client; + }), + } as unknown as Pool, + released, + statements, + }; +} + +function pgError(code: string): DatabaseError { + const error = new DatabaseError( + 'duplicate key value violates unique constraint "example_secret_key"', + 100, + "error", + ); + error.code = code; + error.detail = "Key (secret)=(hunter2) already exists."; + error.hint = "a hint"; + error.table = "example_records"; + error.constraint = "example_secret_key"; + return error; +} + +const pin = + "SET TimeZone TO 'UTC'; SET DateStyle TO 'ISO, MDY'; SET IntervalStyle TO 'postgres'; SET bytea_output TO 'hex'; SET extra_float_digits TO 1"; + +function texts(script: Script): string[] { + return script.statements.map((statement) => statement.text); +} + +describe("createModuleDatabaseExecutor", () => { + it("builds a frozen executor holding exactly the ABI's members, without touching the pool", () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + expect(Object.keys(executor).sort()).toEqual(["abiVersion", "query", "transaction"]); + expect(executor.abiVersion).toBe(1); + expect(Object.isFrozen(executor)).toBe(true); + expect(Object.getPrototypeOf(executor)).toBe(Object.prototype); + for (const forbidden of ["pool", "client", "$client", "_", "session", "dialect", "select"]) { + expect(forbidden in executor).toBe(false); + } + expect(script.connectCount()).toBe(0); + }); + + it("carries its module for Core's attribution, and refuses an invalid module ID", () => { + const executor = createModuleDatabaseExecutor(scriptedPool().pool, { moduleId: "care-team" }); + expect(moduleDatabaseExecutorModuleId(executor)).toBe("care-team"); + expect(moduleDatabaseExecutorModuleId({})).toBeUndefined(); + expect(() => + createModuleDatabaseExecutor(scriptedPool().pool, { moduleId: "Not A Module" }), + ).toThrow(TypeError); + }); + + it("pins the session, runs one extended-protocol statement, and resets before releasing", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + const result = await executor.query({ text: "SELECT 1 AS n", values: [5n] }); + + expect(result).toEqual({ + command: "SELECT", + fields: [{ name: "n", typeOid: 23 }], + rowCount: 1, + rows: [{ n: 1 }], + }); + expect(texts(script)).toEqual([pin, "SELECT 1 AS n", "DISCARD ALL"]); + expect(script.statements[1]).toEqual({ text: "SELECT 1 AS n", values: ["5"] }); + expect(script.released).toEqual([undefined]); + expect(Object.isFrozen(result)).toBe(true); + expect(Object.isFrozen(result.rows)).toBe(true); + expect(Object.isFrozen(result.fields)).toBe(true); + }); + + it("refuses a malformed request before borrowing a connection", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + const query = executor.query as (request: unknown) => Promise; + + for (const request of [ + undefined, + null, + "SELECT 1", + [], + { text: "" }, + { text: 1 }, + { text: "SELECT 1", rowMode: "rows" }, + { text: "SELECT 1", name: "prepared" }, + { text: "SELECT 1", types: [] }, + ]) { + await expectAbiError(query(request), "MODULE_DATABASE_REQUEST_INVALID"); + } + await expectAbiError( + query({ text: "SELECT $1", values: [new Date()] }), + "MODULE_DATABASE_PARAMETER_INVALID", + ); + expect(script.connectCount()).toBe(0); + }); + + it("reports only PostgreSQL's SQLSTATE from a database error, never its text", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + script.failWhen("SELECT 1", pgError("23505")); + const error = await rejection(executor.query({ text: "SELECT 1" })); + + expect(error).toMatchObject({ code: "MODULE_DATABASE_QUERY_FAILED", sqlState: "23505" }); + const serialized = `${JSON.stringify(error)} ${String(error)} ${(error as Error).stack}`; + for (const secret of [ + "hunter2", + "example_secret_key", + "example_records", + "duplicate", + "hint", + ]) { + expect(serialized).not.toContain(secret); + } + expect((error as Error).cause).toBeUndefined(); + // A PostgreSQL error leaves the connection usable, so it is reset and returned. + expect(script.released).toEqual([undefined]); + }); + + it("drops a non-PostgreSQL failure's code and destroys the connection it happened on", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + script.failWhen( + "SELECT 1", + Object.assign(new Error("Connection terminated: postgres://u:p@h/db"), { code: "EPIPE" }), + ); + + const error = await rejection(executor.query({ text: "SELECT 1" })); + expect(error).toMatchObject({ code: "MODULE_DATABASE_QUERY_FAILED" }); + expect((error as { sqlState?: string }).sqlState).toBeUndefined(); + expect(JSON.stringify(error)).not.toContain("postgres://"); + expect(script.released).toEqual([true]); + }); + + it("refuses a top-level transaction-control statement and destroys a connection it left open", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + await expectAbiError(executor.query({ text: "BEGIN" }), "MODULE_DATABASE_REQUEST_INVALID"); + expect(script.released).toEqual([true]); + expect(texts(script)).not.toContain("DISCARD ALL"); + }); +}); + +describe("transaction options", () => { + const invalid: readonly unknown[] = [ + null, + "serializable", + [], + { isolation: "read uncommitted" }, + { isolation: "SERIALIZABLE" }, + { isolation: "constructor" }, + { readOnly: "yes" }, + { readOnly: 1 }, + { deferrable: true }, + { isolation: "serializable", retries: 3 }, + ]; + + for (const options of invalid) { + it(`refuses ${JSON.stringify(options)} before borrowing a connection`, async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + const callback = vi.fn(); + await expectAbiError( + executor.transaction(callback, options as never), + "MODULE_DATABASE_TRANSACTION_OPTION_INVALID", + ); + expect(callback).not.toHaveBeenCalled(); + expect(script.connectCount()).toBe(0); + }); + } + + it("refuses a callback that is not a function", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + await expectAbiError( + executor.transaction("SELECT 1" as never), + "MODULE_DATABASE_REQUEST_INVALID", + ); + expect(script.connectCount()).toBe(0); + }); + + it("begins every transaction with an explicit isolation level and access mode", async () => { + const cases: readonly [unknown, string][] = [ + [undefined, "BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE"], + [{}, "BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE"], + [{ readOnly: true }, "BEGIN ISOLATION LEVEL READ COMMITTED READ ONLY"], + [{ isolation: "repeatable read" }, "BEGIN ISOLATION LEVEL REPEATABLE READ READ WRITE"], + [ + { isolation: "serializable", readOnly: false }, + "BEGIN ISOLATION LEVEL SERIALIZABLE READ WRITE", + ], + ]; + for (const [options, begin] of cases) { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + await executor.transaction(async () => undefined, options as never); + expect(texts(script)).toEqual([pin, begin, "COMMIT", "DISCARD ALL"]); + } + }); +}); + +describe("transaction scopes", () => { + it("commits a resolved callback and returns its value", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + const value = await executor.transaction(async (tx) => { + expect(Object.keys(tx).sort()).toEqual(["abiVersion", "query", "savepoint"]); + expect(Object.isFrozen(tx)).toBe(true); + expect(moduleDatabaseExecutorModuleId(tx)).toBe("example"); + await tx.query({ text: "INSERT INTO example_records VALUES (1)" }); + return "done"; + }); + + expect(value).toBe("done"); + expect(texts(script).slice(1)).toEqual([ + "BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE", + "INSERT INTO example_records VALUES (1)", + "COMMIT", + "DISCARD ALL", + ]); + }); + + it("rolls back a rejected callback and rejects with the callback's own error, untouched", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + const own = new Error("module failure"); + + const error = await rejection( + executor.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records VALUES (1)" }); + throw own; + }), + ); + + expect(error).toBe(own); + expect(texts(script)).toContain("ROLLBACK"); + expect(texts(script)).not.toContain("COMMIT"); + }); + + it("revokes the executor when the callback settles; later use sends nothing", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + let escaped: ModuleDatabaseTransaction | undefined; + await executor.transaction((tx) => { + escaped = tx; + }); + const sent = script.statements.length; + + await expectAbiError( + (escaped as ModuleDatabaseTransaction).query({ text: "SELECT 1" }), + "MODULE_DATABASE_SCOPE_CLOSED", + ); + await expectAbiError( + (escaped as ModuleDatabaseTransaction).savepoint(async () => undefined), + "MODULE_DATABASE_SCOPE_CLOSED", + ); + expect(script.statements).toHaveLength(sent); + }); + + it("never commits a callback that returned with a query still outstanding, even once it succeeds", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + let forgotten: Promise | undefined; + let releaseHeld: (() => void) | undefined; + + const outcome = executor.transaction((tx) => { + releaseHeld = script.holdNext(); + forgotten = tx.query({ text: "INSERT INTO example_records VALUES (1)" }); + }); + await vi.waitFor(() => expect(releaseHeld).toBeDefined()); + // Core waits for the started statement to settle before rolling back the connection. + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(texts(script)).not.toContain("ROLLBACK"); + releaseHeld?.(); + + await expectAbiError(outcome, "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + await expect(forgotten).resolves.toBeDefined(); + expect(texts(script).slice(-2)).toEqual(["ROLLBACK", "DISCARD ALL"]); + expect(texts(script)).not.toContain("COMMIT"); + }); + + it("reports outstanding work even when the callback itself rejected", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + let releaseHeld: (() => void) | undefined; + + const outcome = executor.transaction(async (tx) => { + releaseHeld = script.holdNext(); + void tx.query({ text: "INSERT INTO example_records VALUES (1)" }); + throw new Error("module failure"); + }); + await vi.waitFor(() => expect(releaseHeld).toBeDefined()); + releaseHeld?.(); + + await expectAbiError(outcome, "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + expect(texts(script)).not.toContain("COMMIT"); + }); + + it("refuses a transaction-control statement inside a transaction and never commits it", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + const error = await rejection( + executor.transaction(async (tx) => { + await expectAbiError( + tx.query({ text: "SAVEPOINT mine" }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + return "carried on"; + }), + ); + + expect(error).toMatchObject({ code: "MODULE_DATABASE_REQUEST_INVALID" }); + expect(texts(script)).not.toContain("COMMIT"); + expect(texts(script)).toContain("ROLLBACK"); + }); + + it("reports a COMMIT that PostgreSQL answered with ROLLBACK as a failed transaction", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + const outcome = executor.transaction(async (tx) => { + script.failNext(pgError("22012")); + await expectAbiError( + tx.query({ text: "SELECT 1/0" }), + "MODULE_DATABASE_QUERY_FAILED", + "22012", + ); + }); + + await expectAbiError(outcome, "MODULE_DATABASE_TRANSACTION_FAILED"); + }); +}); + +describe("savepoint scopes", () => { + it("names savepoints itself, releases a resolved one, and rolls back a rejected one", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + const own = new Error("savepoint failure"); + + await executor.transaction(async (tx) => { + await tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records VALUES (1)" }); + }); + expect(await rejection(tx.savepoint(async () => Promise.reject(own)))).toBe(own); + }); + + expect(texts(script).slice(2, -2)).toEqual([ + "SAVEPOINT ekklesia_savepoint_1", + "INSERT INTO example_records VALUES (1)", + "RELEASE SAVEPOINT ekklesia_savepoint_1", + "SAVEPOINT ekklesia_savepoint_2", + "ROLLBACK TO SAVEPOINT ekklesia_savepoint_2", + "RELEASE SAVEPOINT ekklesia_savepoint_2", + ]); + }); + + it("makes the parent inactive while a savepoint is open, so scopes cannot overlap", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + await executor.transaction(async (tx) => { + await tx.savepoint(async (sp) => { + await expectAbiError(tx.query({ text: "SELECT 1" }), "MODULE_DATABASE_SCOPE_INACTIVE"); + await expectAbiError( + tx.savepoint(async () => undefined), + "MODULE_DATABASE_SCOPE_INACTIVE", + ); + await sp.savepoint(async (inner) => { + await expectAbiError(sp.query({ text: "SELECT 1" }), "MODULE_DATABASE_SCOPE_INACTIVE"); + await inner.query({ text: "SELECT 2" }); + }); + await sp.query({ text: "SELECT 3" }); + }); + await tx.query({ text: "SELECT 4" }); + }); + + expect(texts(script)).toContain("SAVEPOINT ekklesia_savepoint_2"); + expect(texts(script)).toContain("COMMIT"); + }); + + it("revokes an escaped savepoint executor", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + await executor.transaction(async (tx) => { + let escaped: ModuleDatabaseTransaction | undefined; + await tx.savepoint((sp) => { + escaped = sp; + }); + await expectAbiError( + (escaped as ModuleDatabaseTransaction).query({ text: "SELECT 1" }), + "MODULE_DATABASE_SCOPE_CLOSED", + ); + await tx.query({ text: "SELECT 2" }); + }); + expect(texts(script)).toContain("COMMIT"); + }); + + it("rolls back, and never releases, a savepoint whose callback left work outstanding", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + + await executor.transaction(async (tx) => { + await expectAbiError( + tx.savepoint((sp) => { + const release = script.holdNext(); + void sp.query({ text: "INSERT INTO example_records VALUES (1)" }); + setTimeout(release, 10); + }), + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", + ); + }); + + expect(texts(script).slice(2, -2)).toEqual([ + "SAVEPOINT ekklesia_savepoint_1", + "INSERT INTO example_records VALUES (1)", + "ROLLBACK TO SAVEPOINT ekklesia_savepoint_1", + ]); + }); + + it("counts an unawaited savepoint as the transaction's outstanding work", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + let forgotten: Promise | undefined; + + await expectAbiError( + executor.transaction((tx) => { + forgotten = tx.savepoint(async (sp) => { + await sp.query({ text: "INSERT INTO example_records VALUES (1)" }); + }); + }), + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", + ); + await expectAbiError(forgotten as Promise, "MODULE_DATABASE_SCOPE_CLOSED"); + expect(texts(script)).not.toContain("COMMIT"); + expect(texts(script).filter((text) => text.startsWith("RELEASE"))).toEqual([]); + expect(texts(script).at(-1)).toBe("DISCARD ALL"); + }); +}); + +describe("the ABI's own sources", () => { + it("import no ORM, query builder, or ORM type", async () => { + for (const file of [ + "../src/modules/database-abi.ts", + "../src/modules/database-abi-values.ts", + ]) { + const source = await readFile(new URL(file, import.meta.url), "utf8"); + const specifiers = [...source.matchAll(/\bfrom\s+"([^"]+)"/gu)].map((match) => match[1]); + expect( + specifiers.every((specifier) => + ["@ekklesia/contracts", "pg", "./database-abi-values.js"].includes(specifier ?? ""), + ), + ).toBe(true); + expect(source).not.toMatch(/drizzle|kysely|NodePg|PgTable|PgColumn/iu); + } + }); +}); diff --git a/docs/architecture/decisions/0050-module-database-abi-v1.md b/docs/architecture/decisions/0050-module-database-abi-v1.md index 5b94889..d107aca 100644 --- a/docs/architecture/decisions/0050-module-database-abi-v1.md +++ b/docs/architecture/decisions/0050-module-database-abi-v1.md @@ -1,6 +1,6 @@ # 0050: Module Database ABI v1: modules and Core exchange SQL, never ORM objects -- Status: Proposed +- Status: Accepted - Date: 2026-10-07 - Supersedes (partial): [0016](0016-module-database-access.md) — for modules that declare ABI v1, only its definition of the module database service as `Pick`, a facade over diff --git a/docs/architecture/decisions/README.md b/docs/architecture/decisions/README.md index 7afc2a4..bccc4e8 100644 --- a/docs/architecture/decisions/README.md +++ b/docs/architecture/decisions/README.md @@ -58,4 +58,4 @@ ADRs capture durable choices that affect more than one package or phase. - [0047: Every module proves itself inside Ekklesia, with one harness the platform publishes](0047-module-ci-harness.md) — Proposed - [0048: Ekklesia ships one generic image, and installed modules live outside it](0048-generic-image-and-installed-module-storage.md) — Accepted - [0049: A module release is one immutable, self-contained package: Ekklesia Module Package v1](0049-module-package-v1.md) — Accepted -- [0050: Module Database ABI v1: modules and Core exchange SQL, never ORM objects](0050-module-database-abi-v1.md) — Proposed +- [0050: Module Database ABI v1: modules and Core exchange SQL, never ORM objects](0050-module-database-abi-v1.md) — Accepted diff --git a/docs/architecture/platform-architecture.md b/docs/architecture/platform-architecture.md index 6703b4f..d50acad 100644 --- a/docs/architecture/platform-architecture.md +++ b/docs/architecture/platform-architecture.md @@ -686,5 +686,7 @@ Module Package v1, a deterministic `.tar.gz` with its SHA-256. and loads the backend once in a short-lived process with an empty environment. Only then does it extract anything. - **What exists today.** Only the `any` target. Drizzle agreement is recorded as transitional, not a - database ABI. [ADR 0050](decisions/0050-module-database-abi-v1.md), still proposed, defines the - Module Database ABI that removes it for modules that declare that ABI. + database ABI. [ADR 0050](decisions/0050-module-database-abi-v1.md) defines the Module Database ABI + that removes it for modules that declare that ABI. Core's ABI v1 executor + (`createModuleDatabaseExecutor`) exists and is proven against PostgreSQL. No module can declare the + ABI or receive the executor yet: that needs manifest version 8. diff --git a/docs/development/testing.md b/docs/development/testing.md index d532911..3cd6214 100644 --- a/docs/development/testing.md +++ b/docs/development/testing.md @@ -509,6 +509,31 @@ an `authenticated` baseline, from platform ownership, or from a user grant. Its answer "yes" to every human question, so any accidental fall-through into a human path would flip those assertions. +A further suite (`core/tests/module-database-abi.integration.ts`, same harness) proves Core's Module +Database ABI v1 executor against real PostgreSQL +([ADR 0050](../architecture/decisions/0050-module-database-abi-v1.md)). Nothing it asserts is mocked, +because every claim is a property of PostgreSQL or of the protocol: +- one statement per `query`, enforced by PostgreSQL's extended protocol rather than by parsing, so + semicolons inside literals, identifiers, and comments are never mistaken for a second statement; +- every allowed parameter category bound exactly, and every refused one refused before PostgreSQL + sees it; +- the decoding table for each type the ADR names; +- PostgreSQL's SQLSTATE preserved while its message, detail, constraint, and values are hidden; +- commit, rollback, isolation levels, read-only, no retries, and an aborted transaction never being + committed; +- callback-scoped savepoints; +- a forgotten `await` rolling back instead of committing; +- revoked executors reaching nothing. + +The suite's database is created with deliberately hostile defaults: a non-UTC time zone, a non-ISO +date style, `sql_standard` intervals, `escape` bytea output, and reduced float digits. So every +decoding assertion also proves the executor pins its own session settings. The pool-safety cases use +a one-connection pool, which proves the next borrower gets a connection with exactly the database's +defaults after every failure class: no transaction, setting, temporary table, listener, or advisory +lock carries over. The same is true after a connection the server terminated. Parameter validation, +decoding from text, option validation, error sanitization, and scope bookkeeping are also unit-tested +without a database in `core/tests/module-database-abi.test.ts`. + Outbound email is proven against a **real SMTP server** rather than a mock (`core/tests/mail-smtp.integration.test.ts`, [ADR 0037](../architecture/decisions/0037-host-provided-outbound-email.md)). `smtp-server` — the reference implementation from the same project as the client — runs in-process, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 6bbe580..2089567 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -4,6 +4,7 @@ export * from "./identity.js"; export * from "./module-configuration.js"; export * from "./module-contribution.js"; export * from "./module-database.js"; +export * from "./module-database-abi.js"; export * from "./module-http.js"; export * from "./module-lifecycle.js"; export * from "./module-manifest.js"; diff --git a/packages/contracts/src/module-database-abi.ts b/packages/contracts/src/module-database-abi.ts new file mode 100644 index 0000000..7c066d1 --- /dev/null +++ b/packages/contracts/src/module-database-abi.ts @@ -0,0 +1,217 @@ +/** + * Module Database ABI v1 (ADR 0050): the stable boundary between Core and a module's database code. + * + * Only PostgreSQL concepts cross it: SQL text, ordered bound parameter values, decoded result rows + * with their column metadata, and transaction scopes that Core controls. **Nothing in this file + * names an ORM, a query builder, or a driver.** A module may build its SQL with Drizzle, Kysely, raw + * text, or anything else, and Core never learns which; Core may change its own ORM without any + * module noticing. + * + * This is a correctness and compatibility boundary, **not a security sandbox**. Installed modules + * remain trusted backend code that runs in the API process with its database role (ADR 0048), and + * nothing here, including table prefixes, `SET ROLE`, or PostgreSQL roles, can stop a module from + * reaching data that is not its own. + */ + +/** The ABI version these types describe. A module declares it; every executor reports it. */ +export const MODULE_DATABASE_ABI_VERSION = 1; +export type ModuleDatabaseAbiVersion = typeof MODULE_DATABASE_ABI_VERSION; + +/** + * A scalar parameter value. Deliberately small: each has exactly one PostgreSQL text encoding. + * + * - `number` is sent as its JavaScript decimal text; `NaN`, `Infinity`, and `-Infinity` as + * PostgreSQL's own spellings. + * - `bigint` is sent as its decimal text. + * - `Uint8Array` is sent as `bytea`. + * + * A JavaScript `Date` is **not** a parameter: whether it means a `timestamp` or a `timestamptz`, and + * in which zone, depends on a column the ABI cannot see. Plain objects are not implicit JSON; encode + * JSON explicitly as a string. Identifiers are never parameters. + */ +export type ModuleDatabaseScalar = null | boolean | number | bigint | string | Uint8Array; + +/** A parameter value: a scalar, or a PostgreSQL array of scalars, nested for more dimensions. */ +export type ModuleDatabaseParameter = ModuleDatabaseScalar | readonly ModuleDatabaseParameter[]; + +/** + * A decoded result value. Values that a generic conversion could change arrive as PostgreSQL's + * exact text: `int8`, `numeric`, every date and time type, `json`, and `jsonb`. `bytea` arrives as a + * `Uint8Array`, and arrays of the built-in types decode element by element. + */ +export type ModuleDatabaseValue = + | null + | boolean + | number + | string + | Uint8Array + | readonly ModuleDatabaseValue[]; + +/** `"object"` rows are keyed by column name; `"array"` rows are positional, in `fields` order. */ +export type ModuleDatabaseRowMode = "object" | "array"; + +export type ModuleDatabaseObjectRow = Readonly>; +export type ModuleDatabaseArrayRow = readonly ModuleDatabaseValue[]; + +/** One query: exactly one PostgreSQL statement with `$1 … $n` placeholders. */ +export interface ModuleDatabaseQuery { + readonly text: string; + readonly values?: readonly ModuleDatabaseParameter[]; + /** Defaults to `"object"`. */ + readonly rowMode?: ModuleDatabaseRowMode; +} + +/** The ordered column metadata of a result, present even when it has no rows. */ +export interface ModuleDatabaseField { + readonly name: string; + /** The column's PostgreSQL type OID. */ + readonly typeOid: number; +} + +export interface ModuleDatabaseQueryResult { + /** PostgreSQL's command tag word, such as `SELECT`, `INSERT`, `UPDATE`, or `DELETE`. */ + readonly command: string; + /** The number of rows PostgreSQL reports as processed, or `null` when it reports none. */ + readonly rowCount: number | null; + readonly rows: readonly Row[]; + readonly fields: readonly ModuleDatabaseField[]; +} + +/** What every executor can do: run one statement. */ +export interface ModuleDatabaseQueryable { + readonly abiVersion: ModuleDatabaseAbiVersion; + query( + query: ModuleDatabaseQuery & { readonly rowMode: "array" }, + ): Promise>; + query( + query: ModuleDatabaseQuery & { readonly rowMode?: "object" }, + ): Promise>; + query( + query: ModuleDatabaseQuery, + ): Promise>; +} + +export type ModuleDatabaseIsolationLevel = "read committed" | "repeatable read" | "serializable"; + +/** Transaction options. Any other key or value is refused, never ignored. Omitted, a transaction is + * `read committed` and read-write. */ +export interface ModuleDatabaseTransactionOptions { + readonly isolation?: ModuleDatabaseIsolationLevel; + readonly readOnly?: boolean; +} + +/** + * The executor inside a transaction or a savepoint. It is valid only while its callback runs: once + * that callback settles it is revoked, and every later call fails without reaching PostgreSQL. + * + * Only the innermost open scope may run statements: while a savepoint is open, its parent refuses + * statements and new savepoints. A callback must await every operation it starts before it returns; + * one that settles with work outstanding is never committed or released. + */ +export interface ModuleDatabaseTransaction extends ModuleDatabaseQueryable { + /** Core names the savepoint. A resolved callback releases it; a rejected one rolls back to it. */ + savepoint(callback: (savepoint: ModuleDatabaseTransaction) => Promise | T): Promise; +} + +/** + * A module's database service under ABI v1. Top-level `query` calls are separate autocommit + * statements on whichever pooled connection is free: no session state carries from one to the + * next, and none survives the operation that made it. + */ +export interface ModuleDatabaseExecutor extends ModuleDatabaseQueryable { + /** Core begins the transaction, commits when `callback` resolves with no work outstanding, and + * rolls back otherwise. Core never retries it. */ + transaction( + callback: (transaction: ModuleDatabaseTransaction) => Promise | T, + options?: ModuleDatabaseTransactionOptions, + ): Promise; +} + +/** + * Every failure category an ABI v1 executor reports. A code is a stable category for programmatic + * handling and keeps its meaning once published; the specific reason is in the static message. + * PostgreSQL's own condition, when there is one, is `sqlState`, a separate field. + */ +export const MODULE_DATABASE_ERROR_CODES = [ + /** The call is not valid ABI use: a malformed query request or callback, a statement text that + * holds no statement, an object-mode result with duplicate column names, or a transaction-control + * statement sent through `query`. A programming error. */ + "MODULE_DATABASE_REQUEST_INVALID", + /** A parameter value is outside the ABI's parameter set. Refused before anything reaches + * PostgreSQL. */ + "MODULE_DATABASE_PARAMETER_INVALID", + /** The statement did not complete. `sqlState` is present when PostgreSQL reported the condition; + * absent when the database could not be reached or the connection failed. */ + "MODULE_DATABASE_QUERY_FAILED", + /** A transaction option or value is not one the ABI defines. Refused before the transaction + * starts. */ + "MODULE_DATABASE_TRANSACTION_OPTION_INVALID", + /** The transaction could not be begun or completed, or a savepoint could not be completed. + * `sqlState` is present when PostgreSQL reported the condition. A transaction left aborted by a + * failed statement ends here too: PostgreSQL rolls it back, and nothing is committed. */ + "MODULE_DATABASE_TRANSACTION_FAILED", + /** The executor's scope has finished, so it is revoked. */ + "MODULE_DATABASE_SCOPE_CLOSED", + /** The executor's scope is open but not active, because a savepoint inside it is open. */ + "MODULE_DATABASE_SCOPE_INACTIVE", + /** A transaction or savepoint callback settled while work it started was still outstanding. The + * scope was rolled back, never committed or released. */ + "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", +] as const; + +export type ModuleDatabaseErrorCode = (typeof MODULE_DATABASE_ERROR_CODES)[number]; + +const databaseErrorMessages: Readonly> = Object.freeze({ + MODULE_DATABASE_PARAMETER_INVALID: "A database parameter value is not supported.", + MODULE_DATABASE_QUERY_FAILED: "The database statement did not complete.", + MODULE_DATABASE_REQUEST_INVALID: "The database request is not valid.", + MODULE_DATABASE_SCOPE_CLOSED: "The database scope has finished and can no longer be used.", + MODULE_DATABASE_SCOPE_INACTIVE: "The database scope cannot be used while a savepoint is open.", + MODULE_DATABASE_SCOPE_WORK_OUTSTANDING: + "The database scope finished with work still outstanding, so it was rolled back.", + MODULE_DATABASE_TRANSACTION_FAILED: "The database transaction could not be completed.", + MODULE_DATABASE_TRANSACTION_OPTION_INVALID: "A database transaction option is not supported.", +}); + +const sqlStatePattern = /^[0-9A-Z]{5}$/u; + +/** + * The one error an ABI v1 executor reports, sanitized by construction: a stable `code`, PostgreSQL's + * `sqlState` when the database supplied one, and a static message. It never carries SQL text, + * parameter values, the driver's message, detail, hint, or object names, a cause, a stack beyond + * its own name and message, or any driver or ORM object. + * + * Recognize it with `isModuleDatabaseError`, never `instanceof`: a module bundles its own copy of + * this package (ADR 0049), so class identity across the boundary is not a contract. + */ +export class ModuleDatabaseError extends Error { + readonly code: ModuleDatabaseErrorCode; + /** Declared, not initialized, so the property exists only when PostgreSQL supplied a SQLSTATE. */ + declare readonly sqlState?: string; + + constructor(code: ModuleDatabaseErrorCode, options: { readonly sqlState?: string } = {}) { + super(databaseErrorMessages[code]); + this.name = "ModuleDatabaseError"; + this.code = code; + if (options.sqlState !== undefined && sqlStatePattern.test(options.sqlState)) { + this.sqlState = options.sqlState; + } + this.stack = `${this.name}: ${this.message}`; + Object.freeze(this); + } +} + +/** Whether `value` is a Module Database ABI error, judged by its shape rather than its class. */ +export function isModuleDatabaseError( + value: unknown, +): value is Pick { + if (!(value instanceof Error) || value.name !== "ModuleDatabaseError") { + return false; + } + const { code, sqlState } = value as { code?: unknown; sqlState?: unknown }; + return ( + typeof code === "string" && + (MODULE_DATABASE_ERROR_CODES as readonly string[]).includes(code) && + (sqlState === undefined || (typeof sqlState === "string" && sqlStatePattern.test(sqlState))) + ); +} diff --git a/packages/contracts/tests/module-database-abi.test.ts b/packages/contracts/tests/module-database-abi.test.ts new file mode 100644 index 0000000..3b3658d --- /dev/null +++ b/packages/contracts/tests/module-database-abi.test.ts @@ -0,0 +1,176 @@ +import { readFile } from "node:fs/promises"; +import { describe, expect, expectTypeOf, it } from "vitest"; +import { + isModuleDatabaseError, + MODULE_DATABASE_ABI_VERSION, + MODULE_DATABASE_ERROR_CODES, + type ModuleDatabaseError, + ModuleDatabaseError as ModuleDatabaseErrorClass, + type ModuleDatabaseExecutor, + type ModuleDatabaseParameter, + type ModuleDatabaseQueryable, + type ModuleDatabaseTransaction, + type ModuleDatabaseValue, +} from "../src/index.js"; + +/** + * Never invoked: proves at typecheck time that the ABI's executors carry exactly the ABI's members, + * and that no ORM, driver, pool, or client escape hatch is reachable through their types. Re-adding + * any of them would make an `@ts-expect-error` below unused, failing `pnpm typecheck`. + */ +function typeOnlyExclusions( + executor: ModuleDatabaseExecutor, + transaction: ModuleDatabaseTransaction, +): void { + // @ts-expect-error No raw pg pool or client. + void executor.pool; + // @ts-expect-error No raw pg client. + void executor.$client; + // @ts-expect-error No ORM internals. + void executor._; + // @ts-expect-error No query builder. + void executor.select; + // @ts-expect-error No relational query object. + void executor.execute; + // @ts-expect-error A transaction executor has no rollback(); a module throws instead. + void transaction.rollback; + // @ts-expect-error Nested transactions are savepoints, not transaction(). + void transaction.transaction; + // @ts-expect-error Savepoints are not named by modules. + void transaction.savepoint("name", async () => undefined); + // @ts-expect-error A Date is not an ABI parameter. + void executor.query({ text: "SELECT $1", values: [new Date()] }); + // @ts-expect-error A plain object is not implicit JSON. + void executor.query({ text: "SELECT $1", values: [{ a: 1 }] }); + // @ts-expect-error undefined is not a parameter. + void executor.query({ text: "SELECT $1", values: [undefined] }); +} +void typeOnlyExclusions; + +describe("Module Database ABI v1 types", () => { + it("names exactly the ABI's members on every executor", () => { + expectTypeOf().toEqualTypeOf< + "abiVersion" | "query" | "transaction" + >(); + expectTypeOf().toEqualTypeOf< + "abiVersion" | "query" | "savepoint" + >(); + expectTypeOf().toEqualTypeOf<"abiVersion" | "query">(); + expectTypeOf().toEqualTypeOf<1>(); + expect(MODULE_DATABASE_ABI_VERSION).toBe(1); + }); + + it("types array-mode rows positionally and object-mode rows by name", () => { + expectTypeOf>["rows"][number]>().toEqualTypeOf< + readonly ModuleDatabaseValue[] + >(); + expectTypeOf>["rows"][number]>().toEqualTypeOf< + Readonly> + >(); + }); + + it("keeps parameter and result values to PostgreSQL wire-safe primitives", () => { + expectTypeOf().not.toMatchTypeOf(); + expectTypeOf>().not.toMatchTypeOf(); + expectTypeOf().not.toMatchTypeOf(); + expectTypeOf<[bigint, Uint8Array, [string | null]]>().toMatchTypeOf(); + expectTypeOf().not.toMatchTypeOf(); + expectTypeOf().not.toMatchTypeOf(); + }); + + it("imports nothing at all, so no ORM or driver type can reach the ABI", async () => { + const source = await readFile( + new URL("../src/module-database-abi.ts", import.meta.url), + "utf8", + ); + expect(source).not.toMatch(/^\s*import\b/mu); + expect(source).not.toMatch(/\bfrom\s+["']/u); + expect(source).not.toMatch(/drizzle|kysely|NodePg|PgTable|PgColumn/u); + }); +}); + +function arrayQuery(executor: ModuleDatabaseExecutor) { + return executor.query({ rowMode: "array", text: "SELECT 1" }); +} + +function objectQuery(executor: ModuleDatabaseExecutor) { + return executor.query({ text: "SELECT 1" }); +} + +describe("ModuleDatabaseError", () => { + it("separates the Ekklesia category from PostgreSQL's own SQLSTATE", () => { + const failed = new ModuleDatabaseErrorClass("MODULE_DATABASE_QUERY_FAILED", { + sqlState: "23505", + }); + expect(failed.code).toBe("MODULE_DATABASE_QUERY_FAILED"); + expect(failed.sqlState).toBe("23505"); + + const closed = new ModuleDatabaseErrorClass("MODULE_DATABASE_SCOPE_CLOSED"); + expect(closed.code).toBe("MODULE_DATABASE_SCOPE_CLOSED"); + expect(closed.sqlState).toBeUndefined(); + expect("sqlState" in closed).toBe(false); + }); + + it("never records anything but a five-character SQLSTATE", () => { + for (const candidate of ["EPIPE1", "2350", "23505 ", "duplicate key", "econn"]) { + const error = new ModuleDatabaseErrorClass("MODULE_DATABASE_QUERY_FAILED", { + sqlState: candidate, + }); + expect(error.sqlState).toBeUndefined(); + } + }); + + it("is sanitized by construction: a static message, its own stack line, no cause, frozen", () => { + for (const code of MODULE_DATABASE_ERROR_CODES) { + const error = new ModuleDatabaseErrorClass(code, { sqlState: "40001" }); + expect(error.name).toBe("ModuleDatabaseError"); + expect(error.stack).toBe(`ModuleDatabaseError: ${error.message}`); + expect(error.cause).toBeUndefined(); + expect(Object.keys(error).sort()).toEqual(["code", "name", "sqlState"]); + expect(Object.isFrozen(error)).toBe(true); + expect(error.message).not.toMatch(/select|insert|\$1|password|postgres:\/\//iu); + } + }); + + it("has one distinct message per code", () => { + const messages = MODULE_DATABASE_ERROR_CODES.map( + (code) => new ModuleDatabaseErrorClass(code).message, + ); + expect(new Set(messages).size).toBe(MODULE_DATABASE_ERROR_CODES.length); + }); + + it("is recognized by shape, so a module's own bundled copy of this package agrees", () => { + const fromAnotherCopy = Object.assign(new Error("The database request is not valid."), { + code: "MODULE_DATABASE_REQUEST_INVALID", + name: "ModuleDatabaseError", + }); + expect(isModuleDatabaseError(fromAnotherCopy)).toBe(true); + expect( + isModuleDatabaseError(new ModuleDatabaseErrorClass("MODULE_DATABASE_SCOPE_CLOSED")), + ).toBe(true); + + expect( + isModuleDatabaseError({ code: "MODULE_DATABASE_SCOPE_CLOSED", name: "ModuleDatabaseError" }), + ).toBe(false); + expect( + isModuleDatabaseError( + Object.assign(new Error("x"), { code: "SOMETHING_ELSE", name: "ModuleDatabaseError" }), + ), + ).toBe(false); + expect( + isModuleDatabaseError( + Object.assign(new Error("x"), { + code: "MODULE_DATABASE_QUERY_FAILED", + name: "ModuleDatabaseError", + sqlState: "not a sqlstate", + }), + ), + ).toBe(false); + expect(isModuleDatabaseError(Object.assign(new Error("x"), { code: "23505" }))).toBe(false); + + const narrowed: unknown = fromAnotherCopy; + if (isModuleDatabaseError(narrowed)) { + expectTypeOf(narrowed.code).toEqualTypeOf(); + } + }); +}); diff --git a/vitest.migrations.integration.config.ts b/vitest.migrations.integration.config.ts index b11897c..c53775c 100644 --- a/vitest.migrations.integration.config.ts +++ b/vitest.migrations.integration.config.ts @@ -14,6 +14,7 @@ export default defineConfig({ "core/tests/module-configuration-store.integration.ts", "core/tests/module-storage.integration.ts", "core/tests/module-delivery.integration.ts", + "core/tests/module-database-abi.integration.ts", "apps/api/tests/api-http.integration.ts", "apps/api/tests/initial-owner.integration.ts", "apps/api/tests/device-pairing.integration.ts", -- 2.54.0 From 317a4ce410cb292257bfc0e7e0ae2f66ade41dc1 Mon Sep 17 00:00:00 2001 From: ekklesia Date: Wed, 7 Oct 2026 16:12:01 -0500 Subject: [PATCH 2/2] fix: ADR 0050 callback-error precedence, and the honest limits of transaction control Narrow contract and implementation adjustments from review. No SDK, manifest, packaging, or Signage work. Callback-error precedence: - A transaction or savepoint callback that rejects keeps its own error, even while sibling work it started is still settling (one failed half of a Promise.all). Core drains that work, rolls back, and never commits or releases. - A callback that resolves with work outstanding (a forgotten await) still fails with MODULE_DATABASE_SCOPE_WORK_OUTSTANDING and rolls back. Transaction-control SQL: - ADR 0050 forbids it through query() as a conformance rule for trusted code, not a security guarantee. Core does not parse SQL, so PostgreSQL has run the statement before Core sees it. Core detects the violation from the command tag or the transaction status, and claims nothing it cannot do: a module's own COMMIT has committed. - After a violation, the transaction's scopes refuse every later statement without sending it (it would otherwise autocommit). Core never commits the transaction itself, and the connection is destroyed rather than reused, for top-level violations too. - The commit and rollback guarantees apply to conforming use. Duplicate object-mode columns: documented as found only after execution, so a mutation can have taken effect; atomicity needs an ABI transaction. Tests prove each of these against PostgreSQL: Promise.all failures in a transaction and a savepoint keep 23505; a forgotten await still rolls back; a module COMMIT's prior work is committed while nothing after it runs; a backend PID changes after every violation; duplicate RETURNING columns persist at the top level and roll back in a transaction. ADR 0050 stays Accepted: the executor conforms to the revised text. Co-Authored-By: Claude Opus 5.5 --- core/src/modules/database-abi.ts | 69 ++++++-- core/tests/module-database-abi.integration.ts | 166 +++++++++++++++--- core/tests/module-database-abi.test.ts | 40 ++++- .../decisions/0050-module-database-abi-v1.md | 52 ++++-- docs/development/testing.md | 7 +- packages/contracts/src/module-database-abi.ts | 28 ++- 6 files changed, 300 insertions(+), 62 deletions(-) diff --git a/core/src/modules/database-abi.ts b/core/src/modules/database-abi.ts index 9019110..ccf7a48 100644 --- a/core/src/modules/database-abi.ts +++ b/core/src/modules/database-abi.ts @@ -49,13 +49,18 @@ import { * not parse SQL. A statement text that holds no statement at all is refused after PostgreSQL reports * it empty. * - * ### Transaction control belongs to Core + * ### Transaction control belongs to the callbacks * - * A statement whose command tag is transaction control (`BEGIN`, `START`, `COMMIT`, `ROLLBACK`, - * `SAVEPOINT`, `RELEASE`, `PREPARE`), or that leaves the connection's transaction status other than - * expected, is refused. PostgreSQL has already run it by then, so inside a transaction it also marks - * the transaction failed: Core rolls back and never commits it. The command tag and the protocol's - * transaction status come from PostgreSQL, not from parsing the SQL. + * Transaction-control SQL (`BEGIN`, `START TRANSACTION`, `COMMIT`, `ROLLBACK`, `SAVEPOINT`, + * `RELEASE SAVEPOINT`, `PREPARE TRANSACTION`, and their aliases) is forbidden through `query`: a + * conformance rule for trusted module code, not a security guarantee. Core cannot refuse such a + * statement beforehand without parsing SQL, which it does not do, and it cannot undo one PostgreSQL + * has already executed: a `COMMIT` sent through a transaction's `query` has committed by the time + * Core sees it. What Core does is detect the violation afterwards, from what PostgreSQL itself + * reports (the statement's command tag, and the protocol's transaction status), and fail safely: + * the call rejects, an enclosing transaction is never committed by Core and rejects too, and the + * connection is destroyed rather than returned to the pool. The commit and rollback guarantees of + * the transaction API apply to conforming use. */ /** The settings the ABI's decoding and determinism depend on, pinned for every ABI operation. */ @@ -281,6 +286,15 @@ function assertActive(scope: Scope): void { } } +/** Once a transaction can no longer be committed, its scopes send nothing more. After a + * transaction-control violation in particular, a later statement would otherwise run outside any + * transaction, as its own autocommitted statement. */ +function assertUsable(state: TransactionState): void { + if (state.failure !== undefined) { + throw new ModuleDatabaseError(state.failure.code); + } +} + function track(state: TransactionState, promise: Promise): Promise { const shadow = promise.then( () => undefined, @@ -379,7 +393,9 @@ export function createModuleDatabaseExecutor( (raw.command !== null && transactionControlCommands.has(raw.command)) || client.getTransactionStatus() !== "I" ) { - // The connection is destroyed below if the statement left a transaction open. + // A transaction-control violation. PostgreSQL has already run it; the connection is + // destroyed below rather than returned in a state Core did not choose. + healthy = false; throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } return buildResult(raw, prepared.rowMode); @@ -393,6 +409,7 @@ export function createModuleDatabaseExecutor( request: unknown, ): Promise> { assertActive(scope); + assertUsable(state); const prepared = prepareQuery(request); scope.pending += 1; try { @@ -409,7 +426,10 @@ export function createModuleDatabaseExecutor( (raw.command !== null && transactionControlCommands.has(raw.command)) || state.client.getTransactionStatus() === "I" ) { + // A transaction-control violation. PostgreSQL has already run it, and Core cannot undo + // that; it can only never commit this transaction itself, and destroy the connection. state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); + state.healthy = false; throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } return buildResult(raw, prepared.rowMode); @@ -422,6 +442,7 @@ export function createModuleDatabaseExecutor( callback: (savepoint: ModuleDatabaseTransaction) => Promise | T, ): Promise { assertActive(scope); + assertUsable(state); if (typeof callback !== "function") { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } @@ -444,20 +465,28 @@ export function createModuleDatabaseExecutor( // The enclosing transaction finished first; it rolls back, and nothing more is sent. throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); } - if (outstanding || "error" in outcome) { + if ("error" in outcome) { + // A rejected callback's own error stays the error, even if sibling work it started was + // still settling: that work has been drained, and is undone here with everything else. try { await control(state, `ROLLBACK TO SAVEPOINT ${name}`); - if (!outstanding) { - await control(state, `RELEASE SAVEPOINT ${name}`); - } + await control(state, `RELEASE SAVEPOINT ${name}`); } catch (error) { state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); throw error; } - if (outstanding) { - throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + throw outcome.error; + } + if (outstanding) { + // A callback that resolved with work outstanding forgot an await. Roll back to the + // savepoint and never release it. + try { + await control(state, `ROLLBACK TO SAVEPOINT ${name}`); + } catch (error) { + state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); + throw error; } - throw (outcome as { readonly error: unknown }).error; + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); } try { await control(state, `RELEASE SAVEPOINT ${name}`); @@ -522,14 +551,18 @@ export function createModuleDatabaseExecutor( await drain(state); state.finished = true; - if (outstanding) { - await rollback(state); - throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); - } if ("error" in outcome) { + // The callback's own error stays primary, even when sibling work it started (say, the + // other half of a `Promise.all`) was still settling. That work has been drained, and rolls + // back with the rest; outstanding work never leads to a commit. await rollback(state); throw outcome.error; } + if (outstanding) { + // The callback resolved with work it never awaited: a forgotten `await`. + await rollback(state); + throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + } if (state.failure !== undefined) { await rollback(state); throw state.failure; diff --git a/core/tests/module-database-abi.integration.ts b/core/tests/module-database-abi.integration.ts index d3476a2..57aedc9 100644 --- a/core/tests/module-database-abi.integration.ts +++ b/core/tests/module-database-abi.integration.ts @@ -228,6 +228,28 @@ describe("query", () => { ); }); + it("finds duplicate object-mode columns only after the statement ran, so atomicity needs a transaction", async () => { + await reset(); + // At the top level, the mutation has happened by the time the result's shape is known. + await expectAbiError( + database.query({ + text: "INSERT INTO example_records (id, value) VALUES (1, 'a') RETURNING id, id", + }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + expect(await storedIds()).toEqual([1]); + // Inside a transaction, the same error rolls the mutation back with everything else. + await expectAbiError( + database.transaction((tx) => + tx.query({ + text: "INSERT INTO example_records (id, value) VALUES (2, 'b') RETURNING id, id", + }), + ), + "MODULE_DATABASE_REQUEST_INVALID", + ); + expect(await storedIds()).toEqual([1]); + }); + it("keeps a column named __proto__ a column", async () => { const result = await database.query({ text: 'SELECT 1 AS "__proto__"' }); const row = result.rows[0] as Record; @@ -838,22 +860,25 @@ describe("transactions", () => { expect(called).toBe(false); }); - it("refuses transaction control sent through query, and never commits the transaction", async () => { - await reset(); - for (const statements of [ - ["SAVEPOINT mine"], - ["SAVEPOINT mine", "RELEASE SAVEPOINT mine"], - ["SAVEPOINT mine", "ROLLBACK TO SAVEPOINT mine"], - ["BEGIN"], - ["START TRANSACTION"], - ]) { + it("detects transaction control sent through query once PostgreSQL has run it, then sends nothing more and commits nothing", async () => { + // Forbidden by ADR 0050, and not refusable beforehand without parsing SQL. Core detects each + // from PostgreSQL's command tag or the protocol's transaction status. + for (const statement of ["SAVEPOINT mine", "BEGIN", "START TRANSACTION", "ROLLBACK", "ABORT"]) { + await reset(); await expectAbiError( database.transaction(async (tx) => { await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); - for (const statement of statements) { - // PostgreSQL has run each statement by the time it is refused. - await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID"); - } + await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID"); + // Anything after the violation is refused without being sent: it would otherwise run + // outside any transaction. + await expectAbiError( + tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + await expectAbiError( + tx.savepoint(async () => undefined), + "MODULE_DATABASE_REQUEST_INVALID", + ); return "carried on"; }), "MODULE_DATABASE_REQUEST_INVALID", @@ -862,15 +887,25 @@ describe("transactions", () => { } }); - it("refuses a COMMIT sent through query and fails the transaction", async () => { - await reset(); - await expectAbiError( - database.transaction(async (tx) => { - await expectAbiError(tx.query({ text: "COMMIT" }), "MODULE_DATABASE_REQUEST_INVALID"); - return "carried on"; - }), - "MODULE_DATABASE_REQUEST_INVALID", - ); + it("cannot undo a COMMIT sent through query: PostgreSQL has committed, and Core reports the violation and commits nothing more", async () => { + for (const statement of ["COMMIT", "END"]) { + await reset(); + await expectAbiError( + database.transaction(async (tx) => { + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID"); + await expectAbiError( + tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + return "carried on"; + }), + "MODULE_DATABASE_REQUEST_INVALID", + ); + // The documented limit: the work before the module's own COMMIT was committed by PostgreSQL + // when it ran that statement. Nothing after it ran. + expect(await storedIds()).toEqual([1]); + } }); it("applies Core's statement bound inside a transaction, and transaction-local settings work", async () => { @@ -917,6 +952,65 @@ describe("outstanding work", () => { expect(await storedIds()).toEqual([]); }); + it("keeps the failing query's own error when one half of a Promise.all fails while the other is still running", async () => { + await reset(); + await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + let sibling: Promise | undefined; + await expectAbiError( + database.transaction(async (tx) => { + const conflicting = tx.query({ + text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')", + }); + sibling = tx.query({ + text: "INSERT INTO example_records (id, value) SELECT 2, 'b' FROM pg_sleep(0.1)", + }); + sibling.catch(() => undefined); + await Promise.all([conflicting, sibling]); + }), + // The PostgreSQL failure, not MODULE_DATABASE_SCOPE_WORK_OUTSTANDING. + "MODULE_DATABASE_QUERY_FAILED", + "23505", + ); + // The sibling was still outstanding when the callback rejected: it settled afterwards (failing, + // because the transaction was already aborted), before Core rolled back. + await expect(sibling).rejects.toMatchObject({ sqlState: "25P02" }); + expect(await storedIds()).toEqual([1]); + }); + + it("keeps a savepoint's failing query error from a Promise.all, and the transaction carries on", async () => { + await reset(); + await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }); + await database.transaction(async (tx) => { + await expectAbiError( + tx.savepoint(async (sp) => { + const sibling = sp.query({ + text: "INSERT INTO example_records (id, value) SELECT 2, 'b' FROM pg_sleep(0.1)", + }); + sibling.catch(() => undefined); + await Promise.all([ + sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')" }), + sibling, + ]); + }), + "MODULE_DATABASE_QUERY_FAILED", + "23505", + ); + await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" }); + }); + expect(await storedIds()).toEqual([1, 3]); + }); + + it("commits a Promise.all whose queries all succeed", async () => { + await reset(); + await database.transaction(async (tx) => { + await Promise.all([ + tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }), + tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }), + ]); + }); + expect(await storedIds()).toEqual([1, 2]); + }); + it("treats an unawaited savepoint as outstanding work", async () => { await reset(); let forgotten: Promise | undefined; @@ -1259,6 +1353,34 @@ describe("pool safety", () => { }); } + it("destroys, never reuses, a connection after transaction-control misuse", async () => { + const pid = async (executor: ModuleDatabaseExecutor | ModuleDatabaseTransaction) => + ( + (await executor.query({ text: "SELECT pg_backend_pid() AS pid" })).rows[0] as { + pid: number; + } + ).pid; + for (const statement of ["COMMIT", "ROLLBACK", "SAVEPOINT mine", "BEGIN"]) { + let used: number | undefined; + await rejection( + single.transaction(async (tx) => { + used = await pid(tx); + await rejection(tx.query({ text: statement })); + }), + ); + expect(used).toBeTypeOf("number"); + expect(await pid(single)).not.toBe(used); + expect(await nextBorrower(onePool)).toEqual(clean); + } + for (const statement of ["BEGIN", "COMMIT"]) { + const before = await pid(single); + // With one connection, the next top-level query would reuse it unless it was destroyed. + await rejection(single.query({ text: statement })); + expect(await pid(single)).not.toBe(before); + expect(await nextBorrower(onePool)).toEqual(clean); + } + }); + it("reports a terminated connection without a SQLSTATE of its own making", async () => { const error = await rejection( single.transaction(async (tx) => { diff --git a/core/tests/module-database-abi.test.ts b/core/tests/module-database-abi.test.ts index 40ead6f..3384e64 100644 --- a/core/tests/module-database-abi.test.ts +++ b/core/tests/module-database-abi.test.ts @@ -654,20 +654,25 @@ describe("transaction scopes", () => { expect(texts(script)).not.toContain("COMMIT"); }); - it("reports outstanding work even when the callback itself rejected", async () => { + it("keeps a rejected callback's own error when sibling work is still settling, and still never commits", async () => { const script = scriptedPool(); const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + const own = new Error("module failure"); let releaseHeld: (() => void) | undefined; const outcome = executor.transaction(async (tx) => { releaseHeld = script.holdNext(); void tx.query({ text: "INSERT INTO example_records VALUES (1)" }); - throw new Error("module failure"); + throw own; }); await vi.waitFor(() => expect(releaseHeld).toBeDefined()); + // The sibling statement is drained before Core rolls back. + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(texts(script)).not.toContain("ROLLBACK"); releaseHeld?.(); - await expectAbiError(outcome, "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); + expect(await rejection(outcome)).toBe(own); + expect(texts(script).slice(-2)).toEqual(["ROLLBACK", "DISCARD ALL"]); expect(texts(script)).not.toContain("COMMIT"); }); @@ -688,6 +693,9 @@ describe("transaction scopes", () => { expect(error).toMatchObject({ code: "MODULE_DATABASE_REQUEST_INVALID" }); expect(texts(script)).not.toContain("COMMIT"); expect(texts(script)).toContain("ROLLBACK"); + // The connection is destroyed, never reset and reused. + expect(texts(script)).not.toContain("DISCARD ALL"); + expect(script.released).toEqual([true]); }); it("reports a COMMIT that PostgreSQL answered with ROLLBACK as a failed transaction", async () => { @@ -794,6 +802,32 @@ describe("savepoint scopes", () => { ]); }); + it("keeps a rejected savepoint callback's own error, rolling back and releasing after its sibling work settles", async () => { + const script = scriptedPool(); + const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); + const own = new Error("savepoint failure"); + + await executor.transaction(async (tx) => { + const error = await rejection( + tx.savepoint(async (sp) => { + const release = script.holdNext(); + void sp.query({ text: "INSERT INTO example_records VALUES (1)" }); + setTimeout(release, 10); + throw own; + }), + ); + expect(error).toBe(own); + }); + + expect(texts(script).slice(2, -2)).toEqual([ + "SAVEPOINT ekklesia_savepoint_1", + "INSERT INTO example_records VALUES (1)", + "ROLLBACK TO SAVEPOINT ekklesia_savepoint_1", + "RELEASE SAVEPOINT ekklesia_savepoint_1", + ]); + expect(texts(script)).toContain("COMMIT"); + }); + it("counts an unawaited savepoint as the transaction's outstanding work", async () => { const script = scriptedPool(); const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" }); diff --git a/docs/architecture/decisions/0050-module-database-abi-v1.md b/docs/architecture/decisions/0050-module-database-abi-v1.md index d107aca..484c176 100644 --- a/docs/architecture/decisions/0050-module-database-abi-v1.md +++ b/docs/architecture/decisions/0050-module-database-abi-v1.md @@ -181,8 +181,13 @@ transaction(callback, options) → the callback's result - `"object"` rows are keyed by column name. - `"array"` rows are positional, in `fields` order. An ORM that selects the same column name twice, as joins do, needs this mode. - - Core rejects an object-mode result that has duplicate column names, rather than silently - dropping one. + - Duplicate column names are invalid in object mode. Core rejects such a result rather than + silently dropping a column. + - With the current PostgreSQL and driver execution path, the result's shape is only known once + the statement has run. So a mutating statement can already have had its effects when the + result-shape error is reported. + - This is not a security guarantee. A caller that needs atomicity runs the mutation inside an ABI + transaction, where the error rolls the mutation back with everything else. - **`command`** is PostgreSQL's command tag word, such as `SELECT`, `INSERT`, `UPDATE`, or `DELETE`. - **`rowCount`** is the number of rows PostgreSQL reports as processed, or `null` when it reports none. @@ -331,15 +336,37 @@ transaction(async (tx) => { … }, { isolation, readOnly }) transaction call rejects with the callback's error. There is no `rollback()` method: to roll back, the module throws. - **All work is awaited inside the callback.** Module code must await every database operation it - starts through `tx` before its callback returns. - - If the callback settles while an operation launched through that scope is still outstanding, the - module has used the scope incorrectly. - - The transaction is then never committed: Core rolls it back, and the transaction call rejects - with the scope-usage `ModuleDatabaseError`. - - A forgotten `await` therefore fails. It never quietly changes what the transaction contains. + starts through `tx` before its callback returns. What happens when work is still outstanding + depends on how the callback settled: + - **Resolved with work outstanding.** The module forgot an `await` and has used the scope + incorrectly. The transaction is never committed: Core rolls it back, and the transaction call + rejects with the scope-usage `ModuleDatabaseError`. A forgotten `await` therefore fails. It never + quietly changes what the transaction contains. + - **Rejected with work outstanding.** The callback's own rejection stays the transaction's error. + Core rolls back and does not replace that error, a PostgreSQL failure included, with the + scope-usage error just because sibling work was still settling. This is what happens when one + query in `await Promise.all([tx.query(a), tx.query(b)])` fails while the other is still running. + - Either way, outstanding work never leads to a commit. - Core may let operations that already started finish, or settle, before it rolls back and releases the connection, so that a pooled connection is never handed on mid-operation. Waiting for them never turns into a commit. +- **Transaction control belongs to the callbacks.** Transaction-control SQL is forbidden through + `query`: `BEGIN`, `START TRANSACTION`, `COMMIT`, `ROLLBACK`, `SAVEPOINT`, `RELEASE SAVEPOINT`, and + every other statement that starts, ends, or marks a transaction. Modules control a transaction's + lifecycle only through the `transaction` and `savepoint` callbacks. + - This is a conformance rule for trusted module code, not a sandbox or a security guarantee. + - Core does not parse SQL to enforce it. Such a statement is valid PostgreSQL, so PostgreSQL + executes it before Core can see its result. + - Core detects a violation where the driver and protocol make that possible: from the statement's + command tag, and from the connection's transaction status. It then fails safely: + - the call rejects; + - the transaction's scopes refuse every later statement without sending it, because it would + otherwise run outside any transaction; + - Core never commits the enclosing transaction itself, and that transaction rejects too; + - the connection is destroyed rather than reused. + - Core does not claim to undo a transaction-control statement PostgreSQL has already executed. A + `COMMIT` sent through a transaction's `query` has already committed the work before it. + - The commit and rollback guarantees in this section apply to conforming use of the ABI. - **Revocation.** When the callback settles, `tx` is revoked, and it stays revoked once the scope has finished. Any later call on it fails with the closed-scope `ModuleDatabaseError` and sends nothing to PostgreSQL. A `tx` that escapes its @@ -362,9 +389,12 @@ tx.savepoint(async (sp) => { … }) - **Naming.** Core generates every savepoint identifier. Modules never provide one. - **Release and rollback.** A resolved callback releases the savepoint. A rejected one rolls back to it, releases it, and rejects with the callback's error. -- **Outstanding work.** The awaiting rule for transactions applies unchanged. A savepoint callback - that settles with operations still outstanding is a scope-usage error. The savepoint is rolled - back, never released, and the call rejects with that error. +- **Outstanding work.** The awaiting rule for transactions applies unchanged. + - A savepoint callback that resolves with operations still outstanding is a scope-usage error. + The savepoint is rolled back, never released, and the call rejects with that error. + - A savepoint callback that rejects keeps its own error. The savepoint is rolled back as for any + rejection. + - Either way, outstanding work never leads to a release. - **One active scope.** Only the innermost open scope may run statements. While a savepoint is open, its parent executor refuses statements and new savepoints, so savepoint scopes can never overlap. diff --git a/docs/development/testing.md b/docs/development/testing.md index 3cd6214..ffa962a 100644 --- a/docs/development/testing.md +++ b/docs/development/testing.md @@ -522,7 +522,12 @@ because every claim is a property of PostgreSQL or of the protocol: - commit, rollback, isolation levels, read-only, no retries, and an aborted transaction never being committed; - callback-scoped savepoints; -- a forgotten `await` rolling back instead of committing; +- a forgotten `await` rolling back instead of committing, while a rejected callback (one half of a + failed `Promise.all`) keeps its own error; +- transaction-control SQL sent through `query` detected only after PostgreSQL has run it: Core + cannot undo a module's own `COMMIT`, so the suite pins what it does do: it refuses everything + after, never commits, and destroys the connection; +- duplicate object-mode columns found only after a mutation ran, unless it ran in a transaction; - revoked executors reaching nothing. The suite's database is created with deliberately hostile defaults: a non-UTC time zone, a non-ISO diff --git a/packages/contracts/src/module-database-abi.ts b/packages/contracts/src/module-database-abi.ts index 7c066d1..8ef9b59 100644 --- a/packages/contracts/src/module-database-abi.ts +++ b/packages/contracts/src/module-database-abi.ts @@ -105,11 +105,17 @@ export interface ModuleDatabaseTransactionOptions { * that callback settles it is revoked, and every later call fails without reaching PostgreSQL. * * Only the innermost open scope may run statements: while a savepoint is open, its parent refuses - * statements and new savepoints. A callback must await every operation it starts before it returns; - * one that settles with work outstanding is never committed or released. + * statements and new savepoints. A callback must await every operation it starts before it returns. + * One that resolves with work outstanding forgot an `await`: it fails with + * `MODULE_DATABASE_SCOPE_WORK_OUTSTANDING` and is rolled back. One that rejects keeps its own error. + * Either way, outstanding work never leads to a commit or a release. + * + * Transaction-control SQL (`BEGIN`, `COMMIT`, `ROLLBACK`, `SAVEPOINT`, `RELEASE SAVEPOINT`, …) is + * forbidden through `query`; the callbacks control the transaction. */ export interface ModuleDatabaseTransaction extends ModuleDatabaseQueryable { - /** Core names the savepoint. A resolved callback releases it; a rejected one rolls back to it. */ + /** Core names the savepoint. A resolved callback with no work outstanding releases it; a rejected + * one rolls back to it and keeps its own error. */ savepoint(callback: (savepoint: ModuleDatabaseTransaction) => Promise | T): Promise; } @@ -120,7 +126,9 @@ export interface ModuleDatabaseTransaction extends ModuleDatabaseQueryable { */ export interface ModuleDatabaseExecutor extends ModuleDatabaseQueryable { /** Core begins the transaction, commits when `callback` resolves with no work outstanding, and - * rolls back otherwise. Core never retries it. */ + * rolls back otherwise; a rejected callback's own error is the transaction's error. Core never + * retries it. These guarantees apply to conforming use: no transaction-control SQL through + * `query`. */ transaction( callback: (transaction: ModuleDatabaseTransaction) => Promise | T, options?: ModuleDatabaseTransactionOptions, @@ -135,7 +143,12 @@ export interface ModuleDatabaseExecutor extends ModuleDatabaseQueryable { export const MODULE_DATABASE_ERROR_CODES = [ /** The call is not valid ABI use: a malformed query request or callback, a statement text that * holds no statement, an object-mode result with duplicate column names, or a transaction-control - * statement sent through `query`. A programming error. */ + * statement sent through `query`. A programming error. + * + * The last two are only detectable once PostgreSQL has run the statement, so its effects may + * already have happened: a mutation with duplicate result columns has been applied (inside a + * transaction, it rolls back with it), and transaction control has taken effect, which Core cannot + * undo. */ "MODULE_DATABASE_REQUEST_INVALID", /** A parameter value is outside the ABI's parameter set. Refused before anything reaches * PostgreSQL. */ @@ -154,8 +167,9 @@ export const MODULE_DATABASE_ERROR_CODES = [ "MODULE_DATABASE_SCOPE_CLOSED", /** The executor's scope is open but not active, because a savepoint inside it is open. */ "MODULE_DATABASE_SCOPE_INACTIVE", - /** A transaction or savepoint callback settled while work it started was still outstanding. The - * scope was rolled back, never committed or released. */ + /** A transaction or savepoint callback resolved while work it started was still outstanding: a + * forgotten `await`. The scope was rolled back, never committed or released. (A callback that + * rejects keeps its own error instead.) */ "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING", ] as const; -- 2.54.0