jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 22:41:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(readState): round-3 v4 conformance — typed outcomes, markAt, restart-safe rollback, auto-retry, non-reentrant drain

Browse Source 
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 <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-08 11:53:40 -04:00
parent 9212fd40b2
commit f6a12809dc
9 changed files with 733 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
desktop/src/features/channels/pendingOverrideIntents.ts
View File

@@ -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;

165
desktop/src/features/channels/readState/readStateDrain.ts
View File

@@ -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<string, AppliedReceipt>;
readonly publishableContextIds: Set<string>;
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",

414
desktop/src/features/channels/readState/readStateManager.test.mjs
View File

@@ -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();
});

90
desktop/src/features/channels/readState/readStateManager.ts
View File

@@ -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<void> {
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<void>((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,
);
}

24
desktop/src/features/channels/readState/readStateStorage.ts
View File

@@ -15,6 +15,7 @@ import type {
PendingIntent,
PendingIntentOp,
} from "@/features/channels/pendingOverrideIntents";
import type { ForcedUnreadEntry } from "@/features/channels/forcedUnreadStore";
export type StoredReadState = {
contexts: Map<string, number>;
@@ -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(

28
desktop/src/features/channels/readState/useReadState.ts
View File

@@ -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;
}

26
desktop/src/features/channels/readStateOverride.ts
View File

@@ -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.

29
desktop/src/features/channels/useUnreadChannels.ts
View File

@@ -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];

122
desktop/src/features/channels/useUnreadChannelsHelpers.ts
View File

@@ -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<ForcedUnreadMap>,
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;
}
}
});