120 lines
10 KiB
TypeScript
120 lines
10 KiB
TypeScript
/** Permanent owned PostgreSQL regression candidate. No source-capture TMP
|
|
* module, app config, application pool or dotenv import is needed. */
|
|
import assert from "node:assert/strict"
|
|
import { randomUUID } from "node:crypto"
|
|
import { basename, resolve } from "node:path"
|
|
import { pathToFileURL } from "node:url"
|
|
import { Pool, type PoolClient } from "pg"
|
|
assert.equal(process.env.MMAPP_OWN_IMPORT_CONTROL, "true")
|
|
assert(process.env.MMAPP_OWN_IMPORT_DIRECTORY && basename(process.env.MMAPP_OWN_IMPORT_DIRECTORY).startsWith("mm-import-control-"))
|
|
const adminUrl = new URL(process.env.MMAPP_OWN_IMPORT_ADMIN_URL ?? "")
|
|
assert.equal(adminUrl.hostname, "127.0.0.1"); assert.equal(adminUrl.pathname, "/postgres")
|
|
assert.equal(process.env.DATABASE_URL, undefined)
|
|
// The candidate runner can point at the real repository while this test stays
|
|
// in TMP. Once integrated, relative production source is the default.
|
|
const sourceRoot = process.env.MMAPP_OWN_IMPORT_SOURCE_ROOT
|
|
const source = (relative: string, ownRelative: string) => sourceRoot ? pathToFileURL(resolve(sourceRoot, "backend/src", relative)).href : new URL(ownRelative, import.meta.url).href
|
|
const { withPartitionWrite, PARTITION_WRITER_ADVISORY_KEY, isPartitionWriteOutcomeUnknown } = await import(source("services/partition-write-transaction.ts", "../services/partition-write-transaction.js"))
|
|
const { writeSqliteImportMarker, resetSqliteImportSequence } = await import(source("db/sqlite-import-control-writes.ts", "./sqlite-import-control-writes.js"))
|
|
const admin = new Pool({ connectionString: adminUrl.toString(), max: 2 })
|
|
const name = "import_control_" + randomUUID().replaceAll("-", "")
|
|
assert(/^import_control_[a-f0-9]{32}$/.test(name))
|
|
await admin.query('CREATE DATABASE "' + name + '" TEMPLATE template0')
|
|
const target = new URL(adminUrl); target.pathname = "/" + name
|
|
const pool = new Pool({ connectionString: target.toString(), max: 8, connectionTimeoutMillis: 3000 })
|
|
const held: PoolClient[] = [], releases: (() => void)[] = [], cases: string[] = []
|
|
async function until(check: () => Promise<boolean>) {
|
|
const end = performance.now() + 4000
|
|
while (!await check()) { if (performance.now() >= end) throw Error("Owned lock state not observed"); await new Promise(r => setTimeout(r, 10)) }
|
|
}
|
|
function fault(mode: "lost-commit" | "aborted-commit" | "rollback-after-setval") {
|
|
const commands: string[] = [], destroyed: boolean[] = []
|
|
let commitTag: string | undefined
|
|
const adapted = { async connect() { const client = await pool.connect(); return {
|
|
release(destroy = false) { destroyed.push(destroy); client.release(destroy) }, async query(text: string, values?: unknown[]) {
|
|
commands.push(text)
|
|
if (text === "COMMIT" && mode === "aborted-commit") await assert.rejects(client.query("SELECT 1/0"), (e: unknown) => !!e && typeof e === "object" && "code" in e && e.code === "22012")
|
|
const result = await client.query(text, values)
|
|
if (text === "COMMIT") commitTag = result.command
|
|
if (text === "COMMIT" && mode === "lost-commit") throw Error("Owned fixture loses actual COMMIT ACK")
|
|
if (text.startsWith("SELECT pg_catalog.setval") && mode === "rollback-after-setval") {
|
|
assert.equal((await client.query("ROLLBACK")).command, "ROLLBACK")
|
|
throw Error("Owned fixture reports failure after actual setval and acknowledged ROLLBACK")
|
|
}
|
|
return result
|
|
},
|
|
} } } as unknown as Pool
|
|
return { pool: adapted, commands, destroyed, tag: () => commitTag }
|
|
}
|
|
async function lastSequence() { return (await pool.query("SELECT last_value::text AS value FROM public.import_rows_id_seq")).rows[0].value }
|
|
async function markerPath() { return (await pool.query("SELECT sqlite_path FROM public.data_migration WHERE id=1")).rows[0].sqlite_path }
|
|
try {
|
|
await pool.query("CREATE TABLE public.data_migration(id integer PRIMARY KEY,sqlite_imported_at timestamptz,sqlite_path text,sqlite_sha256 text,report_json jsonb); INSERT INTO public.data_migration(id) VALUES(1); CREATE TABLE public.import_rows(id bigserial PRIMARY KEY); INSERT INTO public.import_rows(id) VALUES(42); CREATE TABLE public.no_sequence(id integer PRIMARY KEY)")
|
|
const exclusive = await pool.connect(); held.push(exclusive)
|
|
await exclusive.query("SELECT pg_advisory_lock($1::bigint)", [PARTITION_WRITER_ADVISORY_KEY])
|
|
const marker = writeSqliteImportMarker(pool, "owned.sqlite", "a".repeat(64), JSON.stringify({ import_rows: { copied: 1 } }))
|
|
const sequence = resetSqliteImportSequence(pool, "import_rows")
|
|
await until(async () => (await pool.query("SELECT count(*)::int AS n FROM pg_locks WHERE locktype='advisory' AND objid=$1::oid AND NOT granted", [PARTITION_WRITER_ADVISORY_KEY])).rows[0].n === 2)
|
|
assert.equal(await markerPath(), null); assert.equal(await lastSequence(), "1")
|
|
assert.equal((await exclusive.query("SELECT pg_advisory_unlock($1::bigint) AS unlocked", [PARTITION_WRITER_ADVISORY_KEY])).rows[0].unlocked, true)
|
|
await marker; assert.equal(await sequence, "sequence-set"); assert.equal(await lastSequence(), "42")
|
|
const saved = (await pool.query("SELECT sqlite_path,sqlite_sha256,report_json FROM public.data_migration WHERE id=1")).rows[0]
|
|
assert.equal(saved.sqlite_path, "owned.sqlite"); assert.equal(saved.sqlite_sha256, "a".repeat(64)); assert.deepEqual(saved.report_json, { import_rows: { copied: 1 } })
|
|
cases.push("marker and setval use same real shared admission before SQL, wait for exclusive drain and preserve values")
|
|
assert.equal(await resetSqliteImportSequence(pool, "no_sequence"), "sequence-absent")
|
|
cases.push("absent serial identity is explicit compatibility no-op")
|
|
|
|
let entered!: () => void, release!: () => void
|
|
const ready = new Promise<void>(r => { entered = r }), waiting = new Promise<void>(r => { release = r }); releases.push(release)
|
|
const writer = withPartitionWrite(async ({ client }: { client: PoolClient }) => { await client.query("UPDATE public.data_migration SET sqlite_path='already-admitted' WHERE id=1"); entered(); await waiting }, { connect: () => pool.connect() })
|
|
await ready
|
|
const drain = exclusive.query("SELECT pg_advisory_lock($1::bigint)", [PARTITION_WRITER_ADVISORY_KEY])
|
|
await until(async () => (await pool.query("SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype='advisory' AND objid=$1::oid AND NOT granted) AS waiting", [PARTITION_WRITER_ADVISORY_KEY])).rows[0].waiting === true)
|
|
release(); await writer; await drain; assert.equal(await markerPath(), "already-admitted")
|
|
await exclusive.query("SELECT pg_advisory_unlock($1::bigint)", [PARTITION_WRITER_ADVISORY_KEY])
|
|
cases.push("exclusive drain waits for already-admitted writer COMMIT")
|
|
|
|
const lostMarker = fault("lost-commit")
|
|
await assert.rejects(writeSqliteImportMarker(lostMarker.pool, "lost-ack.sqlite", "b".repeat(64), "{}"), isPartitionWriteOutcomeUnknown)
|
|
assert.equal(lostMarker.tag(), "COMMIT"); assert.equal(await markerPath(), "lost-ack.sqlite")
|
|
assert.equal(lostMarker.commands.filter(q => q.startsWith("UPDATE data_migration")).length, 1); assert.equal(lostMarker.commands.filter(q => q === "COMMIT").length, 1)
|
|
assert(!lostMarker.commands.includes("ROLLBACK")); assert.deepEqual(lostMarker.destroyed, [true])
|
|
cases.push("lost actual marker COMMIT ACK is UNKNOWN, one update and no second transition/replay")
|
|
|
|
await pool.query("INSERT INTO public.import_rows(id) VALUES(77)")
|
|
const lostSequence = fault("lost-commit")
|
|
await assert.rejects(resetSqliteImportSequence(lostSequence.pool, "import_rows"), isPartitionWriteOutcomeUnknown)
|
|
assert.equal(await lastSequence(), "77"); assert.equal(lostSequence.commands.filter(q => q.startsWith("SELECT pg_catalog.setval")).length, 1)
|
|
cases.push("lost actual sequence COMMIT ACK propagates UNKNOWN with setval already visible and no replay")
|
|
|
|
await pool.query("INSERT INTO public.import_rows(id) VALUES(88)")
|
|
const rolledSequence = fault("rollback-after-setval")
|
|
await assert.rejects(resetSqliteImportSequence(rolledSequence.pool, "import_rows"), isPartitionWriteOutcomeUnknown)
|
|
assert.equal(await lastSequence(), "88"); assert.equal(rolledSequence.commands.filter(q => q.startsWith("SELECT pg_catalog.setval")).length, 1)
|
|
cases.push("acknowledged actual ROLLBACK does not undo setval; post-sequence failure remains UNKNOWN")
|
|
|
|
const abortedDml = fault("aborted-commit")
|
|
await assert.rejects(withPartitionWrite(async ({ client }: { client: PoolClient }) => { await client.query("INSERT INTO public.no_sequence VALUES(999)") }, { connect: () => abortedDml.pool.connect() }), isPartitionWriteOutcomeUnknown)
|
|
assert.equal(abortedDml.tag(), "ROLLBACK"); assert.equal((await pool.query("SELECT count(*)::int AS n FROM public.no_sequence")).rows[0].n, 0)
|
|
assert.equal(abortedDml.commands.filter(q => q === "COMMIT").length, 1); assert(!abortedDml.commands.includes("ROLLBACK")); assert.deepEqual(abortedDml.destroyed, [true])
|
|
cases.push("real aborted DML COMMIT resolves ROLLBACK and is refused without success/retry/second transition")
|
|
const abortedMarker = fault("aborted-commit")
|
|
await assert.rejects(writeSqliteImportMarker(abortedMarker.pool, "must-not-publish.sqlite", "c".repeat(64), "{}"), isPartitionWriteOutcomeUnknown)
|
|
assert.equal(abortedMarker.tag(), "ROLLBACK"); assert.equal(await markerPath(), "lost-ack.sqlite")
|
|
cases.push("real aborted marker COMMIT=>ROLLBACK cannot publish success")
|
|
await pool.query("INSERT INTO public.import_rows(id) VALUES(99)")
|
|
const abortedSequence = fault("aborted-commit")
|
|
await assert.rejects(resetSqliteImportSequence(abortedSequence.pool, "import_rows"), isPartitionWriteOutcomeUnknown)
|
|
assert.equal(abortedSequence.tag(), "ROLLBACK"); assert.equal(await lastSequence(), "99")
|
|
cases.push("real aborted setval COMMIT=>ROLLBACK refuses success while sequence effect honestly remains")
|
|
await assert.rejects(resetSqliteImportSequence(pool, "missing_import_table"))
|
|
cases.push("non-compatibility SQL errors propagate; blanket catch and replay absent")
|
|
console.log(JSON.stringify({ cases: cases.length, names: cases, actualSql: true, genuineCommitRollbackCases: 3,
|
|
actualCurrentProductionPartitionWriter: true, actualCurrentProductionImporterHelpers: true, sourceCaptureTmpDependency: false,
|
|
allWriterDrainAuthenticated: false, readyAuthorized: false, productionDatabaseTouched: false, hostUsersGroupsServicesChanged: false }, null, 2))
|
|
} finally {
|
|
for (const release of releases) release()
|
|
for (const c of held) c.release(true)
|
|
await pool.end(); await admin.query('DROP DATABASE "' + name + '" WITH (FORCE)'); await admin.end()
|
|
}
|