diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 9ad3f1869a..20c1510577 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -13,6 +13,7 @@ - Runtime settings reconciliation now validates every `web_search.fallback` entry against the declared provider enum instead of accepting unsupported or non-string array items (#3601). - Ultragoal critic-gate, dogfood, review, durable-completion, and runtime test suites now pin `CI_DEV_CHANGED_PATHS` hermetically in their setup/teardown. Their temp checkpoints live inside the enclosing git work tree, so the CI planner's changed paths (which include computer control surface paths on branches that touch them) previously leaked into the computed change set and falsely triggered the mandatory computer red-team suite (`COMPUTER_REDTEAM_CASE_MISSING: … must include kill-switch-bypass`). The production kill-switch-bypass gate is unchanged; only the test fixtures now isolate their own contract from the host branch's diff (#3533). - Ultragoal critic-gate, dogfood, review, durable-completion, and runtime test suites now relocate temp dirs to `os.tmpdir()` (outside the enclosing git work tree) and pin `CI_DEV_CHANGED_PATHS` to a non-computer test path. The prior in-repo temp dirs caused `computeCheckpointChangeSet` to return `captureIncomplete=true` under parallel shard load (git command timeouts), which unconditionally triggered the mandatory computer red-team suite even when no computer surface was touched. The production kill-switch-bypass gate is unchanged; the `.tmp-*` gitignore entry prevents in-repo test artifacts from polluting untracked-file inventory (#3533). +- Telegram topic delete settlement is now fence-epoch bound, two-phase, and durably route-atomic. `TopicRegistry.settleDelete` requires the caller's dispatched authority epoch to still equal both the record's own epoch and the session's current epoch, so a held earlier delete can no longer settle a newer scan/close-started fence for the same session and topic and release its quarantine; it now removes the record but deliberately *retains* the topic-id quarantine and returns a settlement token instead of publishing routes, so no colliding survivor becomes routable and no settled id becomes adoptable while the clear is still only in memory. `commitSettledDelete` publishes the rebuilt inbound routes and releases the quarantine only after the durable topic-state persist resolves, and `rollbackSettledDelete` undoes a failed persist as a compare-and-set that applies only while the post-settlement state is still exactly current, so a stale rollback can no longer resurrect a deleted record over a newer fence. A refused settlement returns no token and is therefore structurally incapable of being rolled back. Authority-epoch advancement is routed through a single saturating helper capped at `Number.MAX_SAFE_INTEGER`, and settlement fails closed (keeping the fence) on a non-safe-integer, negative, or already-saturated epoch instead of settling against an unsound comparison. Telegram's first create-compensation path now marks compensation complete only after that durable clear commits, so a failed persist leaves the fence supervised rather than stranding a cleared memory state against a `delete_pending` disk state. ### Fixed diff --git a/packages/coding-agent/src/sdk/bus/telegram-daemon.ts b/packages/coding-agent/src/sdk/bus/telegram-daemon.ts index dde4527e79..cb0d2b67fb 100644 --- a/packages/coding-agent/src/sdk/bus/telegram-daemon.ts +++ b/packages/coding-agent/src/sdk/bus/telegram-daemon.ts @@ -6814,17 +6814,18 @@ export class TelegramNotificationDaemon { acceptedTopicId = String(tid); this.#malformedTopicCreateEndpoints.delete(sessionId); if (capturedCreationLease && !(await this.#awaitCreationLeaseAuthority(capturedCreationLease))) { - if ( - !this.topics.fenceAcceptedCreateForLease( - sessionId, - acceptedTopicId, - creationLeaseEpoch, - this.opts.now, - name, - creationBinding, - ) - ) - throw new Error("topic authority was revoked during creation"); + const fencedCreate = this.topics.fenceAcceptedCreateForLease( + sessionId, + acceptedTopicId, + creationLeaseEpoch, + this.opts.now, + name, + creationBinding, + ); + if (!fencedCreate) throw new Error("topic authority was revoked during creation"); + // Epoch held when the compensating delete is dispatched below; a + // concurrent re-fence must not be settled by this delete's result. + const fencedCreateEpoch = fencedCreate.authorityEpoch ?? 0; try { await this.persistTopics(); } finally { @@ -6833,11 +6834,27 @@ export class TelegramNotificationDaemon { chat_id: this.opts.chatId, message_thread_id: tid, }); - acceptedTopicCompensated = topicDeleteSettled(deletion); - + // Remote compensation succeeded, but the transaction is not complete + // until the registry clear is durable. `acceptedTopicCompensated` is + // what tells outer recovery to stop supervising the fence, so it is + // set only after the phase-2 commit: a failed clear persist leaves + // the fence supervised instead of stranding a cleared memory state + // against a `delete_pending` disk state. if (topicDeleteSettled(deletion)) { - this.topics.settleDelete(sessionId, acceptedTopicId); - await this.persistTopics(); + const settled = this.topics.settleDelete(sessionId, acceptedTopicId, fencedCreateEpoch); + if (!settled) { + this.#superviseCompensationFence(sessionId); + await this.#persistTopicsWithRetry().catch(() => undefined); + } else + try { + await this.persistTopics(); + this.topics.commitSettledDelete(settled); + acceptedTopicCompensated = true; + } catch { + this.topics.rollbackSettledDelete(settled); + this.#superviseCompensationFence(sessionId); + await this.#persistTopicsWithRetry().catch(() => undefined); + } } else { this.#superviseCompensationFence(sessionId); await this.#persistTopicsWithRetry().catch(() => undefined); @@ -6916,16 +6933,18 @@ export class TelegramNotificationDaemon { ) { // A failed initial commit must never make compensation conditional on // successfully publishing its fence. - if ( - this.topics.fenceAcceptedCreateForLease( - sessionId, - acceptedTopicId, - creationLeaseEpoch, - this.opts.now, - name, - creationBinding, - ) - ) { + const fencedCompensation = this.topics.fenceAcceptedCreateForLease( + sessionId, + acceptedTopicId, + creationLeaseEpoch, + this.opts.now, + name, + creationBinding, + ); + if (fencedCompensation) { + // Epoch held when the compensating delete is dispatched below; a + // concurrent re-fence must not be settled by this delete's result. + const fencedCompensationEpoch = fencedCompensation.authorityEpoch ?? 0; try { await this.#persistTopicsWithRetry(); } catch { @@ -6938,8 +6957,19 @@ export class TelegramNotificationDaemon { message_thread_id: Number(acceptedTopicId), }); if (topicDeleteSettled(deletion)) { - this.topics.settleDelete(sessionId, acceptedTopicId); - await this.persistTopics(); + const settled = this.topics.settleDelete(sessionId, acceptedTopicId, fencedCompensationEpoch); + if (!settled) { + this.#superviseCompensationFence(sessionId); + await this.#persistTopicsWithRetry().catch(() => undefined); + } else + try { + await this.persistTopics(); + this.topics.commitSettledDelete(settled); + } catch { + this.topics.rollbackSettledDelete(settled); + this.#superviseCompensationFence(sessionId); + await this.#persistTopicsWithRetry().catch(() => undefined); + } } else { this.#superviseCompensationFence(sessionId); await this.#persistTopicsWithRetry().catch(() => undefined); @@ -6989,8 +7019,11 @@ export class TelegramNotificationDaemon { socketLease?: { session: SessionSocket; token: number; logicalSessionId: string }, deleteFenceAlreadyPublished = false, ): Promise<"pre_dispatch_cancelled" | "post_dispatch_pending" | "settled"> { - const deleteSnapshot = this.topics.captureDeleteAuthority(sessionId); let record = deleteFenceAlreadyPublished ? this.topics.get(sessionId) : this.topics.beginDelete(sessionId); + // Authority epoch held for this delete. Captured before any dispatch so a + // concurrent scan/close re-fence of the same session cannot be settled by + // this delete's definite result. + const dispatchedAuthorityEpoch = this.topics.authorityEpoch(sessionId); if (socketLease && !this.#deleteLeaseAllows(socketLease)) return "pre_dispatch_cancelled"; await this.persistTopics(); if (socketLease && !this.#deleteLeaseAllows(socketLease)) return "pre_dispatch_cancelled"; @@ -7013,7 +7046,13 @@ export class TelegramNotificationDaemon { await this.flushPool(); if (socketLease && !this.#deleteLeaseAllows(socketLease)) return "pre_dispatch_cancelled"; if (record.topicOrigin === "user_created") { - this.topics.settleDelete(sessionId, record.topicId); + // Phase 1: drop the record but keep the topic id quarantined. Routes are + // only republished by `commitSettledDelete` once the clear is durable. + const settled = this.topics.settleDelete(sessionId, record.topicId, dispatchedAuthorityEpoch); + if (!settled) { + await this.#persistTopicsWithRetry().catch(() => undefined); + return "post_dispatch_pending"; + } for (const k of [...this.liveMessages.keys()]) if (k.startsWith(`${sessionId}:`)) { this.liveMessages.delete(k); @@ -7025,9 +7064,10 @@ export class TelegramNotificationDaemon { this.pendingThreadedFrames.delete(sessionId); try { await this.persistTopics(); + this.topics.commitSettledDelete(settled); return "settled"; } catch { - this.topics.restoreDeleteFence(deleteSnapshot); + this.topics.rollbackSettledDelete(settled); await this.#persistTopicsWithRetry().catch(() => undefined); return "post_dispatch_pending"; } @@ -7037,7 +7077,11 @@ export class TelegramNotificationDaemon { message_thread_id: Number(record.topicId), })) as { ok?: boolean }; if (!topicDeleteSettled(res)) return "post_dispatch_pending"; - this.topics.settleDelete(sessionId, record.topicId); + const settled = this.topics.settleDelete(sessionId, record.topicId, dispatchedAuthorityEpoch); + if (!settled) { + await this.#persistTopicsWithRetry().catch(() => undefined); + return "post_dispatch_pending"; + } for (const k of [...this.liveMessages.keys()]) if (k.startsWith(`${sessionId}:`)) { this.liveMessages.delete(k); @@ -7049,9 +7093,10 @@ export class TelegramNotificationDaemon { this.pendingThreadedFrames.delete(sessionId); try { await this.persistTopics(); + this.topics.commitSettledDelete(settled); return "settled"; } catch { - this.topics.restoreDeleteFence(deleteSnapshot); + this.topics.rollbackSettledDelete(settled); await this.#persistTopicsWithRetry().catch(() => undefined); return "post_dispatch_pending"; } diff --git a/packages/coding-agent/src/sdk/bus/topic-registry.ts b/packages/coding-agent/src/sdk/bus/topic-registry.ts index d18ecdc413..319c100c07 100644 --- a/packages/coding-agent/src/sdk/bus/topic-registry.ts +++ b/packages/coding-agent/src/sdk/bus/topic-registry.ts @@ -94,6 +94,18 @@ export interface TopicDeleteAuthoritySnapshot { record?: TopicRecord; } +/** + * Proof that a definite delete was settled in memory and still owes its durable + * commit. Handed back only by a settlement that actually happened, so a refused + * settlement structurally cannot be followed by a rollback. + */ +export interface TopicSettledDelete { + sessionId: string; + topicId: string; + /** Authority epoch proven current when the record was removed. */ + settledEpoch: number; +} + function isValidBindingString(value: unknown): value is string { return typeof value === "string" && value.trim().length > 0; } @@ -136,6 +148,16 @@ function isValidTopicId(value: unknown): value is string { ); } +/** + * Monotonic epoch successor that saturates instead of leaving the safe-integer + * range. Past `Number.MAX_SAFE_INTEGER` two distinct generations collapse onto + * the same IEEE-754 double, which would let a stale settlement clear a newer + * fence; settlement refuses the saturated value instead of overflowing past it. + */ +function nextAuthorityEpoch(current: number): number { + return current >= Number.MAX_SAFE_INTEGER ? Number.MAX_SAFE_INTEGER : current + 1; +} + export function emptyTopicRegistryState(): TopicRegistryState { return { topics: {} }; } @@ -161,6 +183,8 @@ export class TopicRegistry { private readonly creatingBindings = new Map(); /** Monotonic authority epochs, including deletion fences for absent records. */ private readonly epochs = new Map(); + /** Phase-1 settlements awaiting their durable clear; their topic ids stay quarantined. */ + readonly #settling = new Map(); constructor(state: TopicRegistryState = emptyTopicRegistryState()) { this.topics = new Map(); @@ -173,6 +197,7 @@ export class TopicRegistry { this.byTopic.clear(); this.#ambiguousTopicIds.clear(); this.epochs.clear(); + this.#settling.clear(); this.load(state); } @@ -259,6 +284,10 @@ export class TopicRegistry { private rebuildInboundRoutes(): void { this.byTopic.clear(); this.#ambiguousTopicIds.clear(); + // A settled-but-not-yet-durable clear keeps its topic id quarantined: the + // clear is not authoritative until persisted, so no colliding survivor may + // become routable and no settled id may become adoptable during that write. + for (const settling of this.#settling.values()) this.#ambiguousTopicIds.add(settling.topicId); const activeByTopic = new Map(); for (const [sessionId, record] of this.topics) { @@ -278,6 +307,13 @@ export class TopicRegistry { } } + /** Advance and publish the session authority epoch, saturating at the safe-integer bound. */ + #advanceAuthorityEpoch(sessionId: string, base: number): number { + const epoch = nextAuthorityEpoch(base); + this.epochs.set(sessionId, epoch); + return epoch; + } + /** Resolve the owning session for a topic id (for fail-closed inbound routing). */ sessionForTopic(topicId: string): string | undefined { return this.byTopic.get(topicId); @@ -676,7 +712,7 @@ export class TopicRegistry { /** Restore a failed delete fence only while its exact authority mutation remains current. */ restoreDeleteAuthority(snapshot: TopicDeleteAuthoritySnapshot): boolean { const record = this.topics.get(snapshot.sessionId); - const deleteEpoch = Math.max(snapshot.fenceEpoch ?? 0, snapshot.authorityEpoch ?? 0) + 1; + const deleteEpoch = nextAuthorityEpoch(Math.max(snapshot.fenceEpoch ?? 0, snapshot.authorityEpoch ?? 0)); if (this.epochs.get(snapshot.sessionId) !== deleteEpoch) return false; if (snapshot.topicId === undefined) { if (record) return false; @@ -697,10 +733,17 @@ export class TopicRegistry { return true; } - /** Restore the exact delete fence after a failed compensation publication. */ + /** + * Restore the exact delete fence after a failed compensation publication. + * + * Settlement rebuilds derived routes, which can make a surviving colliding + * record routable. Rebuild again on successful restoration so the reinstated + * fence re-quarantines the topic id instead of leaving inbound routing open to + * the collision partner despite the restored fence. + */ restoreDeleteFence(snapshot: TopicDeleteAuthoritySnapshot): boolean { const record = this.topics.get(snapshot.sessionId); - const deleteEpoch = Math.max(snapshot.fenceEpoch ?? 0, snapshot.authorityEpoch ?? 0) + 1; + const deleteEpoch = nextAuthorityEpoch(Math.max(snapshot.fenceEpoch ?? 0, snapshot.authorityEpoch ?? 0)); if (snapshot.topicId === undefined) { if (record) return false; } else if (!record) { @@ -715,17 +758,19 @@ export class TopicRegistry { } else { record.authorityEpoch = deleteEpoch; record.authorityState = "delete_pending"; - if (this.byTopic.get(record.topicId) === snapshot.sessionId) this.byTopic.delete(record.topicId); } this.epochs.set(snapshot.sessionId, deleteEpoch); + this.rebuildInboundRoutes(); return true; } /** Fence new work before the remote delete starts, including an absent in-flight create. */ beginDelete(sessionId: string): TopicRecord | undefined { const record = this.topics.get(sessionId); - const epoch = Math.max(this.epochs.get(sessionId) ?? 0, record?.authorityEpoch ?? 0) + 1; - this.epochs.set(sessionId, epoch); + const epoch = this.#advanceAuthorityEpoch( + sessionId, + Math.max(this.epochs.get(sessionId) ?? 0, record?.authorityEpoch ?? 0), + ); if (!record) return undefined; record.authorityEpoch = epoch; record.authorityState = "delete_pending"; @@ -799,11 +844,86 @@ export class TopicRegistry { await this.inflight.get(sessionId)?.catch(() => undefined); } - /** Remove only after a definite remote deletion; ambiguity deliberately retains its fence. */ - settleDelete(sessionId: string, topicId: string): boolean { + /** + * Phase 1 of settling a definite remote delete: remove the record while + * deliberately RETAINING its topic-id quarantine. + * + * `dispatchedAuthorityEpoch` is the authority epoch the caller held when it + * dispatched the remote delete. Settlement requires that epoch to still equal + * both the record's own authority epoch and the registry's current epoch for + * the session, so a held earlier delete can never settle a newer fence: if a + * scan or close-started delete re-fenced the same session/topic after this + * delete was dispatched, the stale definite result is refused and the newer + * `delete_pending` record plus its topic-id quarantine stay intact. + * + * An epoch outside the safe-integer range cannot identify a generation, and at + * `Number.MAX_SAFE_INTEGER` epoch advancement has saturated so distinct + * generations are no longer distinguishable. Both fail closed: settlement can + * no longer be proven fresh, so the fence is kept. + * + * Derived routing tables are deliberately NOT rebuilt here. Until the cleared + * snapshot is durably persisted the clear is not authoritative, so publishing + * routes now would make a colliding survivor routable (and the settled id + * adoptable) during the held write, and a failed persist would re-quarantine + * too late. Publish with {@link commitSettledDelete} once the persist + * succeeds, or undo with {@link rollbackSettledDelete} when it fails. + */ + settleDelete(sessionId: string, topicId: string, dispatchedAuthorityEpoch: number): TopicSettledDelete | undefined { + if (!Number.isSafeInteger(dispatchedAuthorityEpoch) || dispatchedAuthorityEpoch < 0) return undefined; + if (this.#settling.has(sessionId)) return undefined; const record = this.topics.get(sessionId); - if (!record || record.topicId !== topicId || record.authorityState !== "delete_pending") return false; + if (!record || record.topicId !== topicId || record.authorityState !== "delete_pending") return undefined; + if ((record.authorityEpoch ?? 0) !== dispatchedAuthorityEpoch) return undefined; + const currentEpoch = this.authorityEpoch(sessionId); + if (currentEpoch !== dispatchedAuthorityEpoch) return undefined; + if (currentEpoch >= Number.MAX_SAFE_INTEGER) return undefined; this.topics.delete(sessionId); + this.#settling.set(sessionId, { topicId, settledEpoch: dispatchedAuthorityEpoch, record: { ...record } }); + this.#ambiguousTopicIds.add(topicId); + if (this.byTopic.get(topicId) === sessionId) this.byTopic.delete(topicId); + return { sessionId, topicId, settledEpoch: dispatchedAuthorityEpoch }; + } + + /** + * Phase 2: publish derived routing tables once the cleared state is durable. + * + * Only here does the settled topic id lose its quarantine, so a surviving + * colliding record becomes routable and a settled id becomes adoptable without + * waiting for a daemon restart. + */ + commitSettledDelete(settled: TopicSettledDelete): boolean { + const pending = this.#settling.get(settled.sessionId); + if (!pending || pending.topicId !== settled.topicId || pending.settledEpoch !== settled.settledEpoch) + return false; + this.#settling.delete(settled.sessionId); + this.rebuildInboundRoutes(); + return true; + } + + /** + * Compare-and-swap undo for a settlement whose durable clear failed. + * + * Restoration applies only while the registry still holds exactly the state + * that this settlement produced: the settlement is still awaiting commit, the + * session still has no record, and the session epoch is still the settled + * epoch. Any mismatch means a newer generation intervened, so the restore is + * refused and the newer state is left untouched. A refused settlement produced + * no token, so it can never reach this path. + */ + rollbackSettledDelete(settled: TopicSettledDelete): boolean { + const pending = this.#settling.get(settled.sessionId); + if (!pending || pending.topicId !== settled.topicId || pending.settledEpoch !== settled.settledEpoch) + return false; + if (this.topics.has(settled.sessionId)) return false; + if (this.authorityEpoch(settled.sessionId) !== settled.settledEpoch) return false; + this.#settling.delete(settled.sessionId); + this.topics.set(settled.sessionId, { + ...pending.record, + authorityEpoch: settled.settledEpoch, + authorityState: "delete_pending", + }); + this.epochs.set(settled.sessionId, settled.settledEpoch); + this.rebuildInboundRoutes(); return true; } @@ -811,7 +931,7 @@ export class TopicRegistry { delete(sessionId: string): boolean { const record = this.topics.get(sessionId); if (!record) return false; - this.epochs.set(sessionId, Math.max(this.epochs.get(sessionId) ?? 0, record.authorityEpoch ?? 0) + 1); + this.#advanceAuthorityEpoch(sessionId, Math.max(this.epochs.get(sessionId) ?? 0, record.authorityEpoch ?? 0)); if (this.byTopic.get(record.topicId) === sessionId) this.byTopic.delete(record.topicId); return this.topics.delete(sessionId); } diff --git a/packages/coding-agent/test/notifications-telegram-daemon.test.ts b/packages/coding-agent/test/notifications-telegram-daemon.test.ts index c6c83a9b13..5386ae8717 100644 --- a/packages/coding-agent/test/notifications-telegram-daemon.test.ts +++ b/packages/coding-agent/test/notifications-telegram-daemon.test.ts @@ -11253,6 +11253,54 @@ test("session_closed revokes persisted ask aliases and pending replies before se releaseDelete.resolve(); await close; }); +test("a concurrent delete re-fence keeps a definite remote delete under durable supervision", async () => { + FakeWs.instances = []; + const agentDir = tempAgentDir(); + const bot = new FakeBotApi(); + const deleteStarted = Promise.withResolvers(); + const releaseDelete = Promise.withResolvers(); + const call = bot.call.bind(bot); + bot.call = async (method, body, options) => { + if (method === "deleteForumTopic") { + bot.calls.push({ method, body, options }); + deleteStarted.resolve(); + await releaseDelete.promise; + return { ok: true, result: true }; + } + return call(method, body, options); + }; + const daemon = new TelegramNotificationDaemon({ + settings: settings(agentDir), + ownerId: "owner", + botToken: "tok", + chatId: "42", + botApi: bot, + WebSocketImpl: FakeWs as any, + rich: { enabled: false }, + }); + daemon.connectSession("S", "ws://s", "token"); + await daemon.handleSessionMessage(daemon.sessions.get("S")!, { + type: "action_needed", + kind: "ask", + id: "ask", + question: "Continue?", + options: ["yes"], + }); + const topicId = String((daemon as any).topics.get("S").topicId); + + const deleting = (daemon as any).deleteTopic("S"); + await deleteStarted.promise; + (daemon as any).topics.beginDelete("S"); + releaseDelete.resolve(); + + await expect(deleting).resolves.toBe("post_dispatch_pending"); + expect((daemon as any).topics.get("S")).toMatchObject({ + topicId, + authorityState: "delete_pending", + }); + const persisted = JSON.parse(fs.readFileSync(path.join(daemonPaths(agentDir).dir, "telegram-topics.json"), "utf8")); + expect(persisted.topics.S).toMatchObject({ topicId, authorityState: "delete_pending" }); +}); test("closing endpoint stays fenced after delete settlement until final persistence and teardown", async () => { FakeWs.instances = []; const agentDir = tempAgentDir(); diff --git a/packages/coding-agent/test/notifications-topic-settle-fence-epoch.test.ts b/packages/coding-agent/test/notifications-topic-settle-fence-epoch.test.ts new file mode 100644 index 0000000000..1ee8174b7a --- /dev/null +++ b/packages/coding-agent/test/notifications-topic-settle-fence-epoch.test.ts @@ -0,0 +1,225 @@ +import { describe, expect, test } from "bun:test"; +import { TopicRegistry, type TopicRegistryState, type TopicSettledDelete } from "../src/sdk/bus/topic-registry"; + +const binding = (sessionId: string) => ({ + chatId: "42", + endpointKey: `ws://${sessionId}`, + endpointDigest: `digest-${sessionId}`, + endpointGeneration: 1, +}); + +/** A persisted record with a complete endpoint binding (pre-binding records are retired on load). */ +const boundRecord = (sessionId: string, topicId: string, authorityEpoch: number, fenced: boolean) => ({ + topicId, + identitySent: false, + createdAt: 1, + authorityEpoch, + ...binding(sessionId), + ...(fenced ? { authorityState: "delete_pending" as const } : {}), +}); + +/** Narrow an accepted phase-1 settlement without weakening the refusal contract. */ +const requireSettled = (settled: TopicSettledDelete | undefined): TopicSettledDelete => { + if (!settled) throw new Error("expected the settlement to be accepted"); + return settled; +}; + +describe("TopicRegistry delete settlement fencing", () => { + test("a settled delete releases the topic-id quarantine so a re-adopted topic routes inbound", async () => { + const state: TopicRegistryState = { + topics: { A: boundRecord("A", "42", 1, true) }, + fences: { A: 1 }, + }; + const reg = new TopicRegistry(state); + + // The delete-pending record quarantines its topic id: not routable, not adoptable. + expect(reg.sessionForTopic("42")).toBeUndefined(); + expect(reg.isTopicIdAvailable("42")).toBe(false); + + const settled = requireSettled(reg.settleDelete("A", "42", reg.authorityEpoch("A"))); + expect(reg.commitSettledDelete(settled)).toBe(true); + + // Once the record is gone and its clear is durable, its topic id no longer + // collides, so it becomes adoptable and routable without a daemon restart. + expect(reg.get("A")).toBeUndefined(); + expect(reg.isTopicIdAvailable("42")).toBe(true); + await reg.getOrCreateTopic( + "B", + async () => "42", + () => 2, + undefined, + binding("B"), + ); + expect(reg.sessionForTopic("42")).toBe("B"); + }); + + test("a stale E1 settlement cannot settle the newer E2 delete fence for the same session and topic", async () => { + const reg = new TopicRegistry(); + await reg.getOrCreateTopic( + "A", + async () => "42", + () => 1, + undefined, + binding("A"), + ); + + // E1 fences the session and dispatches its remote delete under this epoch. + reg.beginDelete("A"); + const dispatchedEpochE1 = reg.authorityEpoch("A"); + + // Before E1's definite result arrives, a scan/close-started E2 delete + // re-fences the same session and topic, superseding E1's authority. + reg.beginDelete("A"); + const dispatchedEpochE2 = reg.authorityEpoch("A"); + expect(dispatchedEpochE2).toBeGreaterThan(dispatchedEpochE1); + + // E1's definite result must not settle E2's fence. + expect(reg.settleDelete("A", "42", dispatchedEpochE1)).toBeUndefined(); + + // E2's delete_pending record and its quarantine survive intact. + expect(reg.get("A")).toMatchObject({ + topicId: "42", + authorityState: "delete_pending", + authorityEpoch: dispatchedEpochE2, + }); + expect(reg.authorityEpoch("A")).toBe(dispatchedEpochE2); + expect(reg.sessionForTopic("42")).toBeUndefined(); + expect(reg.isTopicIdAvailable("42")).toBe(false); + + // The owning E2 epoch still settles normally. + expect(reg.settleDelete("A", "42", dispatchedEpochE2)).toBeDefined(); + }); + + test("restoring the delete fence after a failed persist re-quarantines a colliding topic id", () => { + // Persisted active+pending collision: B is active on the same topic id that + // delete-pending A still holds, so the id is ambiguous and routes nowhere. + const state: TopicRegistryState = { + topics: { A: boundRecord("A", "42", 1, true), B: boundRecord("B", "42", 0, false) }, + fences: { A: 1 }, + }; + const reg = new TopicRegistry(state); + expect(reg.sessionForTopic("42")).toBeUndefined(); + + const snapshot = reg.captureDeleteAuthority("A"); + const settled = requireSettled(reg.settleDelete("A", "42", reg.authorityEpoch("A"))); + expect(reg.commitSettledDelete(settled)).toBe(true); + + // The committed clear rebuilt derived routes, so the surviving colliding + // record is now routable. + expect(reg.sessionForTopic("42")).toBe("B"); + + // A later close-path publication fails and the delete fence is reinstated. + expect(reg.restoreDeleteFence(snapshot)).toBe(true); + + // The restored fence must re-quarantine the topic id; inbound routing to the + // collision partner must not stay open. + expect(reg.get("A")).toMatchObject({ topicId: "42", authorityState: "delete_pending" }); + expect(reg.sessionForTopic("42")).toBeUndefined(); + expect(reg.isTopicIdAvailable("42")).toBe(false); + }); + + test("authority epochs saturate at the safe-integer bound and a saturated fence refuses settlement", () => { + const max = Number.MAX_SAFE_INTEGER; + const state: TopicRegistryState = { + topics: { A: boundRecord("A", "42", max, false) }, + fences: { A: max }, + }; + const reg = new TopicRegistry(state); + expect(reg.authorityEpoch("A")).toBe(max); + + // Fencing at the bound must not produce MAX_SAFE_INTEGER + 1: that value is + // not a safe integer and compares equal to its own successor, so it could + // never distinguish one delete generation from the next. + expect(reg.beginDelete("A")?.authorityEpoch).toBe(max); + expect(reg.authorityEpoch("A")).toBe(max); + expect(Number.isSafeInteger(reg.authorityEpoch("A"))).toBe(true); + expect(reg.serialize().fences?.A).toBe(max); + + // A saturated epoch can no longer prove exclusive authority, so settlement + // fails closed: the fence and the topic-id quarantine are both retained. + expect(reg.settleDelete("A", "42", max)).toBeUndefined(); + expect(reg.get("A")).toMatchObject({ topicId: "42", authorityState: "delete_pending" }); + expect(reg.isTopicIdAvailable("42")).toBe(false); + + // Dispatched epochs that are not non-negative safe integers are rejected + // outright rather than compared numerically. + expect(reg.settleDelete("A", "42", max + 1)).toBeUndefined(); + expect(reg.settleDelete("A", "42", -1)).toBeUndefined(); + expect(reg.settleDelete("A", "42", Number.NaN)).toBeUndefined(); + expect(reg.settleDelete("A", "42", 1.5)).toBeUndefined(); + }); + + test("a rollback refuses any settlement whose post-settlement state no longer holds", async () => { + const reg = new TopicRegistry({ topics: { A: boundRecord("A", "42", 1, true) }, fences: { A: 1 } }); + const settled = requireSettled(reg.settleDelete("A", "42", reg.authorityEpoch("A"))); + + // A concurrent re-fence advances the session epoch past the settlement, so + // the settled state is no longer the state a rollback would be undoing. + reg.beginDelete("A"); + expect(reg.authorityEpoch("A")).toBe(settled.settledEpoch + 1); + + expect(reg.rollbackSettledDelete(settled)).toBe(false); + // The stale record must not resurrect and must not clobber the newer fence. + expect(reg.get("A")).toBeUndefined(); + expect(reg.authorityEpoch("A")).toBe(settled.settledEpoch + 1); + // Fail closed: the clear is still unpublished, so the id stays quarantined. + expect(reg.isTopicIdAvailable("42")).toBe(false); + + // A record recreated for the same session while the clear is still in flight + // is likewise not the post-settlement state a rollback may undo. + const reg2 = new TopicRegistry({ topics: { A: boundRecord("A", "42", 1, true) }, fences: { A: 1 } }); + const settled2 = requireSettled(reg2.settleDelete("A", "42", reg2.authorityEpoch("A"))); + await reg2.getOrCreateTopic( + "A", + async () => "43", + () => 2, + undefined, + binding("A"), + ); + expect(reg2.authorityEpoch("A")).toBe(settled2.settledEpoch); + expect(reg2.rollbackSettledDelete(settled2)).toBe(false); + expect(reg2.get("A")).toMatchObject({ topicId: "43" }); + expect(reg2.get("A")?.authorityState).toBeUndefined(); + }); + + test("a settled delete keeps its topic id quarantined until the clear is durable", async () => { + const reg = new TopicRegistry({ topics: { A: boundRecord("A", "42", 1, true) }, fences: { A: 1 } }); + const settled = requireSettled(reg.settleDelete("A", "42", reg.authorityEpoch("A"))); + + // Phase 1 drops the record but must not publish routes: the clear lives only + // in memory, so the id is neither adoptable nor routable during the write. + expect(reg.get("A")).toBeUndefined(); + expect(reg.isTopicIdAvailable("42")).toBe(false); + expect(reg.sessionForTopic("42")).toBeUndefined(); + + // An adopt racing the held write is admitted as a record but stays unrouted, + // so nothing is delivered against a clear that may still roll back. + await reg.getOrCreateTopic( + "B", + async () => "42", + () => 2, + undefined, + binding("B"), + ); + expect(reg.sessionForTopic("42")).toBeUndefined(); + + // Phase 2 publishes routes only once the clear is durable. + expect(reg.commitSettledDelete(settled)).toBe(true); + expect(reg.sessionForTopic("42")).toBe("B"); + expect(reg.isTopicIdAvailable("42")).toBe(false); + }); + + test("a refused settlement yields no rollback token, so it cannot restore anything", () => { + // The persisted fence is newer than the record's own authority, so this + // dispatched epoch never owned the fence and settlement must be refused. + const reg = new TopicRegistry({ topics: { A: boundRecord("A", "42", 1, true) }, fences: { A: 2 } }); + expect(reg.settleDelete("A", "42", 1)).toBeUndefined(); + + // Refusal is total: fence, record and quarantine are intact, and no token + // exists for any caller to hand back to a rollback. + expect(reg.get("A")).toMatchObject({ topicId: "42", authorityState: "delete_pending" }); + expect(reg.authorityEpoch("A")).toBe(2); + expect(reg.isTopicIdAvailable("42")).toBe(false); + expect(reg.sessionForTopic("42")).toBeUndefined(); + }); +});