test(sqlite): prove publication interruption recovery (#113512)

scripts/bench-sqlite-reliability.ts

@@ -39,6 +39,9 @@ function printProofLines(report: ReliabilityReport): void {
   console.log(
     `SQLITE_RELIABILITY_CRASH_EXIT_SIGNAL=${report.crashRecoveryProof.exit.signal ?? "none"}`,
   );
+  console.log(
+    `SQLITE_RELIABILITY_PUBLICATION_INTERRUPTION=${report.publicationInterruptionProof.beforePublish.recoveryVerified && report.publicationInterruptionProof.afterPublish.targetVerifiedAfterCrash && report.publicationInterruptionProof.afterPublish.recoveryVerified ? "verified" : "missing"}`,
+  );
   console.log(
     `SQLITE_RELIABILITY_WAL_SENTINEL=${report.transactionProof.committedWalSentinel ? "verified" : "missing"}`,
   );

scripts/lib/sqlite-reliability-contract.ts

@@ -78,6 +78,32 @@ export type ReliabilityReport = {
   };
   platform: NodeJS.Platform;
   profile: ProfileId;
+  publicationInterruptionProof: {
+    afterPublish: {
+      existingTargetPreserved: true;
+      exit: {
+        code: number | null;
+        signal: NodeJS.Signals | null;
+      };
+      recoveryVerified: true;
+      sourceStatePreserved: true;
+      stagingEntries: number;
+      targetVerifiedAfterCrash: true;
+      targetVisibleAfterCrash: true;
+    };
+    beforePublish: {
+      exit: {
+        code: number | null;
+        signal: NodeJS.Signals | null;
+      };
+      recoveryVerified: true;
+      retryPublished: true;
+      sourceStatePreserved: true;
+      stagingEntries: number;
+      targetVerifiedAfterCrash: false;
+      targetVisibleAfterCrash: false;
+    };
+  };
   retainedBatches: number;
   restoresVerified: number;
   rowsPerBatch: number;

scripts/lib/sqlite-reliability-publication-worker.ts

@@ -0,0 +1,51 @@
+import fs from "node:fs";
+import { pathToFileURL } from "node:url";
+import { createVerifiedSqliteSnapshot } from "../../src/infra/sqlite-snapshot.js";
+
+type PublicationCrashPoint = "after-publish" | "before-publish";
+
+const SLEEP_BUFFER = new Int32Array(new SharedArrayBuffer(4));
+
+function parseCrashPoint(value: string | undefined): PublicationCrashPoint {
+  if (value === "after-publish" || value === "before-publish") {
+    return value;
+  }
+  throw new Error(`invalid SQLite publication crash point: ${String(value)}`);
+}
+
+function holdAtCrashPoint(markerPath: string, crashPoint: PublicationCrashPoint): never {
+  const marker = fs.openSync(markerPath, "wx", 0o600);
+  try {
+    fs.writeFileSync(marker, `${crashPoint}\n`);
+    fs.fsyncSync(marker);
+  } finally {
+    fs.closeSync(marker);
+  }
+  while (true) {
+    Atomics.wait(SLEEP_BUFFER, 0, 0, 60_000);
+  }
+}
+
+async function main(argv: string[]): Promise<void> {
+  const crashPoint = parseCrashPoint(argv[0]);
+  const sourcePath = argv[1];
+  const targetPath = argv[2];
+  const markerPath = argv[3];
+  if (!sourcePath || !targetPath || !markerPath) {
+    throw new Error("SQLite publication worker requires source, target, and marker paths.");
+  }
+
+  await createVerifiedSqliteSnapshot({
+    sourcePath,
+    targetPath,
+    beforePublish:
+      crashPoint === "before-publish" ? () => holdAtCrashPoint(markerPath, crashPoint) : undefined,
+    afterPublish:
+      crashPoint === "after-publish" ? () => holdAtCrashPoint(markerPath, crashPoint) : undefined,
+  });
+  throw new Error(`SQLite publication worker passed ${crashPoint} without being terminated.`);
+}
+
+if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) {
+  await main(process.argv.slice(2));
+}

scripts/lib/sqlite-reliability-publication.ts

@@ -0,0 +1,243 @@
+import { spawn } from "node:child_process";
+import { createHash } from "node:crypto";
+import fs from "node:fs";
+import path from "node:path";
+import { setTimeout as delay } from "node:timers/promises";
+import { fileURLToPath } from "node:url";
+import { createVerifiedSqliteSnapshot } from "../../src/infra/sqlite-snapshot.js";
+import type { ReliabilityReport, ReliabilityStateProof } from "./sqlite-reliability-contract.js";
+
+type PublicationCrashPoint = "after-publish" | "before-publish";
+type PublicationExit = ReliabilityReport["publicationInterruptionProof"]["beforePublish"]["exit"];
+
+type CrashPointResult = {
+  exit: PublicationExit;
+  sourceStatePreserved: true;
+  stagingEntries: number;
+  targetState: ReliabilityStateProof | null;
+  targetVisibleAfterCrash: boolean;
+};
+
+const PUBLICATION_WORKER_PATH = fileURLToPath(
+  new URL("./sqlite-reliability-publication-worker.ts", import.meta.url),
+);
+const CRASH_POINT_TIMEOUT_MS = 120_000;
+
+function assertSameState(
+  actual: ReliabilityStateProof,
+  expected: ReliabilityStateProof,
+  label: string,
+): void {
+  if (
+    actual.batches !== expected.batches ||
+    actual.rows !== expected.rows ||
+    actual.sha256 !== expected.sha256
+  ) {
+    throw new Error(
+      `${label} changed reliability state: expected batches=${expected.batches} rows=${expected.rows} sha256=${expected.sha256}, got batches=${actual.batches} rows=${actual.rows} sha256=${actual.sha256}`,
+    );
+  }
+}
+
+function assertNoSqliteSidecars(targetPath: string): void {
+  for (const suffix of ["-journal", "-shm", "-wal"]) {
+    if (fs.existsSync(`${targetPath}${suffix}`)) {
+      throw new Error(
+        `publication crash left an unexpected SQLite sidecar: ${targetPath}${suffix}`,
+      );
+    }
+  }
+}
+
+function hashFile(filePath: string): string {
+  return createHash("sha256").update(fs.readFileSync(filePath)).digest("hex");
+}
+
+function listCrashStagingEntries(scratchPath: string): string[] {
+  return fs
+    .readdirSync(scratchPath)
+    .filter(
+      (entry) => entry.startsWith(".sqlite-publish-") || entry.startsWith(".sqlite-snapshot-"),
+    );
+}
+
+async function waitForCrashPoint(params: {
+  child: ReturnType<typeof spawn>;
+  markerPath: string;
+  readStderr: () => string;
+}): Promise<void> {
+  const deadline = Date.now() + CRASH_POINT_TIMEOUT_MS;
+  while (Date.now() < deadline) {
+    if (fs.existsSync(params.markerPath)) {
+      return;
+    }
+    if (params.child.exitCode !== null || params.child.signalCode !== null) {
+      throw new Error(
+        `SQLite publication worker exited before its crash point: ${params.readStderr().trim()}`,
+      );
+    }
+    await delay(5);
+  }
+  throw new Error(`SQLite publication worker did not reach its crash point within 120 seconds.`);
+}
+
+async function runCrashPoint(params: {
+  crashPoint: PublicationCrashPoint;
+  expectedState: ReliabilityStateProof;
+  scratchPath: string;
+  sourcePath: string;
+  verifyDatabase: (databasePath: string) => ReliabilityStateProof;
+}): Promise<CrashPointResult> {
+  const targetPath = path.join(params.scratchPath, `${params.crashPoint}.sqlite`);
+  const markerPath = path.join(params.scratchPath, `${params.crashPoint}.ready`);
+  let stderr = "";
+  const child = spawn(
+    process.execPath,
+    [
+      "--import",
+      "tsx",
+      PUBLICATION_WORKER_PATH,
+      params.crashPoint,
+      params.sourcePath,
+      targetPath,
+      markerPath,
+    ],
+    {
+      cwd: process.cwd(),
+      stdio: ["ignore", "ignore", "pipe"],
+    },
+  );
+  child.stderr.on("data", (chunk) => {
+    stderr += chunk.toString();
+  });
+  const exitPromise = new Promise<PublicationExit>((resolve, reject) => {
+    child.once("error", reject);
+    child.once("exit", (code, signal) => resolve({ code, signal }));
+  });
+
+  let crashStagingEntries: string[];
+  try {
+    await waitForCrashPoint({ child, markerPath, readStderr: () => stderr });
+    const targetVisibleAfterCrash = fs.existsSync(targetPath);
+    crashStagingEntries = listCrashStagingEntries(params.scratchPath);
+    if (crashStagingEntries.length === 0) {
+      throw new Error(`SQLite publication worker reached ${params.crashPoint} without staging.`);
+    }
+    if (!child.kill("SIGKILL")) {
+      throw new Error(`SQLite publication worker could not be terminated at ${params.crashPoint}.`);
+    }
+    const exit = await exitPromise;
+    if (exit.code === null && exit.signal === null) {
+      throw new Error(`SQLite publication worker reported no forced exit at ${params.crashPoint}.`);
+    }
+
+    const sourceState = params.verifyDatabase(params.sourcePath);
+    assertSameState(sourceState, params.expectedState, `${params.crashPoint} source`);
+    let targetState: ReliabilityStateProof | null = null;
+    if (targetVisibleAfterCrash) {
+      assertNoSqliteSidecars(targetPath);
+      targetState = params.verifyDatabase(targetPath);
+      assertSameState(targetState, params.expectedState, `${params.crashPoint} target`);
+    }
+
+    if (params.crashPoint === "before-publish") {
+      await createVerifiedSqliteSnapshot({
+        sourcePath: params.sourcePath,
+        targetPath,
+      });
+      assertNoSqliteSidecars(targetPath);
+      const retryState = params.verifyDatabase(targetPath);
+      assertSameState(retryState, params.expectedState, `${params.crashPoint} retry`);
+    } else {
+      const targetHash = hashFile(targetPath);
+      let retryError: unknown;
+      try {
+        await createVerifiedSqliteSnapshot({
+          sourcePath: params.sourcePath,
+          targetPath,
+        });
+      } catch (error) {
+        retryError = error;
+      }
+      if (!(retryError instanceof Error) || !/target already exists/iu.test(retryError.message)) {
+        throw new Error("SQLite retry did not preserve the already-published target.", {
+          cause: retryError,
+        });
+      }
+      assertNoSqliteSidecars(targetPath);
+      if (hashFile(targetPath) !== targetHash) {
+        throw new Error("SQLite retry changed the already-published target.");
+      }
+      const preservedState = params.verifyDatabase(targetPath);
+      assertSameState(preservedState, params.expectedState, `${params.crashPoint} retry`);
+    }
+    for (const entry of crashStagingEntries) {
+      if (!fs.existsSync(path.join(params.scratchPath, entry))) {
+        throw new Error(`SQLite retry removed crash staging it did not own: ${entry}`);
+      }
+    }
+
+    return {
+      exit,
+      sourceStatePreserved: true,
+      stagingEntries: crashStagingEntries.length,
+      targetState,
+      targetVisibleAfterCrash,
+    };
+  } finally {
+    if (child.exitCode === null && child.signalCode === null) {
+      child.kill("SIGKILL");
+      await exitPromise.catch(() => undefined);
+    }
+    fs.rmSync(markerPath, { force: true });
+    fs.rmSync(targetPath, { force: true });
+    for (const entry of listCrashStagingEntries(params.scratchPath)) {
+      fs.rmSync(path.join(params.scratchPath, entry), { force: true, recursive: true });
+    }
+  }
+}
+
+export async function runPublicationInterruptionProof(params: {
+  expectedState: ReliabilityStateProof;
+  scratchPath: string;
+  sourcePath: string;
+  verifyDatabase: (databasePath: string) => ReliabilityStateProof;
+}): Promise<ReliabilityReport["publicationInterruptionProof"]> {
+  fs.mkdirSync(params.scratchPath, { recursive: true, mode: 0o700 });
+  const beforePublish = await runCrashPoint({
+    ...params,
+    crashPoint: "before-publish",
+  });
+  if (beforePublish.targetVisibleAfterCrash || beforePublish.targetState) {
+    throw new Error("SQLite snapshot became visible before publication.");
+  }
+
+  const afterPublish = await runCrashPoint({
+    ...params,
+    crashPoint: "after-publish",
+  });
+  if (!afterPublish.targetVisibleAfterCrash || !afterPublish.targetState) {
+    throw new Error("SQLite snapshot was not complete after durable publication.");
+  }
+
+  return {
+    afterPublish: {
+      existingTargetPreserved: true,
+      exit: afterPublish.exit,
+      recoveryVerified: true,
+      sourceStatePreserved: true,
+      stagingEntries: afterPublish.stagingEntries,
+      targetVerifiedAfterCrash: true,
+      targetVisibleAfterCrash: true,
+    },
+    beforePublish: {
+      exit: beforePublish.exit,
+      recoveryVerified: true,
+      retryPublished: true,
+      sourceStatePreserved: true,
+      stagingEntries: beforePublish.stagingEntries,
+      targetVerifiedAfterCrash: false,
+      targetVisibleAfterCrash: false,
+    },
+  };
+}

scripts/lib/sqlite-reliability-runner.ts

@@ -27,6 +27,7 @@ import {
   type ReliabilityReport,
   type ReliabilityStateProof,
 } from "./sqlite-reliability-contract.js";
+import { runPublicationInterruptionProof } from "./sqlite-reliability-publication.js";
 import { monitorSqliteWalDuring } from "./sqlite-reliability-wal-monitor.js";
 import {
   crashWriter,
@@ -580,6 +581,25 @@ export async function runReliabilityStress(options: CliOptions): Promise<Reliabi
       throw new Error("SQLite reliability stress did not execute its crash recovery proof.");
     }
     const writerResult = await stopWriter(writer);
+    const stableState = verifyRestoredDatabase({
+      identity: target.identity,
+      path: target.path,
+      rowsPerBatch: profile.rowsPerBatch,
+      uncommittedBatch: null,
+    });
+    const publicationInterruptionProof = await runPublicationInterruptionProof({
+      expectedState: stableState,
+      scratchPath: path.join(runScratch, "publication-interruptions"),
+      sourcePath: target.path,
+      verifyDatabase: (databasePath) =>
+        verifyRestoredDatabase({
+          expectedState: stableState,
+          identity: target.identity,
+          path: databasePath,
+          rowsPerBatch: profile.rowsPerBatch,
+          uncommittedBatch: null,
+        }),
+    });
     const maintenanceProof = await runMaintenanceRoundTrip({
       env,
       repositoryProvider,
@@ -613,6 +633,7 @@ export async function runReliabilityStress(options: CliOptions): Promise<Reliabi
       },
       platform: process.platform,
       profile: options.profile,
+      publicationInterruptionProof,
       retainedBatches: profile.retainedBatches,
       restoresVerified: metrics.length + 1,
       rowsPerBatch: profile.rowsPerBatch,

test/scripts/bench-sqlite-reliability.test.ts

@@ -155,6 +155,7 @@ describe("scripts/bench-sqlite-reliability", () => {
     expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_TARGET=global");
     expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_RESTORES_VERIFIED=5");
     expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_CRASH_RECOVERY=verified");
+    expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_PUBLICATION_INTERRUPTION=verified");
     expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_POST_COMPACT_RESTORE=verified");
     const firstReport = JSON.parse(fs.readFileSync(firstOutput, "utf8")) as {
       concurrentRestoresVerified: number;
@@ -199,6 +200,32 @@ describe("scripts/bench-sqlite-reliability", () => {
         sourceDatabase: string;
         syncedRepository: string;
       };
+      publicationInterruptionProof: {
+        afterPublish: {
+          existingTargetPreserved: boolean;
+          exit: {
+            code: number | null;
+            signal: string | null;
+          };
+          recoveryVerified: boolean;
+          sourceStatePreserved: boolean;
+          stagingEntries: number;
+          targetVerifiedAfterCrash: boolean;
+          targetVisibleAfterCrash: boolean;
+        };
+        beforePublish: {
+          exit: {
+            code: number | null;
+            signal: string | null;
+          };
+          recoveryVerified: boolean;
+          retryPublished: boolean;
+          sourceStatePreserved: boolean;
+          stagingEntries: number;
+          targetVerifiedAfterCrash: boolean;
+          targetVisibleAfterCrash: boolean;
+        };
+      };
       restoresVerified: number;
       transactionProof: {
         committedWalSentinel: boolean;
@@ -226,6 +253,32 @@ describe("scripts/bench-sqlite-reliability", () => {
     expect(firstReport.crashRecoveryProof.stateAfterRecovery).toEqual(
       firstReport.crashRecoveryProof.stateBeforeKill,
     );
+    expect(firstReport.publicationInterruptionProof.beforePublish).toMatchObject({
+      recoveryVerified: true,
+      retryPublished: true,
+      sourceStatePreserved: true,
+      targetVerifiedAfterCrash: false,
+      targetVisibleAfterCrash: false,
+    });
+    expect(firstReport.publicationInterruptionProof.beforePublish.stagingEntries).toBeGreaterThan(
+      0,
+    );
+    expect(
+      firstReport.publicationInterruptionProof.beforePublish.exit.code !== null ||
+        firstReport.publicationInterruptionProof.beforePublish.exit.signal !== null,
+    ).toBe(true);
+    expect(firstReport.publicationInterruptionProof.afterPublish).toMatchObject({
+      existingTargetPreserved: true,
+      recoveryVerified: true,
+      sourceStatePreserved: true,
+      targetVerifiedAfterCrash: true,
+      targetVisibleAfterCrash: true,
+    });
+    expect(firstReport.publicationInterruptionProof.afterPublish.stagingEntries).toBeGreaterThan(0);
+    expect(
+      firstReport.publicationInterruptionProof.afterPublish.exit.code !== null ||
+        firstReport.publicationInterruptionProof.afterPublish.exit.signal !== null,
+    ).toBe(true);
     expect(firstReport.transactionProof.committedWalSentinel).toBe(true);
     expect(firstReport.transactionProof.heldRows).toBeGreaterThan(0);
     expect(firstReport.transactionProof.visibleAfterRestore).toBe(false);