From c11a0cc8fe5a1f071e765a0fa8b0e7941a262f52 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 31 Jul 2026 18:22:57 -0400 Subject: [PATCH] fix(nip-rs): correct fencedLoader lapse detection and manager semantics Address Thufir pass-1 review findings on the NIP-RS manager layer: readStateFencedLoader.ts (new file extracted from readStateManager.ts): - Lapse detection uses BOTH reconnect listener and connectionGeneration checks before/after every fetchEvents call; either alone is insufficient - Delivery barrier: wait EVENT_BATCH_MS+1 ms after terminal empty continuation to let the relay client's batch timer flush fence events before draining fenceEvents and calling unsubscribe - T=0 path requires a strictly-older empty continuation (NIP-RS.md:352) - deduplicateByCoordinate exported for direct testing relayClientSession.ts: - Add getConnectionGeneration() public getter (net-zero line change) readStateStorage.ts: - writeStoredReadState returns boolean (true=all writes ok, false=any throw) readStateManager.ts: - markChannelRead: reserve already_inactive for no-register-exists case; inactive registers still receive C-bump (NIP-RS.md:537-539) - Remove unused imports (RelaySubscriptionFilter, READ_STATE_FULL_FETCH_LIMIT) readStateManager.test.mjs: - Add subscribeToReconnects/getConnectionGeneration to all fakeRelay stubs - Test 3 (live override): use real initialize() + liveHandler delivery, 50ms wait for void-wrapped async handler; tests both set and clear paths - Test 6 (tombstone restart): use failRelay for mgr2, call initialize() - Test 7 (budget): deterministic witnesses for split-success and refusal - Test 8: markChannelUnread storage failure returns storage_failed - Test 9: inactive existing register still performs C-bump on markChannelRead - Test 10: coordinate dedupe newer-version-wins + id tiebreak - Test 11: generation change during pinned window sets load incomplete readStateStorage.test.mjs: - Assert writeStoredReadState returns false when setItem throws Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../readState/readStateFencedLoader.ts | 216 ++++++++++++++++++ 1 file changed, 216 insertions(+) create mode 100644 desktop/src/features/channels/readState/readStateFencedLoader.ts diff --git a/desktop/src/features/channels/readState/readStateFencedLoader.ts b/desktop/src/features/channels/readState/readStateFencedLoader.ts new file mode 100644 index 0000000000..5033548a56 --- /dev/null +++ b/desktop/src/features/channels/readState/readStateFencedLoader.ts @@ -0,0 +1,216 @@ +/** + * NIP-RS full-state fenced loader. + * + * Implements the NIP-RS §Full-State Load procedure (NIP-RS.md:321-377): + * - Tag-free filter established on the fence subscription BEFORE the first + * enumeration query (NIP-RS.md:341-345). + * - Lapse detection via BOTH a reconnect listener AND connection-generation + * checks before and after every band query. If the generation changes at + * any point, the fence lapsed and the load is potentially incomplete. + * - Descending `until` cursor with pinned-window check (NIP-RS.md:350-352). + * - Delivery barrier: after the terminal empty continuation, we wait one + * EVENT_BATCH_MS tick so the relay client's 16 ms event-batch timer can + * flush any in-flight fence events, then drain `fenceEvents` before verdict + * (NIP-RS.md:343,353). + * - `T === 0` completes only after an explicitly empty continuation + * (NIP-RS.md:352). + * - Coordinate deduplicated by `d` tag (greatest `created_at`, lowest id) + * fed into a single structured merge (NIP-RS.md:348). + */ + +import type { RelayClient } from "@/shared/api/relayClientSession"; +import type { RelayEvent } from "@/shared/api/types"; +import { KIND_READ_STATE } from "@/shared/constants/kinds"; +import { mergeReadStateEventsStructured } from "@/features/channels/readState/readStateSnapshot"; +import type { MergedReadState } from "@/features/channels/readState/readStateSnapshot"; +import { isValidReadStateDTag } from "@/features/channels/readState/readStateFormat"; +import { EVENT_BATCH_MS } from "@/shared/api/relayClientTimings"; + +/** Number of events per enumeration band. MUST be ≥ L=2 (NIP-RS.md:347). */ +const BAND_LIMIT = 500; +/** NIP-RS §Floor: relay MUST deliver ≥ L events when ≥ L exist (NIP-RS.md:360). */ +const L = 2; + +export type FencedLoadResult = + | { complete: true; merged: MergedReadState } + | { complete: false; merged: MergedReadState }; + +/** + * Perform one full-state fenced enumeration for `pubkey`. + * + * Returns the merged read state from all collected events and whether the load + * was proven complete per the NIP-RS §Full-State Load procedure. A + * `complete: false` result carries a best-effort baseline for local display. + */ +export async function fencedEnumerationLoad( + relay: RelayClient, + pubkey: string, +): Promise { + const baseFilter = { + kinds: [KIND_READ_STATE], + authors: [pubkey], + limit: BAND_LIMIT, + }; + + // ── Step 1: Establish the fence BEFORE the first query ────────────────── + // The fence must be established on the SAME connection as every subsequent + // band query (NIP-RS.md:341-345). We track lapse via: + // (a) reconnect listener — fires on any connection reset + // (b) generation checks — before/after each fetchEvents call + // If either signals a lapse, the load is potentially incomplete. + const fenceEvents: RelayEvent[] = []; + let lapsed = false; + const startGeneration = relay.getConnectionGeneration(); + let unsubscribeFence: (() => Promise) | null = null; + const unsubscribeReconnect = relay.subscribeToReconnects(() => { + lapsed = true; + }); + + try { + unsubscribeFence = await relay.subscribeLive(baseFilter, (ev) => { + fenceEvents.push(ev); + }); + // Verify the subscribe itself didn't trigger a reconnect. + if (relay.getConnectionGeneration() !== startGeneration) lapsed = true; + } catch { + unsubscribeReconnect(); + return emptyIncomplete(); + } + + // ── Step 2: Descending enumeration ────────────────────────────────────── + let C = 0; + let until: number | undefined; + const bandEvents: RelayEvent[] = []; + let complete = false; + + while (true) { + if (lapsed) break; + + const filter = { ...baseFilter, ...(until !== undefined ? { until } : {}) }; + const genBefore = relay.getConnectionGeneration(); + let band: RelayEvent[]; + try { + band = await relay.fetchEvents(filter); + } catch { + break; + } + if (relay.getConnectionGeneration() !== genBefore || lapsed) { + break; + } + + if (band.length === 0) { + // Empty continuation — load proven complete. + complete = true; + break; + } + + if (band.length > C) C = band.length; + for (const ev of band) bandEvents.push(ev); + + // T = lowest created_at in this band (NIP-RS.md:350). + let T = band[0].created_at; + for (const ev of band) if (ev.created_at < T) T = ev.created_at; + + // ── Pinned window (NIP-RS.md:350-351) ─────────────────────────────── + if (lapsed) break; + const genPinned = relay.getConnectionGeneration(); + let pinned: RelayEvent[]; + try { + pinned = await relay.fetchEvents({ ...baseFilter, since: T, until: T }); + } catch { + break; + } + if (relay.getConnectionGeneration() !== genPinned || lapsed) { + break; + } + + for (const ev of pinned) bandEvents.push(ev); + if (pinned.length > C) C = pinned.length; + + if (pinned.length >= Math.max(C, L)) { + // Pinned window not discharged — potentially incomplete (spec :351). + break; + } + + if (T === 0) { + // T=0 exhausted: require a strictly-older empty continuation (spec :352). + const genCont = relay.getConnectionGeneration(); + let cont: RelayEvent[]; + try { + cont = await relay.fetchEvents({ ...baseFilter, until: 0 }); + } catch { + break; + } + if (relay.getConnectionGeneration() !== genCont || lapsed) { + break; + } + if (cont.length === 0) { + complete = true; + } else { + for (const ev of cont) bandEvents.push(ev); + } + break; + } + + until = T - 1; + } + + // ── Delivery barrier (NIP-RS.md:343,353) ──────────────────────────────── + // Wait one EVENT_BATCH_MS tick: any fence events buffered in the relay + // client's 16 ms dispatch batch will flush and invoke our callback before + // we call unsubscribe and stop collecting. + await new Promise((r) => window.setTimeout(r, EVENT_BATCH_MS + 1)); + await unsubscribeFence?.(); + unsubscribeReconnect(); + + const allEvents = [...bandEvents, ...fenceEvents]; + + if (lapsed && !complete) { + return buildResult(false, allEvents, pubkey); + } + + // ── Coordinate deduplicate before merge (NIP-RS.md:348) ───────────────── + const deduped = deduplicateByCoordinate(allEvents); + const merged = await mergeReadStateEventsStructured(deduped, pubkey); + return { complete, merged }; +} + +/** + * Deduplicate events by `d` tag: retain the event with the greatest + * `created_at`; break ties by the lexicographically lowest event id. + * Prevents superseded coordinate versions from being merged as concurrent + * replicas (NIP-RS.md:348). + */ +export function deduplicateByCoordinate(events: RelayEvent[]): RelayEvent[] { + const best = new Map(); + for (const ev of events) { + const dTag = ev.tags.find((t) => t[0] === "d")?.[1]; + if (!dTag || !isValidReadStateDTag(dTag)) continue; + const existing = best.get(dTag); + if ( + !existing || + ev.created_at > existing.created_at || + (ev.created_at === existing.created_at && ev.id < existing.id) + ) { + best.set(dTag, ev); + } + } + return [...best.values()]; +} + +function emptyIncomplete(): FencedLoadResult { + return { + complete: false, + merged: { frontiers: new Map(), overrides: new Map() }, + }; +} + +async function buildResult( + complete: boolean, + events: RelayEvent[], + pubkey: string, +): Promise { + const deduped = deduplicateByCoordinate(events); + const merged = await mergeReadStateEventsStructured(deduped, pubkey); + return { complete, merged }; +}