feat: Core's Module Database ABI v1 executor, proven against PostgreSQL (ADR 0050)
CI / verify (pull_request) Successful in 59s
CI / infrastructure-smoke (pull_request) Successful in 27s
CI / migration-smoke (pull_request) Successful in 1m53s
CI / image-build (pull_request) Successful in 23s
CI / deploy-smoke (same-origin) (pull_request) Successful in 41s
CI / deploy-smoke (split) (pull_request) Successful in 1m29s
CI / verify (pull_request) Successful in 59s
CI / infrastructure-smoke (pull_request) Successful in 27s
CI / migration-smoke (pull_request) Successful in 1m53s
CI / image-build (pull_request) Successful in 23s
CI / deploy-smoke (same-origin) (pull_request) Successful in 41s
CI / deploy-smoke (split) (pull_request) Successful in 1m29s
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 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
d43031997b
commit
6db50516b7
@@ -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<string> = new Set([
|
||||
"BEGIN",
|
||||
"COMMIT",
|
||||
"PREPARE",
|
||||
"RELEASE",
|
||||
"ROLLBACK",
|
||||
"SAVEPOINT",
|
||||
"START",
|
||||
]);
|
||||
|
||||
const isolationStatements: ReadonlyMap<unknown, string> = 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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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<ModuleDatabaseArrayRow | ModuleDatabaseObjectRow> {
|
||||
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<AbiClient> {
|
||||
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<void> {
|
||||
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<Scope>;
|
||||
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<Promise<void>>;
|
||||
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<T>(state: TransactionState, promise: Promise<T>): Promise<T> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
try {
|
||||
await state.client.query("ROLLBACK");
|
||||
} catch {
|
||||
state.healthy = false;
|
||||
}
|
||||
}
|
||||
|
||||
type Outcome<T> = { readonly value: T } | { readonly error: unknown };
|
||||
|
||||
async function settle<T>(run: () => Promise<T> | T): Promise<Outcome<T>> {
|
||||
try {
|
||||
return { value: await run() };
|
||||
} catch (error) {
|
||||
return { error };
|
||||
}
|
||||
}
|
||||
|
||||
const moduleByExecutor = new WeakMap<object, ModuleId>();
|
||||
|
||||
/** 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<ModuleDatabaseQueryResult<ModuleDatabaseArrayRow | ModuleDatabaseObjectRow>> {
|
||||
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<ModuleDatabaseQueryResult<ModuleDatabaseArrayRow | ModuleDatabaseObjectRow>> {
|
||||
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<T>(
|
||||
callback: (savepoint: ModuleDatabaseTransaction) => Promise<T> | T,
|
||||
): Promise<T> {
|
||||
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<T>(
|
||||
callback: (transaction: ModuleDatabaseTransaction) => Promise<T> | T,
|
||||
options?: unknown,
|
||||
): Promise<T> {
|
||||
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;
|
||||
}
|
||||
Reference in New Issue
Block a user