From f6a12809dcb4f41b9358dc7ea1783dd71bca51b0 Mon Sep 17 00:00:00 2001 From: Duncan Date: Sat, 8 Aug 2026 11:53:40 -0400 Subject: [PATCH] =?UTF-8?q?fix(readState):=20round-3=20v4=20conformance=20?= =?UTF-8?q?=E2=80=94=20typed=20outcomes,=20markAt,=20restart-safe=20rollba?= =?UTF-8?q?ck,=20auto-retry,=20non-reentrant=20drain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six items from Thufir's pass-2 report: 1. Typed DrainOutcome union (applied-read | silent-inactive | applied-unread | genuine-refusal) with exhaustive switch in the hook layer, eliminating the stringly-typed seam that let already_inactive fall outside both branches. 2. already_inactive emits silent-inactive outcome so the hook's source-cleanup callback fires; committed forced-entry ownership is now removed exactly rather than left stale after a successful queued read. 3. Click-time markAt threaded through applyOverrideRead → markChannelRead → PendingIntent.readTarget; drain advances frontier to that exact target before C-bump so pre-ready reads don't lose position. 4. priorForcedEntry persisted in PendingIntent (and in the v2 blob); genuine- refusal DrainOutcome carries it; hook uses the persisted value when the in-memory snapshot map is empty (post-restart rollback is now byte-for-byte exact). 5. initialize() ending with an incomplete verdict schedules the first backoff timer (1 s, then 2 s, 4 s … 60 s via retryLoad's own loop) so pending intents eventually drain without needing an external reconnect signal. New witness: initialize_incompleteVerdict_schedulesAutomaticRetry. 6. Non-reentrant outcome commit: after onDrainOutcome fires, drain re-checks generation; if a newer intent was enqueued inside the callback, all in- memory effects (register, receipt, publishable, extraSlots) are rolled back and the loop continues without consuming the newer intent. drain_compareDeletePreservesNewerIntent strengthened with zero-stale-effects assertion. Also: fix noExplicitAny in readStateStorage.ts (priorForcedEntry now typed as ForcedUnreadEntry); ratchet trimmed to keep useUnreadChannels.ts at limit. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../channels/pendingOverrideIntents.ts | 30 +- .../channels/readState/readStateDrain.ts | 165 +++++-- .../readState/readStateManager.test.mjs | 414 ++++++++++++++++-- .../channels/readState/readStateManager.ts | 90 ++-- .../channels/readState/readStateStorage.ts | 24 +- .../channels/readState/useReadState.ts | 28 +- .../features/channels/readStateOverride.ts | 26 +- .../features/channels/useUnreadChannels.ts | 29 +- .../channels/useUnreadChannelsHelpers.ts | 122 ++++-- 9 files changed, 733 insertions(+), 195 deletions(-) diff --git a/desktop/src/features/channels/pendingOverrideIntents.ts b/desktop/src/features/channels/pendingOverrideIntents.ts index cdf6145a72..4ff9324ac6 100644 --- a/desktop/src/features/channels/pendingOverrideIntents.ts +++ b/desktop/src/features/channels/pendingOverrideIntents.ts @@ -32,6 +32,8 @@ * Old pubkey data is NOT wiped from the v2 blob — the manager owns that. */ +import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; + // ── Types ───────────────────────────────────────────────────────────────────── /** The operation the user intended. */ @@ -40,22 +42,28 @@ export type PendingIntentOp = "unread" | "read"; /** * One pending intent for a channel. * - * `gen` — monotonically increasing per-channel generation counter. - * Incremented each time a new intent replaces the previous one. - * `op` — the user's intended action. - * `sourceScope` — for `read` intents, the exact source (or undefined = all - * sources) whose removal scope was captured at click time. - * Used by the drain to apply source cleanup idempotently. - * `readTarget` — for `read` intents, the frontier value captured at click - * time. Drain advances the frontier to this value before the - * C-bump so that an incomplete-load read later applies at the - * correct logical position. + * `gen` — monotonically increasing per-channel generation counter. + * Incremented each time a new intent replaces the previous one. + * `op` — the user's intended action. + * `sourceScope` — for `read` intents, the exact source (or undefined = all + * sources) whose removal scope was captured at click time. + * Used by the drain to apply source cleanup idempotently. + * `readTarget` — for `read` intents, the frontier value captured at click + * time. Drain advances the frontier to this value before the + * C-bump so that an incomplete-load read later applies at the + * correct logical position. + * `priorForcedEntry` — for `unread` intents, the exact forced-unread entry that + * existed before the optimistic write. Persisted in the v2 + * blob so a post-restart refusal can restore byte-for-byte + * rather than deleting the whole multi-source entry. + * `undefined` means no prior entry existed. */ export type PendingIntent = { gen: number; op: PendingIntentOp; sourceScope?: string; readTarget?: number; + priorForcedEntry?: ForcedUnreadEntry; }; // ── Store ───────────────────────────────────────────────────────────────────── @@ -100,6 +108,7 @@ export class PendingOverrideIntentStore { op: PendingIntentOp, sourceScope?: string, readTarget?: number, + priorForcedEntry?: ForcedUnreadEntry, ): PendingIntent { const gen = this._nextGen++; const intent: PendingIntent = { @@ -107,6 +116,7 @@ export class PendingOverrideIntentStore { op, ...(sourceScope !== undefined ? { sourceScope } : {}), ...(readTarget !== undefined ? { readTarget } : {}), + ...(priorForcedEntry !== undefined ? { priorForcedEntry } : {}), }; this.intents.set(channelId, intent); return intent; diff --git a/desktop/src/features/channels/readState/readStateDrain.ts b/desktop/src/features/channels/readState/readStateDrain.ts index 17626ca3df..cccd5aae76 100644 --- a/desktop/src/features/channels/readState/readStateDrain.ts +++ b/desktop/src/features/channels/readState/readStateDrain.ts @@ -16,6 +16,41 @@ import type { AppliedReceipt } from "@/features/channels/readState/readStateStor import { pendingOverrideIntentStore } from "@/features/channels/pendingOverrideIntents"; import { toast } from "sonner"; import type { MarkResult } from "@/features/channels/readState/readStateManager"; +import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; + +// ── Typed drain outcome ──────────────────────────────────────────────────────── + +/** + * Discriminated union encoding every possible drain outcome. + * + * `applied-read` — read succeeded; `sourceScope` names the exact source + * whose forced-unread entry should be removed (or all + * sources when undefined). Toast-free, receipt-free. + * `silent-inactive` — `already_inactive`; the override was already gone, + * but any forced-entry source must still be cleaned up. + * Toast-free, receipt-free. + * `applied-unread` — unread succeeded; optimistic entry is now committed. + * Discard the pre-mark snapshot. + * `genuine-refusal` — a real failure (`uint32_overflow`, `budget_exhausted`, + * `storage_failed`); show toast, roll back presentation. + * `op` distinguishes the unread vs read rollback path. + * `priorForcedEntry` carries the persisted prior forced + * entry for restart-safe rollback (undefined = no prior). + */ +export type DrainOutcome = + | { kind: "applied-read"; channelId: string; sourceScope?: string } + | { kind: "silent-inactive"; channelId: string; sourceScope?: string } + | { kind: "applied-unread"; channelId: string } + | { + kind: "genuine-refusal"; + channelId: string; + op: "read" | "unread"; + reason: string; + /** For `unread` refusals: the exact prior forced-unread entry persisted + * in the intent, used for restart-safe rollback when the in-memory + * snapshot map is empty (new session after crash/restart). */ + priorForcedEntry?: ForcedUnreadEntry; + }; /** Subset of ReadStateManager state and methods needed for drain operations. */ export interface DrainContext { @@ -37,19 +72,9 @@ export interface DrainContext { markContextRead(channelId: string, unixTimestamp: number): void; tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean; restoreExtraSlotIds(prev: string[]): void; - /** Drain outcome callback — wired by the hook for toast/rollback routing. - * `sourceScope` is set for `read+applied` outcomes; it carries the source - * that was being cleared so the hook can remove exactly that source from the - * forced-unread entry rather than deleting the whole entry. */ - onDrainOutcome: - | (( - channelId: string, - op: string, - status: string, - reason?: string, - sourceScope?: string, - ) => void) - | null; + /** Drain outcome callback — typed discriminated union for exhaustive handling + * in the hook layer. */ + onDrainOutcome: ((outcome: DrainOutcome) => void) | null; } /** @@ -65,15 +90,7 @@ interface ManagerInternals { readonly appliedReceipts: Map; readonly publishableContextIds: Set; readonly extraSlotIds: string[]; - onDrainOutcome: - | (( - channelId: string, - op: string, - status: string, - reason?: string, - sourceScope?: string, - ) => void) - | null; + onDrainOutcome: ((outcome: DrainOutcome) => void) | null; persistLocalState(): boolean; schedulePublish(): void; notifyListeners(): void; @@ -160,6 +177,7 @@ export async function drainPendingIntents( const capturedOp = captured.op; const capturedSourceScope = captured.sourceScope; const capturedReadTarget = captured.readTarget; + const capturedPriorForcedEntry = captured.priorForcedEntry; const capturedPubkey = ctx.pubkey; // Pubkey fence: abort if identity changed mid-drain. @@ -231,32 +249,75 @@ export async function drainPendingIntents( ctx.schedulePublish(); } - // Step 2: source cleanup (idempotent) — capturedSourceScope scopes removal. - // The sourceScope is forwarded to the onDrainOutcome callback so the hook - // layer can remove exactly the right source from the forced-unread entry. - - // Step 3: surface outcomes — toast genuine refusals, route to hook callback. + // 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" && reason !== "load_incomplete") { + 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); - ctx.onDrainOutcome?.(channelId, capturedOp, "refused", reason); + outcome = { + kind: "genuine-refusal", + channelId, + op: capturedOp, + reason, + ...(capturedOp === "unread" && capturedPriorForcedEntry !== undefined + ? { priorForcedEntry: capturedPriorForcedEntry } + : {}), + }; } - // already_inactive: silent definitive success — no badge, no toast. + // load_incomplete: never reaches the user; no outcome emitted. } else if (replayResult.status === "applied") { - ctx.onDrainOutcome?.( - channelId, - capturedOp, - "applied", - undefined, - capturedSourceScope, - ); + outcome = + capturedOp === "unread" + ? { kind: "applied-unread", channelId } + : { + kind: "applied-read", + channelId, + sourceScope: capturedSourceScope, + }; } - // Step 4: atomic cleanup commit — delete receipt + compare-and-delete intent. + // 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()) { @@ -372,13 +433,24 @@ export function markChannelReadDirect( * Public mark-unread: write-ordered forced entry then queue/apply the override. * Queues an intent when load is incomplete; applies immediately otherwise. * Persists the intent atomically in the v2 blob via persistLocalState(). + * + * `priorForcedEntry` is the forced-unread entry that existed before the + * optimistic write. Persisted in the intent so a post-restart refusal can + * restore the exact prior state rather than deleting the whole entry. */ export function markChannelUnread( ctx: DrainContext, channelId: string, + priorForcedEntry?: ForcedUnreadEntry, ): MarkResult { if (!ctx.isLoadComplete) { - pendingOverrideIntentStore.enqueue(channelId, "unread"); + pendingOverrideIntentStore.enqueue( + channelId, + "unread", + undefined, + undefined, + priorForcedEntry, + ); if (!ctx.persistLocalState()) { // Storage failure — intent is in-memory only; session drain still works. // Optimistic presentation continues (forced entry already written). @@ -420,18 +492,27 @@ export function markChannelUnread( /** * Public mark-read: queues an intent when load is incomplete; C-bumps otherwise. - * Captures the current frontier as readTarget when queuing, so the drain can - * advance the frontier to the correct position later. + * Captures `markAt` as readTarget when queuing, so the drain can advance the + * frontier to the authoritative click-time position rather than the partial + * pre-ready frontier. * Captures sourceScope when provided so the drain can perform exact source cleanup. * Persists the intent atomically in the v2 blob via persistLocalState(). + * + * `markAt` should be the authoritative read target computed at click time + * (e.g., the newest observed message timestamp). When undefined/0 the drain + * falls back to the complete-load frontier, which is correct for passive opens. */ export function markChannelRead( ctx: DrainContext, channelId: string, sourceScope?: string, + markAt?: number, ): MarkResult { if (!ctx.isLoadComplete) { - const readTarget = ctx.channelFrontier(channelId); + const readTarget = + markAt !== undefined && markAt > 0 + ? markAt + : ctx.channelFrontier(channelId); pendingOverrideIntentStore.enqueue( channelId, "read", diff --git a/desktop/src/features/channels/readState/readStateManager.test.mjs b/desktop/src/features/channels/readState/readStateManager.test.mjs index ae9f6fb3ab..3e4d950d3c 100644 --- a/desktop/src/features/channels/readState/readStateManager.test.mjs +++ b/desktop/src/features/channels/readState/readStateManager.test.mjs @@ -3285,10 +3285,12 @@ test("drain_readToUnreadInversion_onlySBumpApplied", async () => { mgr.destroy(); }); -// ── Test 33: compare-delete preserves newer intent ──────────────────────────── +// ── Test 33: compare-delete preserves newer intent + zero stale 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. // // We inject gen2 via onDrainOutcome (fires just before compare-delete). globalThis.window.localStorage = makeLocalStorage(); @@ -3313,9 +3315,12 @@ test("drain_compareDeletePreservesNewerIntent", async () => { const gen1 = pendingOverrideIntentStore.get(channelId).gen; // Wire onDrainOutcome to enqueue gen2 BEFORE compare-and-delete runs. - // The onDrainOutcome fires at step 3 (surface outcomes), before step 4 (cleanup commit). - mgr.onDrainOutcome = (chId, _op, _status) => { - if (chId === channelId) { + // The onDrainOutcome fires at step 2 (surface outcomes), before step 3 (cleanup commit). + // Also capture any forced-source effects from gen1 drain. + const outcomes = []; + mgr.onDrainOutcome = (outcome) => { + outcomes.push(outcome); + if (outcome.channelId === channelId) { // Enqueue gen2 between outcome surfacing and compare-delete. pendingOverrideIntentStore.enqueue(channelId, "unread"); } @@ -3335,6 +3340,16 @@ test("drain_compareDeletePreservesNewerIntent", async () => { "surviving intent must be gen2 (newer 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", + ); + // Clean up. pendingOverrideIntentStore.compareAndDelete(channelId, remaining.gen); mgr.destroy(); @@ -3519,39 +3534,184 @@ test("drain_cleanupCommit_intentAndReceiptDeletedAtomically", async () => { mgr.destroy(); }); -// ── Test 36: controller — automatic backoff after incomplete verdict ─────────── -test("retryLoad_incompleteVerdictAdvancesRetryAttempt", async () => { - // After an incomplete verdict, retryAttempt increments so the next retry - // will use exponential backoff. Destroy/cancel before the delayed retry fires. +// ── 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. + // Verifies: (1) timer is scheduled, (2) delay > 0 for second attempt, + // (3) retryAttempt is incremented, (4) destroy cancels the scheduled timer. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "36".repeat(32); + let fetchCallCount = 0; + const fakeRelay = { fetchEvents: async () => [], publishEvent: async () => {}, subscribeToReconnects: () => () => {}, getConnectionGeneration: () => 0, subscribeFenced: async () => { - throw new Error("fence timeout"); + fetchCallCount++; + throw new Error("fence timeout"); // always incomplete }, subscribeLive: async (_f, _h) => () => {}, }; - const mgr = new ReadStateManager(pubkey, fakeRelay); - assert.equal(mgr.retryAttempt, 0, "precondition: retryAttempt is 0"); + const origSetTimeout = globalThis.window.setTimeout; + const origClearTimeout = globalThis.window.clearTimeout; - // First retry (immediate — attempt 0 means delayMs=0). - await mgr.retryLoad(); + const scheduledTimers = new Map(); + let nextFakeId = 10000; - // After incomplete verdict, retryAttempt must be incremented. - assert.ok( - mgr.retryAttempt > 0, - "retryAttempt must be > 0 after incomplete verdict", - ); + globalThis.window.setTimeout = (fn, ms) => { + if (ms > 0) { + const id = nextFakeId++; + scheduledTimers.set(id, { fn, ms }); + return id; + } + return origSetTimeout(fn, ms); + }; + globalThis.window.clearTimeout = (id) => { + if (scheduledTimers.has(id)) { + scheduledTimers.delete(id); + } else { + origClearTimeout(id); + } + }; - // Second retry would use backoff. Destroy before it fires. - mgr.destroy(); - assert.equal(mgr.retryBackoffTimer, null, "timer must be cleared on destroy"); + try { + const mgr = new ReadStateManager(pubkey, fakeRelay); + assert.equal(mgr.retryAttempt, 0, "precondition: retryAttempt is 0"); + + // First retry (immediate — attempt 0 means delayMs=0). + await mgr.retryLoad(); + assert.equal(fetchCallCount, 1, "first fetch must have been called"); + + // After first incomplete verdict, retryAttempt must be > 0. + assert.ok( + mgr.retryAttempt > 0, + "retryAttempt must be > 0 after first incomplete", + ); + + // A backoff timer must have been scheduled (automatic retry loop). + assert.ok( + mgr.retryBackoffTimer !== null, + "retryBackoffTimer must be set — automatic retry was scheduled", + ); + assert.ok( + scheduledTimers.size > 0, + "a timer must be pending in the fake scheduler", + ); + + // Timer delay must be > 0 (bounded backoff). + const [, { ms }] = [...scheduledTimers.entries()][0]; + assert.ok(ms > 0, "backoff timer delay must be > 0 for attempt >= 1"); + + // Destroy must cancel the pending timer. + mgr.destroy(); + assert.equal( + mgr.retryBackoffTimer, + null, + "destroy must clear retryBackoffTimer", + ); + assert.equal( + scheduledTimers.size, + 0, + "fake scheduler must have no pending timers after destroy", + ); + } finally { + globalThis.window.setTimeout = origSetTimeout; + globalThis.window.clearTimeout = origClearTimeout; + } +}); + +// ── Test 36b: controller — initialize() incomplete → schedules automatic retry +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. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "36b".repeat(21).slice(0, 64); + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async () => { + throw new Error("fence timeout"); // always incomplete + }, + subscribeLive: async (_f, _h) => () => {}, + }; + + const origSetTimeout = globalThis.window.setTimeout; + const origClearTimeout = globalThis.window.clearTimeout; + + const scheduledTimers = new Map(); + let nextFakeId = 20000; + + globalThis.window.setTimeout = (fn, ms) => { + if (ms > 0) { + const id = nextFakeId++; + scheduledTimers.set(id, { fn, ms }); + return id; + } + return origSetTimeout(fn, ms); + }; + globalThis.window.clearTimeout = (id) => { + if (scheduledTimers.has(id)) { + scheduledTimers.delete(id); + } else { + origClearTimeout(id); + } + }; + + try { + const mgr = new ReadStateManager(pubkey, fakeRelay); + + await mgr.initialize(); + + // initialize() must have scheduled a retry timer. + assert.ok( + mgr.retryBackoffTimer !== null, + "retryBackoffTimer must be set after incomplete initialize()", + ); + assert.ok( + scheduledTimers.size > 0, + "a timer must be pending in the fake scheduler after incomplete initialize()", + ); + + // retryAttempt must be 1 (initialize = attempt 0 consumed). + assert.equal( + mgr.retryAttempt, + 1, + "retryAttempt must be 1 after incomplete initialize()", + ); + + // Timer delay must be > 0. + const [, { ms }] = [...scheduledTimers.entries()][0]; + assert.ok( + ms > 0, + "backoff timer delay must be > 0 after incomplete initialize()", + ); + + // destroy() must cancel the pending timer. + mgr.destroy(); + assert.equal( + mgr.retryBackoffTimer, + null, + "destroy must clear retryBackoffTimer", + ); + assert.equal( + scheduledTimers.size, + 0, + "fake scheduler must have no pending timers after destroy", + ); + } finally { + globalThis.window.setTimeout = origSetTimeout; + globalThis.window.clearTimeout = origClearTimeout; + } }); // ── Test 37: controller — reconnect during loadInFlight → pendingRetryOnComplete @@ -3876,8 +4036,8 @@ test("drain_queuedUnread_refused_surfacesOutcome", async () => { // Wire outcome callback. const outcomes = []; - mgr.onDrainOutcome = (chId, op, status, reason) => { - outcomes.push({ chId, op, status, reason }); + mgr.onDrainOutcome = (outcome) => { + outcomes.push(outcome); }; // Complete load and drain. @@ -3886,9 +4046,9 @@ test("drain_queuedUnread_refused_surfacesOutcome", async () => { // Drain must have surfaced the refusal. assert.equal(outcomes.length, 1, "exactly one outcome must be emitted"); - assert.equal(outcomes[0].chId, channelId); + assert.equal(outcomes[0].channelId, channelId); + assert.equal(outcomes[0].kind, "genuine-refusal"); assert.equal(outcomes[0].op, "unread"); - assert.equal(outcomes[0].status, "refused"); assert.equal( outcomes[0].reason, "uint32_overflow", @@ -3949,8 +4109,8 @@ test("drain_deferredRefusalAfterRestart_intentHydratedAndRefused", async () => { // Wire outcome callback to capture the refusal. const outcomes = []; - mgr.onDrainOutcome = (chId, op, status, reason) => { - outcomes.push({ chId, op, status, reason }); + mgr.onDrainOutcome = (outcome) => { + outcomes.push(outcome); }; // Complete load and drain — the register is still at uint32-max, so S-bump @@ -3961,8 +4121,8 @@ test("drain_deferredRefusalAfterRestart_intentHydratedAndRefused", async () => { // Drain must surface the refusal via onDrainOutcome. assert.equal(outcomes.length, 1, "exactly one outcome must be emitted"); + assert.equal(outcomes[0].kind, "genuine-refusal"); assert.equal(outcomes[0].op, "unread"); - assert.equal(outcomes[0].status, "refused"); assert.equal(outcomes[0].reason, "uint32_overflow"); // Intent consumed even after refusal (genuine, not load_incomplete). @@ -4042,3 +4202,201 @@ test("markChannelUnread_storageFailed_intentInMemoryOnly", () => { globalThis.window.localStorage.setItem = origSetItem; } }); + +// ── Test 45: authoritative queued-read target captured at click time ────────── +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. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "45".repeat(32); + const channelId = `mark-at-ch-${"m".repeat(53)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + // Partial frontier (simulates what's visible pre-ready). + const partialFrontier = 100; + mgr.effectiveState.set(channelId, partialFrontier); + mgr.isLoadComplete = false; + + // Simulate a register so the read is not already_inactive. + mgr.overrideRegisters.set(channelId, { s: 5, c: 0, b: 100 }); + mgr.publishableContextIds.add(channelId); + + // Click-time markAt — greater than the partial frontier. + const authoritativeMarkAt = 999; + const r = mgr.markChannelRead(channelId, undefined, authoritativeMarkAt); + assert.equal(r.status, "queued", "must queue when load incomplete"); + + // Intent must carry the authoritative markAt as readTarget. + const intent = pendingOverrideIntentStore.get(channelId); + assert.ok(intent, "intent must exist"); + assert.equal( + intent.readTarget, + authoritativeMarkAt, + "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); + + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // Frontier must have advanced to authoritativeMarkAt. + const frontier = mgr.effectiveState.get(channelId) ?? 0; + assert.ok( + frontier >= authoritativeMarkAt, + `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"); + assert.equal( + outcomes[0].kind, + "applied-read", + "outcome must be applied-read", + ); + + mgr.destroy(); +}); + +// ── Test 46: already_inactive — source cleanup runs ────────────────────────── +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. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "46".repeat(32); + const channelId = `already-inactive-ch-${"n".repeat(44)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.effectiveState.set(channelId, 100); + mgr.isLoadComplete = false; + + // Queue a read with a specific 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); + + // 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", + ); + assert.equal( + outcomes[0].sourceScope, + sourceScope, + "sourceScope must be forwarded to silent-inactive outcome", + ); + + // Intent consumed. + assert.equal( + pendingOverrideIntentStore.get(channelId), + undefined, + "intent must be consumed after already_inactive", + ); + + mgr.destroy(); +}); + +// ── Test 47: restart-safe unread rollback — priorForcedEntry persisted ──────── +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. + 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}`; + + // Simulate session 1: {sources:["inbox"]} entry existed, then manual unread + // was attempted (S at max → will overflow), intent queued with priorForcedEntry. + const priorForcedEntry = { markerAtWhenForced: 50, sources: ["inbox"] }; + ls.setItem( + v2Key, + JSON.stringify({ + r: { [channelId]: { s: 0xffffffff, c: 0, b: 50, f: 50 } }, + pi: { + [channelId]: { + gen: 1, + op: "unread", + priorForcedEntry: priorForcedEntry, + }, + }, + ng: 2, + }), + ); + + // Session 2: new manager hydrates. + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.hydrateFromLocalStorage(); + + // Verify intent restored with priorForcedEntry. + const hydrated = pendingOverrideIntentStore.get(channelId); + assert.ok(hydrated, "intent must be hydrated"); + assert.deepEqual( + hydrated.priorForcedEntry, + priorForcedEntry, + "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); + + // 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"); + assert.deepEqual( + outcomes[0].priorForcedEntry, + priorForcedEntry, + "genuine-refusal outcome must carry priorForcedEntry for restart-safe rollback", + ); + + mgr.destroy(); +}); diff --git a/desktop/src/features/channels/readState/readStateManager.ts b/desktop/src/features/channels/readState/readStateManager.ts index cd643e66fb..909ba95897 100644 --- a/desktop/src/features/channels/readState/readStateManager.ts +++ b/desktop/src/features/channels/readState/readStateManager.ts @@ -45,7 +45,9 @@ import { markChannelUnread as markChannelUnreadImpl, markChannelRead as markChannelReadImpl, createDrainContext, + type DrainOutcome, } from "@/features/channels/readState/readStateDrain"; +import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; export type { SlotSplitResult, TrimResult }; export { splitContextsIntoBudgetedSlots, trimContextsToBudget }; @@ -189,15 +191,7 @@ export class ReadStateManager { pendingRetryOnComplete = false; // reconnect arrived during loadInFlight drainScheduled = false; /** Called after each drain intent is settled; wired by the hook for outcome routing. */ - onDrainOutcome: - | (( - channelId: string, - op: string, - status: string, - reason?: string, - sourceScope?: string, - ) => void) - | null = null; + onDrainOutcome: ((outcome: DrainOutcome) => void) | null = null; constructor(pubkey: string, relayClient: RelayClient) { this.pubkey = pubkey; @@ -227,6 +221,17 @@ export class ReadStateManager { if (this.isLoadComplete && !this.drainScheduled) { this.drainScheduled = true; void this.drainPendingIntents(gen); + } else if (!this.isLoadComplete && this.retryBackoffTimer === null) { + // Initial load was incomplete — schedule the first automatic retry with + // bounded backoff so pending intents eventually drain without needing an + // external reconnect signal. Treat initialize() as attempt 0 consumed, + // so the first timed retry uses the attempt-1 delay (1 s). + this.retryAttempt = 1; + const delayMs = 1_000; // Math.min(1000 * 2^0, 60000) + this.retryBackoffTimer = window.setTimeout(() => { + this.retryBackoffTimer = null; + if (!this.destroyed) void this.retryLoad(); + }, delayMs); } await this.startLiveSubscription(); if (this.destroyed) return; @@ -338,13 +343,18 @@ export class ReadStateManager { /** * Retry load: generation-fenced, coalesces in-flight loads. - * Implements bounded exponential backoff (1s→2s→4s…capped at 60s). + * Always starts immediately (no entry delay). After an incomplete verdict + * this method schedules itself via `retryBackoffTimer` with bounded + * exponential backoff (1 s, 2 s, 4 s … 60 s) so pending intents eventually + * drain even without an external reconnect signal. + * * A reconnect during loadInFlight sets pendingRetryOnComplete so the next * complete/incomplete verdict triggers exactly one fresh generation rather * than dropping the reconnect signal. */ async retryLoad(): Promise { if (this.destroyed) return; + // Cancel any pending backoff timer — this call supersedes it. if (this.retryBackoffTimer !== null) { window.clearTimeout(this.retryBackoffTimer); this.retryBackoffTimer = null; @@ -355,36 +365,42 @@ export class ReadStateManager { this.pendingRetryOnComplete = true; return; } - const delayMs = - this.retryAttempt === 0 - ? 0 - : Math.min(1_000 * 2 ** (this.retryAttempt - 1), 60_000); - if (delayMs > 0) { - await new Promise((resolve) => { - this.retryBackoffTimer = window.setTimeout(() => { - this.retryBackoffTimer = null; - resolve(); - }, delayMs); - }); - } - if (this.destroyed || this.loadInFlight) return; - this.retryAttempt++; this.loadInFlight = true; const gen = ++this.loadGeneration; try { this.isLoadComplete = false; await this.fetchAndMerge(gen); if (this.destroyed || gen !== this.loadGeneration) return; - if (this.isLoadComplete) this.retryAttempt = 0; - if (!this.drainScheduled && this.isLoadComplete) { - this.drainScheduled = true; - void this.drainPendingIntents(gen); + if (this.isLoadComplete) { + this.retryAttempt = 0; + if (!this.drainScheduled) { + this.drainScheduled = true; + void this.drainPendingIntents(gen); + } + } else { + // Incomplete verdict — schedule automatic retry with bounded backoff + // so pending intents drain eventually even without a reconnect signal. + // Backoff: 1 s, 2 s, 4 s, … capped at 60 s. + this.retryAttempt++; + const nextDelayMs = Math.min( + 1_000 * 2 ** (this.retryAttempt - 1), + 60_000, + ); + this.retryBackoffTimer = window.setTimeout(() => { + this.retryBackoffTimer = null; + if (!this.destroyed) void this.retryLoad(); + }, nextDelayMs); } } finally { if (gen === this.loadGeneration) { this.loadInFlight = false; if (this.pendingRetryOnComplete && !this.destroyed) { this.pendingRetryOnComplete = false; + // Cancel any scheduled backoff timer — reconnect triggers immediate retry. + if (this.retryBackoffTimer !== null) { + window.clearTimeout(this.retryBackoffTimer); + this.retryBackoffTimer = null; + } void this.retryLoad(); } } @@ -397,15 +413,27 @@ export class ReadStateManager { this.drainScheduled = false; } - markChannelUnread(channelId: string): MarkResult { - return markChannelUnreadImpl(createDrainContext(this), channelId); + markChannelUnread( + channelId: string, + priorForcedEntry?: ForcedUnreadEntry, + ): MarkResult { + return markChannelUnreadImpl( + createDrainContext(this), + channelId, + priorForcedEntry, + ); } - markChannelRead(channelId: string, sourceScope?: string): MarkResult { + markChannelRead( + channelId: string, + sourceScope?: string, + markAt?: number, + ): MarkResult { return markChannelReadImpl( createDrainContext(this), channelId, sourceScope, + markAt, ); } diff --git a/desktop/src/features/channels/readState/readStateStorage.ts b/desktop/src/features/channels/readState/readStateStorage.ts index 7b93d12314..9bb8537d22 100644 --- a/desktop/src/features/channels/readState/readStateStorage.ts +++ b/desktop/src/features/channels/readState/readStateStorage.ts @@ -15,6 +15,7 @@ import type { PendingIntent, PendingIntentOp, } from "@/features/channels/pendingOverrideIntents"; +import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; export type StoredReadState = { contexts: Map; @@ -221,12 +222,14 @@ function readOverrideState(pubkey: string): OverrideStateBlob { if (isPlainRecord(parsed.pi)) { for (const [channelId, entry] of Object.entries(parsed.pi)) { if (!isPlainRecord(entry)) continue; - const { gen, op, sourceScope, readTarget } = entry as { - gen?: unknown; - op?: unknown; - sourceScope?: unknown; - readTarget?: unknown; - }; + const { gen, op, sourceScope, readTarget, priorForcedEntry } = + entry as { + gen?: unknown; + op?: unknown; + sourceScope?: unknown; + readTarget?: unknown; + priorForcedEntry?: unknown; + }; if (typeof gen !== "number" || !Number.isInteger(gen) || gen < 1) continue; if (op !== "unread" && op !== "read") continue; @@ -239,6 +242,11 @@ function readOverrideState(pubkey: string): OverrideStateBlob { op, ...(sourceScope !== undefined ? { sourceScope } : {}), ...(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 } + : {}), }; pendingIntents.set(channelId, intent); } @@ -375,6 +383,7 @@ export function writeStoredReadState( op: PendingIntentOp; sourceScope?: string; readTarget?: number; + priorForcedEntry?: ForcedUnreadEntry; } > = {}; for (const [channelId, intent] of pendingIntents) { @@ -383,6 +392,7 @@ export function writeStoredReadState( op: PendingIntentOp; sourceScope?: string; readTarget?: number; + priorForcedEntry?: ForcedUnreadEntry; } = { gen: intent.gen, op: intent.op, @@ -390,6 +400,8 @@ export function writeStoredReadState( if (intent.sourceScope !== undefined) entry.sourceScope = intent.sourceScope; if (intent.readTarget !== undefined) entry.readTarget = intent.readTarget; + if (intent.priorForcedEntry !== undefined) + entry.priorForcedEntry = intent.priorForcedEntry; piObj[channelId] = entry; } const ok4 = setLocalStorageItemWithRecovery( diff --git a/desktop/src/features/channels/readState/useReadState.ts b/desktop/src/features/channels/readState/useReadState.ts index bba0a10dcb..a88928e03b 100644 --- a/desktop/src/features/channels/readState/useReadState.ts +++ b/desktop/src/features/channels/readState/useReadState.ts @@ -8,6 +8,8 @@ import { import type { OverrideLiveness } from "@/features/channels/readState/readStateFormat"; import type { RelayClient } from "@/shared/api/relayClientSession"; import { pendingOverrideIntentStore } from "@/features/channels/pendingOverrideIntents"; +import type { DrainOutcome } from "@/features/channels/readState/readStateDrain"; +import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore"; const noopGetTimestamp = () => null; const noopMarkRead = () => {}; @@ -24,7 +26,10 @@ const noopSetOnDrainOutcome = (): void => {}; * swept / replaced when a real manager hydrates from storage. Returns the * tri-state `queued` result so `applyOverrideUnread` keeps the forced entry. */ -function noopMarkChannelUnread(channelId: string): MarkResult { +function noopMarkChannelUnread( + channelId: string, + _priorForcedEntry?: ForcedUnreadEntry, +): MarkResult { pendingOverrideIntentStore.enqueue(channelId, "unread"); return { status: "queued" }; } @@ -37,6 +42,7 @@ function noopMarkChannelUnread(channelId: string): MarkResult { function noopMarkChannelRead( channelId: string, sourceScope?: string, + _markAt?: number, ): MarkResult { pendingOverrideIntentStore.enqueue(channelId, "read", sourceScope, undefined); return { status: "queued" }; @@ -127,9 +133,9 @@ export function useReadState( // Override APIs for the NIP-RS manual-unread layer (slice 3 seam). const markChannelUnread = React.useCallback( - (channelId: string): MarkResult => { + (channelId: string, priorForcedEntry?: ForcedUnreadEntry): MarkResult => { return ( - managerRef.current?.markChannelUnread(channelId) ?? { + managerRef.current?.markChannelUnread(channelId, priorForcedEntry) ?? { status: "refused", reason: "load_incomplete", } @@ -139,9 +145,9 @@ export function useReadState( ); const markChannelRead = React.useCallback( - (channelId: string, sourceScope?: string): MarkResult => { + (channelId: string, sourceScope?: string, markAt?: number): MarkResult => { return ( - managerRef.current?.markChannelRead(channelId, sourceScope) ?? { + managerRef.current?.markChannelRead(channelId, sourceScope, markAt) ?? { status: "refused", reason: "load_incomplete", } @@ -164,17 +170,7 @@ export function useReadState( // Wire the drain outcome callback into the current manager. // Called by useUnreadChannels to route rollback/cleanup through the hook layer. const setOnDrainOutcome = React.useCallback( - ( - callback: - | (( - channelId: string, - op: string, - status: string, - reason?: string, - sourceScope?: string, - ) => void) - | null, - ): void => { + (callback: ((outcome: DrainOutcome) => void) | null): void => { if (managerRef.current) { managerRef.current.onDrainOutcome = callback; } diff --git a/desktop/src/features/channels/readStateOverride.ts b/desktop/src/features/channels/readStateOverride.ts index cb22fde328..a77c678931 100644 --- a/desktop/src/features/channels/readStateOverride.ts +++ b/desktop/src/features/channels/readStateOverride.ts @@ -13,6 +13,7 @@ import type { MarkResult } from "@/features/channels/readState/readStateManager" import { forcedUnreadStore, removeForcedUnreadSource, + type ForcedUnreadEntry, type ForcedUnreadMap, type ForcedUnreadSource, } from "@/features/channels/forcedUnreadStore"; @@ -51,8 +52,15 @@ export type OverrideAPIs = { * treated as ready — null liveness after an incomplete load means "not found * in a partial view", not "known absent register". */ isReadStateReady: boolean; - markChannelUnread: (channelId: string) => MarkResult; - markChannelRead: (channelId: string, sourceScope?: string) => MarkResult; + markChannelUnread: ( + channelId: string, + priorForcedEntry?: ForcedUnreadEntry, + ) => MarkResult; + markChannelRead: ( + channelId: string, + sourceScope?: string, + markAt?: number, + ) => MarkResult; getOverrideLiveness: (channelId: string) => OverrideLiveness | null; }; @@ -63,12 +71,16 @@ export type OverrideAPIs = { * * Returns false silently (no toast) when `status: "queued"` — the intent is * enqueued for the next drain; optimistic presentation continues unchanged. + * + * `priorForcedEntry` is the forced-unread entry that existed before the + * optimistic write; forwarded into the intent for restart-safe rollback. */ export function applyOverrideUnread( channelId: string, apis: OverrideAPIs, + priorForcedEntry?: ForcedUnreadEntry, ): boolean { - const result = apis.markChannelUnread(channelId); + const result = apis.markChannelUnread(channelId, priorForcedEntry); if (result.status === "queued") { // Intent enqueued — optimistic forced-entry was already written by the // caller before this call (write ordering per plan ruling 2). No toast. @@ -119,11 +131,17 @@ export type OverrideReadOutcome = "overrideCleared" | "overrideStillActive"; * `sourceScope` is forwarded to markChannelRead so the intent can carry the * exact source being cleared; the drain surfaces it to onDrainOutcome for * exact hook-layer source cleanup (rather than whole-entry deletion). + * + * `markAt` is the authoritative read target computed at click time (e.g., the + * newest observed message timestamp). It is passed to the intent so the drain + * can advance the frontier to the correct logical position even when the load + * completes after the click. */ export function applyOverrideRead( channelId: string, apis: OverrideAPIs, sourceScope?: string, + markAt?: number, ): OverrideReadOutcome { // Step 1: load complete + no register → known absence, cleared. // Guard: only treat null as "known absent" when the load is complete. @@ -138,7 +156,7 @@ export function applyOverrideRead( // register or frontier mutation. Sources are preserved; the intent drains // on the next complete-load transition and rawUnread suppresses presentation // via the pending-read precedence rule. - const result = apis.markChannelRead(channelId, sourceScope); + const result = apis.markChannelRead(channelId, sourceScope, markAt); if (result.status === "queued") { // Intent enqueued — sources preserved, rawUnread handles presentation. diff --git a/desktop/src/features/channels/useUnreadChannels.ts b/desktop/src/features/channels/useUnreadChannels.ts index 075c04d038..8fe28dd045 100644 --- a/desktop/src/features/channels/useUnreadChannels.ts +++ b/desktop/src/features/channels/useUnreadChannels.ts @@ -196,7 +196,6 @@ export function useUnreadChannels( 0, ); - // Wire drain outcome callback; returns the pending-unread-snapshots ref. const pendingUnreadSnapshotsRef = useDrainOutcomeCallback( setOnDrainOutcome, forcedUnreadRef, @@ -260,7 +259,6 @@ export function useUnreadChannels( topLevelOnly?: boolean; } = {}, ) => { - // Frontier advance first (spec order: advance → C-bump → inspect liveness). const observedLatest = topLevelOnly ? undefined : latestByChannelRef.current.get(channelId); @@ -274,7 +272,12 @@ export function useUnreadChannels( // C-bump; clear local presentation only when liveness is confirmed inactive // and the caller has not opted to preserve the forced-unread state. if (!preserveForcedUnread) { - const outcome = applyOverrideRead(channelId, overrideApis); + const outcome = applyOverrideRead( + channelId, + overrideApis, + undefined, + markAt ?? undefined, + ); if (outcome === "overrideCleared") { if (Object.hasOwn(forcedUnreadRef.current, channelId)) { delete forcedUnreadRef.current[channelId]; @@ -311,18 +314,14 @@ export function useUnreadChannels( bumpLatestVersion, ); - // Mark channel unread: write forced entry first (write ordering), then apply override. - // Refused results undo the forced entry. Returns true when accepted (applied or queued). // biome-ignore lint/correctness/useExhaustiveDependencies: pendingUnreadSnapshotsRef is a stable ref; .current.set is Map.prototype.set (stable identity) const markChannelUnread = React.useCallback( (channelId: string, source?: ForcedUnreadSource): boolean => { - // Snapshot the prior entry before the optimistic write for drain rollback. const priorEntry = Object.hasOwn(forcedUnreadRef.current, channelId) ? forcedUnreadRef.current[channelId] : undefined; - // Write forced entry first (write ordering), then apply override. forcedMarkChannelUnread(channelId, source); - if (!applyOverrideUnread(channelId, overrideApis)) { + if (!applyOverrideUnread(channelId, overrideApis, priorEntry)) { // Refused immediately — restore exact prior entry (or remove if none existed). if (priorEntry !== undefined) forcedUnreadRef.current[channelId] = priorEntry; @@ -331,7 +330,6 @@ export function useUnreadChannels( bumpLatestVersion(); return false; } - // Store snapshot; drain will call onDrainOutcome to commit or rollback. pendingUnreadSnapshotsRef.current.set(channelId, priorEntry); return true; }, @@ -834,10 +832,8 @@ export function useUnreadChannels( let unreadChannelNotificationCount = 0; for (const channel of channels) { - const isForcedUnread = - // Manager liveness is authoritative (isReadStateReady). Use - // getOverrideLiveness, not forcedUnreadRef which may be stale. - getOverrideLiveness(channel.id)?.active === true; + // getOverrideLiveness is authoritative when load is complete; forcedUnreadRef may be stale. + const isForcedUnread = getOverrideLiveness(channel.id)?.active === true; if (channel.id === activeChannelId && !isForcedUnread) continue; const observedEvents = observedUnreadEventsByChannelRef.current.get( @@ -946,7 +942,12 @@ export function useUnreadChannels( // observed state. applyOverrideRead returns overrideStillActive on refusal; // only overrideCleared triggers deletion. removeChannel is the sole mutator // of latestByChannelRef and observedUnreadEventsByChannelRef. - const outcome = applyOverrideRead(channelId, overrideApis); + const outcome = applyOverrideRead( + channelId, + overrideApis, + undefined, + unixSeconds ?? undefined, + ); if (outcome === "overrideCleared") { if (Object.hasOwn(forcedUnreadRef.current, channelId)) { delete forcedUnreadRef.current[channelId]; diff --git a/desktop/src/features/channels/useUnreadChannelsHelpers.ts b/desktop/src/features/channels/useUnreadChannelsHelpers.ts index c376a60436..ff9dcfdeec 100644 --- a/desktop/src/features/channels/useUnreadChannelsHelpers.ts +++ b/desktop/src/features/channels/useUnreadChannelsHelpers.ts @@ -21,6 +21,7 @@ import { type ForcedUnreadMap, type ForcedUnreadSource, } from "@/features/channels/forcedUnreadStore"; +import type { DrainOutcome } from "@/features/channels/readState/readStateDrain"; export type UseUnreadChannelsOptions = UseLiveChannelUpdatesOptions & { pubkey?: string; @@ -101,26 +102,17 @@ export function resolveObservedUnreadRootId(tags: string[][]): string | null { * via `setOnDrainOutcome`. Creates and owns the pending-unread-snapshots map. * Returns the snapshots ref for use in markChannelUnread. * - * Handles: - * - `unread + refused` → restore forced-unread entry to its pre-mark snapshot - * - `unread + applied` → discard snapshot (intent succeeded) - * - `read + applied` → remove forced-unread entry (or specific source), mirroring - * the synchronous overrideCleared path + * Handles outcomes via exhaustive switch on the typed DrainOutcome union: + * - `genuine-refusal (unread)` → restore forced-unread entry to its pre-mark snapshot + * - `applied-unread` → discard snapshot (intent succeeded) + * - `applied-read` → remove forced-unread entry (or specific source) + * - `silent-inactive` → same source cleanup as applied-read, no toast + * - `genuine-refusal (read)` → toast already fired in drain; no entry mutation * * Genuine refusal toasts are fired in the drain itself and not repeated here. */ export function useDrainOutcomeCallback( - setOnDrainOutcome: ( - cb: - | (( - channelId: string, - op: string, - status: string, - reason?: string, - sourceScope?: string, - ) => void) - | null, - ) => void, + setOnDrainOutcome: (cb: ((outcome: DrainOutcome) => void) | null) => void, forcedUnreadRef: React.MutableRefObject, pubkey: string | undefined, bumpLatestVersion: () => void, @@ -130,43 +122,85 @@ export function useDrainOutcomeCallback( ); // biome-ignore lint/correctness/useExhaustiveDependencies: stable refs; setOnDrainOutcome is stable React.useEffect(() => { - setOnDrainOutcome((channelId, op, status, _reason, sourceScope) => { - if (op === "unread" && status === "refused") { - // Roll back the optimistic forced-unread entry to its exact prior state. - const prior = pendingUnreadSnapshotsRef.current.get(channelId); - pendingUnreadSnapshotsRef.current.delete(channelId); - if (prior === undefined) { - if (Object.hasOwn(forcedUnreadRef.current, channelId)) { - delete forcedUnreadRef.current[channelId]; + setOnDrainOutcome((outcome) => { + switch (outcome.kind) { + case "genuine-refusal": { + if (outcome.op !== "unread") break; + // Roll back the optimistic forced-unread entry to its exact prior state. + // For same-session rollbacks, use the in-memory snapshot. + // For post-restart rollbacks, fall back to the persisted priorForcedEntry + // from the intent (restart-safe rollback, fix for pass-2 finding 3). + const inMemoryPrior = pendingUnreadSnapshotsRef.current.get( + outcome.channelId, + ); + pendingUnreadSnapshotsRef.current.delete(outcome.channelId); + // Pick: in-memory snapshot if present (same session); otherwise use + // persisted priorForcedEntry; otherwise treat as "no prior entry". + const hasPrior = + inMemoryPrior !== undefined || + outcome.priorForcedEntry !== undefined; + const prior = + inMemoryPrior !== undefined + ? inMemoryPrior + : outcome.priorForcedEntry; + if (!hasPrior) { + // No prior entry existed — remove the whole entry if present. + if (Object.hasOwn(forcedUnreadRef.current, outcome.channelId)) { + delete forcedUnreadRef.current[outcome.channelId]; + if (pubkey) + forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); + } + } else if (prior === undefined) { + // Prior was explicitly "no entry" — delete. + if (Object.hasOwn(forcedUnreadRef.current, outcome.channelId)) { + delete forcedUnreadRef.current[outcome.channelId]; + if (pubkey) + forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); + } + } else { + // Restore exact prior entry byte-for-byte. + forcedUnreadRef.current[outcome.channelId] = prior; if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); bumpLatestVersion(); } - } else { - forcedUnreadRef.current[channelId] = prior; - if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); - bumpLatestVersion(); + break; } - } else if (op === "unread" && status === "applied") { - pendingUnreadSnapshotsRef.current.delete(channelId); - } else if (op === "read" && status === "applied") { - if (Object.hasOwn(forcedUnreadRef.current, channelId)) { - if (sourceScope !== undefined) { - const current = forcedUnreadRef.current[channelId]; - const next = removeForcedUnreadSource( - current, - sourceScope as ForcedUnreadSource, - ); - if (next !== undefined) { - forcedUnreadRef.current[channelId] = next; + case "applied-unread": { + pendingUnreadSnapshotsRef.current.delete(outcome.channelId); + break; + } + case "applied-read": + case "silent-inactive": { + // Both cases: remove the exact source (or whole entry) from forced store. + const { channelId, sourceScope } = outcome; + if (Object.hasOwn(forcedUnreadRef.current, channelId)) { + if (sourceScope !== undefined) { + const current = forcedUnreadRef.current[channelId]; + const next = removeForcedUnreadSource( + current, + sourceScope as ForcedUnreadSource, + ); + if (next !== undefined) { + forcedUnreadRef.current[channelId] = next; + } else { + delete forcedUnreadRef.current[channelId]; + } } else { delete forcedUnreadRef.current[channelId]; } - } else { - delete forcedUnreadRef.current[channelId]; + if (pubkey) + forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); } - if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); - bumpLatestVersion(); + break; + } + default: { + // Exhaustiveness check — TypeScript ensures all cases are handled. + const _exhaustive: never = outcome; + void _exhaustive; } } });