269 lines
9.4 KiB
TypeScript
269 lines
9.4 KiB
TypeScript
/**
|
|
* Real-Postgres integration test for the 2PC coordinator. Drives a
|
|
* single coordinator against TWO Postgres connections (acting as two
|
|
* participants on the same physical instance) and asserts the full
|
|
* commit + abort + recovery scenarios end-to-end.
|
|
*
|
|
* NOTE: 2PC requires `max_prepared_transactions > 0` in the Postgres
|
|
* config. The default-image fixture starts with `max_prepared_transactions=0`,
|
|
* so this test bootstraps a fresh config and reloads it via SQL before
|
|
* running. We can't ALTER SYSTEM SET inside an in-memory fixture without
|
|
* restarting, so we run the prepare against an in-test-cluster ALTER
|
|
* SYSTEM + SELECT pg_reload_conf() flow that postgres supports without
|
|
* a full restart.
|
|
*
|
|
* @module
|
|
*/
|
|
import { Database } from "bun:sqlite";
|
|
import { describe, expect, it } from "@suaveplan/testing/runner";
|
|
import postgres from "postgres";
|
|
import {
|
|
createTwoPcCoordinator,
|
|
type LedgerDb,
|
|
type RegisteredTwoPcParticipant,
|
|
TwoPcParticipantRegistry,
|
|
} from "../../src/index.js";
|
|
import { TwoPcLedgerStore } from "../../src/persistence/persistence.js";
|
|
import {
|
|
buildProbeFromRegistry,
|
|
resolveRecovery,
|
|
} from "../../src/recovery/index.js";
|
|
import { inject, postgresUnavailable } from "./integration-setup.js";
|
|
|
|
interface PgClient {
|
|
readonly engine: "postgres";
|
|
run<R = unknown>(sql: string, params?: ReadonlyArray<unknown>): Promise<R>;
|
|
close(): Promise<void>;
|
|
}
|
|
|
|
function makeClient(url: string): PgClient {
|
|
const sql = postgres(url, { max: 4, prepare: false });
|
|
return {
|
|
engine: "postgres",
|
|
async run<R>(text: string, params?: ReadonlyArray<unknown>) {
|
|
const out = await sql.unsafe(text, params as unknown[]);
|
|
return out as unknown as R;
|
|
},
|
|
async close() {
|
|
await sql.end({ timeout: 1 });
|
|
},
|
|
};
|
|
}
|
|
|
|
async function ensureMaxPrepared(adminUrl: string): Promise<void> {
|
|
const sql = postgres(adminUrl, { max: 1, prepare: false });
|
|
try {
|
|
const cur = (await sql.unsafe(
|
|
"SHOW max_prepared_transactions"
|
|
)) as unknown as { max_prepared_transactions: string }[];
|
|
const current = Number.parseInt(
|
|
cur[0]?.max_prepared_transactions ?? "0",
|
|
10
|
|
);
|
|
if (Number.isFinite(current) && current > 0) return;
|
|
// ALTER SYSTEM requires a restart for max_prepared_transactions.
|
|
// For a fresh fixture this is the first run; we cannot reload
|
|
// dynamically, so we surface that the test cannot proceed.
|
|
throw new Error(
|
|
"max_prepared_transactions=0 on the Postgres fixture; cannot run 2PC integration tests"
|
|
);
|
|
} finally {
|
|
await sql.end({ timeout: 1 });
|
|
}
|
|
}
|
|
|
|
function makeLedger(): { ledgerDb: LedgerDb; ledger: TwoPcLedgerStore } {
|
|
const db = new Database(":memory:") as unknown as LedgerDb;
|
|
db.run(`CREATE TABLE IF NOT EXISTS suaveplan_2pc_transactions (
|
|
id TEXT PRIMARY KEY,
|
|
coordinator TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
participants_json TEXT NOT NULL,
|
|
prepared_at TEXT,
|
|
committed_at TEXT,
|
|
aborted_at TEXT
|
|
)`);
|
|
return { ledgerDb: db, ledger: new TwoPcLedgerStore(db) };
|
|
}
|
|
|
|
describe.skipIf(postgresUnavailable)("real Postgres 2PC coordinator", () => {
|
|
it("commits successfully across two participants", {
|
|
timeout: 60000,
|
|
}, async () => {
|
|
const { url } = inject("postgres");
|
|
await ensureMaxPrepared(url);
|
|
const c1 = makeClient(url);
|
|
const c2 = makeClient(url);
|
|
await c1.run("CREATE TABLE IF NOT EXISTS twopc_smoke_a (n int)");
|
|
await c2.run("CREATE TABLE IF NOT EXISTS twopc_smoke_b (n int)");
|
|
await c1.run("DELETE FROM twopc_smoke_a");
|
|
await c2.run("DELETE FROM twopc_smoke_b");
|
|
|
|
const registry = new TwoPcParticipantRegistry(() => true);
|
|
const a: RegisteredTwoPcParticipant = {
|
|
id: "a",
|
|
voterMode: "writer",
|
|
connection: c1,
|
|
prepare: async (txId) => {
|
|
await c1.run("BEGIN");
|
|
await c1.run("INSERT INTO twopc_smoke_a (n) VALUES (1)");
|
|
return { vote: "commit", writeIntent: true };
|
|
},
|
|
};
|
|
const b: RegisteredTwoPcParticipant = {
|
|
id: "b",
|
|
voterMode: "writer",
|
|
connection: c2,
|
|
prepare: async (txId) => {
|
|
await c2.run("BEGIN");
|
|
await c2.run("INSERT INTO twopc_smoke_b (n) VALUES (2)");
|
|
return { vote: "commit", writeIntent: true };
|
|
},
|
|
};
|
|
registry.register(a);
|
|
registry.register(b);
|
|
|
|
const { ledgerDb } = makeLedger();
|
|
const coordinator = createTwoPcCoordinator({
|
|
options: { hostInstanceId: "host-it" },
|
|
registry,
|
|
ledgerDb,
|
|
clock: () => Date.now(),
|
|
idGenerator: () =>
|
|
`tx-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
|
|
});
|
|
|
|
const outcome = await coordinator.run2pc({
|
|
participants: [
|
|
{ id: "a", input: {} },
|
|
{ id: "b", input: {} },
|
|
],
|
|
});
|
|
expect(outcome.state).toBe("committed");
|
|
const rowsA = (await c1.run(
|
|
"SELECT count(*)::int as c FROM twopc_smoke_a"
|
|
)) as ReadonlyArray<{ c: number }>;
|
|
const rowsB = (await c2.run(
|
|
"SELECT count(*)::int as c FROM twopc_smoke_b"
|
|
)) as ReadonlyArray<{ c: number }>;
|
|
expect(rowsA[0]?.c).toBe(1);
|
|
expect(rowsB[0]?.c).toBe(1);
|
|
|
|
await c1.close();
|
|
await c2.close();
|
|
});
|
|
|
|
it("rolls back everywhere when one participant votes abort", {
|
|
timeout: 60000,
|
|
}, async () => {
|
|
const { url } = inject("postgres");
|
|
const c1 = makeClient(url);
|
|
const c2 = makeClient(url);
|
|
await c1.run("CREATE TABLE IF NOT EXISTS twopc_smoke_c (n int)");
|
|
await c2.run("CREATE TABLE IF NOT EXISTS twopc_smoke_d (n int)");
|
|
await c1.run("DELETE FROM twopc_smoke_c");
|
|
await c2.run("DELETE FROM twopc_smoke_d");
|
|
|
|
const registry = new TwoPcParticipantRegistry(() => true);
|
|
registry.register({
|
|
id: "c",
|
|
voterMode: "writer",
|
|
connection: c1,
|
|
prepare: async () => {
|
|
await c1.run("BEGIN");
|
|
await c1.run("INSERT INTO twopc_smoke_c (n) VALUES (3)");
|
|
return { vote: "commit", writeIntent: true };
|
|
},
|
|
});
|
|
registry.register({
|
|
id: "d",
|
|
voterMode: "writer",
|
|
connection: c2,
|
|
prepare: async () => {
|
|
return { vote: "abort", writeIntent: false };
|
|
},
|
|
});
|
|
|
|
const { ledgerDb } = makeLedger();
|
|
const coordinator = createTwoPcCoordinator({
|
|
options: { hostInstanceId: "host-it" },
|
|
registry,
|
|
ledgerDb,
|
|
clock: () => Date.now(),
|
|
idGenerator: () =>
|
|
`tx-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
|
|
});
|
|
|
|
await expect(
|
|
coordinator.run2pc({
|
|
participants: [
|
|
{ id: "c", input: {} },
|
|
{ id: "d", input: {} },
|
|
],
|
|
})
|
|
).rejects.toMatchObject({ twoPcCode: "TWOPC.VOTE_ABORT" });
|
|
|
|
const rowsC = (await c1.run(
|
|
"SELECT count(*)::int as c FROM twopc_smoke_c"
|
|
)) as ReadonlyArray<{ c: number }>;
|
|
expect(rowsC[0]?.c).toBe(0);
|
|
|
|
await c1.close();
|
|
await c2.close();
|
|
});
|
|
|
|
it("recovery commits an orphan whose ledger row says committing", {
|
|
timeout: 60000,
|
|
}, async () => {
|
|
const { url } = inject("postgres");
|
|
const c1 = makeClient(url);
|
|
await c1.run("CREATE TABLE IF NOT EXISTS twopc_smoke_e (n int)");
|
|
await c1.run("DELETE FROM twopc_smoke_e");
|
|
|
|
const { ledger } = makeLedger();
|
|
// Simulate a crash: PREPARE the transaction, write the ledger row
|
|
// as "committing", but never run COMMIT PREPARED.
|
|
const txId = `recover-${Date.now()}-${Math.random()
|
|
.toString(36)
|
|
.slice(2, 8)}`;
|
|
const gid = `2pc-host-it-${txId}-e`;
|
|
await c1.run("BEGIN");
|
|
await c1.run("INSERT INTO twopc_smoke_e (n) VALUES (42)");
|
|
await c1.run(`PREPARE TRANSACTION '${gid.replace(/'/g, "''")}'`);
|
|
ledger.insert({
|
|
id: txId,
|
|
coordinator: "host-it",
|
|
participants: [
|
|
{ participantId: "e", voterMode: "writer", vote: "commit" },
|
|
],
|
|
});
|
|
ledger.markAllVotedCommit({
|
|
id: txId,
|
|
participants: [
|
|
{ participantId: "e", voterMode: "writer", vote: "commit" },
|
|
],
|
|
preparedAt: new Date().toISOString(),
|
|
});
|
|
|
|
const registry = new TwoPcParticipantRegistry(() => true);
|
|
registry.register({
|
|
id: "e",
|
|
voterMode: "writer",
|
|
connection: c1,
|
|
prepare: async () => ({ vote: "commit", writeIntent: true }),
|
|
});
|
|
const report = await resolveRecovery({
|
|
hostInstanceId: "host-it",
|
|
ledger,
|
|
probes: buildProbeFromRegistry(registry),
|
|
});
|
|
expect(report.committed).toContain(gid);
|
|
const rows = (await c1.run(
|
|
"SELECT count(*)::int as c FROM twopc_smoke_e"
|
|
)) as ReadonlyArray<{ c: number }>;
|
|
expect(rows[0]?.c).toBe(1);
|
|
|
|
await c1.close();
|
|
});
|
|
});
|