diff --git a/src/lab/fabric/observe.ts b/src/lab/fabric/observe.ts index 7e93d13fd..dc843bad4 100644 --- a/src/lab/fabric/observe.ts +++ b/src/lab/fabric/observe.ts @@ -12,7 +12,7 @@ import { fixtureDigest, isSha256Hex, jcsStringify } from "../digest"; import type { ObservationEvent, RouteSubjectV1, TaskSubjectV1 } from "../events/types"; import { LabValidationError } from "../events/errors"; import { assignEventId, validateSubject } from "../events/validate"; -import { appendLabEventIfAbsent } from "../ledger/store"; +import { withLedgerMutation } from "../ledger/store"; import { ensureLabDirs } from "../paths"; import { FABRIC_EVIDENCE_LAYER, @@ -430,9 +430,11 @@ function persistFabricOutcome( const ownsStore = !opts.artifactStore; const store = opts.artifactStore ?? createArtifactStore(paths.artifactsDir); try { - const { event } = observationFromFabricOutcome(outcome, { ...opts, artifactStore: store }); - appendLabEventIfAbsent(paths.ledgerPath, event); - return { event, ledgerPath: paths.ledgerPath }; + return withLedgerMutation(paths.ledgerPath, (ledger) => { + const { event } = observationFromFabricOutcome(outcome, { ...opts, artifactStore: store }); + ledger.appendIfAbsent(event); + return { event, ledgerPath: paths.ledgerPath }; + }); } finally { if (ownsStore) store.close(); } diff --git a/src/lab/ledger/purge.ts b/src/lab/ledger/purge.ts index dd6b95853..6c770d087 100644 --- a/src/lab/ledger/purge.ts +++ b/src/lab/ledger/purge.ts @@ -18,7 +18,7 @@ import { expandSensitiveArtifactEventTargets, } from "./artifact-refs"; import { buildInvalidationIndex } from "./invalidation"; -import { appendLabEvent, replayLabLedger } from "./store"; +import { withLedgerMutation } from "./store"; import { ensureLabDirs } from "../paths"; import { rebuildLabProjection } from "../projection/rebuild"; import { jcsStringify } from "../digest"; @@ -153,73 +153,86 @@ export function purgeSensitiveEvidence(req: SensitivePurgeRequest): PurgeTombsto const targetArtifactDigests = [...(req.targetArtifactDigests ?? [])].sort(); const purgeActions = [...(req.purgeActions ?? PURGE_ACTIONS)].sort(); const explicitSensitive = new Set(targetArtifactDigests); - - const replay = replayLabLedger(paths.ledgerPath); - const index = buildInvalidationIndex(replay.events); - const removeIds = expandSensitiveArtifactEventTargets( - replay.events, - index, - new Set(targetEventIds), - explicitSensitive, - ); - - const tombstonePayload = { - schemaVersion: LAB_EVENT_SCHEMA_VERSION, - eventKind: "purge_tombstone" as const, - recordedAt: req.recordedAt ?? Date.now(), - producer: LAB_PRODUCER, - producerVersion: req.producerVersion ?? LAB_PRODUCER_VERSION, - targetEventIds: [...removeIds].sort(), - targetArtifactDigests, - reason: "sensitive_evidence" as const, - purgeActions, - }; - const tombstone = validateLabEvent(assignEventId(tombstonePayload)) as PurgeTombstoneEvent; - - const deletionPlan = purgeActions.includes("artifact") - ? artifactDeletionPlan(replay.events, index, removeIds, targetArtifactDigests) - : { deletable: [], retainedExplicit: [] }; - - if (deletionPlan.retainedExplicit.length > 0) { - throw new PurgeError( - "sensitive_bytes_retained", - `explicit sensitive artifacts remain required: ${deletionPlan.retainedExplicit.join(",")}`, - ); - } - - let dir: TrustedArtifactDir | null = null; const completed: string[] = []; + try { - if (purgeActions.includes("scratch")) { - purgeBoundedDirectory(paths.scratchDir); - completed.push("scratch"); - } - if (purgeActions.includes("export")) { - purgeBoundedDirectory(paths.exportDir); - completed.push("export"); - } + const tombstone = withLedgerMutation(paths.ledgerPath, (ledger) => { + // Replay and plan under the same lock as every append. Otherwise an event + // appended after this snapshot can be lost by the atomic rename or can + // start referencing an artifact after the deletion plan was calculated. + const replay = ledger.replay(); + const index = buildInvalidationIndex(replay.events); + const removeIds = expandSensitiveArtifactEventTargets( + replay.events, + index, + new Set(targetEventIds), + explicitSensitive, + ); - if (purgeActions.includes("artifact")) { - if (deletionPlan.deletable.length > 0) { - dir = openTrustedArtifactDir(paths.artifactsDir); - deleteArtifactsFailClosed(dir, deletionPlan.deletable); + const tombstonePayload = { + schemaVersion: LAB_EVENT_SCHEMA_VERSION, + eventKind: "purge_tombstone" as const, + recordedAt: req.recordedAt ?? Date.now(), + producer: LAB_PRODUCER, + producerVersion: req.producerVersion ?? LAB_PRODUCER_VERSION, + targetEventIds: [...removeIds].sort(), + targetArtifactDigests, + reason: "sensitive_evidence" as const, + purgeActions, + }; + const tombstone = validateLabEvent(assignEventId(tombstonePayload)) as PurgeTombstoneEvent; + + const deletionPlan = purgeActions.includes("artifact") + ? artifactDeletionPlan(replay.events, index, removeIds, targetArtifactDigests) + : { deletable: [], retainedExplicit: [] }; + + if (deletionPlan.retainedExplicit.length > 0) { + throw new PurgeError( + "sensitive_bytes_retained", + `explicit sensitive artifacts remain required: ${deletionPlan.retainedExplicit.join(",")}`, + ); } - completed.push("artifact"); - } - if (purgeActions.includes("ledger")) { - const kept: LabEvent[] = []; - for (const event of replay.events) { - if (removeIds.has(event.eventId)) continue; - kept.push(event); + if (purgeActions.includes("scratch")) { + purgeBoundedDirectory(paths.scratchDir); + completed.push("scratch"); } - kept.push(tombstone); - atomicRewriteLedger(paths.ledgerPath, kept); - } else { - appendLabEvent(paths.ledgerPath, tombstone); - } - completed.push("ledger"); + if (purgeActions.includes("export")) { + purgeBoundedDirectory(paths.exportDir); + completed.push("export"); + } + + let dir: TrustedArtifactDir | null = null; + try { + if (purgeActions.includes("artifact")) { + if (deletionPlan.deletable.length > 0) { + dir = openTrustedArtifactDir(paths.artifactsDir); + deleteArtifactsFailClosed(dir, deletionPlan.deletable); + } + completed.push("artifact"); + } + + if (purgeActions.includes("ledger")) { + const kept: LabEvent[] = []; + for (const event of replay.events) { + if (removeIds.has(event.eventId)) continue; + kept.push(event); + } + kept.push(tombstone); + atomicRewriteLedger(paths.ledgerPath, kept); + } else { + ledger.append(tombstone); + } + completed.push("ledger"); + } finally { + if (dir) closeTrustedArtifactDir(dir); + } + + return tombstone; + }); + // SQLite is disposable and rebuildLabProjection replays the canonical + // ledger again, so it does not need to extend the mutation lock duration. if (purgeActions.includes("sqlite")) { rebuildLabProjection(req.configDir); completed.push("sqlite"); @@ -235,7 +248,5 @@ export function purgeSensitiveEvidence(req: SensitivePurgeRequest): PurgeTombsto err instanceof Error ? err.message : String(err), completed, ); - } finally { - if (dir) closeTrustedArtifactDir(dir); } } diff --git a/src/lab/ledger/store.ts b/src/lab/ledger/store.ts index 14baf2c94..cd1d37fc4 100644 --- a/src/lab/ledger/store.ts +++ b/src/lab/ledger/store.ts @@ -25,6 +25,12 @@ export interface LedgerStore { replay(): ReplayResult; } +export interface LedgerMutationContext { + replay(): ReplayResult; + append(event: LabEvent): void; + appendIfAbsent(event: LabEvent): boolean; +} + const LEDGER_LOCK_STALE_MS = 60_000; const LEDGER_LOCK_WAIT_MS = 5_000; @@ -36,10 +42,7 @@ interface LedgerLockMeta { /** Block synchronously for the given duration (ledger lock retry only). */ function sleepSyncMs(ms: number): void { - const end = Date.now() + ms; - while (Date.now() < end) { - /* spin */ - } + Bun.sleepSync(ms); } /** Read pid, createdAt, and token metadata from a ledger lock file, if well-formed. */ @@ -83,8 +86,94 @@ function isLedgerLockStale(lockPath: string): boolean { return false; } } - if (isLockHolderAlive(meta.pid)) return false; - return Date.now() - meta.createdAt > LEDGER_LOCK_STALE_MS; + return !isLockHolderAlive(meta.pid); +} + +/** Write lock ownership metadata to a newly created exclusive lock file. */ +function writeLedgerLockMeta(fd: number, token: string): void { + const metadataBytes = Buffer.from(JSON.stringify({ + pid: process.pid, + createdAt: Date.now(), + token, + }), "utf8"); + let written = 0; + while (written < metadataBytes.byteLength) { + const n = writeSync(fd, metadataBytes, written, metadataBytes.byteLength - written); + if (n <= 0) { + throw new LabValidationError("short_write", "ledger lock metadata write incomplete"); + } + written += n; + } +} + +/** Discard a lock whose exclusive creator failed before publishing ownership metadata. */ +function discardUninitialisedLedgerLock(lockPath: string, lockFd: number): void { + try { + closeSync(lockFd); + } catch { + /* ignore */ + } + try { + unlinkSync(lockPath); + } catch { + /* best-effort */ + } +} + +/** Release a lock file only when the token still matches the path owner. */ +function releaseLedgerLock(lockPath: string, lockFd: number, token: string): void { + try { + closeSync(lockFd); + } catch { + /* ignore */ + } + try { + const meta = readLedgerLockMeta(lockPath); + if (meta?.token === token) unlinkSync(lockPath); + } catch { + /* best-effort */ + } +} + +/** + * Recover one stale lock while holding a separate recovery mutex. + * + * The recovery mutex prevents two waiters from both observing the same stale + * owner and then unlinking each other's replacement lock. If a process dies + * while holding the recovery mutex, acquisition fails closed instead of + * guessing ownership of that mutex. + */ +function recoverStaleLedgerLock(lockPath: string): boolean { + const recoveryPath = `${lockPath}.recovery`; + const token = randomBytes(16).toString("hex"); + let recoveryFd: number; + try { + recoveryFd = openSync( + recoveryPath, + fsConstants.O_CREAT | fsConstants.O_EXCL | fsConstants.O_WRONLY, + 0o600, + ); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "EEXIST") return false; + throw error; + } + + try { + writeLedgerLockMeta(recoveryFd, token); + } catch (error) { + discardUninitialisedLedgerLock(recoveryPath, recoveryFd); + throw error; + } + + try { + // Re-check after taking the recovery mutex. Another waiter may already + // have recovered the old lock and installed a live replacement. + if (!existsSync(lockPath) || !isLedgerLockStale(lockPath)) return false; + unlinkSync(lockPath); + return true; + } finally { + releaseLedgerLock(recoveryPath, recoveryFd, token); + } } /** Create a ledger lock file exclusively, recovering stale locks when needed. */ @@ -97,30 +186,19 @@ function tryAcquireLedgerLock(lockPath: string, deadline: number): { fd: number; } catch (error) { if (existsSync(lockPath) && isLedgerLockStale(lockPath)) { try { - unlinkSync(lockPath); - } catch (unlinkError) { - if (Date.now() >= deadline) throw unlinkError; - sleepSyncMs(10); + if (recoverStaleLedgerLock(lockPath)) continue; + } catch (recoveryError) { + if (Date.now() >= deadline) throw recoveryError; } - continue; } if (Date.now() >= deadline) throw error; sleepSyncMs(10); continue; } try { - const metadataBytes = Buffer.from(JSON.stringify({ pid: process.pid, createdAt: Date.now(), token }), "utf8"); - const written = writeSync(fd, metadataBytes); - if (written !== metadataBytes.byteLength) { - throw new LabValidationError("short_write", "ledger lock metadata write incomplete"); - } + writeLedgerLockMeta(fd, token); } catch (error) { - closeSync(fd); - try { - unlinkSync(lockPath); - } catch { - /* best-effort */ - } + discardUninitialisedLedgerLock(lockPath, fd); throw error; } return { fd, token }; @@ -128,21 +206,6 @@ function tryAcquireLedgerLock(lockPath: string, deadline: number): { fd: number; throw new Error("ledger lock acquisition timed out"); } -/** Release a ledger lock only when the token still matches the lock file. */ -function releaseLedgerLock(lockPath: string, lockFd: number, token: string): void { - try { - closeSync(lockFd); - } catch { - /* ignore */ - } - try { - const meta = readLedgerLockMeta(lockPath); - if (meta?.token === token) unlinkSync(lockPath); - } catch { - /* best-effort */ - } -} - /** Run a ledger mutation while holding the compatibility ledger lock file. */ function withLedgerLock(ledgerPath: string, fn: () => T): T { const lockPath = `${ledgerPath}.lock`; @@ -156,11 +219,10 @@ function withLedgerLock(ledgerPath: string, fn: () => T): T { } } -/** Durable append of one validated event as a single JSONL line + fsync. */ -export function appendLabEvent(ledgerPath: string, event: LabEvent): void { - const validated = validateLabEvent(event); +/** Durable append of one already-validated event as a single JSONL line + fsync. */ +function appendValidatedLabEvent(ledgerPath: string, event: LabEvent): void { mkdirSync(dirname(ledgerPath), { recursive: true, mode: 0o700 }); - const line = `${jcsStringify(validated)}\n`; + const line = `${jcsStringify(event)}\n`; const bytes = new TextEncoder().encode(line); const fd = openSync(ledgerPath, "a", 0o600); try { @@ -178,26 +240,78 @@ export function appendLabEvent(ledgerPath: string, event: LabEvent): void { } } +function isThenable(value: unknown): value is PromiseLike { + return ( + (typeof value === "object" && value !== null) || typeof value === "function" + ) && typeof (value as { then?: unknown }).then === "function"; +} + /** - * Append only when eventId is absent. Uses an exclusive lock file plus a - * process-local event-id index refreshed under the lock. + * Serialize a ledger read-modify-write transaction with all ordinary appends. + * The callback is intentionally synchronous. Mutation methods become invalid + * as soon as the callback returns, so an accidental async continuation cannot + * write after the lock has been released. */ -export function appendLabEventIfAbsent(ledgerPath: string, event: LabEvent): boolean { - const validated = validateLabEvent(event); +export function withLedgerMutation( + ledgerPath: string, + fn: (mutation: LedgerMutationContext) => T, +): T { return withLedgerLock(ledgerPath, () => { - // Refresh from disk under the lock so concurrent writers are visible. - const fresh = new Set(); - if (existsSync(ledgerPath)) { - for (const row of replayLabLedger(ledgerPath).events) { - fresh.add(row.eventId); + let active = true; + const requireActive = () => { + if (!active) { + throw new LabValidationError( + "inactive_ledger_mutation", + "ledger mutation context used after its lock was released", + ); } + }; + const replay = () => { + requireActive(); + return replayLabLedger(ledgerPath); + }; + const append = (event: LabEvent) => { + requireActive(); + appendValidatedLabEvent(ledgerPath, validateLabEvent(event)); + }; + const appendIfAbsent = (event: LabEvent): boolean => { + requireActive(); + const validated = validateLabEvent(event); + if (replay().events.some((row) => row.eventId === validated.eventId)) return false; + appendValidatedLabEvent(ledgerPath, validated); + return true; + }; + + try { + const result = fn({ replay, append, appendIfAbsent }); + if (isThenable(result)) { + throw new LabValidationError( + "async_ledger_mutation", + "ledger mutation callback must be synchronous", + ); + } + return result; + } finally { + active = false; } - if (fresh.has(validated.eventId)) return false; - appendLabEvent(ledgerPath, validated); - return true; }); } +/** Durable append of one validated event as a single JSONL line + fsync. */ +export function appendLabEvent(ledgerPath: string, event: LabEvent): void { + withLedgerMutation(ledgerPath, (mutation) => { + mutation.append(event); + }); +} + +/** + * Append only when eventId is absent. Uses the same mutation lock as every + * other ledger writer so the presence check and append are one transaction. + */ +export function appendLabEventIfAbsent(ledgerPath: string, event: LabEvent): boolean { + return withLedgerMutation(ledgerPath, (mutation) => mutation.appendIfAbsent(event)); +} + function processLine( line: string, lineNumber: number, @@ -415,4 +529,4 @@ export function openLedgerStore(configDir?: string): LedgerStore { export function defaultLedgerPath(configDir?: string): string { return labLedgerPath(configDir); -} +} \ No newline at end of file diff --git a/src/lab/observe/from-conformance.ts b/src/lab/observe/from-conformance.ts index ba353f305..486391e04 100644 --- a/src/lab/observe/from-conformance.ts +++ b/src/lab/observe/from-conformance.ts @@ -17,7 +17,7 @@ import { } from "../digest"; import type { ObservationEvent } from "../events/types"; import { assignEventId } from "../events/validate"; -import { appendLabEvent } from "../ledger/store"; +import { withLedgerMutation } from "../ledger/store"; import { ensureLabDirs } from "../paths"; import type { CaseAuthority, @@ -285,12 +285,14 @@ export function persistConformanceResult( const ownsStore = !opts.artifactStore; const store = opts.artifactStore ?? createArtifactStore(paths.artifactsDir); try { - const { event } = observationFromConformanceResult(result, caseRecord, authority, { - ...opts, - artifactStore: store, + return withLedgerMutation(paths.ledgerPath, (ledger) => { + const { event } = observationFromConformanceResult(result, caseRecord, authority, { + ...opts, + artifactStore: store, + }); + ledger.append(event); + return { event, ledgerPath: paths.ledgerPath }; }); - appendLabEvent(paths.ledgerPath, event); - return { event, ledgerPath: paths.ledgerPath }; } finally { if (ownsStore) store.close(); } diff --git a/src/lab/observe/from-live.ts b/src/lab/observe/from-live.ts index 857874ed4..08145d927 100644 --- a/src/lab/observe/from-live.ts +++ b/src/lab/observe/from-live.ts @@ -5,7 +5,7 @@ import { LAB_EVENT_SCHEMA_VERSION, LAB_PRODUCER, LAB_PRODUCER_VERSION, OBSERVATI import { fixtureDigest, scenarioManifestDigest, subjectIdForSubject, suiteManifestDigest } from "../digest"; import type { FailureRecordV1, ObservationEvent } from "../events/types"; import { assignEventId } from "../events/validate"; -import { appendLabEvent } from "../ledger/store"; +import { withLedgerMutation } from "../ledger/store"; import { ensureLabDirs } from "../paths"; import type { CaseAuthority, CaseRecord } from "../conformance/types"; import { trustedLiveResultRetryable } from "../live/executor"; @@ -106,6 +106,12 @@ export function observationFromLiveResult(result: LiveScenarioRunResult, caseRec export function persistLiveResult(result: LiveScenarioRunResult, caseRecord: CaseRecord, authority: CaseAuthority, opts: PersistLiveOptions = {}): PersistedLiveObservation { const paths = ensureLabDirs(opts.configDir); const ownsStore = !opts.artifactStore; const store = opts.artifactStore ?? createArtifactStore(paths.artifactsDir); - try { const { event } = observationFromLiveResult(result, caseRecord, authority, { ...opts, artifactStore: store }); appendLabEvent(paths.ledgerPath, event); return { event, ledgerPath: paths.ledgerPath }; } + try { + return withLedgerMutation(paths.ledgerPath, (ledger) => { + const { event } = observationFromLiveResult(result, caseRecord, authority, { ...opts, artifactStore: store }); + ledger.append(event); + return { event, ledgerPath: paths.ledgerPath }; + }); + } finally { if (ownsStore) store.close(); } } diff --git a/tests/lab-ledger-mutation-lock.test.ts b/tests/lab-ledger-mutation-lock.test.ts new file mode 100644 index 000000000..d659ca4d7 --- /dev/null +++ b/tests/lab-ledger-mutation-lock.test.ts @@ -0,0 +1,263 @@ +import { afterEach, expect, test } from "bun:test"; +import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { + appendLabEvent, + appendLabEventIfAbsent, + assignEventId, + createArtifactStore, + LAB_EVENT_SCHEMA_VERSION, + LAB_PRODUCER, + LAB_PRODUCER_VERSION, + persistConformanceResult, + purgeSensitiveEvidence, + replayLabLedger, + withLedgerMutation, +} from "../src/lab"; +import type { ArtifactStore } from "../src/lab/artifacts/store"; +import { discoverScenarios, loadCaseAuthority } from "../src/lab/conformance/manifest"; +import { resolveProtocolExecutionContext } from "../src/lab/conformance/executor"; +import type { CaseRecord } from "../src/lab/conformance/types"; +import type { InvalidationEvent } from "../src/lab/events/types"; + +const HOMES: string[] = []; + +function tempHome(): string { + const dir = join(tmpdir(), `ocx-lab-ledger-lock-${process.pid}-${Math.random().toString(16).slice(2)}`); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + HOMES.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of HOMES.splice(0)) { + try { + rmSync(dir, { recursive: true, force: true }); + } catch { + /* ignore */ + } + } +}); + +function hash(value: string): string { + return Bun.CryptoHasher.hash("sha256", value, "hex"); +} + +function invalidation(seed: string, recordedAt = 1_700_000_000_000): InvalidationEvent { + return assignEventId({ + schemaVersion: LAB_EVENT_SCHEMA_VERSION, + eventKind: "invalidation" as const, + recordedAt, + producer: LAB_PRODUCER, + producerVersion: LAB_PRODUCER_VERSION, + targetEventIds: [hash(`target:${seed}`)], + reason: "manual_correction" as const, + }) as InvalidationEvent; +} + +function syntheticPassResult(caseRecord: CaseRecord) { + return { + scenarioId: caseRecord.id, + suite: caseRecord.suite, + passed: true, + classification: "inconclusive" as const, + assertionResults: caseRecord.assertions.map((assertion) => ({ + id: assertion.id, + operator: assertion.operator, + required: assertion.required, + passed: true, + observedSummary: "ok", + })), + diagnostics: [], + executionContext: resolveProtocolExecutionContext(caseRecord), + startedAt: 999, + completedAt: 1000, + }; +} + +async function waitForPath(path: string): Promise { + for (let attempt = 0; attempt < 500; attempt += 1) { + if (existsSync(path)) return; + await Bun.sleep(10); + } + throw new Error(`timed out waiting for child marker ${path}`); +} + +async function waitForChild(child: ReturnType): Promise { + const result = await Promise.race([ + child.exited.then((exitCode) => ({ exitCode })), + Bun.sleep(5_000).then(() => null), + ]); + if (!result) { + child.kill(); + await child.exited; + throw new Error("timed out waiting for ledger-lock child"); + } + if (result.exitCode !== 0) { + const stderr = await new Response(child.stderr).text().catch(() => ""); + throw new Error(`ledger-lock child exited ${result.exitCode}: ${stderr}`); + } +} + +function spawnLiveLock( + ledgerPath: string, + readyPath: string, + releaseMarkerPath: string, +): ReturnType { + const lockPath = `${ledgerPath}.lock`; + const childSource = ` + import { mkdirSync, unlinkSync, writeFileSync } from "node:fs"; + import { dirname } from "node:path"; + mkdirSync(dirname(${JSON.stringify(lockPath)}), { recursive: true, mode: 0o700 }); + writeFileSync( + ${JSON.stringify(lockPath)}, + JSON.stringify({ pid: process.pid, createdAt: Date.now(), token: "live-holder" }), + { mode: 0o600 }, + ); + writeFileSync(${JSON.stringify(readyPath)}, "ready"); + Bun.sleepSync(250); + writeFileSync(${JSON.stringify(releaseMarkerPath)}, "releasing"); + unlinkSync(${JSON.stringify(lockPath)}); + `; + return Bun.spawn([process.execPath, "-e", childSource], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); +} + +test("appendLabEvent waits for the shared ledger mutation lock", async () => { + const home = tempHome(); + const ledgerPath = join(home, "lab", "compatibility.jsonl"); + const readyPath = join(home, "holder-ready"); + const releaseMarkerPath = join(home, "holder-releasing"); + const child = spawnLiveLock(ledgerPath, readyPath, releaseMarkerPath); + + try { + await waitForPath(readyPath); + const event = invalidation("append-lock"); + appendLabEvent(ledgerPath, event); + expect(existsSync(releaseMarkerPath)).toBe(true); + expect(replayLabLedger(ledgerPath).events.some((row) => row.eventId === event.eventId)).toBe(true); + } finally { + await waitForChild(child); + } +}); + +test("appendLabEventIfAbsent immediately recovers a lock owned by an exited process", async () => { + const home = tempHome(); + const ledgerPath = join(home, "lab", "compatibility.jsonl"); + const lockPath = `${ledgerPath}.lock`; + const readyPath = join(home, "dead-lock-written"); + mkdirSync(dirname(ledgerPath), { recursive: true, mode: 0o700 }); + + const childSource = ` + import { writeFileSync } from "node:fs"; + writeFileSync( + ${JSON.stringify(lockPath)}, + JSON.stringify({ pid: process.pid, createdAt: Date.now(), token: "dead-holder" }), + { mode: 0o600 }, + ); + writeFileSync(${JSON.stringify(readyPath)}, "ready"); + `; + const child = Bun.spawn([process.execPath, "-e", childSource], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }); + await waitForPath(readyPath); + await waitForChild(child); + + const event = invalidation("dead-lock"); + expect(appendLabEventIfAbsent(ledgerPath, event)).toBe(true); + expect(existsSync(lockPath)).toBe(false); + expect(existsSync(`${lockPath}.recovery`)).toBe(false); + expect(replayLabLedger(ledgerPath).events.some((row) => row.eventId === event.eventId)).toBe(true); +}); + +test("withLedgerMutation rejects async callbacks and invalidates their context", async () => { + const home = tempHome(); + const ledgerPath = join(home, "lab", "compatibility.jsonl"); + const event = invalidation("async-callback"); + let continuation: Promise | undefined; + let continuationError: unknown; + + expect(() => withLedgerMutation(ledgerPath, (mutation) => { + continuation = (async () => { + await Bun.sleep(1); + mutation.append(event); + })().catch((error) => { + continuationError = error; + }); + return continuation; + })).toThrow("ledger mutation callback must be synchronous"); + + await continuation; + expect(continuationError).toBeInstanceOf(Error); + expect((continuationError as Error).message).toContain("after its lock was released"); + expect(replayLabLedger(ledgerPath).events.some((row) => row.eventId === event.eventId)).toBe(false); +}); + +test("canonical persistence publishes artifacts while holding the ledger mutation lock", () => { + const home = tempHome(); + const ledgerPath = join(home, "lab", "compatibility.jsonl"); + const authority = loadCaseAuthority(); + const caseRecord = discoverScenarios(authority, ["responses-core"]).find( + (candidate) => candidate.id === "responses-core.protocol.request-shape", + )!; + const realStore = createArtifactStore(join(home, "lab", "artifacts")); + const guardedStore: ArtifactStore = { + ...realStore, + put(input) { + expect(existsSync(`${ledgerPath}.lock`)).toBe(true); + return realStore.put(input); + }, + }; + + try { + const { event } = persistConformanceResult( + syntheticPassResult(caseRecord), + caseRecord, + authority, + { configDir: home, recordedAt: 1000, artifactStore: guardedStore }, + ); + expect(replayLabLedger(ledgerPath).events.some((row) => row.eventId === event.eventId)).toBe(true); + expect(existsSync(`${ledgerPath}.lock`)).toBe(false); + } finally { + realStore.close(); + } +}); + +test("sensitive purge waits for the ledger mutation lock before rewriting", async () => { + const home = tempHome(); + const ledgerPath = join(home, "lab", "compatibility.jsonl"); + const event = invalidation("purge-lock"); + appendLabEvent(ledgerPath, event); + + const readyPath = join(home, "purge-holder-ready"); + const releaseMarkerPath = join(home, "purge-holder-releasing"); + const child = spawnLiveLock(ledgerPath, readyPath, releaseMarkerPath); + + try { + await waitForPath(readyPath); + purgeSensitiveEvidence({ + configDir: home, + targetEventIds: [event.eventId], + targetArtifactDigests: [], + purgeActions: ["ledger"], + recordedAt: 1_700_000_000_100, + }); + expect(existsSync(releaseMarkerPath)).toBe(true); + const replay = replayLabLedger(ledgerPath); + expect(replay.events.some((row) => row.eventId === event.eventId)).toBe(false); + expect(replay.events.some((row) => row.eventKind === "purge_tombstone")).toBe(true); + expect(existsSync(`${ledgerPath}.lock`)).toBe(false); + } finally { + await waitForChild(child); + } +});