ekklesiaandClaude Opus 5.5
317a4ce410
CI / verify (pull_request) Successful in 1m4s
CI / infrastructure-smoke (pull_request) Successful in 38s
CI / migration-smoke (pull_request) Successful in 1m54s
CI / image-build (pull_request) Successful in 22s
CI / deploy-smoke (same-origin) (pull_request) Successful in 51s
CI / deploy-smoke (split) (pull_request) Successful in 1m37s
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 <[email protected]>
1449 lines
54 KiB
TypeScript
1449 lines
54 KiB
TypeScript
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<string>();
|
|
|
|
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<T>(
|
|
connectionString: string,
|
|
run: (client: Client) => Promise<T>,
|
|
): Promise<T> {
|
|
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<string> {
|
|
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<void> {
|
|
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<void>;
|
|
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<unknown>): Promise<unknown> {
|
|
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<unknown>,
|
|
code: ModuleDatabaseErrorCode,
|
|
sqlState?: string,
|
|
secrets: readonly string[] = [],
|
|
): Promise<void> {
|
|
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<void> {
|
|
await withAdministrativeClient(databaseUrl, async (admin) => {
|
|
await admin.query("TRUNCATE example_records, example_deferred");
|
|
});
|
|
}
|
|
|
|
async function storedIds(): Promise<number[]> {
|
|
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("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<string, unknown>;
|
|
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<string, unknown>;
|
|
|
|
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<string, unknown>;
|
|
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("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')" });
|
|
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",
|
|
);
|
|
expect(await storedIds()).toEqual([]);
|
|
}
|
|
});
|
|
|
|
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 () => {
|
|
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<unknown> | 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("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<unknown> | 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<unknown> | 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<BorrowerState> {
|
|
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<unknown>][] = [
|
|
["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("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) => {
|
|
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" }]);
|
|
});
|
|
});
|