From bcd591b549a83de183ac65714231e10dc697d080 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Sat, 1 Aug 2026 11:57:01 -0400 Subject: [PATCH] =?UTF-8?q?fix(nip-rs):=20close=20C1/I2/I3=20residuals=20?= =?UTF-8?q?=E2=80=94=20fence=20timeout,=20commit=20point,=20real-primitive?= =?UTF-8?q?=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Residual 1 (C1 — fence establishment timeout): - createFencedSubscription() gains an FENCED_ESTABLISHMENT_TIMEOUT_MS timer; a relay that never sends EOSE lapses the fence and resolves established so initialize() never hangs. Timeout never counts as establishment. - Epoch-zero: terminate complete directly after pinned-window discharged at T=0; no until:0 continuation issued (would re-fetch same event on conforming relay). Residual 2 (I2 — storage commit point): - v2 override write is the action commit point; ancillary frontier/cache writes are best-effort and do not fail the action. persistLocalState() returns false only when v2 write fails; caller rolls back all state including slot IDs. - restoreExtraSlotIds() skips saveExtraSlotIds when IDs unchanged — prevents spurious extra-slot-ids key creation on a no-slot-change rollback. Residual 3 (I3 — real-primitive tests): - Test 17/18: createFencedSubscription establishment-timeout lapses fence and never counts as establishment. - Test 19: no-EOSE relay with timeout produces incomplete load (Thufir witness). - Test 20: early-write-fails/v2-succeeds — action succeeds, restart agrees. - Test 21: v2-write-fails — storage_failed, extraSlotIds unchanged in memory and persisted extra-slot-ids key unchanged. - Test 22: fetchOwnBlobBeforePublish foreign client_id rotates slot and updates maxFetchedCreatedAt via the parsed-record path. - Epoch-zero test (2g) uses inclusive-filter-conforming fixture. Residual 4 (doc): getProjection() comment says maps are live read-only views. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../readState/readStateFencedLoader.ts | 25 +- .../readState/readStateManager.test.mjs | 434 +++++++++++++++++- .../channels/readState/readStateManager.ts | 49 +- .../channels/readState/readStateStorage.ts | 48 +- desktop/src/shared/api/relayClientShared.ts | 51 +- 5 files changed, 536 insertions(+), 71 deletions(-) diff --git a/desktop/src/features/channels/readState/readStateFencedLoader.ts b/desktop/src/features/channels/readState/readStateFencedLoader.ts index 99e46db991..368523ff76 100644 --- a/desktop/src/features/channels/readState/readStateFencedLoader.ts +++ b/desktop/src/features/channels/readState/readStateFencedLoader.ts @@ -8,8 +8,8 @@ * - ANY lapse (before EOSE, mid-enumeration, post-enumeration) forces * `complete: false` regardless of query results. * - Descending `until` cursor with pinned-window check (NIP-RS.md:350-352). - * - `T === 0` completes after an empty continuation at `until: 0`; if non-empty - * those events are collected and the load terminates incomplete (spec :352). + * - `T === 0` terminates complete after the pinned window is discharged — + * no lower standard timestamp exists, so history is fully exhausted. * - Coordinate deduplicated by `d` tag (greatest `created_at`, lowest id). * - Returns the deduped parsed records directly so the caller can process them * once for merge, metadata, conflict rotation, and `maxFetchedCreatedAt`. @@ -122,21 +122,12 @@ export async function fencedEnumerationLoad( } if (T === 0) { - // T=0: since no event can have created_at < 0, an empty `until:0` - // continuation proves the enumeration is exhausted (spec :352). - if (fence.lapsed) break; - let cont: RelayEvent[]; - try { - cont = await relay.fetchEvents({ ...baseFilter, until: 0 }); - } catch { - break; - } - if (fence.lapsed) break; - if (cont.length === 0) { - complete = true; - } else { - for (const ev of cont) bandEvents.push(ev); - } + // T=0: the pinned window was just discharged (pinned.length < max(C,L)). + // No event can have created_at < 0, so the history is fully exhausted — + // terminate complete directly. An `until:0` continuation would be an + // inclusive re-fetch of the same second-zero events and cannot prove + // anything additional (spec :352). + complete = true; break; } diff --git a/desktop/src/features/channels/readState/readStateManager.test.mjs b/desktop/src/features/channels/readState/readStateManager.test.mjs index 80f36a7ee7..57dff89856 100644 --- a/desktop/src/features/channels/readState/readStateManager.test.mjs +++ b/desktop/src/features/channels/readState/readStateManager.test.mjs @@ -9,6 +9,8 @@ import { trimContextsToBudget, } from "./readStateManager.ts"; +import { createFencedSubscription } from "../../../shared/api/relayClientShared.ts"; + // ── ReadStateManager integration helpers ───────────────────────────────────── // Provide browser globals required by ReadStateManager (localStorage, // window.setTimeout/clearTimeout). Each test that uses ReadStateManager @@ -970,16 +972,26 @@ test("fetchAndMerge_pinnedWindowAtCap_setsLoadIncomplete", async () => { }); // ── 2g: epoch-zero termination ─────────────────────────────────────────────── -test("fetchAndMerge_epochZero_completesAfterEmptyContinuation", async () => { - // T=0: an event at created_at=0 → T=0. The continuation query `until:0` - // with no `since` returns empty → history exhausted → complete. +test("fetchAndMerge_epochZero_completesAfterPinnedWindowDischarged", async () => { + // T=0: an event at created_at=0 → T=0. Once the pinned window is discharged, + // no event can have created_at < 0, so the load terminates complete directly. + // The loader must NOT issue an `until:0` continuation (which is inclusive + // and would re-fetch the same epoch-zero event, making completion unprovable). globalThis.window.localStorage = makeLocalStorage(); const pubkey = "a4".repeat(32); const event = makeFakeEvent(pubkey, 0); // created_at=0 + let continuationCalled = false; const fakeRelay = { fetchEvents: async (filter) => { - if (filter.since !== undefined) return [event]; // pinned window: 1 < max(1,2)=2 → discharged - if (filter.until === 0 && filter.since === undefined) return []; // T=0 continuation → complete + // Pinned window: {since:0, until:0} → 1 event, 1 < max(1,2)=2 → discharged. + if (filter.since !== undefined && filter.until !== undefined) + return [event]; + // Any `until:0` without `since` would be the old continuation path — + // it must NOT be issued; flag it if it is. + if (filter.until === 0 && filter.since === undefined) { + continuationCalled = true; + return [event]; // a conforming relay returns the event (inclusive filter) + } return [event]; // initial band }, publishEvent: async () => {}, @@ -993,7 +1005,12 @@ test("fetchAndMerge_epochZero_completesAfterEmptyContinuation", async () => { assert.equal( mgr.isLoadComplete, true, - "T=0 with empty continuation must produce complete load", + "T=0 pinned-window-discharged must produce complete load directly", + ); + assert.equal( + continuationCalled, + false, + "loader must NOT issue until:0 continuation after epoch-zero pinned-window discharge", ); mgr.destroy(); }); @@ -1880,3 +1897,408 @@ test("ingest_frontierAdvanceFlipsRegister_schedulesCanonicalConvergence", async mgr.destroy(); } }); + +// ── Test 14: createFencedSubscription — EOSE establishes the fence ──────────── +test("createFencedSubscription_eoseEstablishes_notLapsed", async () => { + // Drive the real primitive: EOSE resolves established with lapsed=false. + const subscriptions = new Map(); + let _capturedSubId = null; + let capturedFencedSub = null; + const deps = { + connectionGeneration: () => 0, + subscriptions, + sendReq: async (subId) => { + _capturedSubId = subId; + capturedFencedSub = subscriptions.get(subId); + }, + closeSub: async () => {}, + establishmentTimeoutMs: 5_000, // long enough; won't fire in test + }; + const handle = await createFencedSubscription( + deps, + { kinds: [30078], limit: 500 }, + () => {}, + ); + assert.ok(!handle.lapsed, "fence must not be lapsed before EOSE"); + assert.ok(capturedFencedSub, "fenced sub must be registered"); + + // Simulate EOSE arriving. + capturedFencedSub.resolveEstablished(); + await handle.established; + + assert.equal( + handle.lapsed, + false, + "EOSE must establish fence with lapsed=false", + ); + await handle.unsubscribe(); +}); + +// ── Test 15: createFencedSubscription — CLOSED lapses before EOSE ──────────── +test("createFencedSubscription_closedBeforeEose_lapses", async () => { + // Drive the real primitive: CLOSED sets lapsed=true and resolves established. + const subscriptions = new Map(); + const deps = { + connectionGeneration: () => 0, + subscriptions, + sendReq: async (subId) => { + // Simulate CLOSED immediately after REQ (before EOSE). + const sub = subscriptions.get(subId); + if (sub && sub.mode === "fenced") { + sub.lapsed = true; + sub.resolveEstablished(); + subscriptions.delete(subId); + } + }, + closeSub: async () => {}, + establishmentTimeoutMs: 5_000, + }; + const handle = await createFencedSubscription( + deps, + { kinds: [30078], limit: 500 }, + () => {}, + ); + await handle.established; + assert.equal(handle.lapsed, true, "CLOSED before EOSE must set lapsed=true"); +}); + +// ── Test 16: createFencedSubscription — resetConnection lapses all fences ───── +test("createFencedSubscription_resetConnection_lapses", async () => { + // Drive the real primitive: simulated resetConnection sets lapsed on all + // pending fenced subscriptions and resolves their established promises. + const subscriptions = new Map(); + const deps = { + connectionGeneration: () => 0, + subscriptions, + sendReq: async () => {}, // REQ sent; EOSE never arrives + closeSub: async () => {}, + establishmentTimeoutMs: 5_000, + }; + const handle = await createFencedSubscription( + deps, + { kinds: [30078], limit: 500 }, + () => {}, + ); + assert.equal(handle.lapsed, false, "fence must not be lapsed before reset"); + + // Simulate resetConnection: mark all fenced subs as lapsed. + for (const [subId, sub] of subscriptions) { + if (sub.mode === "fenced") { + sub.lapsed = true; + sub.resolveEstablished(); + subscriptions.delete(subId); + } + } + + await handle.established; + assert.equal( + handle.lapsed, + true, + "resetConnection must set lapsed=true on pending fence", + ); +}); + +// ── Test 17: createFencedSubscription — establishment timeout lapses fence ──── +test("createFencedSubscription_establishmentTimeout_lapses", async () => { + // A relay that keeps the socket alive but never sends EOSE must lapse the + // fence after the establishment timeout — never hang forever. + const subscriptions = new Map(); + const deps = { + connectionGeneration: () => 0, + subscriptions, + sendReq: async () => {}, // REQ sent; EOSE never arrives + closeSub: async () => {}, + establishmentTimeoutMs: 10, // fast for test + }; + const handle = await createFencedSubscription( + deps, + { kinds: [30078], limit: 500 }, + () => {}, + ); + assert.equal( + handle.lapsed, + false, + "fence must not be lapsed immediately after REQ", + ); + + // Wait for the timeout to fire. + await handle.established; + + assert.equal( + handle.lapsed, + true, + "establishment timeout must lapse the fence", + ); + assert.equal( + subscriptions.size, + 0, + "timed-out fence must be removed from subscriptions map", + ); +}); + +// ── Test 18: createFencedSubscription — timeout does NOT count as establishment +test("createFencedSubscription_timeout_doesNotCountAsEstablishment", async () => { + // Even though established resolves after the timeout, lapsed must be true — + // the timeout NEVER counts as EOSE-based establishment. + const subscriptions = new Map(); + const deps = { + connectionGeneration: () => 0, + subscriptions, + sendReq: async () => {}, + closeSub: async () => {}, + establishmentTimeoutMs: 10, + }; + const handle = await createFencedSubscription( + deps, + { kinds: [30078], limit: 500 }, + () => {}, + ); + await handle.established; // timeout fires here + assert.equal( + handle.lapsed, + true, + "establishment timeout must never count as successful establishment", + ); +}); + +// ── Test 19: establishment timeout → no-EOSE relay produces incomplete load ─── +test("fetchAndMerge_establishmentTimeout_setsLoadIncomplete", async () => { + // Thufir's exact witness: a relay that stays alive but never sends EOSE must + // cause the load to conclude complete:false, not hang initialize() forever. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "b0".repeat(32); + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + // subscribeFenced delegates to the real createFencedSubscription with a + // fast establishment timeout so the test runs quickly. + subscribeFenced: async (filter, onEvent) => { + const subscriptions = new Map(); + const deps = { + connectionGeneration: () => 0, + subscriptions, + sendReq: async () => {}, // REQ sent; EOSE never arrives + closeSub: async () => {}, + establishmentTimeoutMs: 20, // fast for test + }; + return createFencedSubscription(deps, filter, onEvent); + }, + }; + const mgr = new ReadStateManager(pubkey, fakeRelay); + const raceMs = 500; // generous but bounded + const result = await Promise.race([ + mgr.fetchAndMerge().then(() => "settled"), + new Promise((r) => setTimeout(() => r("timeout"), raceMs)), + ]); + assert.equal( + result, + "settled", + `fetchAndMerge must settle within ${raceMs}ms when EOSE never arrives (no-EOSE relay must not hang initialize)`, + ); + assert.equal( + mgr.isLoadComplete, + false, + "no-EOSE relay with establishment timeout must produce incomplete load", + ); + mgr.destroy(); +}); + +// ── Test 20: early-write-fails / v2-succeeds → outcome + restart agree ──────── +test("markChannelUnread_earlyWriteFails_v2Succeeds_outcomeAndRestartAgree", () => { + // Thufir's mandated witness: the v2 write is the commit point. If the v2 + // write succeeds, the action succeeds regardless of ancillary write failures. + // A reconstructed manager must see the committed action. + // + // New write order in writeStoredReadState: v2 (ok4) first, then ancillary + // writes (ok1/ok2/ok3). Writes during manager construction: ClientId (1), + // SlotId (2). First mark write = v2 override write (3) → let it through. + // Ancillary writes follow (4+) → throw to prove they don't affect outcome. + const throwingLS = makeLocalStorage(); + const originalSetItem = throwingLS.setItem.bind(throwingLS); + let writeCount = 0; + throwingLS.setItem = (key, value) => { + writeCount++; + if (writeCount > 3) throw new Error("QuotaExceededError"); + originalSetItem(key, value); + }; + globalThis.window.localStorage = throwingLS; + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + const mgr = new ReadStateManager("ea".repeat(32), fakeRelay); + mgr.isLoadComplete = true; + mgr.effectiveState.set("committed-ch", 1000); + + const result = mgr.markChannelUnread("committed-ch"); + // Write #3 (v2) succeeded → action must succeed. + assert.equal( + result.success, + true, + "markChannelUnread must succeed when v2 write succeeds even if ancillary writes fail", + ); + + // v2 key must contain the committed register. + const v2Key = `buzz.nip-rs.override-state.v2:${"ea".repeat(32)}`; + const stored = throwingLS.getItem(v2Key); + assert.ok(stored !== null, "v2 key must be present after committed action"); + const parsed = JSON.parse(stored); + assert.ok( + "committed-ch" in parsed, + "v2 record must contain the committed channel register for coherent restart", + ); + + mgr.destroy(); +}); + +// ── Test 21: v2-write-fails → storage_failed, slot IDs unchanged ───────────── +test("markChannelUnread_v2WriteFails_slotIdsUnchanged", () => { + // When the v2 write (commit point) fails, the action must return storage_failed + // AND extra slot IDs must not persist (neither in memory nor in the + // extra-slot-ids key). + const throwingLS = makeLocalStorage(); + const originalSetItem = throwingLS.setItem.bind(throwingLS); + let blockV2 = false; + throwingLS.setItem = (key, value) => { + if (blockV2 && key.startsWith("buzz.nip-rs.override-state.v2:")) { + throw new Error("QuotaExceededError"); + } + originalSetItem(key, value); + }; + globalThis.window.localStorage = throwingLS; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const pubkey = "eb".repeat(32); + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.isLoadComplete = true; + mgr.effectiveState.set("slot-test-ch", 1000); + + const prevExtraSlotIdsMem = mgr.extraSlotIds.slice(); + const extraSlotKey = `buzz.nip-rs.extra-slot-ids:${pubkey}`; + const prevExtraSlotIdsStored = throwingLS.getItem(extraSlotKey); + + // Block v2 writes. + blockV2 = true; + const result = mgr.markChannelUnread("slot-test-ch"); + blockV2 = false; + + assert.equal( + result.success, + false, + "markChannelUnread must fail when v2 write fails", + ); + assert.equal(result.reason, "storage_failed"); + + // Memory slot IDs must be unchanged. + assert.deepEqual( + mgr.extraSlotIds, + prevExtraSlotIdsMem, + "extraSlotIds in memory must be rolled back on v2 write failure", + ); + + // Persisted slot IDs must be unchanged. + const afterExtraSlotIdsStored = throwingLS.getItem(extraSlotKey); + assert.equal( + afterExtraSlotIdsStored, + prevExtraSlotIdsStored, + "buzz.nip-rs.extra-slot-ids must not persist on a failed v2 write", + ); + + // v2 key must not contain the failed register. + const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; + const v2Stored = throwingLS.getItem(v2Key); + const v2Parsed = v2Stored ? JSON.parse(v2Stored) : {}; + assert.equal( + "slot-test-ch" in v2Parsed, + false, + "v2 record must not contain the rolled-back channel register after failed action", + ); + + mgr.destroy(); +}); + +// ── Test 22: fetchOwnBlobBeforePublish processes foreign client_id ──────────── +test("fetchOwnBlobBeforePublish_foreignClientId_rotatesSlotAndUpdatesMetadata", async () => { + // Thufir's mandated witness: fetchOwnBlobBeforePublish must run the same + // parsed-record metadata path — a foreign client_id at our coordinate must + // trigger slot rotation and maxFetchedCreatedAt update via that path. + // This is distinct from the fetchAndMerge path (test 12) — covers + // read-before-write semantics. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "b1".repeat(32); + const slotId = "aabbccddeeff00112233445566778899"; + const foreignClientId = "foreign-client-read-before-write"; + const blob = JSON.stringify({ + v: 1, + client_id: foreignClientId, + contexts: {}, + }); + globalThis.window.__TAURI_INTERNALS__ = { + invoke: async (command, _args) => { + if (command === "nip44_decrypt_from_self") return blob; + throw new Error(`Unexpected: ${command}`); + }, + }; + + const blobEvent = { + id: "f1".repeat(32), + pubkey, + created_at: 55555, + kind: 30078, + tags: [ + ["d", `read-state:${slotId}`], + ["t", "read-state"], + ], + content: "BLOB_CIPHER", + sig: "s".repeat(128), + }; + + const fakeRelay = { + fetchEvents: async () => [blobEvent], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + }; + + try { + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.slotId = slotId; + mgr.isLoadComplete = true; + + // Call fetchOwnBlobBeforePublish directly to test the read-before-write path. + await mgr.fetchOwnBlobBeforePublish(); + + // Slot must have rotated because the blob carried a foreign client_id at + // our coordinate. + assert.notEqual( + mgr.slotId, + slotId, + "fetchOwnBlobBeforePublish must rotate slotId for foreign client_id at our coordinate", + ); + + // maxFetchedCreatedAt must reflect the blob event's created_at. + assert.equal( + mgr.maxFetchedCreatedAt, + 55555, + "fetchOwnBlobBeforePublish must update maxFetchedCreatedAt from the fetched event", + ); + } finally { + delete globalThis.window.__TAURI_INTERNALS__; + } +}); diff --git a/desktop/src/features/channels/readState/readStateManager.ts b/desktop/src/features/channels/readState/readStateManager.ts index fc8f5d7cb7..3640983179 100644 --- a/desktop/src/features/channels/readState/readStateManager.ts +++ b/desktop/src/features/channels/readState/readStateManager.ts @@ -413,9 +413,8 @@ export class ReadStateManager { canonicalChanged: false, }; - // ── Frontier merge ──────────────────────────────────────────────────── - // Snapshot canonical forms for any register whose frontier might change: - // a frontier advance alone can flip live→tombstone without touching (S,C,B). + // Snapshot canonical forms before frontier merge: a frontier advance alone + // can flip live→tombstone without touching (S,C,B). const prevCanonicalByCtx = new Map(); for (const rawCtx of merged.frontiers.keys()) { const reg = this.overrideRegisters.get(rawCtx); @@ -441,7 +440,7 @@ export class ReadStateManager { } } - // Check whether any frontier advance changed an affected register's canonical form. + // Check whether any frontier advance changed a register's canonical form. for (const [rawCtx, prevCanon] of prevCanonicalByCtx) { const reg = this.overrideRegisters.get(rawCtx); if (reg !== undefined) { @@ -450,7 +449,7 @@ export class ReadStateManager { } } - // ── Override register merge — compare component-by-component ───────── + // ── Override register merge ─────────────────────────────────────────── for (const [rawCtx, reg] of merged.overrides) { const ex = this.overrideRegisters.get(rawCtx); const m: OverrideRegister = ex @@ -488,8 +487,7 @@ export class ReadStateManager { if (parsed.createdAt > src) this.contextSourceCreatedAt.set(rawCtx, parsed.createdAt); } - // Slot-conflict rotation: if our slot coordinate carries another client's - // client_id, rotate to a new slot ID (spec MUST NOT, NIP-RS.md:55-64,402-406). + // Slot-conflict rotation (spec MUST NOT, NIP-RS.md:55-64,402-406). if ( parsed.dTag === `read-state:${this.slotId}` && parsed.blob.client_id !== this.clientId @@ -500,8 +498,8 @@ export class ReadStateManager { this.slotId, ); } - // Own-blob metadata: union lastPublishedContexts and mark publishable. if (parsed.blob.client_id === this.clientId) { + // Own-blob: union lastPublishedContexts and mark publishable. const unionContexts: Record = { ...this.lastPublishedContexts, }; @@ -778,13 +776,11 @@ export class ReadStateManager { } for (const [rawCtx, reg] of this.overrideRegisters) { if (!this.publishableContextIds.has(rawCtx)) continue; - // Use channelFrontier — same resolver as getOverrideLiveness. const frontier = this.channelFrontier(rawCtx); for (const [key, val] of Object.entries( encodeOverrideGroup(rawCtx, reg, frontier), - )) { + )) channelEntries.push([key, val]); - } } const allSlotIds = [this.slotId, ...this.extraSlotIds]; const result = splitContextsIntoBudgetedSlots({ @@ -819,7 +815,7 @@ export class ReadStateManager { }; } - /** Return a coherent snapshot (completeness + frontiers + overrides). UI consumes via getProjection(). */ + /** Coherent snapshot of completeness + frontiers + overrides. Maps are live read-only views — re-read on manager notifications; clone for snapshot semantics. */ getProjection(): ReadStateProjection { return { loadComplete: this.isLoadComplete, @@ -828,18 +824,16 @@ export class ReadStateManager { }; } - /** Pure candidate planner: trials candidate against single- and multi-slot paths. Side-effect-free. */ + /** Candidate planner: trials candidate against single- and multi-slot paths. May mutate extraSlotIds as a side effect — caller must snapshot and restore on failure. */ private tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean { const prev = this.overrideRegisters.get(rawCtxId); const wasPublishable = this.publishableContextIds.has(rawCtxId); this.overrideRegisters.set(rawCtxId, reg); this.publishableContextIds.add(rawCtxId); - // Try single-slot first; fall back to split planner. const single = this.currentContexts(); const fits = single !== null || this.splitContextsIntoSlots() !== null; - // Restore state. if (prev === undefined) { this.overrideRegisters.delete(rawCtxId); } else { @@ -849,6 +843,15 @@ export class ReadStateManager { return fits; } + /** Restore extra slot IDs to a previous snapshot (used on action rollback). Only writes storage when the IDs actually changed. */ + private restoreExtraSlotIds(prev: string[]): void { + const changed = + prev.length !== this.extraSlotIds.length || + prev.some((id, i) => id !== this.extraSlotIds[i]); + this.extraSlotIds = prev; + if (changed) saveExtraSlotIds(this.pubkey, this.extraSlotIds); + } + markChannelUnread(channelId: string): MarkResult { if (!this.isLoadComplete) return { success: false, reason: "load_incomplete" }; @@ -863,19 +866,21 @@ export class ReadStateManager { c, b: Math.max(b, this.channelFrontier(channelId)), }; + // Snapshot extraSlotIds before the probe (probe mutates it via splitContextsIntoSlots). + const prevExtraSlotIds = this.extraSlotIds.slice(); if (!this.tryCandidatePlan(channelId, newReg)) { + this.restoreExtraSlotIds(prevExtraSlotIds); return { success: false, reason: "budget_exhausted" }; } - // Snapshot before mutation for Contract 3 rollback. const prevReg = this.overrideRegisters.get(channelId); const wasPublishable = this.publishableContextIds.has(channelId); this.overrideRegisters.set(channelId, newReg); this.publishableContextIds.add(channelId); if (!this.persistLocalState()) { - // Rollback: restore pre-mutation state. if (prevReg === undefined) this.overrideRegisters.delete(channelId); else this.overrideRegisters.set(channelId, prevReg); if (!wasPublishable) this.publishableContextIds.delete(channelId); + this.restoreExtraSlotIds(prevExtraSlotIds); return { success: false, reason: "storage_failed" }; } this.notifyListeners(); @@ -888,11 +893,8 @@ export class ReadStateManager { return { success: false, reason: "load_incomplete" }; const reg = this.overrideRegisters.get(channelId); const effectiveFrontier = this.channelFrontier(channelId); - // No register at all — nothing to clear. if (!reg) return { success: false, reason: "already_inactive" }; - // Register exists: always attempt C-bump (spec NIP-RS.md:537-539 — explicit - // read advances monotone frontier AND increments C; frontier-only deactivation - // is the success fallback when increment would overflow, not a skip condition). + // Always attempt C-bump (spec NIP-RS.md:537-539). const newC = Math.max(reg.s, reg.c) + 1; if (newC > 0xffffffff) return { success: false, reason: "uint32_overflow" }; const newReg: OverrideRegister = { s: reg.s, c: newC, b: reg.b }; @@ -902,12 +904,11 @@ export class ReadStateManager { ); return { success: false, reason: "already_inactive" }; } - // Snapshot before mutation for Contract 3 rollback. const wasPublishable = this.publishableContextIds.has(channelId); this.overrideRegisters.set(channelId, newReg); this.publishableContextIds.add(channelId); if (!this.persistLocalState()) { - // Rollback: restore pre-mutation state. + // v2 write failed — roll back. this.overrideRegisters.set(channelId, reg); if (!wasPublishable) this.publishableContextIds.delete(channelId); return { success: false, reason: "storage_failed" }; @@ -941,7 +942,7 @@ export class ReadStateManager { this.persistLocalState(); } - /** Persist local state. Returns false if any write failed (mark-action must fail). */ + /** Persist local state. Returns false only if the v2 override write failed (commit point). */ private persistLocalState(): boolean { const entries = new Map( [...this.overrideRegisters].map(([ctx, r]) => [ diff --git a/desktop/src/features/channels/readState/readStateStorage.ts b/desktop/src/features/channels/readState/readStateStorage.ts index ea2fcba610..245d5d9726 100644 --- a/desktop/src/features/channels/readState/readStateStorage.ts +++ b/desktop/src/features/channels/readState/readStateStorage.ts @@ -240,29 +240,12 @@ export function writeStoredReadState( state[contextId] = new Date(timestamp * 1_000).toISOString(); } - const ok1 = setLocalStorageItemWithRecovery( - localReadStateKey(pubkey), - JSON.stringify(state), - ); - const ok2 = setLocalStorageItemWithRecovery( - localPublishableContextKey(pubkey), - JSON.stringify([...publishableContextIds].filter((id) => pruned.has(id))), - ); - - const sourceState: Record = {}; - for (const [contextId, createdAt] of contextSourceCreatedAt) { - if (pruned.has(contextId)) { - sourceState[contextId] = createdAt; - } - } - const ok3 = setLocalStorageItemWithRecovery( - localSourceCreatedAtKey(pubkey), - JSON.stringify(sourceState), - ); - // Persist override registers atomically with their frontier timestamps (v2). // Registers and frontiers in one JSON blob — a single write ensures they are // never torn: a register cannot be present without its associated frontier. + // This is the ACTION COMMIT POINT: if ok4 is true the action is durably + // committed; ancillary frontier/cache write failures (ok1-ok3) do not fail + // the action. If ok4 is false, the caller must roll back ALL state. const overrideState: Record< string, { s: number; c: number; b: number; f: number } @@ -275,5 +258,28 @@ export function writeStoredReadState( JSON.stringify(overrideState), ); - return ok1 && ok2 && ok3 && ok4; + // Ancillary writes: frontier cache, publishable set, source timestamps. + // Best-effort — failures do not fail the action. + setLocalStorageItemWithRecovery( + localReadStateKey(pubkey), + JSON.stringify(state), + ); + setLocalStorageItemWithRecovery( + localPublishableContextKey(pubkey), + JSON.stringify([...publishableContextIds].filter((id) => pruned.has(id))), + ); + + const sourceState: Record = {}; + for (const [contextId, createdAt] of contextSourceCreatedAt) { + if (pruned.has(contextId)) { + sourceState[contextId] = createdAt; + } + } + setLocalStorageItemWithRecovery( + localSourceCreatedAtKey(pubkey), + JSON.stringify(sourceState), + ); + + // The commit point: only the v2 override write determines action success. + return ok4; } diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index b1aa0ef6b1..cb6f48cd73 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -120,14 +120,22 @@ export function buildFenceHandle( }; } +/** + * Millis to wait for EOSE after the REQ is sent. If the relay keeps the + * socket alive but never sends this subscription's EOSE, the fence lapses and + * the load concludes `complete: false`. Reconnect / `retryLoad` owns recovery. + * Must NEVER count as establishment. + */ +export const FENCED_ESTABLISHMENT_TIMEOUT_MS = 30_000; + /** * Core of `RelayClient.subscribeFenced()` extracted for testability. * * Registers a fenced subscription, sends REQ, and returns a `FenceHandle` * whose `established` resolves ONLY on EOSE — never on a fallback timer or - * CLOSED. Any lapse (generation mismatch, send failure, `resetConnection`) - * sets `lapsed = true` and resolves `established` so no caller suspends - * forever. + * CLOSED. Any lapse (generation mismatch, send failure, `resetConnection`, + * or establishment timeout) sets `lapsed = true` and resolves `established` + * so no caller suspends forever. */ export async function createFencedSubscription( deps: { @@ -135,6 +143,8 @@ export async function createFencedSubscription( subscriptions: Map; sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise; closeSub: (subId: string) => Promise; + /** Override the establishment timeout (ms). Defaults to FENCED_ESTABLISHMENT_TIMEOUT_MS. */ + establishmentTimeoutMs?: number; }, filter: RelaySubscriptionFilter, onEvent: (event: RelayEvent) => void, @@ -168,7 +178,42 @@ export async function createFencedSubscription( } if (deps.connectionGeneration() !== generation || fencedSub.lapsed) return lapseAndReturn(); + + // ── Establishment timeout ─────────────────────────────────────────────── + // A relay that stays alive but never delivers this subscription's EOSE must + // not hang initialize() forever. The timeout lapses the fence exactly like + // a CLOSED or reconnect — it NEVER counts as establishment. + const timeoutMs = + deps.establishmentTimeoutMs ?? FENCED_ESTABLISHMENT_TIMEOUT_MS; + let establishmentTimer: ReturnType | null = null; + + // Wrap resolveEstablished so the timer is cancelled on normal EOSE. + const originalResolve = fencedSub.resolveEstablished; + fencedSub.resolveEstablished = () => { + if (establishmentTimer !== null) { + clearTimeout(establishmentTimer); + establishmentTimer = null; + } + originalResolve(); + }; + // Also patch the closed reference so the promise resolves via the wrapper. + resolveEstablished = fencedSub.resolveEstablished; + + establishmentTimer = setTimeout(() => { + establishmentTimer = null; + // Only lapse if the fence hasn't already been resolved (EOSE or prior lapse). + if (!fencedSub.lapsed && deps.subscriptions.get(subId) === fencedSub) { + fencedSub.lapsed = true; + fencedSub.resolveEstablished(); + deps.subscriptions.delete(subId); + } + }, timeoutMs); + return buildFenceHandle(fencedSub, established, async () => { + if (establishmentTimer !== null) { + clearTimeout(establishmentTimer); + establishmentTimer = null; + } if (deps.subscriptions.get(subId) !== fencedSub) return; deps.subscriptions.delete(subId); await deps.closeSub(subId);