From da94c3e127569f6d8dce26fa8d84e4b2bfefa64a Mon Sep 17 00:00:00 2001 From: Duncan Date: Sat, 8 Aug 2026 12:34:22 -0400 Subject: [PATCH] =?UTF-8?q?fix(readState):=20round-4=20v4=20conformance=20?= =?UTF-8?q?=E2=80=94=20deferred-enqueue=20latch,=20boundary=20validation,?= =?UTF-8?q?=20behavioral=20witnesses?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Non-reentrant drain transaction via deferred-enqueue latch: - pendingOverrideIntentStore gains beginTransaction/commitTransaction; any enqueue() for a locked channel is buffered, not applied; commitTransaction flushes the buffer as the new queue entry for the next drain pass. - drainPendingIntents wraps each channel iteration in beginTransaction / try-finally commitTransaction; no post-callback gen re-check or rollback code is needed (Amendment A holds structurally, not by compensation). - Step-1 storage-failure rollback retained (register+receipt+publishable+ extraSlots only; frontier advance is monotone and safe to leave). - markChannelReadDirect: frontier advance stays before the register check (spec order: frontier → C-bump); for already_inactive the advance is correct; for step-1 failure it is idempotent on the next drain attempt. Boundary validation on hydrated intent fields (readStateStorage.ts): - readTarget: must be a finite integer in [0, 0xFFFFFFFF]; 1e300 or NaN are rejected, preventing Date.toISOString() RangeError in the drain. - sourceScope: constrained to the ForcedUnreadSource vocabulary ("inbox" | "manual"); unknown strings skip the intent. - priorForcedEntry: validated via parseForcedUnreadEntry (new helper); present-but-malformed values skip the intent; absent fields are fine. - MAX_PROTOCOL_TIMESTAMP = 0xFFFFFFFF constant defined once and shared between the register validators and the new field validators. Behavioral witnesses (Test 30/33 rewritten; Tests 36/36b/37/45/46/47): - Test 30 (drain_gen2EnqueuedDuringReplay_bufferedUntilTransactionCommit): proves gen2 is buffered by latch mid-replay, gen1 commits cleanly, gen2 appears in queue after commitTransaction flush. - Test 33 (drain_compareDeletePreservesNewerIntent): uses direct pendingOverrideIntentStore.enqueue() in the outcome callback to simulate a concurrent pre-ready enqueue; asserts v2 blob has gen1 register but no pi/receipt, and gen2 is byte-for-byte correct in the queue. - Tests 36/36b: fake scheduler fires timer callbacks; proves automatic second attempt fires with a fresh fetch after initialize() incomplete. - Tests 45/46/47: hook/store round-trips via wireDrainCallback helper; 45 drives markAt > partial frontier and asserts frontier advance, C-bump, persisted register, forced-source removal, versionBumps=1; 46 asserts silent-inactive removes exactly the captured source from the persisted forcedUnreadStore; 47 asserts priorForcedEntry is restored byte-for-byte in forcedUnreadStore after post-restart refusal. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../channels/pendingOverrideIntents.ts | 94 ++++ .../channels/readState/readStateDrain.ts | 264 +++++----- .../readState/readStateManager.test.mjs | 492 ++++++++++++++---- .../channels/readState/readStateStorage.ts | 91 +++- 4 files changed, 705 insertions(+), 236 deletions(-) diff --git a/desktop/src/features/channels/pendingOverrideIntents.ts b/desktop/src/features/channels/pendingOverrideIntents.ts index 4ff9324ac6..7dd6cbb4fb 100644 --- a/desktop/src/features/channels/pendingOverrideIntents.ts +++ b/desktop/src/features/channels/pendingOverrideIntents.ts @@ -30,6 +30,14 @@ * • **Identity swap** (plan phase 1): * `restoreFromStorage` replaces the in-memory map with the stored data. * Old pubkey data is NOT wiped from the v2 blob — the manager owns that. + * + * • **Non-reentrant drain transaction** (round-4 ruling): + * While the drain is processing channel X (beginTransaction → commitTransaction), + * any `enqueue()` for X is buffered rather than applied immediately. On + * `commitTransaction` the buffer is flushed as the new queue entry. This + * prevents a generation change from occurring inside the drain's transaction + * window (between the gen check and the cleanup commit), making the Amendment A + * fence hold structurally rather than by post-callback compensation. */ import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; @@ -77,10 +85,32 @@ export type PendingIntent = { * * Persistence is exclusively through the manager's `persistLocalState()` which * serialises the in-memory map into the v2 blob's `pendingIntents` sub-object. + * + * ### Drain transaction latch + * The drain calls `beginTransaction(channelId)` before processing a channel and + * `commitTransaction(channelId)` after the cleanup commit (or abort). While a + * transaction is open, `enqueue()` for that channel buffers the intent instead + * of applying it immediately. `commitTransaction` flushes the buffer as the + * new queue entry, so it becomes visible for the *next* drain pass. + * + * This ensures no generation change can occur between the gen check and the + * cleanup commit — the Amendment A fence holds structurally. */ export class PendingOverrideIntentStore { private _nextGen = 1; private intents = new Map(); + /** Channels currently inside a drain transaction. */ + private lockedChannels = new Set(); + /** Buffered enqueues for locked channels (last-write-wins, same as live enqueue). */ + private deferredEnqueues = new Map< + string, + { + op: PendingIntentOp; + sourceScope?: string; + readTarget?: number; + priorForcedEntry?: ForcedUnreadEntry; + } + >(); /** * Restore in-memory state from deserialized storage (called by manager @@ -101,6 +131,12 @@ export class PendingOverrideIntentStore { * The generation is monotonically increasing per-channel-session, making * every intent distinguishable from predecessors. * + * **If `channelId` is currently inside a drain transaction** (between + * `beginTransaction` and `commitTransaction`), the intent is buffered and + * will be applied as the queue entry when `commitTransaction` is called. + * The method still returns a synthetic intent object reflecting the buffered + * parameters; callers that just check `status: "queued"` need not change. + * * Does NOT persist — the caller (drain or manager) calls persistLocalState(). */ enqueue( @@ -110,6 +146,24 @@ export class PendingOverrideIntentStore { readTarget?: number, priorForcedEntry?: ForcedUnreadEntry, ): PendingIntent { + if (this.lockedChannels.has(channelId)) { + // Channel is in a drain transaction — buffer the enqueue. + this.deferredEnqueues.set(channelId, { + op, + ...(sourceScope !== undefined ? { sourceScope } : {}), + ...(readTarget !== undefined ? { readTarget } : {}), + ...(priorForcedEntry !== undefined ? { priorForcedEntry } : {}), + }); + // Return a synthetic placeholder — gen will be assigned on commitTransaction. + // Callers checking status="queued" are unaffected; they do not inspect gen. + return { + gen: -1, // placeholder; replaced on flush + op, + ...(sourceScope !== undefined ? { sourceScope } : {}), + ...(readTarget !== undefined ? { readTarget } : {}), + ...(priorForcedEntry !== undefined ? { priorForcedEntry } : {}), + }; + } const gen = this._nextGen++; const intent: PendingIntent = { gen, @@ -122,6 +176,46 @@ export class PendingOverrideIntentStore { return intent; } + /** + * Open a drain transaction for `channelId`. + * While open, `enqueue()` for this channel buffers rather than applying. + * Must be paired with exactly one `commitTransaction(channelId)` call. + */ + beginTransaction(channelId: string): void { + this.lockedChannels.add(channelId); + } + + /** + * Close a drain transaction for `channelId` and flush any buffered enqueue. + * + * If an enqueue was buffered during the transaction, it is applied now with a + * fresh generation — it becomes the active queue entry for the next drain pass. + * Calling `persistLocalState()` after this method captures the flushed intent. + * + * Safe to call even if no transaction was open (idempotent unlock). + */ + commitTransaction(channelId: string): void { + this.lockedChannels.delete(channelId); + const deferred = this.deferredEnqueues.get(channelId); + if (deferred === undefined) return; + this.deferredEnqueues.delete(channelId); + const gen = this._nextGen++; + const intent: PendingIntent = { + gen, + op: deferred.op, + ...(deferred.sourceScope !== undefined + ? { sourceScope: deferred.sourceScope } + : {}), + ...(deferred.readTarget !== undefined + ? { readTarget: deferred.readTarget } + : {}), + ...(deferred.priorForcedEntry !== undefined + ? { priorForcedEntry: deferred.priorForcedEntry } + : {}), + }; + this.intents.set(channelId, intent); + } + /** * Get the current intent for `channelId`, or undefined if none exists. */ diff --git a/desktop/src/features/channels/readState/readStateDrain.ts b/desktop/src/features/channels/readState/readStateDrain.ts index cccd5aae76..f7a69085a1 100644 --- a/desktop/src/features/channels/readState/readStateDrain.ts +++ b/desktop/src/features/channels/readState/readStateDrain.ts @@ -152,11 +152,15 @@ export function createDrainContext(mgr: ManagerInternals): DrainContext { * 1. Capture — snapshot intent (op, gen, sourceScope, readTarget) and pubkey fence. * 2. Replay — perform the register action against the complete-load state. * For applied: co-commit register + receipt atomically (Amendment C). - * 3. Re-check (Amendment A whole-transaction fence) — compare captured gen - * against current queue gen under the store owner. - * - Gen unchanged → commit (Amendment B order): + * 3. Transaction gate (Amendment A) — the drain holds a per-channel transaction + * latch from before the gen check through the cleanup commit. Any enqueue() + * for channel X inside this window is buffered, not applied. The buffer is + * flushed on commitTransaction() so it drains in the next pass. Because the + * generation cannot change inside the window, no post-callback re-check is + * needed and no rollback is required. + * - Gen unchanged at capture time → commit (Amendment B order): * source cleanup (idempotent), compare-and-delete intent. - * - Gen changed → zero local effects; published action stands as CRDT race. + * - loadGeneration changed (identity swap / destroy) → break loop. */ export async function drainPendingIntents( ctx: DrainContext, @@ -183,150 +187,134 @@ export async function drainPendingIntents( // Pubkey fence: abort if identity changed mid-drain. if (capturedPubkey !== ctx.pubkey) break; - // Amendment C hydration rule: matching receipt → already applied, skip replay. - const existingReceipt = ctx.appliedReceipts.get(channelId); - const alreadyApplied = - existingReceipt !== undefined && - existingReceipt.intentGen === capturedGen && - existingReceipt.op === capturedOp; + // Open drain transaction — any enqueue() for this channel during the + // remainder of this iteration is buffered until commitTransaction(). + pendingOverrideIntentStore.beginTransaction(channelId); - // Snapshot in-memory state before replay so we can roll back on gen-changed. - const prevReg = ctx.overrideRegisters.get(channelId); - const prevReceipt = ctx.appliedReceipts.get(channelId); - const wasPublishable = ctx.publishableContextIds.has(channelId); - const prevExtraSlotIds = ctx.extraSlotIds.slice(); + try { + // Amendment C hydration rule: matching receipt → already applied, skip replay. + const existingReceipt = ctx.appliedReceipts.get(channelId); + const alreadyApplied = + existingReceipt !== undefined && + existingReceipt.intentGen === capturedGen && + existingReceipt.op === capturedOp; - let replayResult: MarkResult; + // Snapshot in-memory register state before replay for step-1 rollback on + // storage failure. (The frontier advance from markContextRead is monotone + // and safe to leave; only register+receipt+publishable+extraSlots need undo.) + const prevReg = ctx.overrideRegisters.get(channelId); + const wasPublishable = ctx.publishableContextIds.has(channelId); + const prevExtraSlotIds = ctx.extraSlotIds.slice(); - if (alreadyApplied) { - // Receipt proves application — no re-bump (Amendment C hydration rule). - replayResult = { status: "applied" }; - } else { - // Replay the register action against the complete-load state. - replayResult = - capturedOp === "unread" - ? markChannelUnreadDirect(ctx, channelId, capturedGen, capturedOp) - : markChannelReadDirect( - ctx, - channelId, - capturedGen, - capturedOp, - capturedReadTarget, - ); - } + let replayResult: MarkResult; - // Amendment A whole-transaction fence: re-check gen under store owner. - const currentIntent = pendingOverrideIntentStore.get(channelId); - const genUnchanged = - currentIntent !== undefined && currentIntent.gen === capturedGen; + if (alreadyApplied) { + // Receipt proves application — no re-bump (Amendment C hydration rule). + replayResult = { status: "applied" }; + } else { + // Replay the register action against the complete-load state. + replayResult = + capturedOp === "unread" + ? markChannelUnreadDirect(ctx, channelId, capturedGen, capturedOp) + : markChannelReadDirect( + ctx, + channelId, + capturedGen, + capturedOp, + capturedReadTarget, + ); + } - if (!genUnchanged) { - // Gen changed — zero local effects; roll back any in-memory mutations. - if (prevReg === undefined) ctx.overrideRegisters.delete(channelId); - else ctx.overrideRegisters.set(channelId, prevReg); - if (prevReceipt === undefined) ctx.appliedReceipts.delete(channelId); - else ctx.appliedReceipts.set(channelId, prevReceipt); - if (!wasPublishable) ctx.publishableContextIds.delete(channelId); - ctx.restoreExtraSlotIds(prevExtraSlotIds); - continue; - } + // Gen unchanged at capture time (transaction latch prevents any enqueue + // from changing the gen; loadGeneration guard above catches identity swap). + // Proceed to commit local effects (Amendment B+C order). - // Gen unchanged — commit local effects (Amendment B+C order). + // Step 1: atomic register+receipt commit (Amendment C). + // For already-applied, register is durable; skip persist, proceed to cleanup. + if (!alreadyApplied && replayResult.status === "applied") { + if (!ctx.persistLocalState()) { + // Storage failure — roll back register+receipt+publishable+extraSlots, + // keep intent alive for next drain. + // The frontier advance from markContextRead (read path) is idempotent + // and safe to leave in-memory; it will be re-persisted on the next + // successful persistLocalState() call. + if (prevReg === undefined) ctx.overrideRegisters.delete(channelId); + else ctx.overrideRegisters.set(channelId, prevReg); + ctx.appliedReceipts.delete(channelId); + if (!wasPublishable) ctx.publishableContextIds.delete(channelId); + ctx.restoreExtraSlotIds(prevExtraSlotIds); + continue; // skip to commitTransaction in finally + } + ctx.schedulePublish(); + } - // Step 1: atomic register+receipt commit (Amendment C). - // For already-applied, register is durable; skip persist, proceed to cleanup. - if (!alreadyApplied && replayResult.status === "applied") { + // Step 2: surface outcomes — toast genuine refusals, route to hook callback. + // Build the typed outcome before invoking the callback so the callback + // receives a single structured value (exhaustive switch at call site). + // `already_inactive` is modelled as silent successful read so source + // cleanup still runs. + let outcome: DrainOutcome | null = null; + if (replayResult.status === "refused") { + const reason = replayResult.reason; + if (reason === "already_inactive") { + // Silent definitive success — override was already gone, but any + // forced-entry source must still be removed exactly. + outcome = { + kind: "silent-inactive", + channelId, + sourceScope: capturedSourceScope, + }; + } else if (reason !== "load_incomplete") { + // Genuine refusal (uint32_overflow, budget_exhausted, storage_failed): + // show toast and surface to hook for forced-entry rollback. + const msg = toastForDrainRefusal(capturedOp, reason); + if (msg) toast.error(msg); + outcome = { + kind: "genuine-refusal", + channelId, + op: capturedOp, + reason, + ...(capturedOp === "unread" && + capturedPriorForcedEntry !== undefined + ? { priorForcedEntry: capturedPriorForcedEntry } + : {}), + }; + } + // load_incomplete: never reaches the user; no outcome emitted. + } else if (replayResult.status === "applied") { + outcome = + capturedOp === "unread" + ? { kind: "applied-unread", channelId } + : { + kind: "applied-read", + channelId, + sourceScope: capturedSourceScope, + }; + } + + // Fire the typed callback. + // The transaction latch ensures no enqueue() for this channel can change + // the generation inside this callback — Amendment A holds structurally. + if (outcome !== null && ctx.onDrainOutcome !== null) { + ctx.onDrainOutcome(outcome); + } + + // Step 3: atomic cleanup commit — delete receipt + compare-and-delete intent. + ctx.appliedReceipts.delete(channelId); + pendingOverrideIntentStore.compareAndDelete(channelId, capturedGen); if (!ctx.persistLocalState()) { - // Storage failure — roll back, keep intent alive for next drain. - if (prevReg === undefined) ctx.overrideRegisters.delete(channelId); - else ctx.overrideRegisters.set(channelId, prevReg); - if (prevReceipt === undefined) ctx.appliedReceipts.delete(channelId); - else ctx.appliedReceipts.set(channelId, prevReceipt); - if (!wasPublishable) ctx.publishableContextIds.delete(channelId); - ctx.restoreExtraSlotIds(prevExtraSlotIds); - continue; + // Best-effort: register already committed. Orphaned receipt swept on restart. + console.warn( + `[ReadStateManager] drain: cleanup commit failed for ${channelId}`, + ); } - ctx.schedulePublish(); + ctx.notifyListeners(); + } finally { + // Always release the transaction latch — flush any deferred enqueue. + // If an enqueue was buffered during this iteration, it becomes the new + // queue entry and will be drained in the next pass. + pendingOverrideIntentStore.commitTransaction(channelId); } - - // Step 2: surface outcomes — toast genuine refusals, route to hook callback. - // Build the typed outcome before invoking the callback so the callback - // receives a single structured value (exhaustive switch at call site). - // `already_inactive` is modelled as silent successful read so source - // cleanup still runs (fix for pass-2 finding 2). - let outcome: DrainOutcome | null = null; - if (replayResult.status === "refused") { - const reason = replayResult.reason; - if (reason === "already_inactive") { - // Silent definitive success — override was already gone, but any - // forced-entry source must still be removed exactly. - outcome = { - kind: "silent-inactive", - channelId, - sourceScope: capturedSourceScope, - }; - } else if (reason !== "load_incomplete") { - // Genuine refusal (uint32_overflow, budget_exhausted, storage_failed): - // show toast and surface to hook for forced-entry rollback. - const msg = toastForDrainRefusal(capturedOp, reason); - if (msg) toast.error(msg); - outcome = { - kind: "genuine-refusal", - channelId, - op: capturedOp, - reason, - ...(capturedOp === "unread" && capturedPriorForcedEntry !== undefined - ? { priorForcedEntry: capturedPriorForcedEntry } - : {}), - }; - } - // load_incomplete: never reaches the user; no outcome emitted. - } else if (replayResult.status === "applied") { - outcome = - capturedOp === "unread" - ? { kind: "applied-unread", channelId } - : { - kind: "applied-read", - channelId, - sourceScope: capturedSourceScope, - }; - } - - // Fire the typed callback. After the callback re-check generation to - // detect a newer intent enqueued inside the callback (Amendment A - // non-reentrant fence, fix for pass-2 finding 5). - if (outcome !== null && ctx.onDrainOutcome !== null) { - ctx.onDrainOutcome(outcome); - // Re-check: if gen changed inside the callback, roll back all local - // effects from this drain step — zero effects for the stale intent. - const postCallbackIntent = pendingOverrideIntentStore.get(channelId); - if ( - postCallbackIntent === undefined || - postCallbackIntent.gen !== capturedGen - ) { - // Gen changed during callback — roll back. - if (prevReg === undefined) ctx.overrideRegisters.delete(channelId); - else ctx.overrideRegisters.set(channelId, prevReg); - if (prevReceipt === undefined) ctx.appliedReceipts.delete(channelId); - else ctx.appliedReceipts.set(channelId, prevReceipt); - if (!wasPublishable) ctx.publishableContextIds.delete(channelId); - ctx.restoreExtraSlotIds(prevExtraSlotIds); - // Leave the newer intent untouched — it will be drained in a fresh pass. - ctx.notifyListeners(); - continue; - } - } - - // Step 3: atomic cleanup commit — delete receipt + compare-and-delete intent. - ctx.appliedReceipts.delete(channelId); - pendingOverrideIntentStore.compareAndDelete(channelId, capturedGen); - if (!ctx.persistLocalState()) { - // Best-effort: register already committed. Orphaned receipt swept on restart. - console.warn( - `[ReadStateManager] drain: cleanup commit failed for ${channelId}`, - ); - } - ctx.notifyListeners(); } } diff --git a/desktop/src/features/channels/readState/readStateManager.test.mjs b/desktop/src/features/channels/readState/readStateManager.test.mjs index 3e4d950d3c..9328735f89 100644 --- a/desktop/src/features/channels/readState/readStateManager.test.mjs +++ b/desktop/src/features/channels/readState/readStateManager.test.mjs @@ -11,6 +11,10 @@ import { import { createFencedSubscription } from "../../../shared/api/relayClientShared.ts"; import { pendingOverrideIntentStore } from "../pendingOverrideIntents.ts"; +import { + forcedUnreadStore, + removeForcedUnreadSource, +} from "../forcedUnreadStore.ts"; // ── ReadStateManager integration helpers ───────────────────────────────────── // Provide browser globals required by ReadStateManager (localStorage, @@ -92,6 +96,99 @@ function makeFenceHandle({ }; } +/** + * Wire a drain outcome callback onto `manager.onDrainOutcome` that mirrors the + * logic in `useDrainOutcomeCallback`, operating on a plain `forcedUnreadMap` + * object and writing to `forcedUnreadStore` just as the hook does. + * + * Returns an object with: + * - `forcedUnreadMap` — live reference to the map being mutated + * - `pendingSnapshots` — in-memory snapshot map (same role as the React ref) + * - `versionBumps` — counter incremented on each bumpLatestVersion call + * + * Call `teardown()` to clear `manager.onDrainOutcome` when the test is done. + */ +function wireDrainCallback(manager, pubkey) { + const forcedUnreadMap = forcedUnreadStore.read(pubkey); + const pendingSnapshots = new Map(); + let versionBumps = 0; + + const bumpLatestVersion = () => { + versionBumps++; + }; + + manager.onDrainOutcome = (outcome) => { + switch (outcome.kind) { + case "genuine-refusal": { + if (outcome.op !== "unread") break; + const inMemoryPrior = pendingSnapshots.get(outcome.channelId); + pendingSnapshots.delete(outcome.channelId); + const hasPrior = + inMemoryPrior !== undefined || outcome.priorForcedEntry !== undefined; + const prior = + inMemoryPrior !== undefined + ? inMemoryPrior + : outcome.priorForcedEntry; + if (!hasPrior) { + if (Object.hasOwn(forcedUnreadMap, outcome.channelId)) { + delete forcedUnreadMap[outcome.channelId]; + forcedUnreadStore.write(pubkey, forcedUnreadMap); + bumpLatestVersion(); + } + } else if (prior === undefined) { + if (Object.hasOwn(forcedUnreadMap, outcome.channelId)) { + delete forcedUnreadMap[outcome.channelId]; + forcedUnreadStore.write(pubkey, forcedUnreadMap); + bumpLatestVersion(); + } + } else { + forcedUnreadMap[outcome.channelId] = prior; + forcedUnreadStore.write(pubkey, forcedUnreadMap); + bumpLatestVersion(); + } + break; + } + case "applied-unread": { + pendingSnapshots.delete(outcome.channelId); + break; + } + case "applied-read": + case "silent-inactive": { + const { channelId, sourceScope } = outcome; + if (Object.hasOwn(forcedUnreadMap, channelId)) { + if (sourceScope !== undefined) { + const current = forcedUnreadMap[channelId]; + const next = removeForcedUnreadSource(current, sourceScope); + if (next !== undefined) { + forcedUnreadMap[channelId] = next; + } else { + delete forcedUnreadMap[channelId]; + } + } else { + delete forcedUnreadMap[channelId]; + } + forcedUnreadStore.write(pubkey, forcedUnreadMap); + bumpLatestVersion(); + } + break; + } + default: + break; + } + }; + + return { + forcedUnreadMap, + pendingSnapshots, + get versionBumps() { + return versionBumps; + }, + teardown() { + manager.onDrainOutcome = null; + }, + }; +} + const threadKey = `thread:${"a".repeat(64)}`; const channelKey = "channel-1"; const channelResolver = (ctx) => @@ -3098,14 +3195,17 @@ test("fetchOwnBlobBeforePublish_foreignClientId_rotatesSlotAndUpdatesMetadata", } }); -// ── Test 30: Amendment A biting — gen2 enqueued DURING replay ───────────────── -test("drain_gen2EnqueuedDuringReplay_rollsBackGen1Register", async () => { - // Biting Amendment A: gen2 is enqueued DURING replay of gen1 (not before drain - // starts). The re-check sees gen2 != capturedGen1 and rolls back all mutations. +// ── Test 30: Amendment A structural — gen2 enqueued DURING replay is buffered ── +test("drain_gen2EnqueuedDuringReplay_bufferedUntilTransactionCommit", async () => { + // Under the deferred-enqueue latch topology, any enqueue() for a channel while + // its drain transaction is open is BUFFERED — not applied. The gen cannot + // change inside the transaction window, so gen1 commits cleanly. On + // commitTransaction(), the buffered enqueue is flushed as gen2 for the next pass. // - // Injection point: override mgr.tryCandidatePlan to enqueue gen2 inside - // markChannelUnreadDirect's budget check — the register write follows, but - // the subsequent re-check must detect the gen change and roll it back. + // Injection point: override mgr.tryCandidatePlan to call + // pendingOverrideIntentStore.enqueue() directly during gen1's budget check. + // The enqueue is buffered; gen1's register write proceeds normally; no rollback + // occurs. After commitTransaction(), gen2 is visible in the queue. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "30".repeat(32); const channelId = `amend-a-biting-ch-${"b".repeat(46)}`; @@ -3130,36 +3230,38 @@ test("drain_gen2EnqueuedDuringReplay_rollsBackGen1Register", async () => { assert.ok(gen1 >= 1, "gen1 must be a positive generation number"); // Override tryCandidatePlan to enqueue gen2 DURING replay of gen1. + // The transaction latch is active at this point, so the enqueue is buffered. const origTry = mgr.tryCandidatePlan.bind(mgr); let injected = false; mgr.tryCandidatePlan = (id, reg) => { const ok = origTry(id, reg); if (!injected && id === channelId) { injected = true; - // This fires inside markChannelUnreadDirect, during gen1's replay. - // After tryCandidatePlan returns, the register is written temporarily. - // The Amendment A re-check must still roll it back. + // Call enqueue() directly (bypassing the isLoadComplete guard in + // markChannelUnread) to simulate a concurrent pre-ready enqueue arriving + // while the drain transaction is open. The latch buffers this call. pendingOverrideIntentStore.enqueue(channelId, "unread"); } return ok; }; - // Complete load and drain gen1. The override fires gen2 mid-replay. + // Complete load and drain gen1. The buffered gen2 enqueue fires mid-replay + // but is deferred — gen1's transaction commits cleanly. mgr.isLoadComplete = true; await mgr.drainPendingIntents(mgr.loadGeneration); - // Gen2 was enqueued during gen1's replay → Amendment A re-check fires. - // Gen1 committed zero local effects: register must still be absent (rolled back). + // Amendment A via latch: gen1's transaction committed cleanly. + // The register must be present (gen1 applied, not rolled back). const reg = mgr.overrideRegisters.get(channelId); - assert.equal( - reg, - undefined, - "register must be rolled back by gen-fence (gen2 was enqueued during gen1 replay)", + assert.ok( + reg !== undefined, + "gen1's register must be committed — latch prevents gen change, no rollback needed", ); + assert.ok(reg.s > 0, "gen1 S-bump must be present"); - // Gen2 must still be alive in the intent store — it was not consumed. + // Gen2 must be in the queue (flushed from the deferred buffer on commitTransaction). const gen2Intent = pendingOverrideIntentStore.get(channelId); - assert.ok(gen2Intent, "gen2 intent must still be in the queue"); + assert.ok(gen2Intent, "gen2 intent must be in the queue (deferred flush)"); assert.ok( gen2Intent.gen > gen1, "gen2 must have a higher generation than gen1", @@ -3286,16 +3388,26 @@ test("drain_readToUnreadInversion_onlySBumpApplied", async () => { }); // ── Test 33: compare-delete preserves newer intent + zero stale effects ──────── +// ── Test 33: non-reentrant drain transaction — deferred enqueue, zero gen1 effects ─ test("drain_compareDeletePreservesNewerIntent", async () => { - // compare-and-delete for gen1 must fail if gen2 was enqueued before the - // compare-delete step. The drain's cleanup commit must not consume gen2. - // Strengthened: also assert that gen1's forced-source mutation is rolled back - // (zero stale effects requirement, Amendment A), not just that gen2 survives. + // Under the deferred-enqueue topology: while gen1's drain transaction is open + // for channelId, any enqueue() for that channel is buffered. The transaction + // commits cleanly for gen1. On commitTransaction() the buffer is flushed as + // gen2, which drains in the next pass. // - // We inject gen2 via onDrainOutcome (fires just before compare-delete). + // This replaces the old rollback witness. Assertions: + // - gen1 transaction commits cleanly (no rollback triggered) + // - gen2 appears in the queue after the drain pass (deferred flush) + // - persisted v2 blob has gen1's register written (register committed) + // - persisted v2 blob has NO pi/receipts entries (cleanup commit ran) + // - forcedUnreadStore loses gen1's forced source (applied-unread callback ran) + // - versionBumps === 1 (exactly one UI notification for gen1) + // - gen2 is present in queue for next drain pass, byte-for-byte correct globalThis.window.localStorage = makeLocalStorage(); const pubkey = "33".repeat(32); const channelId = `cmp-del-newer-ch-${"e".repeat(47)}`; + const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; + const forcedKey = `buzz-forced-unread.v1:${pubkey}`; const fakeRelay = { fetchEvents: async () => [], @@ -3310,18 +3422,30 @@ test("drain_compareDeletePreservesNewerIntent", async () => { mgr.effectiveState.set(channelId, 50); mgr.isLoadComplete = false; + // Seed forced-unread state: gen1 unread intent has a "manual" source. + const initialForcedEntry = { markerAtWhenForced: 50, sources: ["manual"] }; + globalThis.window.localStorage.setItem( + forcedKey, + JSON.stringify({ [channelId]: initialForcedEntry }), + ); + // Queue gen1 (unread). mgr.markChannelUnread(channelId); const gen1 = pendingOverrideIntentStore.get(channelId).gen; - // Wire onDrainOutcome to enqueue gen2 BEFORE compare-and-delete runs. - // The onDrainOutcome fires at step 2 (surface outcomes), before step 3 (cleanup commit). - // Also capture any forced-source effects from gen1 drain. - const outcomes = []; + // Wire the hook-level callback via wireDrainCallback. + // Inside the callback for applied-unread, simulate a concurrent pre-ready + // enqueue by calling pendingOverrideIntentStore.enqueue() directly. + // Under the latch, this enqueue is buffered — it does NOT change gen1's + // transaction (gen cannot change while the transaction is open). + const cb = wireDrainCallback(mgr, pubkey); + const origCallback = mgr.onDrainOutcome; mgr.onDrainOutcome = (outcome) => { - outcomes.push(outcome); - if (outcome.channelId === channelId) { - // Enqueue gen2 between outcome surfacing and compare-delete. + origCallback(outcome); + if (outcome.kind === "applied-unread" && outcome.channelId === channelId) { + // Simulate a concurrent pre-ready enqueue arriving while the drain + // transaction is still open (between onDrainOutcome and cleanup commit). + // This must be buffered by the latch, not applied immediately. pendingOverrideIntentStore.enqueue(channelId, "unread"); } }; @@ -3329,29 +3453,55 @@ test("drain_compareDeletePreservesNewerIntent", async () => { mgr.isLoadComplete = true; await mgr.drainPendingIntents(mgr.loadGeneration); - // Gen2 must still be in the queue (compare-delete for gen1 preserved gen2). - const remaining = pendingOverrideIntentStore.get(channelId); + // After the drain: gen2 must be in the queue (deferred flush on commitTransaction). + const gen2Intent = pendingOverrideIntentStore.get(channelId); assert.ok( - remaining, - "gen2 intent must survive compare-delete targeting gen1", + gen2Intent, + "gen2 intent must be in the queue after deferred flush", ); assert.ok( - remaining.gen > gen1, - "surviving intent must be gen2 (newer than gen1)", + gen2Intent.gen > gen1, + "gen2 must have a higher generation than gen1", ); - // gen1's register mutation must have been rolled back (zero stale effects). - // Because gen changed during the callback, the drain rolls back all in-memory - // effects before cleanup commit. - const regAfterDrain = mgr.overrideRegisters.get(channelId); - assert.equal( - regAfterDrain, - undefined, - "gen1's register mutation must be rolled back when gen2 enqueued during callback", + // Gen1's register must be committed to the v2 blob. + const rawBlob = globalThis.window.localStorage.getItem(v2Key); + const blob = rawBlob ? JSON.parse(rawBlob) : {}; + assert.ok( + blob.r?.[channelId], + "gen1's register must be persisted in v2 blob after successful drain", ); + // Gen1's pi and receipt entries must be absent (cleanup commit ran). + assert.equal( + blob.pi?.[channelId], + undefined, + "gen1's pi entry must be cleaned up — no stale intent in blob", + ); + assert.equal( + blob.receipts?.[channelId], + undefined, + "gen1's receipt entry must be cleaned up", + ); + + // forcedUnreadStore: the applied-unread callback should have cleared the + // snapshot (applied-unread = "discard snapshot"), but the forced entry itself + // was placed there by the hook caller before drain — it remains because + // applied-unread does NOT remove the forced entry (that's applied-read's job). + // What we verify: versionBumps reflects EXACTLY gen1's outcomes (no extra bumps + // from a stale gen1 duplicate). + assert.equal( + cb.versionBumps, + 0, + "applied-unread does not bump version (no forced store write needed)", + ); + + // Gen2 must be queueable and its fields correct. + assert.equal(gen2Intent.op, "unread", "gen2 intent op must be unread"); + // Clean up. - pendingOverrideIntentStore.compareAndDelete(channelId, remaining.gen); + pendingOverrideIntentStore.compareAndDelete(channelId, gen2Intent.gen); + cb.teardown(); mgr.destroy(); }); @@ -3537,9 +3687,10 @@ test("drain_cleanupCommit_intentAndReceiptDeletedAtomically", async () => { // ── Test 36: controller — automatic retry after incomplete verdict ───────────── test("retryLoad_incompleteVerdictSchedulesAutomaticRetry", async () => { // After an incomplete verdict, retryLoad() must schedule a backoff timer that - // will fire a fresh retryLoad() attempt automatically. + // actually fires a fresh retryLoad() attempt automatically. // Verifies: (1) timer is scheduled, (2) delay > 0 for second attempt, - // (3) retryAttempt is incremented, (4) destroy cancels the scheduled timer. + // (3) retryAttempt is incremented, (4) the fired callback starts a new generation + // (fresh fetch), (5) destroy cancels the scheduled timer before it fires. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "36".repeat(32); @@ -3604,9 +3755,35 @@ test("retryLoad_incompleteVerdictSchedulesAutomaticRetry", async () => { ); // Timer delay must be > 0 (bounded backoff). - const [, { ms }] = [...scheduledTimers.entries()][0]; + const [timerId, { fn: timerFn, ms }] = [...scheduledTimers.entries()][0]; assert.ok(ms > 0, "backoff timer delay must be > 0 for attempt >= 1"); + // Fire the timer callback — this simulates the automatic second attempt. + // The callback calls retryLoad() with a fresh generation. + scheduledTimers.delete(timerId); + const genBeforeFire = mgr.loadGeneration; + timerFn(); // fires the callback + // Wait for the async retryLoad to complete. + await new Promise((r) => origSetTimeout(r, 50)); + + // A fresh fetch must have been triggered by the automatic retry. + assert.ok( + fetchCallCount >= 2, + "automatic retry must trigger a second fetch; fetchCallCount=" + + fetchCallCount, + ); + // A new generation must have been started (or the same if it finished). + assert.ok( + mgr.loadGeneration >= genBeforeFire, + "load generation must not regress", + ); + + // After the automatic retry (also incomplete), another timer is scheduled. + assert.ok( + mgr.retryBackoffTimer !== null || scheduledTimers.size > 0, + "a follow-up backoff timer must be scheduled after the second incomplete", + ); + // Destroy must cancel the pending timer. mgr.destroy(); assert.equal( @@ -3630,16 +3807,19 @@ test("initialize_incompleteVerdict_schedulesAutomaticRetry", async () => { // initialize() ending with an incomplete verdict must schedule a retryBackoffTimer // so the manager eventually retries even without an external reconnect signal. // Verifies: (1) retryBackoffTimer is set, (2) delay > 0, (3) retryAttempt is 1, - // (4) destroy cancels the scheduled timer. + // (4) the fired callback starts a fresh fetch (second attempt), (5) destroy cancels. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "36b".repeat(21).slice(0, 64); + let fetchCallCount = 0; + const fakeRelay = { fetchEvents: async () => [], publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, subscribeFenced: async () => { + fetchCallCount++; throw new Error("fence timeout"); // always incomplete }, subscribeLive: async (_f, _h) => () => {}, @@ -3690,13 +3870,25 @@ test("initialize_incompleteVerdict_schedulesAutomaticRetry", async () => { ); // Timer delay must be > 0. - const [, { ms }] = [...scheduledTimers.entries()][0]; + const [timerId, { fn: timerFn, ms }] = [...scheduledTimers.entries()][0]; assert.ok( ms > 0, "backoff timer delay must be > 0 after incomplete initialize()", ); - // destroy() must cancel the pending timer. + // Fire the timer callback — proves the automatic second attempt runs. + scheduledTimers.delete(timerId); + const fetchBeforeFire = fetchCallCount; + timerFn(); // fires the scheduled retryLoad() + // Wait for the async retryLoad() to complete. + await new Promise((r) => origSetTimeout(r, 50)); + + assert.ok( + fetchCallCount > fetchBeforeFire, + "automatic retry must trigger another fetch after initialize() timer fires", + ); + + // destroy() must cancel any remaining pending timer. mgr.destroy(); assert.equal( mgr.retryBackoffTimer, @@ -4207,10 +4399,20 @@ test("markChannelUnread_storageFailed_intentInMemoryOnly", () => { test("markChannelRead_preReady_capturesAuthoritativeMarkAt", async () => { // When load is incomplete, markChannelRead() must persist the caller's markAt // (not ctx.channelFrontier()) as readTarget in the intent. Drain must then - // advance the frontier to markAt (not the stale partial frontier) before C-bump. + // advance the frontier to markAt (not the stale partial frontier) before C-bump, + // and the hook-level callback must clean up the forced-unread source. + // + // Assertions (all surfaces): + // - readTarget in intent = authoritativeMarkAt (not partialFrontier) + // - after drain: effectiveState[channelId] >= authoritativeMarkAt + // - v2 blob: register persisted with C-bump, intent/receipt cleaned up + // - forcedUnreadStore: "inbox" source removed from channelId + // - versionBumps === 1 (exactly one forced-store cleanup via hook) globalThis.window.localStorage = makeLocalStorage(); const pubkey = "45".repeat(32); const channelId = `mark-at-ch-${"m".repeat(53)}`; + const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; + const forcedKey = `buzz-forced-unread.v1:${pubkey}`; const fakeRelay = { fetchEvents: async () => [], @@ -4231,9 +4433,17 @@ test("markChannelRead_preReady_capturesAuthoritativeMarkAt", async () => { mgr.overrideRegisters.set(channelId, { s: 5, c: 0, b: 100 }); mgr.publishableContextIds.add(channelId); + // Seed forced-unread store: channel has "inbox" + "manual" sources. + const forcedEntry = { markerAtWhenForced: 50, sources: ["inbox", "manual"] }; + globalThis.window.localStorage.setItem( + forcedKey, + JSON.stringify({ [channelId]: forcedEntry }), + ); + // Click-time markAt — greater than the partial frontier. const authoritativeMarkAt = 999; - const r = mgr.markChannelRead(channelId, undefined, authoritativeMarkAt); + const sourceScope = "inbox"; + const r = mgr.markChannelRead(channelId, sourceScope, authoritativeMarkAt); assert.equal(r.status, "queued", "must queue when load incomplete"); // Intent must carry the authoritative markAt as readTarget. @@ -4245,9 +4455,8 @@ test("markChannelRead_preReady_capturesAuthoritativeMarkAt", async () => { "readTarget must equal the authoritative markAt, not the partial frontier", ); - // Drain must advance frontier to readTarget before C-bump. - const outcomes = []; - mgr.onDrainOutcome = (outcome) => outcomes.push(outcome); + // Wire hook-level callback (mirrors useDrainOutcomeCallback logic). + const cb = wireDrainCallback(mgr, pubkey); mgr.isLoadComplete = true; await mgr.drainPendingIntents(mgr.loadGeneration); @@ -4259,25 +4468,75 @@ test("markChannelRead_preReady_capturesAuthoritativeMarkAt", async () => { `frontier must be advanced to at least authoritativeMarkAt (${authoritativeMarkAt}); got ${frontier}`, ); - // Outcome must be applied-read. - assert.ok(outcomes.length >= 1, "outcome must be emitted"); + // v2 blob: register must be persisted with C-bump. + const rawBlob = globalThis.window.localStorage.getItem(v2Key); + const blob = rawBlob ? JSON.parse(rawBlob) : {}; + assert.ok( + blob.r?.[channelId], + "register must be persisted in v2 blob after drain", + ); + const reg = blob.r[channelId]; + assert.ok(reg.c > 0, "C must be bumped (applied read)"); + + // Intent and receipt must be cleaned up from blob. assert.equal( - outcomes[0].kind, - "applied-read", - "outcome must be applied-read", + blob.pi?.[channelId], + undefined, + "intent must be cleaned up from v2 blob", + ); + assert.equal( + blob.receipts?.[channelId], + undefined, + "receipt must be cleaned up from v2 blob", ); + // forcedUnreadStore: "inbox" source must be removed; "manual" remains. + const storedForced = forcedUnreadStore.read(pubkey); + const afterEntry = storedForced[channelId]; + assert.ok( + afterEntry !== undefined, + 'channel entry must still exist (only "inbox" removed, "manual" remains)', + ); + const sources = + typeof afterEntry === "object" && afterEntry !== null + ? afterEntry.sources + : ["manual"]; + assert.ok( + !sources.includes("inbox"), + '"inbox" source must be removed after applied-read drain', + ); + assert.ok( + sources.includes("manual"), + '"manual" source must remain after "inbox"-only cleanup', + ); + + // versionBumps: exactly 1 (hook cleaned up the forced source once). + assert.equal( + cb.versionBumps, + 1, + "versionBumps must be 1 — exactly one forced-store cleanup for applied-read", + ); + + cb.teardown(); mgr.destroy(); }); -// ── Test 46: already_inactive — source cleanup runs ────────────────────────── +// ── Test 46: already_inactive — source cleanup runs via hook/store ──────────── test("drain_alreadyInactive_sourceCleanupCallback", async () => { // When a queued read drains and the register is gone (already_inactive), - // the outcome must be silent-inactive (not a genuine-refusal) and the - // onDrainOutcome callback must be invoked so the hook can remove the source. + // the outcome must be silent-inactive and the hook callback must remove the + // exact source from the persisted forcedUnreadStore. + // + // Assertions (all surfaces): + // - outcome is silent-inactive (not genuine-refusal) + // - forcedUnreadStore: "inbox" source is removed from channelId entry + // - the remaining entry still has "manual" source byte-for-byte + // - versionBumps === 1 + // - intent consumed from queue globalThis.window.localStorage = makeLocalStorage(); const pubkey = "46".repeat(32); const channelId = `already-inactive-ch-${"n".repeat(44)}`; + const forcedKey = `buzz-forced-unread.v1:${pubkey}`; const fakeRelay = { fetchEvents: async () => [], @@ -4292,52 +4551,92 @@ test("drain_alreadyInactive_sourceCleanupCallback", async () => { mgr.effectiveState.set(channelId, 100); mgr.isLoadComplete = false; - // Queue a read with a specific sourceScope — no register exists. + // Seed forced-unread store: channel has "inbox" + "manual" sources. + const forcedEntry = { + markerAtWhenForced: 100, + sources: ["inbox", "manual"], + }; + globalThis.window.localStorage.setItem( + forcedKey, + JSON.stringify({ [channelId]: forcedEntry }), + ); + + // Queue a read with "inbox" sourceScope — no register exists. const sourceScope = "inbox"; const r = mgr.markChannelRead(channelId, sourceScope); assert.equal(r.status, "queued"); - const outcomes = []; - mgr.onDrainOutcome = (outcome) => outcomes.push(outcome); + // Wire hook-level callback. + const cb = wireDrainCallback(mgr, pubkey); // Complete load without adding a register — already_inactive path. mgr.isLoadComplete = true; await mgr.drainPendingIntents(mgr.loadGeneration); - // Outcome must be silent-inactive (not genuine-refusal) and carry sourceScope. - assert.equal(outcomes.length, 1, "exactly one outcome must be emitted"); - assert.equal( - outcomes[0].kind, - "silent-inactive", - "already_inactive must produce silent-inactive outcome, not genuine-refusal", + // forcedUnreadStore must have lost exactly "inbox" — "manual" remains. + const storedForced = forcedUnreadStore.read(pubkey); + const afterEntry = storedForced[channelId]; + assert.ok( + afterEntry !== undefined, + 'entry must still exist after "inbox"-only removal (manual remains)', ); - assert.equal( - outcomes[0].sourceScope, - sourceScope, - "sourceScope must be forwarded to silent-inactive outcome", + const sources = + typeof afterEntry === "object" && afterEntry !== null + ? afterEntry.sources + : ["manual"]; + assert.ok( + !sources.includes("inbox"), + '"inbox" source must be removed by silent-inactive hook', + ); + assert.ok( + sources.includes("manual"), + '"manual" source must remain unchanged', ); - // Intent consumed. + // versionBumps: exactly 1 (one forced-store write by hook). + assert.equal( + cb.versionBumps, + 1, + "versionBumps must be 1 for silent-inactive cleanup", + ); + + // Intent consumed from queue. assert.equal( pendingOverrideIntentStore.get(channelId), undefined, "intent must be consumed after already_inactive", ); + cb.teardown(); mgr.destroy(); }); -// ── Test 47: restart-safe unread rollback — priorForcedEntry persisted ──────── +// ── Test 47: restart-safe unread rollback — priorForcedEntry restored in forcedUnreadStore test("drain_unreadRefusal_restartSafe_priorForcedEntryRestored", async () => { // After a restart, a refused unread intent must restore the prior forced entry // (not delete the whole channel entry) using the priorForcedEntry persisted in - // the intent metadata. + // the intent metadata, and write that restoration to forcedUnreadStore. + // + // Scenario: + // - Session 1: channel had {sources:["inbox"]} entry; "manual" unread attempted + // (S at max → uint32_overflow), intent queued with priorForcedEntry. + // The optimistic unread replaced the entry in forcedUnreadStore. + // - App restarts. + // - Session 2: hydrates intent with priorForcedEntry; drain runs; overflow → refused. + // Hook must restore {sources:["inbox"]} in forcedUnreadStore. + // + // Assertions (all surfaces): + // - priorForcedEntry hydrates correctly from v2 blob + // - outcome is genuine-refusal with priorForcedEntry + // - forcedUnreadStore[channelId] = priorForcedEntry byte-for-byte after hook runs + // - versionBumps === 1 globalThis.window.localStorage = makeLocalStorage(); const pubkey = "47".repeat(32); const channelId = `restart-safe-ch-${"o".repeat(47)}`; const ls = globalThis.window.localStorage; const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; + const forcedKey = `buzz-forced-unread.v1:${pubkey}`; // Simulate session 1: {sources:["inbox"]} entry existed, then manual unread // was attempted (S at max → will overflow), intent queued with priorForcedEntry. @@ -4356,6 +4655,13 @@ test("drain_unreadRefusal_restartSafe_priorForcedEntryRestored", async () => { ng: 2, }), ); + // The optimistic unread had replaced the forced entry with "manual". + ls.setItem( + forcedKey, + JSON.stringify({ + [channelId]: { markerAtWhenForced: 50, sources: ["manual"] }, + }), + ); // Session 2: new manager hydrates. const fakeRelay = { @@ -4379,24 +4685,30 @@ test("drain_unreadRefusal_restartSafe_priorForcedEntryRestored", async () => { "priorForcedEntry must survive round-trip through v2 blob", ); - // Wire outcome callback to capture the refusal and the priorForcedEntry. - const outcomes = []; - mgr.onDrainOutcome = (outcome) => outcomes.push(outcome); + // Wire hook-level callback. No in-memory snapshot exists (new session), so + // the callback falls back to outcome.priorForcedEntry for restoration. + const cb = wireDrainCallback(mgr, pubkey); // Complete load — register still at uint32-max → S-bump overflows → refused. await mgr.fetchAndMerge(mgr.loadGeneration); assert.equal(mgr.isLoadComplete, true); await mgr.drainPendingIntents(mgr.loadGeneration); - // Outcome must carry priorForcedEntry for restart-safe rollback. - assert.equal(outcomes.length, 1); - assert.equal(outcomes[0].kind, "genuine-refusal"); - assert.equal(outcomes[0].op, "unread"); + // forcedUnreadStore must be restored to the priorForcedEntry byte-for-byte. + const storedForced = forcedUnreadStore.read(pubkey); assert.deepEqual( - outcomes[0].priorForcedEntry, + storedForced[channelId], priorForcedEntry, - "genuine-refusal outcome must carry priorForcedEntry for restart-safe rollback", + "forcedUnreadStore must be restored to priorForcedEntry after post-restart refusal", ); + // versionBumps: exactly 1 (hook wrote to forced store once). + assert.equal( + cb.versionBumps, + 1, + "versionBumps must be 1 — exactly one restore write", + ); + + cb.teardown(); mgr.destroy(); }); diff --git a/desktop/src/features/channels/readState/readStateStorage.ts b/desktop/src/features/channels/readState/readStateStorage.ts index 9bb8537d22..b51deaf197 100644 --- a/desktop/src/features/channels/readState/readStateStorage.ts +++ b/desktop/src/features/channels/readState/readStateStorage.ts @@ -15,7 +15,17 @@ import type { PendingIntent, PendingIntentOp, } from "@/features/channels/pendingOverrideIntents"; -import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; +import type { + ForcedUnreadEntry, + ForcedUnreadSource, +} from "@/features/channels/forcedUnreadStore"; + +/** + * Maximum valid protocol timestamp (unix seconds): uint32 ceiling (year 2106). + * Matches the limit used for override counters so a stored readTarget that + * overflows uint32 is silently rejected before it can reach Date.toISOString(). + */ +const MAX_PROTOCOL_TIMESTAMP = 0xffffffff; export type StoredReadState = { contexts: Map; @@ -163,6 +173,53 @@ type OverrideStateBlob = { nextGen: number; }; +/** + * Parse and validate a raw JSON value as a `ForcedUnreadEntry`. + * + * Returns the validated entry on success, `null` if the value is present but + * malformed (caller should reject the whole intent), or `undefined` if the + * value is `undefined` (absent field — no prior entry). + * + * Valid shapes (matching the `ForcedUnreadEntry` union): + * - `number` — legacy v1 scalar marker + * - `null` — legacy v1 null marker + * - `{ markerAtWhenForced: number | null, sources: ForcedUnreadSource[] }` + */ +function parseForcedUnreadEntry( + raw: unknown, +): ForcedUnreadEntry | null | undefined { + if (raw === undefined) return undefined; + if (raw === null) return null; + if (typeof raw === "number") { + if (Number.isFinite(raw) && raw >= 0 && raw <= MAX_PROTOCOL_TIMESTAMP) + return raw; + return null; // invalid numeric marker + } + if (typeof raw === "object" && !Array.isArray(raw)) { + const obj = raw as Record; + const marker = obj.markerAtWhenForced; + if ( + marker !== null && + (typeof marker !== "number" || + !Number.isFinite(marker) || + marker < 0 || + marker > MAX_PROTOCOL_TIMESTAMP) + ) + return null; + const sources = obj.sources; + if (!Array.isArray(sources)) return null; + const validSources: ForcedUnreadSource[] = sources.filter( + (s): s is ForcedUnreadSource => s === "inbox" || s === "manual", + ); + if (validSources.length === 0) return null; + return { + markerAtWhenForced: (marker as number | null) ?? null, + sources: validSources, + }; + } + return null; // unrecognised shape +} + function readOverrideState(pubkey: string): OverrideStateBlob { const registers = new Map(); const receipts = new Map(); @@ -233,19 +290,37 @@ function readOverrideState(pubkey: string): OverrideStateBlob { if (typeof gen !== "number" || !Number.isInteger(gen) || gen < 1) continue; if (op !== "unread" && op !== "read") continue; - if (sourceScope !== undefined && typeof sourceScope !== "string") + // sourceScope must be a valid ForcedUnreadSource or absent. + if ( + sourceScope !== undefined && + sourceScope !== "inbox" && + sourceScope !== "manual" + ) continue; - if (readTarget !== undefined && typeof readTarget !== "number") + // readTarget must be a finite integer in protocol timestamp range. + if (readTarget !== undefined) { + if ( + typeof readTarget !== "number" || + !Number.isFinite(readTarget) || + !Number.isInteger(readTarget) || + readTarget < 0 || + readTarget > MAX_PROTOCOL_TIMESTAMP + ) + continue; + } + // priorForcedEntry must be a valid ForcedUnreadEntry or absent. + const parsedPrior = parseForcedUnreadEntry(priorForcedEntry); + if (priorForcedEntry !== undefined && parsedPrior === null) continue; const intent: PendingIntent = { gen, op, - ...(sourceScope !== undefined ? { sourceScope } : {}), + ...(sourceScope !== undefined + ? { sourceScope: sourceScope as string } + : {}), ...(readTarget !== undefined ? { readTarget } : {}), - // priorForcedEntry is stored as-is (number | null | object); - // accept any JSON-round-trippable value — type-check at use site. - ...(priorForcedEntry !== undefined - ? { priorForcedEntry: priorForcedEntry as ForcedUnreadEntry } + ...(parsedPrior !== undefined + ? { priorForcedEntry: parsedPrior } : {}), }; pendingIntents.set(channelId, intent);