ekklesiaandClaude Opus 5.5
98e59d4ba0
CI / verify (pull_request) Successful in 1m5s
CI / infrastructure-smoke (pull_request) Successful in 35s
CI / migration-smoke (pull_request) Successful in 1m52s
CI / package-release (pull_request) Successful in 35s
CI / image-build (pull_request) Successful in 31s
CI / deploy-smoke (same-origin) (pull_request) Successful in 45s
CI / deploy-smoke (split) (pull_request) Successful in 1m30s
CI / package-publication (pull_request) Successful in 44s
Every platform workspace package and every public @ekklesia/* package declares 0.2.0, and the deploy README's image build example names it. templates/module keeps its own version. 0.2.0, not 0.1.3: before 1.0 a minor version is the breaking one (docs/development/releases.md, ADR 0054), and since 0.1.2 main has broken modules, module authors, and operators: - #64 (ADR 0048): one generic image; installed modules live outside it; - #66, #72 (ADR 0049, 0050): modules install as Module Package v1, and move to the Module Database ABI; - #71: the manifest v8 cutover; v7 manifests are refused; - #79 (ADR 0053): the curated SDK drops defineModuleSchema and the legacy ModuleDatabase type; plus #67 (AGPL), #74, #77 (licensing, native targets), and the public package train (#80, #81). It is also the first package train to publish: 0.1.2 never will be. Because 0.2.0 is outside ^0.1.0, the same change moves to ^0.2.0: - the standalone fixture's @ekklesia ranges and its exact module-pack pin; - the compatibility ranges of the modules built for this release: the template, the standalone fixture, and the native proof module; - the test fixtures that model a module of the current release, or the current platform's versions; - the module-development guide's examples, and the release guide, which now names what a minor bump moves. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2046 lines
77 KiB
TypeScript
2046 lines
77 KiB
TypeScript
import { createHash, randomBytes } from "node:crypto";
|
|
import { cp, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
|
import net from "node:net";
|
|
import { tmpdir } from "node:os";
|
|
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,
|
|
discoverAllMigrations,
|
|
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, retained: [] });
|
|
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("retains an absent module's history and still applies later work (ADR 0048)", async () => {
|
|
await withIsolatedDatabase("absent_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 });
|
|
// History left by a module that is no longer installed: kept, reported, never fatal.
|
|
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 plan = await planModuleMigrations({
|
|
aggregate: completeInventory,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(plan.retained).toEqual([{ appliedCount: 1, moduleId: "ghost" }]);
|
|
expect(plan.pendingCount).toBe(1);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: completeInventory,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((migration) => migration.identity)).toEqual([later.identity]);
|
|
expect(result.retained).toEqual([{ appliedCount: 1, moduleId: "ghost" }]);
|
|
expect(await columnExists(database.client, "people_records", "email")).toBe(true);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
"ghost:0000_unknown",
|
|
later.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("rejects an unknown applied identity of an installed module 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)
|
|
`,
|
|
["people:0099_unknown", "people", "0099_unknown", 99, "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,
|
|
"people:0099_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, retained: [] });
|
|
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);
|
|
});
|
|
});
|
|
});
|
|
|
|
/**
|
|
* ADR 0048 end to end, with real discovery: Core's committed migrations, the module template
|
|
* installed in a modules directory outside the repository, and a real PostgreSQL. Core's newest
|
|
* migration is withheld from a copy of `core/drizzle` at first, so a genuinely later Core migration
|
|
* can be applied while the module is absent.
|
|
*/
|
|
describe.sequential("live PostgreSQL retained history for an uninstalled module", () => {
|
|
const repositoryRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..", "..");
|
|
const templateModule = path.join(repositoryRoot, "templates", "module");
|
|
const versions = { moduleSdkVersion: "0.2.0", platformVersion: "0.2.0" } as const;
|
|
|
|
async function coreWithoutNewestMigration(directory: string): Promise<string> {
|
|
const target = path.join(directory, "core-drizzle");
|
|
await cp(coreMigrationsDirectory, target, { recursive: true });
|
|
const journalPath = path.join(target, "meta", "_journal.json");
|
|
const policyPath = path.join(target, "meta", "_ekklesia.json");
|
|
const journal = JSON.parse(await readFile(journalPath, "utf8")) as {
|
|
entries: { idx: number; tag: string }[];
|
|
};
|
|
const policy = JSON.parse(await readFile(policyPath, "utf8")) as { entries: unknown[] };
|
|
const newest = journal.entries.pop();
|
|
policy.entries.pop();
|
|
if (!newest) {
|
|
throw harnessError("Core has no migration to withhold.");
|
|
}
|
|
await writeFile(journalPath, `${JSON.stringify(journal, null, 2)}\n`);
|
|
await writeFile(policyPath, `${JSON.stringify(policy, null, 2)}\n`);
|
|
await rm(path.join(target, `${newest.tag}.sql`));
|
|
await rm(path.join(target, "meta", `${String(newest.idx).padStart(4, "0")}_snapshot.json`));
|
|
return target;
|
|
}
|
|
|
|
async function installExample(modulesDirectory: string): Promise<void> {
|
|
for (const entry of ["package.json", "ekklesia.module.json", "drizzle", "dist"]) {
|
|
await cp(path.join(templateModule, entry), path.join(modulesDirectory, "example", entry), {
|
|
recursive: true,
|
|
});
|
|
}
|
|
}
|
|
|
|
async function inventory(coreDirectory: string, modulesDirectory: string) {
|
|
return discoverAllMigrations({
|
|
coreMigrationsDirectory: coreDirectory,
|
|
moduleDiscovery: { ...versions, modulesDirectory, modulesDirectoryRequired: true },
|
|
});
|
|
}
|
|
|
|
async function exampleValues(client: Client): Promise<readonly string[]> {
|
|
const result = await safeQuery<{ readonly value: string } & QueryResultRow>(
|
|
client,
|
|
"SELECT value FROM public.example_records ORDER BY value",
|
|
);
|
|
return result.rows.map((row) => row.value);
|
|
}
|
|
|
|
it("keeps an uninstalled module's data and history, applies later Core work, and restores it", async () => {
|
|
const scratch = await mkdtemp(path.join(tmpdir(), "ekklesia-retained-history-"));
|
|
try {
|
|
const earlyCore = await coreWithoutNewestMigration(scratch);
|
|
const modulesDirectory = path.join(scratch, "modules");
|
|
await mkdir(modulesDirectory);
|
|
await installExample(modulesDirectory);
|
|
|
|
await withIsolatedDatabase("retained_module", async (database) => {
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
|
|
// 1. Core (minus its newest migration) and the installed module, applied together.
|
|
const first = await applyModuleMigrations({
|
|
aggregate: await inventory(earlyCore, modulesDirectory),
|
|
database: migrationDatabase,
|
|
});
|
|
expect(first.applied.map((migration) => migration.identity)).toContain(
|
|
"example:0000_initial",
|
|
);
|
|
expect(first.retained).toEqual([]);
|
|
|
|
// 2. Module-owned data.
|
|
await safeQuery(
|
|
database.client,
|
|
"INSERT INTO public.example_records (id, value) VALUES (gen_random_uuid(), 'kept')",
|
|
);
|
|
|
|
// 3. The module is uninstalled: its directory is gone from module storage.
|
|
await rm(path.join(modulesDirectory, "example"), { recursive: true });
|
|
|
|
// 4. A later Core migration plans and applies while the module is absent.
|
|
const absentInventory = await inventory(coreMigrationsDirectory, modulesDirectory);
|
|
const absentPlan = await planModuleMigrations({
|
|
aggregate: absentInventory,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(absentPlan.retained).toEqual([{ appliedCount: 1, moduleId: "example" }]);
|
|
expect(
|
|
absentPlan.entries.filter((entry) => entry.status === "pending").map((e) => e.moduleId),
|
|
).toEqual(["core"]);
|
|
const later = await applyModuleMigrations({
|
|
aggregate: absentInventory,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(later.applied).toHaveLength(1);
|
|
expect(later.applied[0]?.moduleId).toBe("core");
|
|
expect(later.retained).toEqual([{ appliedCount: 1, moduleId: "example" }]);
|
|
|
|
// 5. The module's table and rows are untouched, and so is its history.
|
|
expect(await exampleValues(database.client)).toEqual(["kept"]);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toContain(
|
|
"example:0000_initial",
|
|
);
|
|
|
|
// 6. The module is reinstalled.
|
|
await installExample(modulesDirectory);
|
|
const restored = await inventory(coreMigrationsDirectory, modulesDirectory);
|
|
|
|
// 7. Its history is strict again, and valid; its data is where it left it.
|
|
const restoredPlan = await planModuleMigrations({
|
|
aggregate: restored,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(restoredPlan.retained).toEqual([]);
|
|
expect(restoredPlan.pendingCount).toBe(0);
|
|
expect(
|
|
restoredPlan.entries.find((entry) => entry.identity === "example:0000_initial")?.status,
|
|
).toBe("applied");
|
|
const noOp = await applyModuleMigrations({
|
|
aggregate: restored,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(noOp.noOp).toBe(true);
|
|
expect(await exampleValues(database.client)).toEqual(["kept"]);
|
|
|
|
// A returning package whose applied migration changed is refused, exactly as before.
|
|
const sqlPath = path.join(modulesDirectory, "example", "drizzle", "0000_initial.sql");
|
|
await writeFile(sqlPath, `${await readFile(sqlPath, "utf8")}-- edited\n`);
|
|
await captureRunnerError(
|
|
planModuleMigrations({
|
|
aggregate: await inventory(coreMigrationsDirectory, modulesDirectory),
|
|
database: migrationDatabase,
|
|
}),
|
|
"MIGRATION_CHECKSUM_MISMATCH",
|
|
);
|
|
expect(await exampleValues(database.client)).toEqual(["kept"]);
|
|
});
|
|
} finally {
|
|
await rm(scratch, { recursive: true, force: true });
|
|
}
|
|
});
|
|
});
|