From e4228d9a50fbd515ad8cff2bf92ba3fcf5477312 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 16:45:36 +0200 Subject: [PATCH 1/7] =?UTF-8?q?=F0=9F=A4=96=20tests:=20await=20token-budge?= =?UTF-8?q?t=20warning=20visibility?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wait for the warning's visible state after opening it so the assertion can tolerate the app entrance transition without accepting persistent invisibility. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ieeb9c6b96af089dab27156d43a7f54fcb9b38dbe --- src/browser/stories/App.tokenBudget.stories.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/browser/stories/App.tokenBudget.stories.tsx b/src/browser/stories/App.tokenBudget.stories.tsx index 283ae0e4466..288b43496b0 100644 --- a/src/browser/stories/App.tokenBudget.stories.tsx +++ b/src/browser/stories/App.tokenBudget.stories.tsx @@ -133,7 +133,7 @@ export const Rollover: AppStory = { await expect(canvas.queryByText(WARNING)).not.toBeInTheDocument(); const warning = await canvas.findByRole("button", { name: /Context budget warning/ }); await userEvent.click(warning); - await expect(canvas.getByText(WARNING)).toBeVisible(); + await waitFor(() => expect(canvas.getByText(WARNING)).toBeVisible()); await userEvent.click(warning); const tool = await canvas.findByText("session_history", { exact: true }); await userEvent.click(tool); From 086e60d84e9ff4d96fef5d96e5a0acf96e21a829 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 14:39:44 +0200 Subject: [PATCH 2/7] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20define=20inactiv?= =?UTF-8?q?e=20compaction=20cancellation=20state?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep cancellation nonce and retry ownership across failed persistence. Model witnessed retirement and deletion debt without activating runtime behavior; a later adapter supplies shared-lock storage and acceptance witnesses. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ic12b78c45520044eba01a6857ada8179c91a62d7 --- .../services/compactionCancellation.test.ts | 483 ++++++++++++++++++ src/node/services/compactionCancellation.ts | 283 ++++++++++ 2 files changed, 766 insertions(+) create mode 100644 src/node/services/compactionCancellation.test.ts create mode 100644 src/node/services/compactionCancellation.ts diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts new file mode 100644 index 00000000000..0e2577e4afb --- /dev/null +++ b/src/node/services/compactionCancellation.test.ts @@ -0,0 +1,483 @@ +import { describe, expect, it, mock } from "bun:test"; +import assert from "node:assert/strict"; +import { + CompactionCancellation, + MalformedCompactionCancellationError, + matchesCompactionCancellation, + type CompactionCancellationMutation, + type CompactionCancellationMutationOutcome, + type CompactionCancellationPublication, + type CompactionCancellationRecord, + type CompactionCancellationStorage, + type CompactionCancellationSummary, +} from "./compactionCancellation"; + +const summary: CompactionCancellationSummary = { + id: "summary-a", + sequence: 3, + pendingFollowUp: { text: "Continue", options: { model: "test" } }, +}; + +function cancellation(nonce: string): CompactionCancellationRecord { + return { version: 1, nonce, scope: { kind: "unresolved" } }; +} + +function harness() { + const shared: { record: CompactionCancellationRecord | null } = { record: null }; + // Scripted adapter outcomes exercise the core's response to committed, failed and + // superseded writes. This is not a simulation/proof of filesystem CAS or history repair. + const storage = { + read: mock(() => Promise.resolve(structuredClone(shared.record))), + mutate: mock( + ( + mutation: CompactionCancellationMutation, + isCurrent: () => boolean + ): Promise => { + if (!isCurrent()) return Promise.resolve("superseded"); + shared.record = mutation.kind === "retire" ? null : structuredClone(mutation.record); + return Promise.resolve("applied"); + } + ), + repair: mock( + ( + _isCurrent: () => boolean, + _onCommitted: () => void + ): Promise => { + return Promise.reject(new Error("Unexpected repair")); + } + ), + } satisfies CompactionCancellationStorage; + return { shared, storage, state: new CompactionCancellation(storage) }; +} + +describe("inactive cancellation state core", () => { + it("retains a failed publication's nonce and advanced frontier through an exact retry", async () => { + const { state, storage, shared } = harness(); + const apply = storage.mutate.getMockImplementation()!; + const frontier = { nonce: null, generation: "advanced-a" }; + let publication: CompactionCancellationPublication | undefined; + storage.mutate.mockImplementationOnce((mutation) => { + assert(mutation.kind === "publish"); + publication = mutation.publication; + publication.predecessor = frontier; + return Promise.reject(new Error("sidecar publication failed after advancement")); + }); + await assert.rejects(state.cancel(), /sidecar publication failed/); + const first = await state.read(); + expect(state.blocksRecovery).toBe(true); + expect(storage.read).not.toHaveBeenCalled(); + await assert.rejects(state.narrow(first!.nonce, summary), /sidecar publication failed/); + storage.mutate.mockImplementationOnce(async (mutation, current) => { + assert(mutation.kind === "publish"); + expect(mutation.record.nonce).toBe(first!.nonce); + expect(mutation.publication).toBe(publication!); + expect(mutation.publication.predecessor).toBe(frontier); + expect(mutation.publication.attempts).toBe(2); + return apply(mutation, current); + }); + expect(await state.retry()).toBe("applied"); + expect(shared.record?.nonce).toBe(first!.nonce); + expect(state.needsPersistence).toBe(false); + }); + + it("refreshes a foreign successor after an adapter rejects the retry frontier", async () => { + const { state, storage, shared } = harness(); + storage.mutate.mockRejectedValueOnce(new Error("failed Stop")); + await assert.rejects(state.cancel(), /failed Stop/); + shared.record = cancellation("foreign-b"); + storage.mutate.mockResolvedValueOnce("superseded"); + expect(await state.retry()).toBe("superseded"); + expect(state.needsPersistence).toBe(false); + expect(await state.readForReplacement()).toEqual(shared.record); + expect(storage.mutate).toHaveBeenCalledTimes(2); + }); + + it("queued local Stops cannot publish or acknowledge an obsolete nonce", async () => { + const { state, storage } = harness(); + const first = state.cancel(); + const firstRecord = state.read(); + const second = state.cancel(); + const secondRecord = state.read(); + expect((await firstRecord)?.nonce).not.toBe((await secondRecord)?.nonce); + expect(await first).toBe("superseded"); + expect(await second).toBe("applied"); + expect(await state.read()).toEqual(await secondRecord); + expect(storage.mutate).toHaveBeenCalledTimes(1); + expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({ + kind: "publish", + record: await secondRecord, + }); + }); + + it("a Stop admitted during an old write invalidates that write's final guard", async () => { + const { state, storage } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const apply = storage.mutate.getMockImplementation()!; + storage.mutate.mockImplementationOnce(async (mutation, current) => { + entered.resolve(); + await release.promise; + expect(current()).toBe(false); + return apply(mutation, current); + }); + const first = state.cancel(); + await entered.promise; + const second = state.cancel(); + const expected = await state.read(); + release.resolve(); + expect(await first).toBe("superseded"); + expect(await second).toBe("applied"); + expect(await state.read()).toEqual(expected); + }); + + it("failed retirement retries deletion while maintaining conservative exclusion", async () => { + const { state, storage } = harness(); + await state.cancel(); + const record = (await state.read())!; + storage.mutate.mockRejectedValueOnce(new Error("unlink failed")); + await assert.rejects(state.retire(record.nonce), /unlink failed/); + expect(state.blocksRecovery).toBe(true); + expect(await state.read()).toEqual(record); + await assert.rejects(state.flush(), /unlink failed/); + expect(await state.retry()).toBe("applied"); + expect(storage.mutate.mock.calls.at(-1)?.[0]).toEqual({ kind: "retire", nonce: record.nonce }); + expect(await state.read()).toBeNull(); + }); + + it("late acknowledgment of a committed Stop cannot settle its pending successor", async () => { + const { state, storage } = harness(); + const committed = Promise.withResolvers(); + const acknowledge = Promise.withResolvers(); + const successorEntered = Promise.withResolvers(); + const releaseSuccessor = Promise.withResolvers(); + const apply = storage.mutate.getMockImplementation()!; + storage.mutate.mockImplementationOnce(async (mutation, current) => { + const outcome = await apply(mutation, current); + committed.resolve(); + await acknowledge.promise; + return outcome; + }); + storage.mutate.mockImplementationOnce(async (mutation, current) => { + successorEntered.resolve(); + await releaseSuccessor.promise; + return apply(mutation, current); + }); + const first = state.cancel(); + await committed.promise; + const second = state.cancel(); + const expected = await state.read(); + acknowledge.resolve(); + await successorEntered.promise; + expect(await first).toBe("applied"); + expect(state.blocksRecovery).toBe(true); + expect(await state.read()).toEqual(expected); + releaseSuccessor.resolve(); + expect(await second).toBe("applied"); + expect(state.needsPersistence).toBe(false); + }); + + it("failed narrowing retains unresolved exclusion until the exact retry commits", async () => { + const { state, storage } = harness(); + await state.cancel(); + const record = (await state.read())!; + storage.mutate.mockRejectedValueOnce(new Error("narrowing failed")); + await assert.rejects(state.narrow(record.nonce, summary), /narrowing failed/); + expect(await state.read()).toEqual(record); + expect(state.blocksRecovery).toBe(true); + await state.retry(); + expect((await state.read())?.scope).toEqual({ kind: "summary", ...summary }); + }); + + it.each([ + { witnessed: false, failed: false }, + { witnessed: true, failed: false }, + { witnessed: false, failed: true }, + { witnessed: true, failed: true }, + ])( + "narrowing cannot replace a concurrently queued retirement (witnessed=$witnessed, failed=$failed)", + async ({ witnessed, failed }) => { + const { state, storage, shared } = harness(); + await state.cancel(); + const record = (await state.read())!; + if (failed) storage.mutate.mockRejectedValueOnce(new Error("retirement failed")); + + // narrow yields at its publication join. Retirement must keep ownership when + // that continuation resumes, on both sides of the adapter's deletion result. + const narrowing = state.narrow(record.nonce, summary); + const retirement = witnessed + ? state.retireReplacement({ nonce: record.nonce }) + : state.retire(record.nonce); + const results = await Promise.allSettled([narrowing, retirement]); + expect(results[0]).toEqual({ status: "fulfilled", value: undefined }); + expect(results[1].status).toBe(failed ? "rejected" : "fulfilled"); + expect(storage.mutate.mock.calls.map(([mutation]) => mutation.kind)).toEqual([ + "publish", + "retire", + ]); + expect(state.needsPersistence).toBe(failed); + expect(state.blocksRecovery).toBe(failed && !witnessed); + expect(shared.record).toEqual(failed ? record : null); + + if (failed) { + expect(await state.retry()).toBe("applied"); + expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({ + kind: "retire", + nonce: record.nonce, + }); + } + expect(shared.record).toBeNull(); + expect(state.needsPersistence).toBe(false); + } + ); + + it("requires a matching replacement witness to retire retained cancellation", async () => { + const { state, storage } = harness(); + await state.cancel({ retainUntilReplacement: true }); + const record = (await state.read())!; + await state.narrow(record.nonce, summary); + await state.retire(record.nonce); + await state.retireReplacement({ nonce: "unrelated" }); + expect(storage.mutate).toHaveBeenCalledTimes(1); + expect(await state.read()).toEqual(record); + expect(await state.retireReplacement({ nonce: record.nonce })).toBe("applied"); + expect(await state.read()).toBeNull(); + }); + + it("a subsequent Stop carries a retained full-clear obligation", async () => { + const { state } = harness(); + await state.cancel({ retainUntilReplacement: true }); + const first = (await state.read())!; + await state.cancel(); + const second = (await state.read())!; + expect(second.nonce).not.toBe(first.nonce); + expect(second.retainUntilReplacement).toBe(true); + await state.retireReplacement({ nonce: first.nonce }); + expect(await state.read()).toEqual(second); + }); + + it("witnessed deletion debt allows fresh reads without adopting a foreign Stop for retry", async () => { + const { state, storage, shared } = harness(); + await state.cancel({ retainUntilReplacement: true }); + const record = (await state.read())!; + storage.mutate.mockRejectedValueOnce(new Error("witnessed unlink failed")); + await assert.rejects(state.retireReplacement({ nonce: record.nonce }), /unlink failed/); + expect(state.needsPersistence).toBe(true); + expect(state.blocksRecovery).toBe(false); + await state.flush(); + expect(await state.readForReplacement()).toBeNull(); + shared.record = cancellation("foreign-b"); + expect(await state.readForReplacement()).toEqual(shared.record); + expect(state.needsPersistence).toBe(true); + storage.mutate.mockResolvedValueOnce("superseded"); + await state.retry(); + expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({ + kind: "retire", + nonce: record.nonce, + }); + expect(await state.read()).toEqual(shared.record); + expect(state.needsPersistence).toBe(false); + }); + + it("a stale witness cannot make a newer failed Stop non-blocking", async () => { + const { state, storage } = harness(); + await state.cancel(); + const first = (await state.read())!; + await state.retireReplacement({ nonce: first.nonce }); + storage.mutate.mockRejectedValueOnce(new Error("new Stop failed")); + await assert.rejects(state.cancel(), /new Stop failed/); + const second = await state.read(); + await state.retireReplacement({ nonce: first.nonce }); + expect(state.blocksRecovery).toBe(true); + expect(await state.read()).toEqual(second); + await assert.rejects(state.flush(), /new Stop failed/); + }); + + it("ordinary cleanup cannot downgrade witnessed deletion debt", async () => { + const { state, storage } = harness(); + await state.cancel(); + const record = (await state.read())!; + storage.mutate.mockRejectedValue(new Error("unlink still failed")); + await assert.rejects(state.retireReplacement({ nonce: record.nonce }), /unlink/); + await assert.rejects(state.retire(record.nonce), /unlink/); + expect(state.needsPersistence).toBe(true); + expect(state.blocksRecovery).toBe(false); + await state.flush(); + expect(await state.readForReplacement()).toBeNull(); + }); + + it("readers joining an active failed publication report it without retrying", async () => { + const { state, storage } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.mutate.mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + throw new Error("active publication failed"); + }); + const stopping = state.cancel(); + await entered.promise; + const readers = [state.readForReplacement(), state.readForReplacement()]; + release.resolve(); + const results = await Promise.allSettled([stopping, ...readers]); + expect(results.map((result) => result.status)).toEqual(["rejected", "rejected", "rejected"]); + expect(storage.mutate).toHaveBeenCalledTimes(1); + expect(state.blocksRecovery).toBe(true); + }); + + it.each([false, true])( + "replacement readers share a retry and its outcome (failure=%s)", + async (failed) => { + const { state, storage } = harness(); + storage.mutate.mockRejectedValueOnce(new Error("initial failure")); + await assert.rejects(state.cancel(), /initial failure/); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const apply = storage.mutate.getMockImplementation()!; + storage.mutate.mockImplementationOnce(async (mutation, current) => { + entered.resolve(); + await release.promise; + if (failed) throw new Error("retry failure"); + return apply(mutation, current); + }); + const readers = [state.readForReplacement(), state.readForReplacement()]; + await entered.promise; + const retryA = state.retry(); + expect(state.retry()).toBe(retryA); + release.resolve(); + const results = await Promise.allSettled(readers); + expect(results.map((result) => result.status)).toEqual( + failed ? ["rejected", "rejected"] : ["fulfilled", "fulfilled"] + ); + expect(storage.mutate).toHaveBeenCalledTimes(2); + expect(state.needsPersistence).toBe(failed); + if (!failed) expect(await readers[0]).toEqual(await readers[1]); + } + ); + + it.each(["old record", "malformed", "I/O"] as const)( + "ignores an obsolete read after Stop (%s)", + async (result) => { + const { state, storage } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.read.mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + if (result === "malformed") throw new MalformedCompactionCancellationError(); + if (result === "I/O") throw new Error("read failed"); + return cancellation("obsolete"); + }); + const reading = state.read(); + await entered.promise; + await state.cancel(); + const expected = await state.read(); + release.resolve(); + expect(await reading).toEqual(expected); + expect(storage.repair).not.toHaveBeenCalled(); + } + ); + + it.each([false, true])( + "repairs malformed state with guarded commit evidence (superseded=%s)", + async (superseded) => { + const { state, storage, shared } = harness(); + storage.read.mockRejectedValueOnce(new MalformedCompactionCancellationError()); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.repair.mockImplementationOnce(async (current, committed) => { + entered.resolve(); + await release.promise; + if (!current()) return null; + committed(); + shared.record = null; + return null; + }); + const reading = state.read(); + await entered.promise; + if (superseded) await state.cancel(); + release.resolve(); + expect(await reading).toEqual(shared.record); + expect(state.repairRevision).toBe(superseded ? 0 : 1); + } + ); + + it("accepts a newer valid record returned by repair without claiming a history rewrite", async () => { + const { state, storage } = harness(); + storage.read.mockRejectedValueOnce(new MalformedCompactionCancellationError()); + storage.repair.mockResolvedValueOnce(cancellation("newer-valid")); + expect(await state.read()).toEqual(cancellation("newer-valid")); + expect(state.repairRevision).toBe(0); + }); + + it("ordinary read failure never repairs; explicit replacement publishes a retained fence", async () => { + const { state, storage } = harness(); + storage.read.mockRejectedValueOnce(new Error("permission denied")); + await assert.rejects(state.read(), /permission denied/); + expect(storage.repair).not.toHaveBeenCalled(); + expect(storage.mutate).not.toHaveBeenCalled(); + storage.read.mockRejectedValueOnce(new Error("permission denied")); + expect(await state.readForReplacement()).toMatchObject({ retainUntilReplacement: true }); + expect(storage.repair).not.toHaveBeenCalled(); + expect(storage.mutate).toHaveBeenCalledTimes(1); + }); + + it("failed explicit fence publication stays blocking and is reported", async () => { + const { state, storage } = harness(); + storage.read.mockRejectedValueOnce(new Error("read failed")); + storage.mutate.mockRejectedValueOnce(new Error("write failed")); + await assert.rejects(state.readForReplacement(), /write failed/); + expect(state.blocksRecovery).toBe(true); + expect(await state.read()).toMatchObject({ retainUntilReplacement: true }); + }); + + it("narrowing snapshots inputs before awaits and read results cannot mutate state", async () => { + const { state } = harness(); + await state.cancel(); + const record = (await state.read())!; + const input = structuredClone(summary); + const narrowing = state.narrow(record.nonce, input); + input.pendingFollowUp.text = "changed"; + await narrowing; + const narrowed = (await state.read())!; + expect(matchesCompactionCancellation(narrowed, summary)).toBe(true); + assert(narrowed.scope.kind === "summary"); + narrowed.scope.pendingFollowUp.text = "also changed"; + expect(matchesCompactionCancellation((await state.read())!, summary)).toBe(true); + }); + + it.each(["id", "sequence", "request"] as const)( + "summary cancellation matches exact identity (%s changes)", + (field) => { + const record: CompactionCancellationRecord = { + ...cancellation("narrowed"), + scope: { kind: "summary", ...summary }, + }; + const changed = structuredClone(summary); + if (field === "id") changed.id = "other"; + if (field === "sequence") changed.sequence = 4; + if (field === "request") changed.pendingFollowUp.options = { model: "other" }; + expect(matchesCompactionCancellation(record, changed)).toBe(false); + expect(matchesCompactionCancellation(cancellation("unresolved"), changed)).toBe(true); + } + ); + + it("flush follows the latest Stop through an obsolete publication failure", async () => { + const { state, storage } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.mutate.mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + throw new Error("obsolete failure"); + }); + const first = state.cancel(); + await entered.promise; + const flushing = state.flush(); + const second = state.cancel(); + release.resolve(); + await assert.rejects(first, /obsolete failure/); + await flushing; + expect(await second).toBe("applied"); + expect(state.needsPersistence).toBe(false); + }); +}); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts new file mode 100644 index 00000000000..aa813da603c --- /dev/null +++ b/src/node/services/compactionCancellation.ts @@ -0,0 +1,283 @@ +import { randomUUID } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; + +export interface CompactionCancellationSummary { + id: string; + sequence?: number; + pendingFollowUp: Record; +} + +export interface CompactionCancellationRecord { + version: 1; + nonce: string; + retainUntilReplacement?: boolean; + scope: { kind: "unresolved" } | ({ kind: "summary" } & CompactionCancellationSummary); +} + +export interface CompactionCancellationPublication { + attempts: number; + // The adapter records each admitted/advanced frontier BEFORE any subsequent failing await. + // Retries reuse this object; an unobserved attempt must never adopt a foreign frontier. + predecessor?: { nonce: string | null | undefined; generation: string | undefined }; +} + +/** Issued only after an exact replacement row is durably committed or found in history. */ +export interface CompactionCancellationReplacementWitness { + readonly nonce: string; +} + +export type CompactionCancellationMutation = + | { + kind: "publish"; + record: CompactionCancellationRecord; + publication: CompactionCancellationPublication; + } + | { kind: "narrow"; record: CompactionCancellationRecord } + | { + kind: "retire"; + nonce: string; + replacementWitness?: CompactionCancellationReplacementWitness; + }; + +export type CompactionCancellationMutationOutcome = "applied" | "superseded"; + +/** Only successfully read bytes with invalid JSON/schema authorize automatic repair. */ +export class MalformedCompactionCancellationError extends Error {} + +export interface CompactionCancellationStorage { + /** Fresh shared state; absence and unreadable I/O must remain distinguishable. */ + read(): Promise; + /** + * Atomically compare nonce/frontier and mutate under the shared history lock, checking + * isCurrent immediately before publication. Preserve inherited retention, including an + * unreadable predecessor. Retirement requires the exact nonce and, for retained records, + * a verified replacement witness. Superseded means no authority to apply this mutation. + */ + mutate( + mutation: CompactionCancellationMutation, + isCurrent: () => boolean + ): Promise; + /** + * Re-read under the lock; preserve newer valid records. Neutralize obsolete recovery + * before removing malformed bytes, preserve privacy floors, and call onCommitted + * synchronously when repair commits. Never repair an ordinary read/I/O failure. + */ + repair( + isCurrent: () => boolean, + onCommitted: () => void + ): Promise; +} + +/** + * Inactive cancellation state core. Stop's retry identity outlives failed turn preparation; + * a later delivery must supply the storage adapter and activate every recovery consumer. + * Injected-adapter tests establish state invariants, not filesystem or cross-process CAS. + */ +export class CompactionCancellation { + private current?: CompactionCancellationRecord | null; + private replacementNonce?: string; + private mutation?: CompactionCancellationMutation; + private pending: Promise = + Promise.resolve(undefined); + private unsettled = false; + private inFlight = false; + private repairedHistoryRevision = 0; + + constructor(private readonly storage: CompactionCancellationStorage) {} + + get needsPersistence(): boolean { + return this.unsettled; + } + + get blocksRecovery(): boolean { + return this.unsettled && !this.isWitnessedRetirement(); + } + + get repairRevision(): number { + return this.repairedHistoryRevision; + } + + private isWitnessedRetirement(): boolean { + return ( + this.mutation?.kind === "retire" && + this.mutation.replacementWitness?.nonce === this.mutation.nonce + ); + } + + private effectiveRecord(): CompactionCancellationRecord | null { + // Callers cannot mutate a captured cancellation or its exact pending-request identity. + return structuredClone( + this.current?.nonce === this.replacementNonce ? null : (this.current ?? null) + ); + } + + async read(): Promise { + if (this.blocksRecovery) return this.effectiveRecord(); + const mutation = this.mutation; + const pending = this.pending; + const isCurrent = () => mutation === this.mutation && pending === this.pending; + try { + const record = await this.storage.read().catch((error: unknown) => { + if (!(error instanceof MalformedCompactionCancellationError) || !isCurrent()) throw error; + return this.storage.repair(isCurrent, () => { + this.repairedHistoryRevision++; + }); + }); + if (isCurrent()) this.current = structuredClone(record); + } catch (error) { + // A stale read/repair cannot hide a newer local Stop or trigger its replacement. + if (isCurrent()) throw error; + } + return this.effectiveRecord(); + } + + cancel(options?: { + retainUntilReplacement?: boolean; + }): Promise { + this.current = { + version: 1, + nonce: randomUUID(), + scope: { kind: "unresolved" }, + ...(options?.retainUntilReplacement || this.current?.retainUntilReplacement + ? { retainUntilReplacement: true } + : {}), + }; + return this.persist({ kind: "publish", record: this.current, publication: { attempts: 0 } }); + } + + async readForReplacement(): Promise { + for (;;) { + if (this.blocksRecovery) { + const pending = this.pending; + const retryFailed = !this.inFlight; + try { + await pending; + } catch (error) { + // Retry already-failed debt; readers joining an in-flight attempt share its + // outcome instead of turning one failure into a chain of additional retries. + if (pending !== this.pending) continue; + if (!retryFailed) throw error; + const retried = this.retry(); + try { + await retried; + } catch (error) { + if (retried === this.pending) throw error; + } + } + continue; + } + try { + const record = await this.read(); + if (!this.blocksRecovery) return record; + } catch { + // Explicit intervention may replace unreadable state, but cannot lose an unknown + // full-clear obligation. Failed publication remains blocking and visible. + await this.cancel({ retainUntilReplacement: true }); + } + } + } + + async narrow(nonce: string, summary: CompactionCancellationSummary) { + const captured = structuredClone(summary); + const mutation = this.mutation; + const pending = this.pending; + await pending; + // Retirement can claim the same nonce during this join. Narrowing must not + // supersede its deletion or discard witnessed cleanup debt on resumption. + if ( + this.mutation !== mutation || + this.pending !== pending || + this.replacementNonce === nonce || + this.current?.nonce !== nonce || + this.current.scope.kind !== "unresolved" || + this.current.retainUntilReplacement + ) + return; + // Failed narrowing must retain the broader exclusion until persistence succeeds. + return this.persist({ + kind: "narrow", + record: { ...this.current, scope: { kind: "summary", ...captured } }, + }); + } + + retire(nonce: string) { + if (this.current?.nonce !== nonce) return Promise.resolve(undefined); + // A later cleanup request cannot downgrade already-witnessed deletion debt. + if (this.replacementNonce === nonce) return this.retireReplacement({ nonce }); + if (this.current.retainUntilReplacement) return Promise.resolve(undefined); + return this.persist({ kind: "retire", nonce }); + } + + retireReplacement(witness: CompactionCancellationReplacementWitness) { + if (this.current?.nonce !== witness.nonce) return Promise.resolve(undefined); + this.replacementNonce = witness.nonce; + return this.persist({ + kind: "retire", + nonce: witness.nonce, + replacementWitness: { ...witness }, + }); + } + + retry(): Promise { + return this.unsettled && !this.inFlight && this.mutation + ? this.persist(this.mutation) + : this.pending; + } + + async flush(): Promise { + for (;;) { + const pending = this.pending; + try { + await pending; + } catch (error) { + if (pending !== this.pending) continue; + if (!this.isWitnessedRetirement()) throw error; + } + if (pending === this.pending) return; + } + } + + private persist( + mutation: CompactionCancellationMutation + ): Promise { + this.mutation = mutation; + this.unsettled = true; + this.inFlight = true; + const isCurrent = () => this.mutation === mutation; + const result = this.pending + .catch(() => undefined) + .then(async (): Promise => { + if (!isCurrent()) return "superseded"; + if (mutation.kind === "publish") mutation.publication.attempts++; + const outcome = await this.storage.mutate(mutation, isCurrent); + if (isCurrent()) { + this.unsettled = false; + this.current = + outcome === "superseded" + ? undefined + : mutation.kind === "retire" + ? null + : structuredClone(mutation.record); + } + return outcome; + }); + this.pending = result; + const settled = () => { + if (this.pending === result) this.inFlight = false; + }; + result.then(settled, settled); + return result; + } +} + +export function matchesCompactionCancellation( + record: CompactionCancellationRecord, + summary: CompactionCancellationSummary +): boolean { + return ( + record.scope.kind === "unresolved" || + (record.scope.id === summary.id && + record.scope.sequence === summary.sequence && + isDeepStrictEqual(record.scope.pendingFollowUp, summary.pendingFollowUp)) + ); +} From ee130bd9a6dd16ab8ef8c3e86ff075c1a6b4b920 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 16:13:00 +0200 Subject: [PATCH 3/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20preserve=20cancellati?= =?UTF-8?q?on=20ownership=20across=20rejected=20joins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recheck fallback ownership after read rejection and let foreign cancellation narrowing bypass obsolete witnessed deletion debt. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ib8a1b50afce050febe46cbb9a110bc1686832d69 --- .../services/compactionCancellation.test.ts | 88 ++++++++++++++++++- src/node/services/compactionCancellation.ts | 19 +++- 2 files changed, 105 insertions(+), 2 deletions(-) diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts index 0e2577e4afb..1f37c7b19aa 100644 --- a/src/node/services/compactionCancellation.test.ts +++ b/src/node/services/compactionCancellation.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, mock } from "bun:test"; +import { describe, expect, it, mock, spyOn } from "bun:test"; import assert from "node:assert/strict"; import { CompactionCancellation, @@ -278,6 +278,37 @@ describe("inactive cancellation state core", () => { expect(state.needsPersistence).toBe(false); }); + it.each([false, true])( + "foreign narrowing is independent of witnessed deletion debt (narrow failure=%s)", + async (failed) => { + const { state, storage, shared } = harness(); + await state.cancel(); + const first = (await state.read())!; + storage.mutate.mockRejectedValueOnce(new Error("old unlink failed")); + await assert.rejects(state.retireReplacement({ nonce: first.nonce }), /old unlink failed/); + shared.record = cancellation("foreign-b"); + expect(await state.readForReplacement()).toEqual(shared.record); + + if (failed) storage.mutate.mockRejectedValueOnce(new Error("new narrowing failed")); + const narrowing = state.narrow("foreign-b", summary); + if (failed) { + await assert.rejects(narrowing, /new narrowing failed/); + expect(state.blocksRecovery).toBe(true); + expect(await state.read()).toEqual(cancellation("foreign-b")); + expect(await state.retry()).toBe("applied"); + } else expect(await narrowing).toBe("applied"); + + expect(shared.record).toEqual({ + ...cancellation("foreign-b"), + scope: { kind: "summary", ...summary }, + }); + expect(storage.mutate.mock.calls.slice(2).map(([mutation]) => mutation.kind)).toEqual( + failed ? ["narrow", "narrow"] : ["narrow"] + ); + expect(state.needsPersistence).toBe(false); + } + ); + it("a stale witness cannot make a newer failed Stop non-blocking", async () => { const { state, storage } = harness(); await state.cancel(); @@ -421,6 +452,61 @@ describe("inactive cancellation state core", () => { expect(storage.mutate).toHaveBeenCalledTimes(1); }); + it("fallback cannot replace a Stop admitted after the read's final error check", async () => { + const { state, storage } = harness(); + storage.read.mockRejectedValueOnce(new Error("read unavailable")); + const read = state.read.bind(state); + let stopping: ReturnType | undefined; + let newerRecord: ReturnType | undefined; + // Enter the promise boundary after read() has checked ownership and rejected, + // before readForReplacement() receives that rejection and considers fallback. + const checkedRead = spyOn(state, "read").mockImplementationOnce(() => + read().catch((error: unknown) => { + stopping = state.cancel(); + newerRecord = read(); + throw error; + }) + ); + try { + const replacement = await state.readForReplacement(); + expect(await stopping).toBe("applied"); + assert(newerRecord); + expect(replacement).toEqual(await newerRecord); + expect(storage.mutate).toHaveBeenCalledTimes(1); + } finally { + checkedRead.mockRestore(); + } + }); + + it("fallback cannot replace a same-mutation retry started after the checked read fails", async () => { + const { state, storage, shared } = harness(); + await state.cancel(); + const first = (await state.read())!; + storage.mutate.mockRejectedValueOnce(new Error("unlink failed")); + await assert.rejects(state.retireReplacement({ nonce: first.nonce }), /unlink failed/); + storage.read.mockRejectedValueOnce(new Error("read unavailable")); + const read = state.read.bind(state); + let retry: ReturnType | undefined; + const checkedRead = spyOn(state, "read").mockImplementationOnce(() => + read().catch((error: unknown) => { + retry = state.retry(); + throw error; + }) + ); + try { + expect(await state.readForReplacement()).toBeNull(); + expect(await retry).toBe("applied"); + expect(shared.record).toBeNull(); + expect(storage.mutate.mock.calls.map(([mutation]) => mutation.kind)).toEqual([ + "publish", + "retire", + "retire", + ]); + } finally { + checkedRead.mockRestore(); + } + }); + it("failed explicit fence publication stays blocking and is reported", async () => { const { state, storage } = harness(); storage.read.mockRejectedValueOnce(new Error("read failed")); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index aa813da603c..bd8855a25ca 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -166,10 +166,15 @@ export class CompactionCancellation { } continue; } + const mutation = this.mutation; + const pending = this.pending; try { const record = await this.read(); if (!this.blocksRecovery) return record; } catch { + // read() checks before rejecting, but a newer Stop/retry can enter before + // this rejection resumes. Fallback must still own that exact failed read. + if (this.mutation !== mutation || this.pending !== pending) continue; // Explicit intervention may replace unreadable state, but cannot lose an unknown // full-clear obligation. Failed publication remains blocking and visible. await this.cancel({ retainUntilReplacement: true }); @@ -181,7 +186,19 @@ export class CompactionCancellation { const captured = structuredClone(summary); const mutation = this.mutation; const pending = this.pending; - await pending; + try { + await pending; + } catch (error) { + if (this.mutation !== mutation || this.pending !== pending) return; + // An old witnessed unlink is ancillary once a fresh read discovers B. + // Its failure cannot block B's narrowing; B's own failed writes still do. + if ( + mutation?.kind !== "retire" || + mutation.replacementWitness?.nonce !== mutation.nonce || + mutation.nonce === nonce + ) + throw error; + } // Retirement can claim the same nonce during this join. Narrowing must not // supersede its deletion or discard witnessed cleanup debt on resumption. if ( From 65826f476d9c28f53adad4eb3401812cb68b109e Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 17:05:55 +0200 Subject: [PATCH 4/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20retain=20cancellation?= =?UTF-8?q?=20reads=20until=20a=20newer=20result=20is=20accepted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reject stale completions after an accepted successor without treating an in-flight read as evidence that cancellation is absent. Preserve error, repair, and replacement fallback ownership across the same boundary. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Icd8b3f1eb6e009c8f843712a08a4ac4c75ecd836 --- .../services/compactionCancellation.test.ts | 158 ++++++++++++++++++ src/node/services/compactionCancellation.ts | 28 +++- 2 files changed, 181 insertions(+), 5 deletions(-) diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts index 1f37c7b19aa..ca4e3f5a91c 100644 --- a/src/node/services/compactionCancellation.test.ts +++ b/src/node/services/compactionCancellation.test.ts @@ -385,6 +385,144 @@ describe("inactive cancellation state core", () => { } ); + it.each(["read", "replacement"] as const)( + "a pending newer read cannot hide an earlier durable cancellation (%s)", + async (reader) => { + const { state, storage } = harness(); + const older = cancellation("older-a"); + const newer = cancellation("newer-b"); + const firstRead = Promise.withResolvers(); + const secondRead = Promise.withResolvers(); + storage.read.mockReturnValueOnce(firstRead.promise).mockReturnValueOnce(secondRead.promise); + const earlier = reader === "read" ? state.read() : state.readForReplacement(); + const later = state.read(); + firstRead.resolve(older); + try { + expect(await earlier).toEqual(older); + expect(storage.mutate).not.toHaveBeenCalled(); + } finally { + secondRead.resolve(newer); + expect(await later).toEqual(newer); + } + } + ); + + it.each([false, true])( + "a pending newer read does not suppress repair or its failure (repair failure=%s)", + async (failed) => { + const { state, storage } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const secondRead = Promise.withResolvers(); + storage.read + .mockRejectedValueOnce(new MalformedCompactionCancellationError()) + .mockReturnValueOnce(secondRead.promise); + storage.repair.mockImplementationOnce(async (current, committed) => { + entered.resolve(); + await release.promise; + if (failed) throw new Error("repair failed"); + if (current()) committed(); + return null; + }); + const earlier = state.read(); + await entered.promise; + const later = state.read(); + release.resolve(); + try { + if (failed) await assert.rejects(earlier, /repair failed/); + else expect(await earlier).toBeNull(); + expect(state.repairRevision).toBe(failed ? 0 : 1); + } finally { + secondRead.resolve(null); + await later; + } + } + ); + + it.each(["old record", "absence", "malformed", "I/O"] as const)( + "a late read cannot replace a newer accepted read (%s)", + async (result) => { + const { state, storage, shared } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.read.mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + if (result === "malformed") throw new MalformedCompactionCancellationError(); + if (result === "I/O") throw new Error("obsolete read failure"); + return result === "absence" ? null : cancellation("older-a"); + }); + const earlier = state.read(); + await entered.promise; + shared.record = cancellation("newer-b"); + expect(await state.read()).toEqual(shared.record); + release.resolve(); + expect(await earlier).toEqual(shared.record); + expect(storage.repair).not.toHaveBeenCalled(); + expect(storage.mutate).not.toHaveBeenCalled(); + } + ); + + it.each([false, true])( + "a newer read invalidates an in-flight repair (repair failure=%s)", + async (failed) => { + const { state, storage, shared } = harness(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.read.mockRejectedValueOnce(new MalformedCompactionCancellationError()); + storage.repair.mockImplementationOnce(async (current, committed) => { + entered.resolve(); + await release.promise; + if (failed) throw new Error("obsolete repair failure"); + if (current()) { + shared.record = null; + committed(); + } + return null; + }); + const earlier = state.read(); + await entered.promise; + const newer = cancellation("newer-b"); + shared.record = newer; + expect(await state.read()).toEqual(newer); + release.resolve(); + expect(await earlier).toEqual(newer); + expect(shared.record).toEqual(newer); + expect(state.repairRevision).toBe(0); + } + ); + + it("out-of-order reads preserve the original witnessed retirement retry", async () => { + const { state, storage, shared } = harness(); + await state.cancel(); + const first = (await state.read())!; + storage.mutate.mockRejectedValueOnce(new Error("unlink failed")); + await assert.rejects(state.retireReplacement({ nonce: first.nonce }), /unlink failed/); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + storage.read.mockImplementationOnce(async () => { + entered.resolve(); + await release.promise; + return first; + }); + const earlier = state.read(); + await entered.promise; + shared.record = cancellation("newer-b"); + expect(await state.readForReplacement()).toEqual(shared.record); + release.resolve(); + expect(await earlier).toEqual(shared.record); + expect(state.needsPersistence).toBe(true); + expect(state.blocksRecovery).toBe(false); + storage.mutate.mockResolvedValueOnce("superseded"); + await state.retry(); + expect(storage.mutate.mock.calls.at(-1)?.[0]).toMatchObject({ + kind: "retire", + nonce: first.nonce, + }); + expect(shared.record).toEqual(cancellation("newer-b")); + expect(state.needsPersistence).toBe(false); + }); + it.each(["old record", "malformed", "I/O"] as const)( "ignores an obsolete read after Stop (%s)", async (result) => { @@ -507,6 +645,26 @@ describe("inactive cancellation state core", () => { } }); + it("fallback cannot replace a newer read accepted after the checked read fails", async () => { + const { state, storage, shared } = harness(); + storage.read.mockRejectedValueOnce(new Error("read unavailable")); + const newer = cancellation("newer-b"); + const read = state.read.bind(state); + const checkedRead = spyOn(state, "read").mockImplementationOnce(() => + read().catch(async (error: unknown) => { + shared.record = newer; + expect(await read()).toEqual(newer); + throw error; + }) + ); + try { + expect(await state.readForReplacement()).toEqual(newer); + expect(storage.mutate).not.toHaveBeenCalled(); + } finally { + checkedRead.mockRestore(); + } + }); + it("failed explicit fence publication stays blocking and is reported", async () => { const { state, storage } = harness(); storage.read.mockRejectedValueOnce(new Error("read failed")); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index bd8855a25ca..dfcaf00f058 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -81,6 +81,8 @@ export class CompactionCancellation { Promise.resolve(undefined); private unsettled = false; private inFlight = false; + private readGeneration = 0; + private acceptedReadGeneration = 0; private repairedHistoryRevision = 0; constructor(private readonly storage: CompactionCancellationStorage) {} @@ -115,7 +117,13 @@ export class CompactionCancellation { if (this.blocksRecovery) return this.effectiveRecord(); const mutation = this.mutation; const pending = this.pending; - const isCurrent = () => mutation === this.mutation && pending === this.pending; + // Only an accepted newer read displaces an earlier snapshot/error/repair. + // A pending successor is not evidence of absence and cannot hide a valid Stop. + const generation = ++this.readGeneration; + const isCurrent = () => + generation >= this.acceptedReadGeneration && + mutation === this.mutation && + pending === this.pending; try { const record = await this.storage.read().catch((error: unknown) => { if (!(error instanceof MalformedCompactionCancellationError) || !isCurrent()) throw error; @@ -123,7 +131,10 @@ export class CompactionCancellation { this.repairedHistoryRevision++; }); }); - if (isCurrent()) this.current = structuredClone(record); + if (isCurrent()) { + this.current = structuredClone(record); + this.acceptedReadGeneration = generation; + } } catch (error) { // A stale read/repair cannot hide a newer local Stop or trigger its replacement. if (isCurrent()) throw error; @@ -168,13 +179,20 @@ export class CompactionCancellation { } const mutation = this.mutation; const pending = this.pending; + const reading = this.read(); + const generation = this.readGeneration; try { - const record = await this.read(); + const record = await reading; if (!this.blocksRecovery) return record; } catch { - // read() checks before rejecting, but a newer Stop/retry can enter before + // read() checks before rejecting, but a newer Stop/retry/read can enter before // this rejection resumes. Fallback must still own that exact failed read. - if (this.mutation !== mutation || this.pending !== pending) continue; + if ( + this.acceptedReadGeneration > generation || + this.mutation !== mutation || + this.pending !== pending + ) + continue; // Explicit intervention may replace unreadable state, but cannot lose an unknown // full-clear obligation. Failed publication remains blocking and visible. await this.cancel({ retainUntilReplacement: true }); From f374bf0296154b4e7249cae6f361b522f98ba434 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 18:45:26 +0200 Subject: [PATCH 5/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20apply=20cancellation?= =?UTF-8?q?=20state=20at=20its=20durable=20commit=20receipt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Storage supplies the exact committed record, including inherited retention, through a synchronous receipt before lock release or cleanup. Apply cache and read invalidation there so delayed settlement cannot resurrect retired state or discard a newer foreign cancellation. Preserve existing public mutation outcomes. All 60 core tests and full static checks pass. Regressions cover inherited retention, stale predecessor reads, both foreign-successor completion orders, and failed or superseded publication. Independent review approved. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: Ie0a730120182d8ccce38ae4e2b3ae4ad74c85cb9 --- .../services/compactionCancellation.test.ts | 160 ++++++++++++++++-- src/node/services/compactionCancellation.ts | 25 ++- 2 files changed, 165 insertions(+), 20 deletions(-) diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts index ca4e3f5a91c..e4c5eb8e558 100644 --- a/src/node/services/compactionCancellation.test.ts +++ b/src/node/services/compactionCancellation.test.ts @@ -31,10 +31,12 @@ function harness() { mutate: mock( ( mutation: CompactionCancellationMutation, - isCurrent: () => boolean + isCurrent: () => boolean, + onCommitted: (record: CompactionCancellationRecord | null) => undefined ): Promise => { if (!isCurrent()) return Promise.resolve("superseded"); shared.record = mutation.kind === "retire" ? null : structuredClone(mutation.record); + onCommitted(shared.record); return Promise.resolve("applied"); } ), @@ -67,13 +69,13 @@ describe("inactive cancellation state core", () => { expect(state.blocksRecovery).toBe(true); expect(storage.read).not.toHaveBeenCalled(); await assert.rejects(state.narrow(first!.nonce, summary), /sidecar publication failed/); - storage.mutate.mockImplementationOnce(async (mutation, current) => { + storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => { assert(mutation.kind === "publish"); expect(mutation.record.nonce).toBe(first!.nonce); expect(mutation.publication).toBe(publication!); expect(mutation.publication.predecessor).toBe(frontier); expect(mutation.publication.attempts).toBe(2); - return apply(mutation, current); + return apply(mutation, current, onCommitted); }); expect(await state.retry()).toBe("applied"); expect(shared.record?.nonce).toBe(first!.nonce); @@ -114,11 +116,11 @@ describe("inactive cancellation state core", () => { const entered = Promise.withResolvers(); const release = Promise.withResolvers(); const apply = storage.mutate.getMockImplementation()!; - storage.mutate.mockImplementationOnce(async (mutation, current) => { + storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => { entered.resolve(); await release.promise; expect(current()).toBe(false); - return apply(mutation, current); + return apply(mutation, current, onCommitted); }); const first = state.cancel(); await entered.promise; @@ -151,16 +153,16 @@ describe("inactive cancellation state core", () => { const successorEntered = Promise.withResolvers(); const releaseSuccessor = Promise.withResolvers(); const apply = storage.mutate.getMockImplementation()!; - storage.mutate.mockImplementationOnce(async (mutation, current) => { - const outcome = await apply(mutation, current); + storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => { + const outcome = await apply(mutation, current, onCommitted); committed.resolve(); await acknowledge.promise; return outcome; }); - storage.mutate.mockImplementationOnce(async (mutation, current) => { + storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => { successorEntered.resolve(); await releaseSuccessor.promise; - return apply(mutation, current); + return apply(mutation, current, onCommitted); }); const first = state.cancel(); await committed.promise; @@ -255,6 +257,142 @@ describe("inactive cancellation state core", () => { expect(await state.read()).toEqual(second); }); + it("preserves retention inherited by the adapter without first reading the predecessor", async () => { + const { state, storage, shared } = harness(); + shared.record = { ...cancellation("retained-predecessor"), retainUntilReplacement: true }; + const apply = storage.mutate.getMockImplementation()!; + storage.mutate.mockImplementationOnce((mutation, current, onCommitted) => { + assert(mutation.kind === "publish"); + expect(mutation.record.retainUntilReplacement).toBeUndefined(); + // Inheritance happens under the adapter lock, without mutating the submitted record. + return apply( + { ...mutation, record: { ...mutation.record, retainUntilReplacement: true } }, + current, + onCommitted + ); + }); + await state.cancel(); + const committed = structuredClone(shared.record); + assert(committed); + await state.narrow(committed.nonce, summary); + await state.retire(committed.nonce); + expect(shared.record).toEqual(committed); + expect(storage.mutate).toHaveBeenCalledTimes(1); + expect(storage.read).not.toHaveBeenCalled(); + await state.cancel(); + expect(shared.record?.retainUntilReplacement).toBe(true); + }); + + it.each(["record", "malformed", "I/O"])( + "a read spanning committed witnessed retirement cannot restore its retention or error (%s)", + async (outcome) => { + const { state, storage, shared } = harness(); + await state.cancel({ retainUntilReplacement: true }); + const record = structuredClone(shared.record); + assert(record); + const snapshot = Promise.withResolvers(); + storage.read.mockReturnValueOnce(snapshot.promise); + const retirement = state.retireReplacement({ nonce: record.nonce }); + const reading = state.read(); + await retirement; + expect(shared.record).toBeNull(); + if (outcome === "record") snapshot.resolve(record); + else + snapshot.reject( + outcome === "malformed" + ? new MalformedCompactionCancellationError("old bytes") + : new Error("old I/O failure") + ); + expect(await reading).toBeNull(); + expect(storage.repair).not.toHaveBeenCalled(); + await state.cancel(); + expect(shared.record?.retainUntilReplacement).toBeUndefined(); + } + ); + + it.each([ + { outcome: "failed", completion: "before" }, + { outcome: "failed", completion: "after" }, + { outcome: "superseded", completion: "before" }, + { outcome: "superseded", completion: "after" }, + ])( + "uncommitted retirement preserves a foreign read ($outcome, completion=$completion)", + async ({ outcome, completion }) => { + const { state, storage, shared } = harness(); + await state.cancel({ retainUntilReplacement: true }); + const record = shared.record; + assert(record); + const snapshot = Promise.withResolvers(); + storage.read.mockReturnValueOnce(snapshot.promise); + const acknowledge = Promise.withResolvers(); + storage.mutate.mockReturnValueOnce(acknowledge.promise); + const retirement = state.retireReplacement({ nonce: record.nonce }); + const reading = state.read(); + const foreign = { ...cancellation("foreign-b"), retainUntilReplacement: true }; + if (completion === "before") { + snapshot.resolve(foreign); + expect(await reading).toEqual(foreign); + } + if (outcome === "failed") acknowledge.reject(new Error("unlink failed")); + else acknowledge.resolve("superseded"); + if (outcome === "failed") await assert.rejects(retirement, /unlink failed/); + else await retirement; + if (completion === "after") { + snapshot.resolve(foreign); + expect(await reading).toEqual(foreign); + } + await state.cancel(); + expect(shared.record?.retainUntilReplacement).toBe(true); + } + ); + + it.each([ + { completion: "before", failedCleanup: false }, + { completion: "after", failedCleanup: false }, + { completion: "before", failedCleanup: true }, + { completion: "after", failedCleanup: true }, + ])( + "preserves a post-deletion foreign read (completion=$completion, failed cleanup=$failedCleanup)", + async ({ completion, failedCleanup }) => { + const { state, storage, shared } = harness(); + await state.cancel({ retainUntilReplacement: true }); + const record = shared.record; + assert(record); + const deleted = Promise.withResolvers(); + const acknowledge = Promise.withResolvers(); + storage.mutate.mockImplementationOnce(async (_mutation, _current, onCommitted) => { + shared.record = null; + onCommitted(null); + deleted.resolve(); + await acknowledge.promise; + if (failedCleanup) throw new Error("cleanup failed after deletion"); + return "applied"; + }); + const retirement = state.retireReplacement({ nonce: record.nonce }); + await deleted.promise; + const foreign = { ...cancellation("foreign-b"), retainUntilReplacement: true }; + shared.record = foreign; + const snapshot = Promise.withResolvers(); + storage.read.mockReturnValueOnce(snapshot.promise); + const reading = state.read(); + if (completion === "before") { + snapshot.resolve(foreign); + expect(await reading).toEqual(foreign); + } + acknowledge.resolve(); + if (failedCleanup) await assert.rejects(retirement, /cleanup failed/); + else await retirement; + expect(state.needsPersistence).toBe(failedCleanup); + if (completion === "after") { + snapshot.resolve(foreign); + expect(await reading).toEqual(foreign); + } + // No refresh is needed for the next Stop to carry B's full-clear obligation. + await state.cancel(); + expect(shared.record?.retainUntilReplacement).toBe(true); + } + ); + it("witnessed deletion debt allows fresh reads without adopting a foreign Stop for retry", async () => { const { state, storage, shared } = harness(); await state.cancel({ retainUntilReplacement: true }); @@ -364,11 +502,11 @@ describe("inactive cancellation state core", () => { const entered = Promise.withResolvers(); const release = Promise.withResolvers(); const apply = storage.mutate.getMockImplementation()!; - storage.mutate.mockImplementationOnce(async (mutation, current) => { + storage.mutate.mockImplementationOnce(async (mutation, current, onCommitted) => { entered.resolve(); await release.promise; if (failed) throw new Error("retry failure"); - return apply(mutation, current); + return apply(mutation, current, onCommitted); }); const readers = [state.readForReplacement(), state.readForReplacement()]; await entered.promise; diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index dfcaf00f058..0778dd9b2c6 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -52,10 +52,14 @@ export interface CompactionCancellationStorage { * isCurrent immediately before publication. Preserve inherited retention, including an * unreadable predecessor. Retirement requires the exact nonce and, for retained records, * a verified replacement witness. Superseded means no authority to apply this mutation. + * Call onCommitted synchronously at the durable commit/confirmation, before releasing + * the lock or awaiting cleanup, with inherited retention or null after retirement. + * Every applied outcome requires this receipt; later failure cannot undo the commit. */ mutate( mutation: CompactionCancellationMutation, - isCurrent: () => boolean + isCurrent: () => boolean, + onCommitted: (record: CompactionCancellationRecord | null) => undefined ): Promise; /** * Re-read under the lock; preserve newer valid records. Neutralize obsolete recovery @@ -117,7 +121,7 @@ export class CompactionCancellation { if (this.blocksRecovery) return this.effectiveRecord(); const mutation = this.mutation; const pending = this.pending; - // Only an accepted newer read displaces an earlier snapshot/error/repair. + // Only an accepted newer read or committed mutation displaces a snapshot/error/repair. // A pending successor is not evidence of absence and cannot hide a valid Stop. const generation = ++this.readGeneration; const isCurrent = () => @@ -278,21 +282,24 @@ export class CompactionCancellation { this.mutation = mutation; this.unsettled = true; this.inFlight = true; + const generation = this.acceptedReadGeneration; const isCurrent = () => this.mutation === mutation; const result = this.pending .catch(() => undefined) .then(async (): Promise => { if (!isCurrent()) return "superseded"; if (mutation.kind === "publish") mutation.publication.attempts++; - const outcome = await this.storage.mutate(mutation, isCurrent); + const outcome = await this.storage.mutate(mutation, isCurrent, (record) => { + if (!isCurrent()) return; + this.current = structuredClone(record); + // Commit invalidates pre-deletion reads before lock release. A later foreign + // read must survive acknowledgment delayed by adapter cleanup. + this.acceptedReadGeneration = ++this.readGeneration; + }); if (isCurrent()) { this.unsettled = false; - this.current = - outcome === "superseded" - ? undefined - : mutation.kind === "retire" - ? null - : structuredClone(mutation.record); + if (outcome === "superseded" && this.acceptedReadGeneration === generation) + this.current = undefined; } return outcome; }); From 0a367ba75042c9e6adce43a01f84fa91be6abe10 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 19:44:14 +0200 Subject: [PATCH 6/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20bound=20cancellation?= =?UTF-8?q?=20refresh=20without=20returning=20tentative=20state?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserve unknown state after superseded reads and refresh once before replacement. Reject another overlap or unresolved persistence; unsupported storage records bypass fallback. Validation: 71 targeted tests, full static checks, and independent review pass. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: I3581c075367f653fba39e6dedf8f62c350952652 --- .../services/compactionCancellation.test.ts | 152 ++++++++++++++++++ src/node/services/compactionCancellation.ts | 26 ++- 2 files changed, 176 insertions(+), 2 deletions(-) diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts index e4c5eb8e558..33471d2fed1 100644 --- a/src/node/services/compactionCancellation.test.ts +++ b/src/node/services/compactionCancellation.test.ts @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import { CompactionCancellation, MalformedCompactionCancellationError, + CompactionCancellationReadRefusedError, matchesCompactionCancellation, type CompactionCancellationMutation, type CompactionCancellationMutationOutcome, @@ -716,6 +717,16 @@ describe("inactive cancellation state core", () => { expect(state.repairRevision).toBe(0); }); + it("replacement propagates a read refusal without repair or fallback publication", async () => { + const { state, storage } = harness(); + const refusal = new CompactionCancellationReadRefusedError("Read refused"); + storage.read.mockRejectedValueOnce(refusal); + await assert.rejects(state.readForReplacement(), (error) => error === refusal); + expect(storage.repair).not.toHaveBeenCalled(); + expect(storage.mutate).not.toHaveBeenCalled(); + expect(state.needsPersistence).toBe(false); + }); + it("ordinary read failure never repairs; explicit replacement publishes a retained fence", async () => { const { state, storage } = harness(); storage.read.mockRejectedValueOnce(new Error("permission denied")); @@ -728,6 +739,147 @@ describe("inactive cancellation state core", () => { expect(storage.mutate).toHaveBeenCalledTimes(1); }); + it.each([ + { completion: "before", freshFailure: false }, + { completion: "after", freshFailure: false }, + { completion: "before", freshFailure: true }, + { completion: "after", freshFailure: true }, + ])( + "replacement refreshes unknown state (error=$completion supersession, fresh failure=$freshFailure)", + async ({ completion, freshFailure }) => { + const { state, storage, shared } = harness(); + shared.record = { ...cancellation("foreign-b"), retainUntilReplacement: true }; + const failure = Promise.withResolvers(); + const superseded = Promise.withResolvers(); + const readFinished = Promise.withResolvers(); + const deliver = Promise.withResolvers(); + storage.read.mockReturnValueOnce(failure.promise); + if (freshFailure) storage.read.mockRejectedValueOnce(new Error("fresh read unavailable")); + storage.mutate.mockReturnValueOnce(superseded.promise); + const read = state.read.bind(state); + const checkedRead = spyOn(state, "read").mockImplementationOnce(async () => { + try { + const record = await read(); + readFinished.resolve(); + if (completion === "before") await deliver.promise; + return record; + } finally { + readFinished.resolve(); + } + }); + try { + const replacement = state.readForReplacement(); + const stopping = state.cancel(); + if (completion === "before") { + failure.reject(new Error("old read failed")); + await readFinished.promise; + } + superseded.resolve("superseded"); + expect(await stopping).toBe("superseded"); + deliver.resolve(); + if (completion === "after") failure.reject(new Error("old read failed")); + if (freshFailure) await assert.rejects(replacement, /fresh read unavailable/); + else expect(await replacement).toEqual(shared.record); + expect(storage.read).toHaveBeenCalledTimes(2); + expect(storage.mutate).toHaveBeenCalledTimes(1); + } finally { + checkedRead.mockRestore(); + } + } + ); + + it.each([ + { overlap: false, readFailure: true }, + { overlap: true, readFailure: true }, + { overlap: true, readFailure: false }, + ])( + "replacement bounds its refresh (another overlap=$overlap, read failure=$readFailure)", + async ({ overlap, readFailure }) => { + const { state, storage, shared } = harness(); + shared.record = cancellation("foreign-b"); + const failure = Promise.withResolvers(); + const refresh = Promise.withResolvers(); + const refreshEntered = Promise.withResolvers(); + storage.read.mockReturnValueOnce(failure.promise).mockImplementationOnce(() => { + refreshEntered.resolve(); + return refresh.promise; + }); + storage.mutate.mockResolvedValue("superseded"); + const replacement = state.readForReplacement(); + await state.cancel(); + failure.reject(new Error("old read failed")); + // The race allows no absence decision before the fresh authoritative read starts. + await Promise.race([refreshEntered.promise, replacement]); + expect(storage.read).toHaveBeenCalledTimes(2); + if (overlap) await state.cancel(); + if (readFailure) { + refresh.reject(new Error("fresh read unavailable")); + await assert.rejects(replacement, /fresh read unavailable/); + } else { + refresh.resolve(null); + await assert.rejects(replacement); + } + expect(storage.read).toHaveBeenCalledTimes(2); + expect(storage.mutate).toHaveBeenCalledTimes(overlap ? 2 : 1); + expect(shared.record).toEqual(cancellation("foreign-b")); + } + ); + + it("joining a superseded Stop surfaces a failed authoritative read without fallback", async () => { + const { state, storage, shared } = harness(); + shared.record = cancellation("foreign-b"); + storage.read.mockRejectedValue(new Error("fresh read unavailable")); + storage.mutate.mockResolvedValueOnce("superseded"); + storage.mutate.mockRejectedValue(new Error("unexpected fallback")); + const stopping = state.cancel(); + const replacement = state.readForReplacement(); + expect(await stopping).toBe("superseded"); + await assert.rejects(replacement, /fresh read unavailable/); + expect(storage.read).toHaveBeenCalledTimes(1); + expect(storage.mutate).toHaveBeenCalledTimes(1); + }); + + it.each([false, true])( + "a bounded refresh refuses another Stop overlapping its read (publication settled=%s)", + async (settled) => { + const { state, storage, shared } = harness(); + shared.record = cancellation("foreign-a"); + const failedRead = Promise.withResolvers(); + const refresh = Promise.withResolvers(); + const refreshEntered = Promise.withResolvers(); + const publish = Promise.withResolvers(); + storage.read.mockReturnValueOnce(failedRead.promise).mockImplementationOnce(() => { + refreshEntered.resolve(); + return refresh.promise; + }); + const apply = storage.mutate.getMockImplementation()!; + storage.mutate + .mockResolvedValueOnce("superseded") + .mockImplementationOnce(async (mutation, current, onCommitted) => { + await publish.promise; + return apply(mutation, current, onCommitted); + }); + const replacement = state.readForReplacement(); + await state.cancel(); + failedRead.reject(new Error("old read unavailable")); + await refreshEntered.promise; + const stopping = state.cancel(); + if (settled) { + publish.resolve(); + expect(await stopping).toBe("applied"); + } + refresh.resolve(shared.record); + await assert.rejects(replacement); + expect(state.blocksRecovery).toBe(!settled); + expect(state.needsPersistence).toBe(!settled); + expect(storage.read).toHaveBeenCalledTimes(2); + publish.resolve(); + expect(await stopping).toBe("applied"); + expect(await state.readForReplacement()).toEqual(shared.record); + expect(state.blocksRecovery).toBe(false); + } + ); + it("fallback cannot replace a Stop admitted after the read's final error check", async () => { const { state, storage } = harness(); storage.read.mockRejectedValueOnce(new Error("read unavailable")); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index 0778dd9b2c6..f3ad1660f17 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -44,6 +44,9 @@ export type CompactionCancellationMutationOutcome = "applied" | "superseded"; /** Only successfully read bytes with invalid JSON/schema authorize automatic repair. */ export class MalformedCompactionCancellationError extends Error {} +/** Unsupported or oversized records must be preserved instead of repaired or overwritten. */ +export class CompactionCancellationReadRefusedError extends Error {} + export interface CompactionCancellationStorage { /** Fresh shared state; absence and unreadable I/O must remain distinguishable. */ read(): Promise; @@ -141,8 +144,10 @@ export class CompactionCancellation { } } catch (error) { // A stale read/repair cannot hide a newer local Stop or trigger its replacement. - if (isCurrent()) throw error; + if (isCurrent() || this.current === undefined) throw error; } + // Supersession without a receipt leaves unknown state, never evidence of absence. + if (this.current === undefined) throw new Error("Cancellation state changed during read"); return this.effectiveRecord(); } @@ -179,6 +184,7 @@ export class CompactionCancellation { if (retried === this.pending) throw error; } } + if (this.current === undefined) return this.refreshForReplacement(); continue; } const mutation = this.mutation; @@ -187,8 +193,14 @@ export class CompactionCancellation { const generation = this.readGeneration; try { const record = await reading; + if (this.current === undefined) return this.refreshForReplacement(); if (!this.blocksRecovery) return record; - } catch { + } catch (error) { + if (error instanceof CompactionCancellationReadRefusedError) throw error; + // Refresh unknown state once; propagate that read's failure instead of repeatedly + // publishing fallback Stops that a foreign cancellation keeps superseding. + if (this.current === undefined && (this.mutation !== mutation || this.pending !== pending)) + return this.refreshForReplacement(); // read() checks before rejecting, but a newer Stop/retry/read can enter before // this rejection resumes. Fallback must still own that exact failed read. if ( @@ -204,6 +216,16 @@ export class CompactionCancellation { } } + private async refreshForReplacement(): Promise { + const pending = this.pending; + await this.read(); + // One refresh cannot turn another Stop's tentative state into replacement authority. + // Further overlap requires a new request rather than an unbounded refresh/retry loop. + if (this.pending !== pending || this.blocksRecovery || this.current === undefined) + throw new Error("Cancellation changed during replacement refresh"); + return this.effectiveRecord(); + } + async narrow(nonce: string, summary: CompactionCancellationSummary) { const captured = structuredClone(summary); const mutation = this.mutation; From 4684034bb73043461536c0ddb50123e6a1b9a539 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 8 Sep 2026 20:29:42 +0200 Subject: [PATCH 7/7] =?UTF-8?q?=F0=9F=A4=96=20fix:=20preserve=20cancellati?= =?UTF-8?q?on=20read=20authority=20across=20commit=20receipts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Return current cancellation state after replacement reads and invalidate earlier reads synchronously when repair commits. Preserve successor state and bounded refresh behavior. Both races reproduced before the fix. All 77 core tests, full static checks, and independent review pass. Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: I77e0089454adfee985ba8aaa72cf51c2a28f5100 --- .../services/compactionCancellation.test.ts | 81 +++++++++++++++++++ src/node/services/compactionCancellation.ts | 10 ++- 2 files changed, 89 insertions(+), 2 deletions(-) diff --git a/src/node/services/compactionCancellation.test.ts b/src/node/services/compactionCancellation.test.ts index 33471d2fed1..0ab8cb38c57 100644 --- a/src/node/services/compactionCancellation.test.ts +++ b/src/node/services/compactionCancellation.test.ts @@ -631,6 +631,64 @@ describe("inactive cancellation state core", () => { } ); + it.each([ + { completion: "before", successor: "cancel" }, + { completion: "after", successor: "cancel" }, + { completion: "before", successor: "read" }, + { completion: "after", successor: "read" }, + ])( + "repair receipt fences retained reads before acknowledgment (read=$completion, successor=$successor)", + async ({ completion, successor }) => { + const { state, storage, shared } = harness(); + const retained = { ...cancellation("older-a"), retainUntilReplacement: true }; + const olderSnapshot = Promise.withResolvers(); + const entered = Promise.withResolvers(); + const commit = Promise.withResolvers(); + const committed = Promise.withResolvers(); + const acknowledge = Promise.withResolvers(); + storage.read + .mockReturnValueOnce(olderSnapshot.promise) + .mockRejectedValueOnce(new MalformedCompactionCancellationError()); + storage.repair.mockImplementationOnce(async (current, onCommitted) => { + entered.resolve(); + await commit.promise; + assert(current()); + shared.record = null; + onCommitted(); + committed.resolve(); + await acknowledge.promise; + return null; + }); + const older = state.read(); + const repairing = state.read(); + await entered.promise; + if (completion === "before") { + olderSnapshot.resolve(retained); + expect(await older).toEqual(retained); + } + commit.resolve(); + await committed.promise; + expect(state.repairRevision).toBe(1); + try { + if (completion === "after") { + olderSnapshot.resolve(retained); + expect(await older).toBeNull(); + } + if (successor === "cancel") { + await state.cancel(); + expect(shared.record?.retainUntilReplacement).toBeUndefined(); + } else { + shared.record = cancellation("newer-b"); + expect(await state.read()).toEqual(shared.record); + } + } finally { + acknowledge.resolve(); + await repairing; + } + expect(await repairing).toEqual(shared.record); + } + ); + it("out-of-order reads preserve the original witnessed retirement retry", async () => { const { state, storage, shared } = harness(); await state.cancel(); @@ -880,6 +938,29 @@ describe("inactive cancellation state core", () => { } ); + it.each(["record", "absence"])( + "replacement returns the committed Stop admitted after a successful read (%s)", + async (snapshot) => { + const { state, storage, shared } = harness(); + shared.record = snapshot === "record" ? cancellation("older-a") : null; + const read = state.read.bind(state); + const checkedRead = spyOn(state, "read").mockImplementationOnce(() => + read().then(async (record) => { + await state.cancel(); + return record; + }) + ); + try { + expect(await state.readForReplacement()).toEqual(shared.record); + expect(shared.record).not.toBeNull(); + expect(state.blocksRecovery).toBe(false); + expect(storage.read).toHaveBeenCalledTimes(1); + } finally { + checkedRead.mockRestore(); + } + } + ); + it("fallback cannot replace a Stop admitted after the read's final error check", async () => { const { state, storage } = harness(); storage.read.mockRejectedValueOnce(new Error("read unavailable")); diff --git a/src/node/services/compactionCancellation.ts b/src/node/services/compactionCancellation.ts index f3ad1660f17..9e906f20a82 100644 --- a/src/node/services/compactionCancellation.ts +++ b/src/node/services/compactionCancellation.ts @@ -136,6 +136,11 @@ export class CompactionCancellation { if (!(error instanceof MalformedCompactionCancellationError) || !isCurrent()) throw error; return this.storage.repair(isCurrent, () => { this.repairedHistoryRevision++; + if (!isCurrent()) return; + // Removal is already committed; pre-repair reads must not restore retention + // while the adapter is still finishing lock cleanup. + this.current = null; + this.acceptedReadGeneration = ++this.readGeneration; }); }); if (isCurrent()) { @@ -192,9 +197,10 @@ export class CompactionCancellation { const reading = this.read(); const generation = this.readGeneration; try { - const record = await reading; + await reading; if (this.current === undefined) return this.refreshForReplacement(); - if (!this.blocksRecovery) return record; + // A Stop or newer read can commit after reading resolves but before we resume. + if (!this.blocksRecovery) return this.effectiveRecord(); } catch (error) { if (error instanceof CompactionCancellationReadRefusedError) throw error; // Refresh unknown state once; propagate that read's failure instead of repeatedly