From 2b19ae1f00eec0f72f41375589b100ae298bb4d3 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sat, 25 Jul 2026 18:50:50 +0800 Subject: [PATCH] fix(snapshot): recover complete pending sqlite snapshots (#113607) --- scripts/bench-sqlite-reliability.ts | 2 +- scripts/lib/sqlite-reliability-contract.ts | 6 +- scripts/lib/sqlite-reliability-repository.ts | 10 +- src/snapshot/local-repository.test.ts | 177 ++++++++++++++++++ src/snapshot/local-repository.ts | 131 ++++++++++++- test/scripts/bench-sqlite-reliability.test.ts | 10 +- 6 files changed, 317 insertions(+), 19 deletions(-) diff --git a/scripts/bench-sqlite-reliability.ts b/scripts/bench-sqlite-reliability.ts index e0c36e43d86b..f371f988402a 100644 --- a/scripts/bench-sqlite-reliability.ts +++ b/scripts/bench-sqlite-reliability.ts @@ -46,7 +46,7 @@ function printProofLines(report: ReliabilityReport): void { `SQLITE_RELIABILITY_RESTORE_INTERRUPTION=${report.maintenanceProof.restoreInterruption.beforePublish.recoveryVerified && report.maintenanceProof.restoreInterruption.beforePublish.retryRestored && report.maintenanceProof.restoreInterruption.afterPublish.targetVerifiedAfterCrash && report.maintenanceProof.restoreInterruption.afterPublish.existingTargetPreserved ? "verified" : "missing"}`, ); console.log( - `SQLITE_RELIABILITY_REPOSITORY_INTERRUPTION=${report.maintenanceProof.repositoryInterruption.beforePending.repositoryVerified && report.maintenanceProof.repositoryInterruption.beforePending.retryCreated && report.maintenanceProof.repositoryInterruption.pending.repositoryVerified && report.maintenanceProof.repositoryInterruption.pending.retryCreated && report.maintenanceProof.repositoryInterruption.afterCommit.crashSnapshotVerifiedAfterCrash && report.maintenanceProof.repositoryInterruption.afterCommit.retryCreated ? "verified" : "missing"}`, + `SQLITE_RELIABILITY_REPOSITORY_INTERRUPTION=${report.maintenanceProof.repositoryInterruption.beforePending.repositoryVerified && report.maintenanceProof.repositoryInterruption.beforePending.retryCreated && report.maintenanceProof.repositoryInterruption.pending.crashSnapshotVerifiedAfterCrash && report.maintenanceProof.repositoryInterruption.pending.crashSnapshotVisibleAfterCrash && report.maintenanceProof.repositoryInterruption.pending.incompleteEntries === 0 && report.maintenanceProof.repositoryInterruption.pending.retryCreated && report.maintenanceProof.repositoryInterruption.afterCommit.crashSnapshotVerifiedAfterCrash && report.maintenanceProof.repositoryInterruption.afterCommit.retryCreated ? "verified" : "missing"}`, ); console.log( `SQLITE_RELIABILITY_WAL_SENTINEL=${report.transactionProof.committedWalSentinel ? "verified" : "missing"}`, diff --git a/scripts/lib/sqlite-reliability-contract.ts b/scripts/lib/sqlite-reliability-contract.ts index 27e8d21e80cc..b41b10447a9e 100644 --- a/scripts/lib/sqlite-reliability-contract.ts +++ b/scripts/lib/sqlite-reliability-contract.ts @@ -135,13 +135,13 @@ export type ReliabilityReport = { visibleSnapshotsAfterCrash: number; }; pending: { - crashSnapshotVerifiedAfterCrash: false; - crashSnapshotVisibleAfterCrash: false; + crashSnapshotVerifiedAfterCrash: true; + crashSnapshotVisibleAfterCrash: true; exit: { code: number | null; signal: NodeJS.Signals | null; }; - incompleteEntries: 1; + incompleteEntries: 0; payload: { bytes: number; idSum: number; diff --git a/scripts/lib/sqlite-reliability-repository.ts b/scripts/lib/sqlite-reliability-repository.ts index 7689e60ccaa7..7aaa827d577c 100644 --- a/scripts/lib/sqlite-reliability-repository.ts +++ b/scripts/lib/sqlite-reliability-repository.ts @@ -235,7 +235,7 @@ async function runCrashPoint(params: { const crashSnapshots = visibleAfter.filter( (snapshot) => !visiblePathsBefore.has(path.resolve(snapshot.ref.path)), ); - const expectedVisibleCrashSnapshots = params.crashPoint === "after-commit" ? 1 : 0; + const expectedVisibleCrashSnapshots = params.crashPoint === "before-pending" ? 0 : 1; if (crashSnapshots.length !== expectedVisibleCrashSnapshots) { throw new Error( `SQLite repository exposed ${crashSnapshots.length} snapshot(s) at ${params.crashPoint}; expected ${expectedVisibleCrashSnapshots}.`, @@ -259,7 +259,7 @@ async function runCrashPoint(params: { if (stagingEntries === 0) { throw new Error(`SQLite repository worker left no staging at ${params.crashPoint}.`); } - const expectedIncompleteEntries = params.crashPoint === "after-commit" ? 0 : 1; + const expectedIncompleteEntries = params.crashPoint === "before-pending" ? 1 : 0; if (incompleteEntries !== expectedIncompleteEntries) { throw new Error( `SQLite repository left ${incompleteEntries} incomplete final entries at ${params.crashPoint}; expected ${expectedIncompleteEntries}.`, @@ -366,9 +366,9 @@ export async function runRepositoryInterruptionProof(params: { }, pending: { ...pending, - crashSnapshotVerifiedAfterCrash: false, - crashSnapshotVisibleAfterCrash: false, - incompleteEntries: 1, + crashSnapshotVerifiedAfterCrash: true, + crashSnapshotVisibleAfterCrash: true, + incompleteEntries: 0, repositoryVerified: true, retryCreated: true, sourcePayloadPreserved: true, diff --git a/src/snapshot/local-repository.test.ts b/src/snapshot/local-repository.test.ts index 10d4a2027ce0..65a6d2e687d2 100644 --- a/src/snapshot/local-repository.test.ts +++ b/src/snapshot/local-repository.test.ts @@ -20,6 +20,7 @@ import { } from "./snapshot-provider.js"; const durabilityTestState = vi.hoisted(() => ({ + beforePin: undefined as ((directoryPath: string) => void | Promise) | undefined, beforeSync: undefined as ((directoryPath: string) => void | Promise) | undefined, pinnedSyncOutcome: undefined as | { status: "synced" } @@ -32,6 +33,10 @@ vi.mock("@openclaw/fs-safe/durability", async (importOriginal) => { return { ...actual, pinDirectory: async (...args: Parameters) => { + const directory = args[0]; + await durabilityTestState.beforePin?.( + typeof directory === "string" ? path.resolve(directory) : directory.path, + ); const pinned = await actual.pinDirectory(...args); return { receipt: pinned.receipt, @@ -61,6 +66,7 @@ const DURABLE_PLUGIN_BLOB_MARKER = "durable-plugin-blob-control"; const STATE_LEASE_MARKER = "snapshot-must-not-retain-active-lease"; afterEach(() => { + durabilityTestState.beforePin = undefined; durabilityTestState.beforeSync = undefined; durabilityTestState.pinnedSyncOutcome = undefined; }); @@ -552,6 +558,177 @@ describe("local SQLite snapshot repository", () => { await expect(provider.list()).resolves.toEqual([second, first]); }); + it("recovers a complete snapshot left pending after a crash", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + createGenericDatabase(sourcePath); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "recover-complete-pending" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + await fs.writeFile(pendingPath, ""); + + await expect(provider.list()).resolves.toEqual([snapshot]); + await expect(fs.access(pendingPath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(provider.verify(snapshot.ref)).resolves.toMatchObject({ ok: true }); + }); + + it("recovers a complete pending snapshot through direct verify and restore", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + const restorePath = path.join(tempDir, "restore", "source.sqlite"); + createGenericDatabase(sourcePath, { values: ["durable"] }); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "direct-pending-recovery" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + + await fs.writeFile(pendingPath, ""); + await expect(provider.verify(snapshot.ref)).resolves.toEqual({ + ok: true, + manifest: snapshot.manifest, + }); + await expect(fs.access(pendingPath)).rejects.toMatchObject({ code: "ENOENT" }); + + await fs.writeFile(pendingPath, ""); + await expect(provider.restoreFresh(snapshot.ref, restorePath)).resolves.toEqual({ + ok: true, + manifest: snapshot.manifest, + }); + await expect(fs.access(pendingPath)).rejects.toMatchObject({ code: "ENOENT" }); + const sqlite = requireNodeSqlite(); + const restored = new sqlite.DatabaseSync(restorePath, { readOnly: true }); + try { + expect(restored.prepare("SELECT value FROM entries").all()).toEqual([{ value: "durable" }]); + } finally { + restored.close(); + } + }); + + it("allows concurrent callers to recover the same complete pending snapshot", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + createGenericDatabase(sourcePath); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "concurrent-pending-recovery" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + await fs.writeFile(pendingPath, ""); + + let syncArrivals = 0; + let releaseSyncs: (() => void) | undefined; + const syncBarrier = new Promise((resolve) => { + releaseSyncs = resolve; + }); + durabilityTestState.beforeSync = async (directoryPath) => { + if (directoryPath !== snapshot.ref.path || syncArrivals >= 2) { + return; + } + syncArrivals += 1; + if (syncArrivals === 2) { + releaseSyncs?.(); + } + await syncBarrier; + }; + + await expect(Promise.all([provider.list(), provider.list()])).resolves.toEqual([ + [snapshot], + [snapshot], + ]); + expect(syncArrivals).toBe(2); + await expect(fs.access(pendingPath)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("accepts a concurrent commit after classifying a complete pending snapshot", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + createGenericDatabase(sourcePath); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "concurrent-pending-commit" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + await fs.writeFile(pendingPath, ""); + durabilityTestState.beforePin = async (directoryPath) => { + if (directoryPath === snapshot.ref.path) { + durabilityTestState.beforePin = undefined; + await fs.unlink(pendingPath); + } + }; + + await expect(provider.list()).resolves.toEqual([snapshot]); + await expect(provider.verify(snapshot.ref)).resolves.toMatchObject({ ok: true }); + }); + + it("preserves the pending marker when recovery cannot guarantee directory durability", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + createGenericDatabase(sourcePath); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "unsupported-pending-recovery" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + await fs.writeFile(pendingPath, ""); + durabilityTestState.pinnedSyncOutcome = { status: "unsupported", code: "ENOTSUP" }; + + await expect(provider.list()).rejects.toThrow(/crash-durable directory synchronization/u); + await expect(fs.access(pendingPath)).resolves.toBeUndefined(); + + durabilityTestState.pinnedSyncOutcome = undefined; + await expect(provider.list()).resolves.toEqual([snapshot]); + }); + + it("rejects a hardlinked pending marker without committing the snapshot", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + createGenericDatabase(sourcePath); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "hardlinked-pending-recovery" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + const markerSourcePath = path.join(tempDir, "pending-marker"); + await fs.writeFile(markerSourcePath, ""); + await fs.link(markerSourcePath, pendingPath); + + await expect(provider.list()).rejects.toThrow(/pending marker is unsafe/u); + await expect(fs.access(pendingPath)).resolves.toBeUndefined(); + }); + + it("preserves a complete pending snapshot that fails recovery verification", async () => { + const tempDir = await createTempDir(); + const sourcePath = path.join(tempDir, "source.sqlite"); + const repositoryPath = path.join(tempDir, "snapshots"); + createGenericDatabase(sourcePath); + const provider = createLocalSqliteSnapshotProvider({ repositoryPath }); + const snapshot = await provider.create({ + path: sourcePath, + identity: { role: "generic", id: "reject-invalid-pending" }, + }); + const pendingPath = path.join(snapshot.ref.path, ".pending"); + await fs.writeFile(pendingPath, ""); + await fs.appendFile(path.join(snapshot.ref.path, SNAPSHOT_SQLITE_FILENAME), "corrupt"); + + await expect(provider.list()).rejects.toThrow(/size mismatch|hash mismatch/u); + await expect(fs.access(pendingPath)).resolves.toBeUndefined(); + }); + it.each([SNAPSHOT_SQLITE_FILENAME, SNAPSHOT_MANIFEST_FILENAME])( "rejects markerless partial snapshot directories containing only %s", async (entryName) => { diff --git a/src/snapshot/local-repository.ts b/src/snapshot/local-repository.ts index 1129f49958c9..b07c68bd5fd9 100644 --- a/src/snapshot/local-repository.ts +++ b/src/snapshot/local-repository.ts @@ -572,11 +572,20 @@ class LocalSqliteSnapshotProvider implements SqliteSnapshotProvider { ); } const snapshotPath = path.join(this.#repositoryPath, entry.name); - if (await isIncompleteSnapshotDirectory(snapshotPath)) { + const snapshotState = await classifySnapshotDirectory(snapshotPath); + if (snapshotState === "incomplete") { continue; } - await assertExactSnapshotContents(snapshotPath); - const manifest = await readSnapshotManifest(snapshotPath); + const manifest = + snapshotState === "complete-pending" + ? await recoverCompletePendingSnapshot({ + allowedDatabaseRoles: this.#allowedDatabaseRoles, + repositoryIdentity: repositoryStat, + repositoryPath: this.#repositoryPath, + snapshotPath, + validationRootPath: this.#validationRootPath, + }) + : await readVerifiedSnapshotManifest(snapshotPath); assertAllowedDatabaseRole(manifest, this.#allowedDatabaseRoles); snapshots.push({ ref: { path: snapshotPath }, @@ -601,6 +610,18 @@ class LocalSqliteSnapshotProvider implements SqliteSnapshotProvider { assertDirectory(repositoryStat, this.#repositoryPath, "SQLite snapshot repository"); const snapshotStat = await fs.lstat(snapshotDir); assertDirectory(snapshotStat, snapshotDir, "SQLite snapshot"); + if (await lstatIfExists(path.join(snapshotDir, SNAPSHOT_PENDING_FILENAME))) { + const snapshotState = await classifySnapshotDirectory(snapshotDir); + if (snapshotState === "complete-pending") { + await recoverCompletePendingSnapshot({ + allowedDatabaseRoles: this.#allowedDatabaseRoles, + repositoryIdentity: repositoryStat, + repositoryPath: this.#repositoryPath, + snapshotPath: snapshotDir, + validationRootPath: this.#validationRootPath, + }); + } + } return snapshotDir; } } @@ -1026,7 +1047,9 @@ async function assertSnapshotContents(snapshotDir: string, expected: Set } } -async function isIncompleteSnapshotDirectory(snapshotDir: string): Promise { +type SnapshotDirectoryState = "committed" | "complete-pending" | "incomplete"; + +async function classifySnapshotDirectory(snapshotDir: string): Promise { const entries = await fs.readdir(snapshotDir, { withFileTypes: true }); const knownEntries = new Set([ SNAPSHOT_MANIFEST_FILENAME, @@ -1041,7 +1064,94 @@ async function isIncompleteSnapshotDirectory(snapshotDir: string): Promise entry.name)); - return names.size === 0 || names.has(SNAPSHOT_PENDING_FILENAME); + if (names.size === 0) { + return "incomplete"; + } + if (!names.has(SNAPSHOT_PENDING_FILENAME)) { + return "committed"; + } + const complete = names.has(SNAPSHOT_MANIFEST_FILENAME) && names.has(SNAPSHOT_SQLITE_FILENAME); + return complete ? "complete-pending" : "incomplete"; +} + +async function recoverCompletePendingSnapshot(params: { + allowedDatabaseRoles: readonly SnapshotDatabaseIdentity["role"][] | undefined; + repositoryIdentity: Stats; + repositoryPath: string; + snapshotPath: string; + validationRootPath: string; +}): Promise { + const trustedRepositoryPath = await assertTrustedStagingRoot( + params.repositoryIdentity, + params.repositoryPath, + ); + await assertDirectoryIdentity(trustedRepositoryPath, params.repositoryIdentity); + const snapshotDirectory = await pinDirectory(params.snapshotPath, { + label: "SQLite pending snapshot directory", + }); + try { + const snapshotIdentity = snapshotDirectory.receipt.identity; + await assertPrivateStagingDirectory(snapshotIdentity, params.snapshotPath); + await snapshotDirectory.assertCurrent(); + const snapshotState = await classifySnapshotDirectory(params.snapshotPath); + if (snapshotState === "incomplete") { + throw new Error(`SQLite snapshot is incomplete: ${params.snapshotPath}`); + } + const manifest = await readSnapshotManifest(params.snapshotPath); + assertAllowedDatabaseRole(manifest, params.allowedDatabaseRoles); + const artifact = await hashSnapshotArtifact(params.snapshotPath); + const artifactPath = path.join(params.snapshotPath, SNAPSHOT_SQLITE_FILENAME); + assertArtifactMatchesManifest(artifactPath, artifact, manifest); + await verifySnapshotDatabaseFile( + artifactPath, + artifact.stat, + manifest, + params.validationRootPath, + ); + requireDirectorySync(await snapshotDirectory.sync(), "SQLite pending snapshot directory"); + + const pendingPath = path.join(params.snapshotPath, SNAPSHOT_PENDING_FILENAME); + const pendingIdentity = lstatIfExistsSync(pendingPath); + if (pendingIdentity) { + if ( + pendingIdentity.isSymbolicLink() || + !pendingIdentity.isFile() || + pendingIdentity.nlink > 1 + ) { + throw new Error(`SQLite snapshot pending marker is unsafe: ${pendingPath}`); + } + await snapshotDirectory.assertCurrent(); + const currentPendingIdentity = lstatIfExistsSync(pendingPath); + if (currentPendingIdentity) { + if (!sameFileIdentity(pendingIdentity, currentPendingIdentity)) { + throw new Error(`SQLite snapshot pending marker changed: ${pendingPath}`); + } + try { + fsSync.unlinkSync(pendingPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + throw error; + } + } + } + } + + // Both durable payload files already exist. Removing the exact marker and + // syncing this directory completes the interrupted repository commit. + // A concurrent recovery may win the unlink; syncing here still commits it. + requireDirectorySync(await snapshotDirectory.sync(), "SQLite pending snapshot directory"); + await snapshotDirectory.assertCurrent(); + const committedManifest = await readVerifiedSnapshotManifest(params.snapshotPath); + if (!isDeepStrictEqual(committedManifest, manifest)) { + throw new Error(`SQLite snapshot manifest changed during recovery: ${params.snapshotPath}`); + } + const committedArtifact = await hashSnapshotArtifact(params.snapshotPath); + assertArtifactMatchesManifest(artifactPath, committedArtifact, committedManifest); + await assertDirectoryIdentity(trustedRepositoryPath, params.repositoryIdentity); + return committedManifest; + } finally { + await snapshotDirectory.close().catch(() => undefined); + } } async function assertFreshRestorePathsAbsent(databasePath: string): Promise { @@ -1081,6 +1191,17 @@ async function lstatIfExists(pathname: string): Promise { } } +function lstatIfExistsSync(pathname: string): Stats | undefined { + try { + return fsSync.lstatSync(pathname); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return undefined; + } + throw error; + } +} + async function removePrivateDirectoryIfOwned( directoryPath: string, expectedIdentity: Stats, diff --git a/test/scripts/bench-sqlite-reliability.test.ts b/test/scripts/bench-sqlite-reliability.test.ts index 6930bb93eff9..2d4c559cdb8c 100644 --- a/test/scripts/bench-sqlite-reliability.test.ts +++ b/test/scripts/bench-sqlite-reliability.test.ts @@ -246,14 +246,14 @@ describe("scripts/bench-sqlite-reliability", () => { firstReport.maintenanceProof.repositoryInterruption.beforePending.exit.signal !== null, ).toBe(true); expect(firstReport.maintenanceProof.repositoryInterruption.pending).toMatchObject({ - crashSnapshotVerifiedAfterCrash: false, - crashSnapshotVisibleAfterCrash: false, - incompleteEntries: 1, + crashSnapshotVerifiedAfterCrash: true, + crashSnapshotVisibleAfterCrash: true, + incompleteEntries: 0, repositoryVerified: true, retryCreated: true, sourcePayloadPreserved: true, sourceStatePreserved: true, - visibleSnapshotsAfterCrash: 2, + visibleSnapshotsAfterCrash: 3, }); expect( firstReport.maintenanceProof.repositoryInterruption.pending.stagingEntries, @@ -270,7 +270,7 @@ describe("scripts/bench-sqlite-reliability", () => { retryCreated: true, sourcePayloadPreserved: true, sourceStatePreserved: true, - visibleSnapshotsAfterCrash: 4, + visibleSnapshotsAfterCrash: 5, }); expect( firstReport.maintenanceProof.repositoryInterruption.afterCommit.stagingEntries,