ekklesiaandClaude Opus 5
6912680113
pg-connection-string retains arbitrary query keys, so the driver's own configuration key connectionTimeoutMillis reaches ConnectionParameters straight from a URL — and it was not on the rejected list, while spellings that do nothing were. Rejecting it closes that gap. The precedence claims that guard was built on were partly wrong, so every source was measured against the installed [email protected].0, a real blackhole listener, and a real PostgreSQL rather than inferred from the merge: statement_timeout=0 / =900000 server reported 0 and 15min — wins outright query_timeout=0 overrides the field, but the parsed value is the string "0", which is truthy, so every query gets a 0 ms read timer and fails nondeterministically — it breaks the client rather than loosening the bound connectionTimeoutMillis=0/900000 pollutes connectionParameters.connect_timeout (10 -> 0 or 900 s), but that field only feeds getLibpqConnectionString(), which the pure-JS client never calls; the effective timer was unchanged, with an explicit 500 ms still expiring at ~500 ms either way connect_timeout=0 no effect: recomputed from the explicit field options=-c statement_timeout=… no effect, including a backslash-escaped spelling — the named startup parameter wins PGOPTIONS / PGCONNECT_TIMEOUT no effect unrelated parameters no effect, deliberately untouched Only statement_timeout is a genuine downgrade. The rest are rejected as unsupported configuration and defense in depth: no legitimate DATABASE_URL carries a driver option, the polluted connect_timeout would take effect on the native binding, and query_timeout produces intermittent failures an operator could not diagnose. That distinction is now stated where it was previously asserted wrongly — the earlier comment and ADR text claimed connect_timeout overrode the explicit field, which measurement disproved. Rejection still happens at the one shared construction boundary, before any pg.Client exists, with the existing fixed POSTGRES_MIGRATION_CONFIGURATION_INVALID and no URL, host, port, credential, parameter, or value in the message. The 10 s defaults, narrower test overrides, Pass 2 TLS policy and its URL-conflict checks, and every unrelated parameter are unchanged. Co-Authored-By: Claude Opus 5 <[email protected]>
1854 lines
68 KiB
TypeScript
1854 lines
68 KiB
TypeScript
import { createHash, randomBytes } from "node:crypto";
|
|
import net from "node:net";
|
|
import path from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { Client, type QueryResult, type QueryResultRow } from "pg";
|
|
import { afterAll, describe, expect, it, vi } from "vitest";
|
|
import { reportMigrationCommandError } from "../src/cli/migration-command-errors.js";
|
|
import {
|
|
applyModuleMigrations,
|
|
createPostgresMigrationDatabase,
|
|
discoverCoreMigrations,
|
|
type MigrationApproval,
|
|
type MigrationApprovalProvider,
|
|
MigrationRunnerError,
|
|
type MigrationRunnerErrorCode,
|
|
planModuleMigrations,
|
|
PostgresMigrationDatabaseError,
|
|
} from "../src/migrations/index.js";
|
|
import type {
|
|
DiscoveredModuleMigration,
|
|
ModuleMigrationAggregate,
|
|
} from "../src/modules/migration-discovery.js";
|
|
|
|
const coreMigrationsDirectory = path.join(
|
|
path.dirname(fileURLToPath(import.meta.url)),
|
|
"..",
|
|
"drizzle",
|
|
);
|
|
|
|
const integrationDatabaseUrlEnvironment = "EKKLESIA_MIGRATION_TEST_DATABASE_URL";
|
|
const historyTable = "public.ekklesia_core_migrations";
|
|
const migrationLockNamespace = 1_162_562_380;
|
|
const migrationLockKey = 1_296_648_018;
|
|
const concurrencyBarrierNamespace = 1_162_562_380;
|
|
const concurrencyBarrierKey = 1_414_743_380;
|
|
const trackedDatabases = new Set<string>();
|
|
let databaseCounter = 0;
|
|
|
|
class IntegrationHarnessError extends Error {
|
|
constructor(message: string) {
|
|
super(message);
|
|
this.name = "IntegrationHarnessError";
|
|
this.stack = `${this.name}: ${this.message}`;
|
|
}
|
|
}
|
|
|
|
function harnessError(message: string): IntegrationHarnessError {
|
|
return new IntegrationHarnessError(message);
|
|
}
|
|
|
|
function baseDatabaseUrl(): string {
|
|
const value = process.env[integrationDatabaseUrlEnvironment];
|
|
if (typeof value !== "string" || value.trim() === "") {
|
|
throw harnessError("The migration 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 harnessError("The migration integration database configuration is invalid.");
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function quoteDatabaseName(databaseName: string): string {
|
|
if (!/^[a-z][a-z0-9_]{0,62}$/u.test(databaseName)) {
|
|
throw harnessError("The generated integration database name is invalid.");
|
|
}
|
|
return `"${databaseName}"`;
|
|
}
|
|
|
|
function makeDatabaseName(label: string): string {
|
|
const safeLabel = label
|
|
.toLowerCase()
|
|
.replaceAll(/[^a-z0-9]+/gu, "_")
|
|
.replaceAll(/^_+|_+$/gu, "")
|
|
.slice(0, 18);
|
|
databaseCounter += 1;
|
|
return `ekk_mig_${process.pid}_${databaseCounter}_${safeLabel || "scenario"}_${randomBytes(4).toString("hex")}`;
|
|
}
|
|
|
|
function databaseUrl(databaseName: string, applicationName?: string): string {
|
|
const parsed = new URL(baseDatabaseUrl());
|
|
parsed.pathname = `/${databaseName}`;
|
|
if (applicationName === undefined) {
|
|
parsed.searchParams.delete("application_name");
|
|
} else {
|
|
parsed.searchParams.set("application_name", applicationName);
|
|
}
|
|
return parsed.toString();
|
|
}
|
|
|
|
async function openAdministrativeClient(
|
|
connectionString: string,
|
|
applicationName: string,
|
|
): Promise<Client> {
|
|
const client = new Client({ application_name: applicationName, connectionString });
|
|
client.on("error", () => undefined);
|
|
try {
|
|
await client.connect();
|
|
return client;
|
|
} catch {
|
|
try {
|
|
await client.end();
|
|
} catch {
|
|
// The public harness failure remains static when connection cleanup also fails.
|
|
}
|
|
throw harnessError("The migration integration database connection failed.");
|
|
}
|
|
}
|
|
|
|
async function safeQuery<Row extends QueryResultRow>(
|
|
client: Client,
|
|
text: string,
|
|
values: unknown[] = [],
|
|
failureMessage = "An integration database administration operation failed.",
|
|
): Promise<QueryResult<Row>> {
|
|
try {
|
|
return await client.query<Row>(text, values);
|
|
} catch {
|
|
throw harnessError(failureMessage);
|
|
}
|
|
}
|
|
|
|
async function safeEnd(client: Client): Promise<void> {
|
|
try {
|
|
await client.end();
|
|
} catch {
|
|
throw harnessError("An integration database connection could not be closed.");
|
|
}
|
|
}
|
|
|
|
async function createDatabase(databaseName: string): Promise<void> {
|
|
const admin = await openAdministrativeClient(baseDatabaseUrl(), "ekklesia_migration_test_setup");
|
|
let operationError: unknown;
|
|
try {
|
|
await safeQuery(admin, `CREATE DATABASE ${quoteDatabaseName(databaseName)} TEMPLATE template0`);
|
|
} catch (error) {
|
|
operationError = error;
|
|
}
|
|
try {
|
|
await safeEnd(admin);
|
|
} catch (error) {
|
|
if (operationError === undefined) {
|
|
operationError = error;
|
|
}
|
|
}
|
|
if (operationError !== undefined) {
|
|
throw operationError;
|
|
}
|
|
}
|
|
|
|
async function dropDatabase(databaseName: string): Promise<void> {
|
|
const admin = await openAdministrativeClient(
|
|
baseDatabaseUrl(),
|
|
"ekklesia_migration_test_cleanup",
|
|
);
|
|
let operationError: unknown;
|
|
try {
|
|
await safeQuery(
|
|
admin,
|
|
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()",
|
|
[databaseName],
|
|
);
|
|
await safeQuery(
|
|
admin,
|
|
`DROP DATABASE IF EXISTS ${quoteDatabaseName(databaseName)} WITH (FORCE)`,
|
|
);
|
|
trackedDatabases.delete(databaseName);
|
|
} catch (error) {
|
|
operationError = error;
|
|
}
|
|
try {
|
|
await safeEnd(admin);
|
|
} catch (error) {
|
|
if (operationError === undefined) {
|
|
operationError = error;
|
|
}
|
|
}
|
|
if (operationError !== undefined) {
|
|
throw operationError;
|
|
}
|
|
}
|
|
|
|
interface ScenarioDatabase {
|
|
readonly client: Client;
|
|
readonly name: string;
|
|
readonly url: string;
|
|
}
|
|
|
|
async function withIsolatedDatabase<T>(
|
|
label: string,
|
|
operation: (database: ScenarioDatabase) => Promise<T>,
|
|
): Promise<T> {
|
|
const name = makeDatabaseName(label);
|
|
await createDatabase(name);
|
|
trackedDatabases.add(name);
|
|
|
|
let client: Client | undefined;
|
|
let outcome:
|
|
| { readonly error: unknown; readonly success: false }
|
|
| { readonly success: true; readonly value: T };
|
|
try {
|
|
const url = databaseUrl(name);
|
|
client = await openAdministrativeClient(url, "ekklesia_migration_test_inspection");
|
|
outcome = { success: true, value: await operation(Object.freeze({ client, name, url })) };
|
|
} catch (error) {
|
|
outcome = { error, success: false };
|
|
}
|
|
|
|
let cleanupFailed = false;
|
|
if (client !== undefined) {
|
|
try {
|
|
await safeEnd(client);
|
|
} catch {
|
|
cleanupFailed = true;
|
|
}
|
|
}
|
|
try {
|
|
await dropDatabase(name);
|
|
} catch {
|
|
cleanupFailed = true;
|
|
}
|
|
|
|
if (!outcome.success) {
|
|
throw outcome.error;
|
|
}
|
|
if (cleanupFailed) {
|
|
throw harnessError("An isolated migration integration database could not be removed.");
|
|
}
|
|
return outcome.value;
|
|
}
|
|
|
|
afterAll(async () => {
|
|
let cleanupFailed = false;
|
|
for (const databaseName of [...trackedDatabases]) {
|
|
try {
|
|
await dropDatabase(databaseName);
|
|
} catch {
|
|
cleanupFailed = true;
|
|
}
|
|
}
|
|
if (cleanupFailed) {
|
|
throw harnessError("Migration integration database cleanup was incomplete.");
|
|
}
|
|
});
|
|
|
|
interface MigrationOptions {
|
|
readonly approvalRequired?: boolean;
|
|
readonly sql: string;
|
|
readonly tag: string;
|
|
}
|
|
|
|
interface ModuleFixture {
|
|
readonly id: string;
|
|
readonly migrations: readonly DiscoveredModuleMigration[];
|
|
}
|
|
|
|
function migration(
|
|
moduleId: string,
|
|
index: number,
|
|
options: MigrationOptions,
|
|
): DiscoveredModuleMigration {
|
|
const approvalRequired = options.approvalRequired ?? false;
|
|
return Object.freeze({
|
|
approvalRequired,
|
|
breakpoints: true,
|
|
checksum: createHash("sha256").update(options.sql).digest("hex"),
|
|
destructive: Object.freeze(
|
|
approvalRequired
|
|
? {
|
|
mode: "approval-required" as const,
|
|
reason: "This integration fixture requires an exact external approval.",
|
|
}
|
|
: { mode: "none" as const },
|
|
),
|
|
file: `${options.tag}.sql`,
|
|
identity: `${moduleId}:${options.tag}`,
|
|
index,
|
|
moduleId,
|
|
sql: options.sql,
|
|
tag: options.tag,
|
|
timestamp: 1_800_000_000_000 + index,
|
|
});
|
|
}
|
|
|
|
function aggregate(...moduleFixtures: readonly ModuleFixture[]): ModuleMigrationAggregate {
|
|
const modules = Object.freeze(
|
|
moduleFixtures.map((moduleFixture) =>
|
|
Object.freeze({
|
|
migrations: Object.freeze([...moduleFixture.migrations]),
|
|
moduleId: moduleFixture.id,
|
|
}),
|
|
),
|
|
);
|
|
return Object.freeze({
|
|
discoveredModules: Object.freeze([]),
|
|
migrations: Object.freeze(modules.flatMap((moduleFixture) => moduleFixture.migrations)),
|
|
modules,
|
|
});
|
|
}
|
|
|
|
function moduleFixture(
|
|
id: string,
|
|
migrations: readonly DiscoveredModuleMigration[],
|
|
): ModuleFixture {
|
|
return Object.freeze({ id, migrations: Object.freeze([...migrations]) });
|
|
}
|
|
|
|
const peopleInitialSql = `CREATE TABLE "people_records" (
|
|
"id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
|
"display_name" text NOT NULL
|
|
);\n`;
|
|
const calendarInitialSql = `CREATE TABLE "calendar_events" (
|
|
"id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
|
"owner_id" bigint NOT NULL,
|
|
"title" text NOT NULL
|
|
);\n`;
|
|
const peopleLaterSql = `ALTER TABLE "people_records" ADD COLUMN "email" text;\n`;
|
|
|
|
function peopleInitial(sql = peopleInitialSql): DiscoveredModuleMigration {
|
|
return migration("people", 0, { sql, tag: "0000_initial" });
|
|
}
|
|
|
|
function calendarInitial(): DiscoveredModuleMigration {
|
|
return migration("calendar", 0, { sql: calendarInitialSql, tag: "0000_initial" });
|
|
}
|
|
|
|
function peopleLater(sql = peopleLaterSql): DiscoveredModuleMigration {
|
|
return migration("people", 1, { sql, tag: "0001_add_email" });
|
|
}
|
|
|
|
interface RelationExistenceRow extends QueryResultRow {
|
|
readonly relation_name: string | null;
|
|
}
|
|
|
|
interface UserObjectRow extends QueryResultRow {
|
|
readonly object_kind: string;
|
|
readonly object_name: string;
|
|
readonly schema_name: string;
|
|
}
|
|
|
|
interface ColumnRow extends QueryResultRow {
|
|
readonly column_default: string | null;
|
|
readonly column_name: string;
|
|
readonly data_type: string;
|
|
readonly is_nullable: string;
|
|
readonly ordinal_position: number;
|
|
readonly table_name: string;
|
|
readonly table_schema: string;
|
|
}
|
|
|
|
interface ConstraintRow extends QueryResultRow {
|
|
readonly constraint_definition: string;
|
|
readonly constraint_name: string;
|
|
readonly table_name: string;
|
|
readonly table_schema: string;
|
|
}
|
|
|
|
interface SequenceRow extends QueryResultRow {
|
|
readonly cache_size: string;
|
|
readonly cycle: boolean;
|
|
readonly increment_by: string;
|
|
readonly last_value: string | null;
|
|
readonly sequence_name: string;
|
|
readonly sequence_schema: string;
|
|
}
|
|
|
|
interface HistoryRow extends QueryResultRow {
|
|
readonly applied_at: Date;
|
|
readonly applied_order: number;
|
|
readonly checksum: string;
|
|
readonly identity: string;
|
|
readonly migration_index: number;
|
|
readonly migration_tag: string;
|
|
readonly module_id: string;
|
|
}
|
|
|
|
interface BooleanRow extends QueryResultRow {
|
|
readonly value: boolean;
|
|
}
|
|
|
|
interface CountRow extends QueryResultRow {
|
|
readonly value: number;
|
|
}
|
|
|
|
interface DatabaseState {
|
|
readonly columns: readonly ColumnRow[];
|
|
readonly constraints: readonly ConstraintRow[];
|
|
readonly history: readonly HistoryRow[];
|
|
readonly objects: readonly UserObjectRow[];
|
|
readonly sequences: readonly SequenceRow[];
|
|
readonly supplementalObjects: readonly UserObjectRow[];
|
|
}
|
|
|
|
async function relationExists(client: Client, qualifiedName: string): Promise<boolean> {
|
|
const result = await safeQuery<RelationExistenceRow>(
|
|
client,
|
|
"SELECT to_regclass($1) AS relation_name",
|
|
[qualifiedName],
|
|
);
|
|
return result.rows[0]?.relation_name !== null && result.rows[0]?.relation_name !== undefined;
|
|
}
|
|
|
|
async function readHistory(client: Client): Promise<readonly HistoryRow[]> {
|
|
if (!(await relationExists(client, historyTable))) {
|
|
return Object.freeze([]);
|
|
}
|
|
const result = await safeQuery<HistoryRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
applied_order::integer AS applied_order,
|
|
identity,
|
|
module_id,
|
|
migration_tag,
|
|
migration_index,
|
|
checksum,
|
|
applied_at
|
|
FROM public.ekklesia_core_migrations
|
|
ORDER BY applied_order ASC
|
|
`,
|
|
);
|
|
return Object.freeze(result.rows.map((row) => Object.freeze({ ...row })));
|
|
}
|
|
|
|
async function databaseState(client: Client): Promise<DatabaseState> {
|
|
const objects = await safeQuery<UserObjectRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
namespace.nspname AS schema_name,
|
|
relation.relname AS object_name,
|
|
relation.relkind::text AS object_kind
|
|
FROM pg_catalog.pg_class AS relation
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = relation.relnamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
ORDER BY namespace.nspname, relation.relname, relation.relkind
|
|
`,
|
|
[],
|
|
"Integration user-object inspection failed.",
|
|
);
|
|
const columns = await safeQuery<ColumnRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
table_schema,
|
|
table_name,
|
|
column_name,
|
|
ordinal_position,
|
|
data_type,
|
|
is_nullable,
|
|
column_default
|
|
FROM information_schema.columns
|
|
WHERE table_schema NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND table_schema <> 'information_schema'
|
|
ORDER BY table_schema, table_name, ordinal_position
|
|
`,
|
|
[],
|
|
"Integration column inspection failed.",
|
|
);
|
|
const constraints = await safeQuery<ConstraintRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
namespace.nspname AS table_schema,
|
|
relation.relname AS table_name,
|
|
catalog_constraint.conname AS constraint_name,
|
|
pg_catalog.pg_get_constraintdef(catalog_constraint.oid, true) AS constraint_definition
|
|
FROM pg_catalog.pg_constraint AS catalog_constraint
|
|
INNER JOIN pg_catalog.pg_class AS relation
|
|
ON relation.oid = catalog_constraint.conrelid
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
ORDER BY namespace.nspname, relation.relname, catalog_constraint.conname
|
|
`,
|
|
[],
|
|
"Integration constraint inspection failed.",
|
|
);
|
|
const sequences = await safeQuery<SequenceRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
schemaname AS sequence_schema,
|
|
sequencename AS sequence_name,
|
|
increment_by::text,
|
|
cycle,
|
|
cache_size::text,
|
|
last_value::text
|
|
FROM pg_catalog.pg_sequences
|
|
WHERE schemaname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND schemaname <> 'information_schema'
|
|
ORDER BY schemaname, sequencename
|
|
`,
|
|
[],
|
|
"Integration sequence inspection failed.",
|
|
);
|
|
const supplementalObjects = await safeQuery<UserObjectRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
'schema' AS object_kind,
|
|
namespace.nspname AS object_name,
|
|
'' AS schema_name
|
|
FROM pg_catalog.pg_namespace AS namespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
UNION ALL
|
|
SELECT
|
|
'extension' AS object_kind,
|
|
catalog_extension.extname AS object_name,
|
|
namespace.nspname AS schema_name
|
|
FROM pg_catalog.pg_extension AS catalog_extension
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = catalog_extension.extnamespace
|
|
UNION ALL
|
|
SELECT
|
|
'routine' AS object_kind,
|
|
catalog_routine.oid::regprocedure::text AS object_name,
|
|
namespace.nspname AS schema_name
|
|
FROM pg_catalog.pg_proc AS catalog_routine
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = catalog_routine.pronamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
UNION ALL
|
|
SELECT
|
|
CASE user_type.typtype
|
|
WHEN 'd' THEN 'domain'
|
|
WHEN 'e' THEN 'enum'
|
|
WHEN 'm' THEN 'multirange'
|
|
WHEN 'r' THEN 'range'
|
|
END AS object_kind,
|
|
user_type.typname AS object_name,
|
|
namespace.nspname AS schema_name
|
|
FROM pg_catalog.pg_type AS user_type
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = user_type.typnamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
AND user_type.typtype IN ('d', 'e', 'm', 'r')
|
|
ORDER BY object_kind, schema_name, object_name
|
|
`,
|
|
[],
|
|
"Integration supplemental-object inspection failed.",
|
|
);
|
|
const history = await readHistory(client);
|
|
|
|
return Object.freeze({
|
|
columns: Object.freeze(columns.rows.map((row) => Object.freeze({ ...row }))),
|
|
constraints: Object.freeze(constraints.rows.map((row) => Object.freeze({ ...row }))),
|
|
history,
|
|
objects: Object.freeze(objects.rows.map((row) => Object.freeze({ ...row }))),
|
|
sequences: Object.freeze(sequences.rows.map((row) => Object.freeze({ ...row }))),
|
|
supplementalObjects: Object.freeze(
|
|
supplementalObjects.rows.map((row) => Object.freeze({ ...row })),
|
|
),
|
|
});
|
|
}
|
|
|
|
async function columnExists(
|
|
client: Client,
|
|
tableName: string,
|
|
columnName: string,
|
|
): Promise<boolean> {
|
|
const result = await safeQuery<BooleanRow>(
|
|
client,
|
|
`
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM information_schema.columns
|
|
WHERE table_schema = 'public' AND table_name = $1 AND column_name = $2
|
|
) AS value
|
|
`,
|
|
[tableName, columnName],
|
|
);
|
|
return result.rows[0]?.value === true;
|
|
}
|
|
|
|
async function captureRunnerError(
|
|
operation: Promise<unknown>,
|
|
code: MigrationRunnerErrorCode,
|
|
): Promise<MigrationRunnerError> {
|
|
try {
|
|
await operation;
|
|
} catch (error) {
|
|
if (!(error instanceof MigrationRunnerError)) {
|
|
throw harnessError("A migration operation returned an unsafe error type.");
|
|
}
|
|
const migrationError = error as MigrationRunnerError;
|
|
expect(migrationError.code).toBe(code);
|
|
return migrationError;
|
|
}
|
|
throw harnessError("A migration operation unexpectedly succeeded.");
|
|
}
|
|
|
|
function approvalFor(migrationFixture: DiscoveredModuleMigration): MigrationApproval {
|
|
return Object.freeze({
|
|
checksum: migrationFixture.checksum,
|
|
identity: migrationFixture.identity,
|
|
moduleId: migrationFixture.moduleId,
|
|
});
|
|
}
|
|
|
|
function approvalProvider(approvals: readonly MigrationApproval[]): MigrationApprovalProvider {
|
|
const frozenApprovals = Object.freeze(
|
|
approvals.map((approval) => Object.freeze({ ...approval })),
|
|
);
|
|
return Object.freeze({
|
|
getApprovals: () => Promise.resolve(frozenApprovals),
|
|
});
|
|
}
|
|
|
|
async function acquireAdministrativeLock(
|
|
client: Client,
|
|
namespace: number,
|
|
key: number,
|
|
): Promise<void> {
|
|
await safeQuery(client, "SELECT pg_advisory_lock($1::integer, $2::integer)", [namespace, key]);
|
|
}
|
|
|
|
async function releaseAdministrativeLock(
|
|
client: Client,
|
|
namespace: number,
|
|
key: number,
|
|
): Promise<void> {
|
|
const result = await safeQuery<BooleanRow>(
|
|
client,
|
|
"SELECT pg_advisory_unlock($1::integer, $2::integer) AS value",
|
|
[namespace, key],
|
|
);
|
|
if (result.rows[0]?.value !== true) {
|
|
throw harnessError("An integration advisory lock was not released.");
|
|
}
|
|
}
|
|
|
|
async function expectMigrationLockAvailable(client: Client): Promise<void> {
|
|
const result = await safeQuery<BooleanRow>(
|
|
client,
|
|
"SELECT pg_try_advisory_lock($1::integer, $2::integer) AS value",
|
|
[migrationLockNamespace, migrationLockKey],
|
|
);
|
|
expect(result.rows[0]?.value).toBe(true);
|
|
if (result.rows[0]?.value === true) {
|
|
await releaseAdministrativeLock(client, migrationLockNamespace, migrationLockKey);
|
|
}
|
|
}
|
|
|
|
async function waitForDatabaseCondition(
|
|
check: () => Promise<boolean>,
|
|
timeoutMs = 5_000,
|
|
): Promise<void> {
|
|
const deadline = performance.now() + timeoutMs;
|
|
while (performance.now() < deadline) {
|
|
if (await check()) {
|
|
return;
|
|
}
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 25));
|
|
}
|
|
throw harnessError("A bounded integration database condition was not observed.");
|
|
}
|
|
|
|
describe.sequential("live PostgreSQL module migration execution", () => {
|
|
it("keeps fresh planning read-only", async () => {
|
|
await withIsolatedDatabase("fresh_plan", async (database) => {
|
|
const people = peopleInitial();
|
|
const calendar = calendarInitial();
|
|
const inventory = aggregate(
|
|
moduleFixture("people", [people]),
|
|
moduleFixture("calendar", [calendar]),
|
|
);
|
|
const before = await databaseState(database.client);
|
|
|
|
const plan = await planModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(plan.entries.map((entry) => [entry.identity, entry.status])).toEqual([
|
|
[people.identity, "pending"],
|
|
[calendar.identity, "pending"],
|
|
]);
|
|
expect(plan).toMatchObject({ appliedCount: 0, approvalRequiredCount: 0, pendingCount: 2 });
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.people_records")).toBe(false);
|
|
expect(await relationExists(database.client, "public.calendar_events")).toBe(false);
|
|
});
|
|
});
|
|
|
|
it("applies a fresh dependency-ordered inventory and records exact history", async () => {
|
|
await withIsolatedDatabase("fresh_apply", async (database) => {
|
|
const people = peopleInitial();
|
|
const calendar = calendarInitial();
|
|
const inventory = aggregate(
|
|
moduleFixture("people", [people]),
|
|
moduleFixture("calendar", [calendar]),
|
|
);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([
|
|
people.identity,
|
|
calendar.identity,
|
|
]);
|
|
expect(await relationExists(database.client, "public.people_records")).toBe(true);
|
|
expect(await relationExists(database.client, "public.calendar_events")).toBe(true);
|
|
const history = await readHistory(database.client);
|
|
expect(history.map(({ applied_at: _appliedAt, ...row }) => row)).toEqual([
|
|
{
|
|
applied_order: 1,
|
|
checksum: people.checksum,
|
|
identity: people.identity,
|
|
migration_index: 0,
|
|
migration_tag: people.tag,
|
|
module_id: people.moduleId,
|
|
},
|
|
{
|
|
applied_order: 2,
|
|
checksum: calendar.checksum,
|
|
identity: calendar.identity,
|
|
migration_index: 0,
|
|
migration_tag: calendar.tag,
|
|
module_id: calendar.moduleId,
|
|
},
|
|
]);
|
|
expect(history).toHaveLength(2);
|
|
expect(
|
|
history.every(
|
|
(row) => row.applied_at instanceof Date && !Number.isNaN(row.applied_at.getTime()),
|
|
),
|
|
).toBe(true);
|
|
});
|
|
});
|
|
|
|
it("performs a genuine no-op without changing schema, data, sequences, or history", async () => {
|
|
await withIsolatedDatabase("no_op", async (database) => {
|
|
const initial = peopleInitial();
|
|
const inventory = aggregate(moduleFixture("people", [initial]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: inventory, database: migrationDatabase });
|
|
await safeQuery(
|
|
database.client,
|
|
"INSERT INTO public.people_records (display_name) VALUES ($1)",
|
|
["No-op sentinel"],
|
|
);
|
|
const before = await databaseState(database.client);
|
|
const beforeData = await safeQuery<QueryResultRow>(
|
|
database.client,
|
|
"SELECT id::integer, display_name FROM public.people_records ORDER BY id",
|
|
);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
const afterData = await safeQuery<QueryResultRow>(
|
|
database.client,
|
|
"SELECT id::integer, display_name FROM public.people_records ORDER BY id",
|
|
);
|
|
expect(afterData.rows).toEqual(beforeData.rows);
|
|
});
|
|
});
|
|
|
|
it("executes only a newly presented later migration", async () => {
|
|
await withIsolatedDatabase("later", async (database) => {
|
|
const initial = peopleInitial();
|
|
const later = peopleLater();
|
|
const initialInventory = aggregate(moduleFixture("people", [initial]));
|
|
const laterInventory = aggregate(moduleFixture("people", [initial, later]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: initialInventory, database: migrationDatabase });
|
|
const originalHistory = await readHistory(database.client);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: laterInventory,
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([later.identity]);
|
|
expect(await columnExists(database.client, "people_records", "email")).toBe(true);
|
|
const history = await readHistory(database.client);
|
|
expect(history).toHaveLength(2);
|
|
expect(history[0]).toEqual(originalHistory[0]);
|
|
expect(history[1]).toMatchObject({
|
|
applied_order: 2,
|
|
checksum: later.checksum,
|
|
identity: later.identity,
|
|
migration_index: 1,
|
|
migration_tag: later.tag,
|
|
module_id: later.moduleId,
|
|
});
|
|
});
|
|
});
|
|
|
|
it("rejects edited applied contents during both planning and applying", async () => {
|
|
await withIsolatedDatabase("edited", async (database) => {
|
|
const initial = peopleInitial();
|
|
const edited = peopleInitial(
|
|
`CREATE TABLE "people_records" ("id" bigint PRIMARY KEY, "edited_marker" text);\n`,
|
|
);
|
|
const originalInventory = aggregate(moduleFixture("people", [initial]));
|
|
const editedInventory = aggregate(moduleFixture("people", [edited]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: originalInventory, database: migrationDatabase });
|
|
const before = await databaseState(database.client);
|
|
|
|
const planError = await captureRunnerError(
|
|
planModuleMigrations({ aggregate: editedInventory, database: migrationDatabase }),
|
|
"MIGRATION_CHECKSUM_MISMATCH",
|
|
);
|
|
expect(planError.migrationIdentity).toBe(initial.identity);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
|
|
const applyError = await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: editedInventory, database: migrationDatabase }),
|
|
"MIGRATION_CHECKSUM_MISMATCH",
|
|
);
|
|
expect(applyError.migrationIdentity).toBe(initial.identity);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await columnExists(database.client, "people_records", "edited_marker")).toBe(false);
|
|
});
|
|
});
|
|
|
|
it("rejects an unknown applied identity without repair or further execution", async () => {
|
|
await withIsolatedDatabase("unknown_history", async (database) => {
|
|
const initial = peopleInitial();
|
|
const later = peopleLater();
|
|
const initialInventory = aggregate(moduleFixture("people", [initial]));
|
|
const completeInventory = aggregate(moduleFixture("people", [initial, later]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: initialInventory, database: migrationDatabase });
|
|
await safeQuery(
|
|
database.client,
|
|
`
|
|
INSERT INTO public.ekklesia_core_migrations
|
|
(identity, module_id, migration_tag, migration_index, checksum)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
`,
|
|
["ghost:0000_unknown", "ghost", "0000_unknown", 0, "f".repeat(64)],
|
|
);
|
|
const corrupted = await databaseState(database.client);
|
|
|
|
await captureRunnerError(
|
|
planModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_UNKNOWN_APPLIED",
|
|
);
|
|
await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_UNKNOWN_APPLIED",
|
|
);
|
|
|
|
expect(await databaseState(database.client)).toEqual(corrupted);
|
|
expect(await columnExists(database.client, "people_records", "email")).toBe(false);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
"ghost:0000_unknown",
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("rejects an invalid per-module history prefix without silent repair", async () => {
|
|
await withIsolatedDatabase("history_gap", async (database) => {
|
|
const initial = peopleInitial();
|
|
const later = peopleLater();
|
|
const unreachable = migration("people", 2, {
|
|
sql: `CREATE TABLE "people_unreachable" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0002_unreachable",
|
|
});
|
|
const appliedInventory = aggregate(moduleFixture("people", [initial, later]));
|
|
const completeInventory = aggregate(moduleFixture("people", [initial, later, unreachable]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: appliedInventory, database: migrationDatabase });
|
|
await safeQuery(
|
|
database.client,
|
|
"DELETE FROM public.ekklesia_core_migrations WHERE identity = $1",
|
|
[initial.identity],
|
|
);
|
|
const corrupted = await databaseState(database.client);
|
|
|
|
await captureRunnerError(
|
|
planModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_HISTORY_INVALID",
|
|
);
|
|
await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_HISTORY_INVALID",
|
|
);
|
|
|
|
expect(await databaseState(database.client)).toEqual(corrupted);
|
|
expect(await relationExists(database.client, "public.people_unreachable")).toBe(false);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
later.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("rolls back the entire current pending batch after a real PostgreSQL failure", async () => {
|
|
await withIsolatedDatabase("rollback", async (database) => {
|
|
const committed = migration("rollback", 0, {
|
|
sql: `CREATE TABLE "rollback_committed" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_committed",
|
|
});
|
|
const pendingFirst = migration("rollback", 1, {
|
|
sql: `CREATE TABLE "rollback_pending_one" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0001_pending_one",
|
|
});
|
|
const pendingFailure = migration("rollback", 2, {
|
|
sql: `CREATE TABLE "rollback_pending_two" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
|
|
tag: "0002_pending_failure",
|
|
});
|
|
const committedInventory = aggregate(moduleFixture("rollback", [committed]));
|
|
const completeInventory = aggregate(
|
|
moduleFixture("rollback", [committed, pendingFirst, pendingFailure]),
|
|
);
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: committedInventory, database: migrationDatabase });
|
|
|
|
const error = await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
|
|
expect(error.migrationIdentity).toBe(pendingFailure.identity);
|
|
expect(await relationExists(database.client, "public.rollback_committed")).toBe(true);
|
|
expect(await relationExists(database.client, "public.rollback_pending_one")).toBe(false);
|
|
expect(await relationExists(database.client, "public.rollback_pending_two")).toBe(false);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
committed.identity,
|
|
]);
|
|
await expectMigrationLockAvailable(database.client);
|
|
});
|
|
});
|
|
|
|
it("honors Drizzle statement breakpoints without splitting procedural semicolons", async () => {
|
|
await withIsolatedDatabase("breakpoints", async (database) => {
|
|
const sql = `CREATE TABLE "breakpoints_steps" (
|
|
"id" integer PRIMARY KEY,
|
|
"note" text NOT NULL
|
|
);
|
|
--> statement-breakpoint
|
|
DO $body$
|
|
BEGIN
|
|
INSERT INTO "breakpoints_steps" ("id", "note") VALUES (1, 'first;still-one-statement');
|
|
END
|
|
$body$;
|
|
--> statement-breakpoint
|
|
ALTER TABLE "breakpoints_steps" ADD COLUMN "sequence_value" integer;
|
|
UPDATE "breakpoints_steps" SET "sequence_value" = 1 WHERE "id" = 1;
|
|
INSERT INTO "breakpoints_steps" ("id", "note", "sequence_value")
|
|
SELECT 2, 'second', "sequence_value" + 1 FROM "breakpoints_steps" WHERE "id" = 1;
|
|
`;
|
|
const initial = migration("breakpoints", 0, { sql, tag: "0000_ordered_steps" });
|
|
|
|
await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("breakpoints", [initial])),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
const rows = await safeQuery<QueryResultRow>(
|
|
database.client,
|
|
`SELECT id, note, sequence_value FROM public.breakpoints_steps ORDER BY id`,
|
|
);
|
|
expect(rows.rows).toEqual([
|
|
{ id: 1, note: "first;still-one-statement", sequence_value: 1 },
|
|
{ id: 2, note: "second", sequence_value: 2 },
|
|
]);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("serializes two real migrators with the session advisory lock", async () => {
|
|
await withIsolatedDatabase("concurrency", async (database) => {
|
|
await acquireAdministrativeLock(
|
|
database.client,
|
|
concurrencyBarrierNamespace,
|
|
concurrencyBarrierKey,
|
|
);
|
|
let barrierHeld = true;
|
|
const sql = `SELECT pg_advisory_xact_lock(${concurrencyBarrierNamespace}::integer, ${concurrencyBarrierKey}::integer);\n--> statement-breakpoint\nCREATE TABLE "concurrency_records" ("id" bigint PRIMARY KEY);\n`;
|
|
const initial = migration("concurrency", 0, { sql, tag: "0000_serialized" });
|
|
const inventory = aggregate(moduleFixture("concurrency", [initial]));
|
|
const applicationName = "ekklesia_migration_test_concurrency";
|
|
const productionUrl = databaseUrl(database.name, applicationName);
|
|
const firstApply = applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(productionUrl),
|
|
lockTimeoutMs: 5_000,
|
|
});
|
|
|
|
try {
|
|
await waitForDatabaseCondition(async () => {
|
|
const result = await safeQuery<CountRow>(
|
|
database.client,
|
|
`
|
|
SELECT count(*)::integer AS value
|
|
FROM pg_catalog.pg_locks AS lock
|
|
INNER JOIN pg_catalog.pg_stat_activity AS activity ON activity.pid = lock.pid
|
|
WHERE lock.locktype = 'advisory'
|
|
AND lock.classid = $1::oid
|
|
AND lock.objid = $2::oid
|
|
AND lock.objsubid = 2
|
|
AND lock.granted
|
|
AND activity.datname = current_database()
|
|
AND activity.application_name = $3
|
|
`,
|
|
[migrationLockNamespace, migrationLockKey, applicationName],
|
|
);
|
|
return result.rows[0]?.value === 1;
|
|
});
|
|
|
|
const secondError = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(productionUrl),
|
|
lockTimeoutMs: 250,
|
|
}),
|
|
"MIGRATION_LOCK_TIMEOUT",
|
|
);
|
|
expect(secondError.migrationIdentity).toBeUndefined();
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.concurrency_records")).toBe(false);
|
|
|
|
await releaseAdministrativeLock(
|
|
database.client,
|
|
concurrencyBarrierNamespace,
|
|
concurrencyBarrierKey,
|
|
);
|
|
barrierHeld = false;
|
|
const firstResult = await firstApply;
|
|
expect(firstResult.applied.map((entry) => entry.identity)).toEqual([initial.identity]);
|
|
} finally {
|
|
if (barrierHeld) {
|
|
await releaseAdministrativeLock(
|
|
database.client,
|
|
concurrencyBarrierNamespace,
|
|
concurrencyBarrierKey,
|
|
);
|
|
}
|
|
await Promise.allSettled([firstApply]);
|
|
}
|
|
|
|
expect(await relationExists(database.client, "public.concurrency_records")).toBe(true);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("times out on the real lock, releases it, and permits a later apply", async () => {
|
|
await withIsolatedDatabase("lock_timeout", async (database) => {
|
|
const initial = migration("locks", 0, {
|
|
sql: `CREATE TABLE "locks_records" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_initial",
|
|
});
|
|
const inventory = aggregate(moduleFixture("locks", [initial]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await acquireAdministrativeLock(database.client, migrationLockNamespace, migrationLockKey);
|
|
let lockHeld = true;
|
|
const before = await databaseState(database.client);
|
|
const startedAt = performance.now();
|
|
try {
|
|
await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
lockTimeoutMs: 250,
|
|
}),
|
|
"MIGRATION_LOCK_TIMEOUT",
|
|
);
|
|
} finally {
|
|
await releaseAdministrativeLock(database.client, migrationLockNamespace, migrationLockKey);
|
|
lockHeld = false;
|
|
}
|
|
const elapsedMs = performance.now() - startedAt;
|
|
expect(elapsedMs).toBeGreaterThanOrEqual(150);
|
|
expect(elapsedMs).toBeLessThan(2_000);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(lockHeld).toBe(false);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([initial.identity]);
|
|
expect(await relationExists(database.client, "public.locks_records")).toBe(true);
|
|
});
|
|
});
|
|
|
|
it("does not strand the migration lock after execution failure", async () => {
|
|
await withIsolatedDatabase("failed_lock_release", async (database) => {
|
|
const failing = migration("lock_failure", 0, {
|
|
sql: `CREATE TABLE "lock_failure_records" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
|
|
tag: "0000_initial",
|
|
});
|
|
const corrected = migration("lock_failure", 0, {
|
|
sql: `CREATE TABLE "lock_failure_records" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_initial",
|
|
});
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
|
|
await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("lock_failure", [failing])),
|
|
database: migrationDatabase,
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
await expectMigrationLockAvailable(database.client);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.lock_failure_records")).toBe(false);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("lock_failure", [corrected])),
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([corrected.identity]);
|
|
});
|
|
});
|
|
|
|
it("requires an exact destructive approval before any persistent mutation", async () => {
|
|
await withIsolatedDatabase("approvals", async (database) => {
|
|
const destructive = migration("destructive", 0, {
|
|
approvalRequired: true,
|
|
sql: `CREATE TABLE "destructive_records" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_replace_records",
|
|
});
|
|
const inventory = aggregate(moduleFixture("destructive", [destructive]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
const exact = approvalFor(destructive);
|
|
const rejectedApprovals: readonly (readonly MigrationApproval[] | undefined)[] = [
|
|
undefined,
|
|
[Object.freeze({ ...exact, moduleId: "wrong-module" })],
|
|
[Object.freeze({ ...exact, identity: "destructive:0000_wrong" })],
|
|
[Object.freeze({ ...exact, checksum: "0".repeat(64) })],
|
|
];
|
|
|
|
for (const supplied of rejectedApprovals) {
|
|
const before = await databaseState(database.client);
|
|
const operation =
|
|
supplied === undefined
|
|
? applyModuleMigrations({ aggregate: inventory, database: migrationDatabase })
|
|
: applyModuleMigrations({
|
|
aggregate: inventory,
|
|
approvals: approvalProvider(supplied),
|
|
database: migrationDatabase,
|
|
});
|
|
const error = await captureRunnerError(operation, "MIGRATION_APPROVAL_REQUIRED");
|
|
expect(error.migrationIdentity).toBe(destructive.identity);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.destructive_records")).toBe(false);
|
|
}
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
approvals: approvalProvider([exact]),
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([destructive.identity]);
|
|
expect(await relationExists(database.client, "public.destructive_records")).toBe(true);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
destructive.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("sanitizes a real PostgreSQL failure and normal command reporting", async () => {
|
|
await withIsolatedDatabase("sanitization", async (database) => {
|
|
const hostilePassword = "fake-password-7f7e620d";
|
|
const hostileUrl = `postgresql://hostile:${hostilePassword}@private-db.invalid/secret`;
|
|
const hostileHostname = "private-db.invalid";
|
|
const hostilePath = "/srv/ekklesia/private/migration.sql";
|
|
const hostileSqlMarker = "HOSTILE_RAW_SQL_MARKER_45f28a";
|
|
const sql = `DO $body$
|
|
BEGIN
|
|
RAISE EXCEPTION '${hostilePassword} ${hostileUrl} ${hostileHostname} ${hostilePath} ${hostileSqlMarker}';
|
|
END
|
|
$body$;\n`;
|
|
const failing = migration("sanitization", 0, {
|
|
sql,
|
|
tag: "0000_hostile_failure",
|
|
});
|
|
const capturedOutput: string[] = [];
|
|
const errorSpy = vi
|
|
.spyOn(console, "error")
|
|
.mockImplementation((...values: unknown[]) =>
|
|
capturedOutput.push(values.map(String).join(" ")),
|
|
);
|
|
const logSpy = vi
|
|
.spyOn(console, "log")
|
|
.mockImplementation((...values: unknown[]) =>
|
|
capturedOutput.push(values.map(String).join(" ")),
|
|
);
|
|
let publicError: MigrationRunnerError;
|
|
try {
|
|
publicError = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("sanitization", [failing])),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
reportMigrationCommandError(publicError);
|
|
} finally {
|
|
errorSpy.mockRestore();
|
|
logSpy.mockRestore();
|
|
}
|
|
|
|
const publicText = [
|
|
String(publicError),
|
|
publicError.message,
|
|
publicError.stack ?? "",
|
|
JSON.stringify(publicError),
|
|
...capturedOutput,
|
|
].join("\n");
|
|
expect(publicError.migrationIdentity).toBe(failing.identity);
|
|
expect(publicText.includes("MIGRATION_EXECUTION_FAILED")).toBe(true);
|
|
expect(publicText.includes(failing.identity)).toBe(true);
|
|
const hostileValues = [
|
|
hostilePassword,
|
|
hostileUrl,
|
|
hostileHostname,
|
|
hostilePath,
|
|
hostileSqlMarker,
|
|
"RAISE EXCEPTION",
|
|
database.url,
|
|
new URL(database.url).password,
|
|
new URL(database.url).hostname,
|
|
];
|
|
expect(hostileValues.some((hostileValue) => publicText.includes(hostileValue))).toBe(false);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
});
|
|
});
|
|
});
|
|
|
|
const coreIdentityTables = Object.freeze([
|
|
"ekklesia_core_user",
|
|
"ekklesia_core_session",
|
|
"ekklesia_core_account",
|
|
"ekklesia_core_verification",
|
|
]);
|
|
|
|
async function loadRealCoreMigrations(): Promise<readonly DiscoveredModuleMigration[]> {
|
|
const migrations = await discoverCoreMigrations({ coreMigrationsDirectory });
|
|
if (migrations.length === 0) {
|
|
throw harnessError("The committed Core identity migration is missing.");
|
|
}
|
|
return migrations;
|
|
}
|
|
|
|
interface IndexRow extends QueryResultRow {
|
|
readonly indexname: string;
|
|
readonly tablename: string;
|
|
}
|
|
|
|
async function coreIdentityIndexNames(client: Client): Promise<readonly string[]> {
|
|
const result = await safeQuery<IndexRow>(
|
|
client,
|
|
`SELECT tablename, indexname FROM pg_catalog.pg_indexes
|
|
WHERE schemaname = 'public' AND tablename = ANY($1::text[])
|
|
ORDER BY tablename, indexname`,
|
|
[coreIdentityTables],
|
|
);
|
|
return Object.freeze(result.rows.map((row) => row.indexname));
|
|
}
|
|
|
|
describe.sequential("live PostgreSQL Core identity migration execution", () => {
|
|
it("plans the committed Core identity migration on a fresh database without mutation", async () => {
|
|
await withIsolatedDatabase("core_fresh_plan", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const inventory = aggregate(moduleFixture("core", coreMigrations));
|
|
const before = await databaseState(database.client);
|
|
|
|
const plan = await planModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(plan.entries.map((entry) => entry.status)).toEqual(
|
|
coreMigrations.map(() => "pending"),
|
|
);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(false);
|
|
}
|
|
});
|
|
});
|
|
|
|
it("applies the committed Core identity migration and creates exactly the expected tables and indexes", async () => {
|
|
await withIsolatedDatabase("core_apply", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const inventory = aggregate(moduleFixture("core", coreMigrations));
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual(
|
|
coreMigrations.map((entry) => entry.identity),
|
|
);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(true);
|
|
}
|
|
expect(await coreIdentityIndexNames(database.client)).toEqual([
|
|
"ekklesia_core_account_pkey",
|
|
"ekklesia_core_account_user_id_idx",
|
|
"ekklesia_core_session_pkey",
|
|
"ekklesia_core_session_token_unique",
|
|
"ekklesia_core_session_user_id_idx",
|
|
"ekklesia_core_user_email_unique",
|
|
"ekklesia_core_user_pkey",
|
|
"ekklesia_core_verification_identifier_idx",
|
|
"ekklesia_core_verification_pkey",
|
|
]);
|
|
|
|
const history = await readHistory(database.client);
|
|
expect(history).toHaveLength(coreMigrations.length);
|
|
expect(history.map(({ applied_at: _appliedAt, ...row }) => row)).toEqual(
|
|
coreMigrations.map((entry, index) => ({
|
|
applied_order: index + 1,
|
|
checksum: entry.checksum,
|
|
identity: entry.identity,
|
|
migration_index: entry.index,
|
|
migration_tag: entry.tag,
|
|
module_id: "core",
|
|
})),
|
|
);
|
|
});
|
|
});
|
|
|
|
it("performs a genuine no-op on a second Core identity apply", async () => {
|
|
await withIsolatedDatabase("core_no_op", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const inventory = aggregate(moduleFixture("core", coreMigrations));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: inventory, database: migrationDatabase });
|
|
const before = await databaseState(database.client);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
});
|
|
});
|
|
|
|
it("coexists with a fixture module migration, executing Core first as documented", async () => {
|
|
await withIsolatedDatabase("core_and_module", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const people = peopleInitial();
|
|
const inventory = aggregate(
|
|
moduleFixture("core", coreMigrations),
|
|
moduleFixture("people", [people]),
|
|
);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([
|
|
...coreMigrations.map((entry) => entry.identity),
|
|
people.identity,
|
|
]);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(true);
|
|
}
|
|
expect(await relationExists(database.client, "public.people_records")).toBe(true);
|
|
const history = await readHistory(database.client);
|
|
expect(history.map((row) => row.module_id)).toEqual([
|
|
...coreMigrations.map(() => "core"),
|
|
"people",
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("does not corrupt previously valid module-only history when Core is introduced later", async () => {
|
|
await withIsolatedDatabase("core_added_later", async (database) => {
|
|
const people = peopleInitial();
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("people", [people])),
|
|
database: migrationDatabase,
|
|
});
|
|
const historyBeforeCore = await readHistory(database.client);
|
|
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const result = await applyModuleMigrations({
|
|
aggregate: aggregate(
|
|
moduleFixture("core", coreMigrations),
|
|
moduleFixture("people", [people]),
|
|
),
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual(
|
|
coreMigrations.map((entry) => entry.identity),
|
|
);
|
|
const historyAfterCore = await readHistory(database.client);
|
|
const peopleRowBefore = historyBeforeCore.find((row) => row.module_id === "people");
|
|
const peopleRowAfter = historyAfterCore.find((row) => row.module_id === "people");
|
|
expect(peopleRowAfter).toEqual(peopleRowBefore);
|
|
expect(historyAfterCore).toHaveLength(historyBeforeCore.length + coreMigrations.length);
|
|
});
|
|
});
|
|
|
|
it("does not damage the committed Core identity migration when a later module migration fails", async () => {
|
|
await withIsolatedDatabase("core_rollback_safety", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("core", coreMigrations)),
|
|
database: migrationDatabase,
|
|
});
|
|
const historyAfterCore = await readHistory(database.client);
|
|
|
|
const failing = migration("broken", 0, {
|
|
sql: `CREATE TABLE "broken_records" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
|
|
tag: "0000_broken",
|
|
});
|
|
|
|
const error = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate(
|
|
moduleFixture("core", coreMigrations),
|
|
moduleFixture("broken", [failing]),
|
|
),
|
|
database: migrationDatabase,
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
|
|
expect(error.migrationIdentity).toBe(failing.identity);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(true);
|
|
}
|
|
expect(await relationExists(database.client, "public.broken_records")).toBe(false);
|
|
expect(await readHistory(database.client)).toEqual(historyAfterCore);
|
|
await expectMigrationLockAvailable(database.client);
|
|
});
|
|
});
|
|
});
|
|
|
|
/**
|
|
* Pass 4: the migration commands' finite timeout defaults, and whole-batch SQL execution, proven
|
|
* against real PostgreSQL and a real TCP endpoint rather than a fake.
|
|
*
|
|
* Every listener, client, and disposable database here is closed in `finally`/`afterAll`, including
|
|
* after a failed assertion.
|
|
*/
|
|
describe.sequential("live PostgreSQL migration command hardening", () => {
|
|
it("bounds a connection to an endpoint that accepts TCP but never completes PostgreSQL startup", async () => {
|
|
// A real blackhole listener bound only to loopback: it accepts the socket and then says
|
|
// nothing, which is exactly the case `connectionTimeoutMillis` exists for.
|
|
const sockets: net.Socket[] = [];
|
|
const listener = net.createServer((socket) => {
|
|
sockets.push(socket);
|
|
});
|
|
try {
|
|
await new Promise<void>((resolve, reject) => {
|
|
listener.once("error", reject);
|
|
listener.listen(0, "127.0.0.1", () => resolve());
|
|
});
|
|
const address = listener.address();
|
|
if (!address || typeof address !== "object") {
|
|
throw harnessError("The blackhole listener has no bound address.");
|
|
}
|
|
|
|
// A short, test-only override; production keeps MIGRATION_CONNECTION_TIMEOUT_MS unchanged.
|
|
const blackholeDatabase = createPostgresMigrationDatabase(
|
|
`postgresql://ekklesia:[email protected]:${address.port}/ekklesia`,
|
|
{ connectionTimeoutMs: 300, operationTimeoutMs: 300 },
|
|
);
|
|
|
|
const startedAt = Date.now();
|
|
let captured: unknown;
|
|
try {
|
|
await blackholeDatabase.openSession();
|
|
} catch (error) {
|
|
captured = error;
|
|
}
|
|
|
|
expect(captured).toBeInstanceOf(PostgresMigrationDatabaseError);
|
|
expect(captured).toMatchObject({ code: "POSTGRES_MIGRATION_CONNECTION_FAILED" });
|
|
// Finite, and nowhere near an unbounded wait.
|
|
expect(Date.now() - startedAt).toBeLessThan(10_000);
|
|
const publicText = `${String(captured)}\n${(captured as Error).stack ?? ""}`;
|
|
expect(publicText).not.toContain("127.0.0.1");
|
|
expect(publicText).not.toContain("ekklesia:ekklesia");
|
|
expect(publicText).not.toContain(String(address.port));
|
|
} finally {
|
|
for (const socket of sockets) {
|
|
socket.destroy();
|
|
}
|
|
await new Promise<void>((resolve) => {
|
|
listener.close(() => resolve());
|
|
});
|
|
}
|
|
});
|
|
|
|
it("aborts a migration that exceeds the operation deadline and commits no history row", async () => {
|
|
await withIsolatedDatabase("operation_timeout", async (database) => {
|
|
const slow = migration("people", 0, {
|
|
sql: "SELECT pg_sleep(5);",
|
|
tag: "0000_slow",
|
|
});
|
|
// The operation deadline is the behavior under test, so a bounded server-side delay is the
|
|
// only way to exercise it.
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url, {
|
|
connectionTimeoutMs: 5_000,
|
|
operationTimeoutMs: 250,
|
|
});
|
|
|
|
const error = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate({ id: "people", migrations: [slow] }),
|
|
database: migrationDatabase,
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
|
|
expect(error.migrationIdentity).toBe(slow.identity);
|
|
const publicText = `${String(error)}\n${error.stack ?? ""}`;
|
|
expect(publicText).not.toContain("pg_sleep");
|
|
expect(publicText).not.toContain(database.name);
|
|
|
|
// Nothing was committed: the transaction rolled back with its history insert.
|
|
const history = await safeQuery<{ identity: string }>(
|
|
database.client,
|
|
`select identity from public.ekklesia_core_migrations`,
|
|
).catch(() => undefined);
|
|
expect(history?.rows ?? []).toHaveLength(0);
|
|
});
|
|
});
|
|
|
|
it("executes marker bytes inside literals, dollar-quoted bodies, and comments intact", async () => {
|
|
await withIsolatedDatabase("intact_sql", async (database) => {
|
|
const marker = "--> statement-breakpoint";
|
|
const sql = [
|
|
`CREATE TABLE "people_notes" ("id" integer primary key, "body" text not null);`,
|
|
marker,
|
|
`INSERT INTO "people_notes" ("id", "body") VALUES (1, 'keep this: ${marker} intact');`,
|
|
marker,
|
|
`CREATE FUNCTION people_touch() RETURNS integer AS $$`,
|
|
`BEGIN`,
|
|
` -- ${marker}`,
|
|
` PERFORM 1;`,
|
|
` RETURN 42;`,
|
|
`END;`,
|
|
`$$ LANGUAGE plpgsql;`,
|
|
marker,
|
|
`/* block comment with ${marker} */`,
|
|
`CREATE TABLE "odd${marker}name" ("id" integer);`,
|
|
].join("\n");
|
|
const intact = migration("people", 0, { sql, tag: "0000_intact" });
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: aggregate({ id: "people", migrations: [intact] }),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
expect(result.appliedCount).toBe(1);
|
|
|
|
// The literal survived byte-for-byte, so the marker was never treated as a split point.
|
|
const stored = await safeQuery<{ body: string }>(
|
|
database.client,
|
|
`select body from "people_notes" where id = 1`,
|
|
);
|
|
expect(stored.rows[0]?.body).toBe(`keep this: ${marker} intact`);
|
|
|
|
// The dollar-quoted body executed as one function, not as fragments.
|
|
const called = await safeQuery<{ result: number }>(
|
|
database.client,
|
|
`select people_touch() as result`,
|
|
);
|
|
expect(Number(called.rows[0]?.result)).toBe(42);
|
|
|
|
// The quoted identifier kept the marker in its name.
|
|
const relations = await safeQuery<{ table_name: string }>(
|
|
database.client,
|
|
`select table_name from information_schema.tables
|
|
where table_schema = 'public' and table_name like 'odd%'`,
|
|
);
|
|
expect(relations.rows.map((row) => row.table_name)).toEqual([`odd${marker}name`]);
|
|
});
|
|
});
|
|
|
|
it("rolls back the whole multi-statement migration when a later statement fails", async () => {
|
|
await withIsolatedDatabase("intact_rollback", async (database) => {
|
|
const failing = migration("people", 0, {
|
|
sql: [
|
|
`CREATE TABLE "people_partial" ("id" integer primary key);`,
|
|
`--> statement-breakpoint`,
|
|
`INSERT INTO "people_partial" ("id") VALUES (1);`,
|
|
`--> statement-breakpoint`,
|
|
`INSERT INTO "people_partial" ("id") VALUES (1);`,
|
|
].join("\n"),
|
|
tag: "0000_rollback",
|
|
});
|
|
|
|
const error = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate({ id: "people", migrations: [failing] }),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
expect(error.migrationIdentity).toBe(failing.identity);
|
|
|
|
// The earlier statements in the same batch rolled back with it.
|
|
const relations = await safeQuery(
|
|
database.client,
|
|
`select table_name from information_schema.tables
|
|
where table_schema = 'public' and table_name = 'people_partial'`,
|
|
);
|
|
expect(relations.rows).toHaveLength(0);
|
|
});
|
|
});
|
|
});
|
|
|
|
/**
|
|
* Pass 4 correction: what each timeout-shaped connection-string key and inherited environment
|
|
* variable *actually* does to the pinned `[email protected]`, measured rather than inferred from its merge.
|
|
*
|
|
* The distinction matters: the boundary rejects several keys, but only one of them genuinely
|
|
* weakens the selected policy. Documenting a key as "overrides the explicit field" when it does not
|
|
* is the kind of claim this suite exists to prevent.
|
|
*
|
|
* Every listener, socket, client, and disposable database is closed in `finally`.
|
|
*/
|
|
describe.sequential("live PostgreSQL migration timeout precedence", () => {
|
|
const productionClientFields = Object.freeze({
|
|
connectionTimeoutMillis: 10_000,
|
|
query_timeout: 10_000,
|
|
statement_timeout: 10_000,
|
|
});
|
|
|
|
it("does not let a URL weaken the effective connect timer, whichever spelling is used", async () => {
|
|
const sockets: net.Socket[] = [];
|
|
const listener = net.createServer((socket) => {
|
|
sockets.push(socket);
|
|
});
|
|
try {
|
|
await new Promise<void>((resolve, reject) => {
|
|
listener.once("error", reject);
|
|
listener.listen(0, "127.0.0.1", () => resolve());
|
|
});
|
|
const address = listener.address();
|
|
if (!address || typeof address !== "object") {
|
|
throw harnessError("The blackhole listener has no bound address.");
|
|
}
|
|
const base = `postgresql://ekklesia:[email protected]:${address.port}/ekklesia`;
|
|
|
|
// A short explicit deadline stands in for the production 10 s so the measurement is fast.
|
|
for (const query of [
|
|
"",
|
|
"?connectionTimeoutMillis=0",
|
|
"?connectionTimeoutMillis=900000",
|
|
"?connect_timeout=0",
|
|
]) {
|
|
const client = new Client({
|
|
connectionString: `${base}${query}`,
|
|
connectionTimeoutMillis: 500,
|
|
});
|
|
client.on("error", () => undefined);
|
|
const startedAt = Date.now();
|
|
let settled = false;
|
|
try {
|
|
await client.connect();
|
|
} catch {
|
|
settled = true;
|
|
}
|
|
const elapsed = Date.now() - startedAt;
|
|
try {
|
|
await client.end();
|
|
} catch {
|
|
// Already unusable.
|
|
}
|
|
|
|
// The pure-JS client reads `connectionTimeoutMillis` from the object handed to its
|
|
// constructor, before any connection-string merge; the merged value only reaches
|
|
// `ConnectionParameters.connect_timeout`, which solely feeds `getLibpqConnectionString()`.
|
|
expect(settled, `expected the connect timer to fire for ${JSON.stringify(query)}`).toBe(
|
|
true,
|
|
);
|
|
expect(elapsed, `unexpected timer for ${JSON.stringify(query)}`).toBeLessThan(3_000);
|
|
}
|
|
|
|
// ...but the driver-visible field really is polluted, which is why the boundary rejects the
|
|
// key rather than relying on the pure-JS path staying this way. `connectionParameters` is
|
|
// real on the pinned client but absent from its published types, so it is read through a
|
|
// narrow structural view rather than asserted away.
|
|
const connectTimeoutOf = (connectionString: string): string => {
|
|
const client = new Client({ connectionString, ...productionClientFields }) as unknown as {
|
|
readonly connectionParameters: { readonly connect_timeout: unknown };
|
|
};
|
|
(client as unknown as Client).on("error", () => undefined);
|
|
return String(client.connectionParameters.connect_timeout);
|
|
};
|
|
expect(connectTimeoutOf(`${base}?connectionTimeoutMillis=0`)).toBe("0");
|
|
expect(connectTimeoutOf(base)).toBe("10");
|
|
} finally {
|
|
for (const socket of sockets) {
|
|
socket.destroy();
|
|
}
|
|
await new Promise<void>((resolve) => {
|
|
listener.close(() => resolve());
|
|
});
|
|
}
|
|
});
|
|
|
|
it("records what each timeout source does to the real server's statement_timeout", async () => {
|
|
await withIsolatedDatabase("timeout_sources", async (database) => {
|
|
async function effectiveStatementTimeout(
|
|
query: string,
|
|
environment: Readonly<Record<string, string>> = {},
|
|
): Promise<string> {
|
|
const saved = new Map<string, string | undefined>();
|
|
for (const [key, value] of Object.entries(environment)) {
|
|
saved.set(key, process.env[key]);
|
|
process.env[key] = value;
|
|
}
|
|
const client = new Client({
|
|
connectionString: `${database.url}${query}`,
|
|
...productionClientFields,
|
|
});
|
|
client.on("error", () => undefined);
|
|
try {
|
|
await client.connect();
|
|
const result = await client.query<{ statement_timeout: string }>(
|
|
"show statement_timeout",
|
|
);
|
|
return result.rows[0]?.statement_timeout ?? "unknown";
|
|
} catch {
|
|
return "connection failed";
|
|
} finally {
|
|
try {
|
|
await client.end();
|
|
} catch {
|
|
// Already unusable.
|
|
}
|
|
for (const [key, value] of saved) {
|
|
if (value === undefined) {
|
|
delete process.env[key];
|
|
} else {
|
|
process.env[key] = value;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// The baseline the policy selects.
|
|
expect(await effectiveStatementTimeout("")).toBe("10s");
|
|
|
|
// The one key that genuinely wins, and the reason it must stay rejected.
|
|
expect(await effectiveStatementTimeout("?statement_timeout=0")).toBe("0");
|
|
expect(await effectiveStatementTimeout("?statement_timeout=900000")).toBe("15min");
|
|
|
|
// `query_timeout` from the URL *does* override the explicit field, but it does not weaken the
|
|
// bound — it breaks the client. `client.query` computes its read timer as
|
|
// `queryConfig.query_timeout || connectionParameters.query_timeout`, and the parsed URL value
|
|
// is the *string* `"0"`, which is truthy, so every query gets a 0 ms read timer and fails
|
|
// nondeterministically ("Query read timeout") depending on which wins the race. The
|
|
// deterministic, assertable fact is the parameter the URL installed.
|
|
const racyClient = new Client({
|
|
connectionString: `${database.url}?query_timeout=0`,
|
|
...productionClientFields,
|
|
}) as unknown as { readonly connectionParameters: { readonly query_timeout: unknown } };
|
|
(racyClient as unknown as Client).on("error", () => undefined);
|
|
expect(racyClient.connectionParameters.query_timeout).toBe("0");
|
|
|
|
// `options` does *not* override the named startup parameter — including a backslash-escaped
|
|
// spelling — so rejecting it is unsupported-configuration hygiene, not a precedence fix.
|
|
expect(
|
|
await effectiveStatementTimeout(`?options=${encodeURIComponent("-c statement_timeout=0")}`),
|
|
).toBe("10s");
|
|
expect(
|
|
await effectiveStatementTimeout(
|
|
`?options=${encodeURIComponent("-c statement_timeout=900s")}`,
|
|
),
|
|
).toBe("10s");
|
|
expect(
|
|
await effectiveStatementTimeout(
|
|
`?options=${encodeURIComponent("-c statement_timeout\\=0")}`,
|
|
),
|
|
).toBe("10s");
|
|
|
|
// Inherited environment cannot reach the selected policy either.
|
|
expect(await effectiveStatementTimeout("", { PGOPTIONS: "-c statement_timeout=0" })).toBe(
|
|
"10s",
|
|
);
|
|
expect(await effectiveStatementTimeout("", { PGCONNECT_TIMEOUT: "0" })).toBe("10s");
|
|
|
|
// An unrelated parameter changes nothing.
|
|
expect(await effectiveStatementTimeout("?application_name=ekklesia")).toBe("10s");
|
|
});
|
|
});
|
|
|
|
it("refuses every rejected key at the shared boundary before a client is constructed", async () => {
|
|
await withIsolatedDatabase("timeout_boundary", async (database) => {
|
|
for (const query of [
|
|
"?connectionTimeoutMillis=0",
|
|
"?connectionTimeoutMillis=900000",
|
|
"?connect_timeout=0",
|
|
"?query_timeout=0",
|
|
"?statement_timeout=0",
|
|
`?options=${encodeURIComponent("-c statement_timeout=0")}`,
|
|
]) {
|
|
let captured: unknown;
|
|
try {
|
|
createPostgresMigrationDatabase(`${database.url}${query}`);
|
|
} catch (error) {
|
|
captured = error;
|
|
}
|
|
expect(captured, `expected ${query} rejected`).toBeInstanceOf(
|
|
PostgresMigrationDatabaseError,
|
|
);
|
|
expect(captured).toMatchObject({ code: "POSTGRES_MIGRATION_CONFIGURATION_INVALID" });
|
|
const publicText = `${String(captured)}\n${(captured as Error).stack ?? ""}`;
|
|
expect(publicText).not.toContain(database.name);
|
|
expect(publicText).not.toContain("127.0.0.1");
|
|
}
|
|
|
|
// The clean URL the commands actually use still works, end to end, with the real defaults.
|
|
const applied = await applyModuleMigrations({
|
|
aggregate: aggregate({
|
|
id: "people",
|
|
migrations: [migration("people", 0, { sql: "SELECT 1;", tag: "0000_ok" })],
|
|
}),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
expect(applied.appliedCount).toBe(1);
|
|
});
|
|
});
|
|
});
|