From d53ef253965f433b1444c1953b3d9ff3cb292137 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Fri, 31 Jul 2026 21:19:40 -0400 Subject: [PATCH] fix(nip-rs): add fence primitive, parsed-record loader, transactional storage (contracts 1-3) Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- .../readState/readStateFencedLoader.ts | 136 ++++++++---------- .../channels/readState/readStateStorage.ts | 89 +++++++++--- desktop/src/shared/api/relayClientSession.ts | 85 ++++++----- desktop/src/shared/api/relayClientShared.ts | 114 ++++++++++++++- desktop/src/shared/api/relayClosedRecovery.ts | 17 +++ 5 files changed, 299 insertions(+), 142 deletions(-) diff --git a/desktop/src/features/channels/readState/readStateFencedLoader.ts b/desktop/src/features/channels/readState/readStateFencedLoader.ts index 5033548a56..99e46db991 100644 --- a/desktop/src/features/channels/readState/readStateFencedLoader.ts +++ b/desktop/src/features/channels/readState/readStateFencedLoader.ts @@ -2,29 +2,27 @@ * 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. + * - EOSE-established fence via `relay.subscribeFenced()` — resolves only on + * this subscription's own EOSE, never via the 250 ms fallback or CLOSED + * (NIP-RS.md:341-345). Events are delivered synchronously; no timer drain. + * - 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). - * - 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). + * - `T === 0` completes after an empty continuation at `until: 0`; if non-empty + * those events are collected and the load terminates incomplete (spec :352). + * - 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`. */ 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 { + parseReadStateEvent, + type ParsedReadStateEvent, +} 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; @@ -32,15 +30,15 @@ const BAND_LIMIT = 500; const L = 2; export type FencedLoadResult = - | { complete: true; merged: MergedReadState } - | { complete: false; merged: MergedReadState }; + | { complete: true; events: ParsedReadStateEvent[] } + | { complete: false; events: ParsedReadStateEvent[] }; /** * 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. + * Returns the coordinate-deduped parsed records and whether the load was proven + * complete per the NIP-RS §Full-State Load procedure. A `complete: false` + * result carries best-effort events for baseline display. */ export async function fencedEnumerationLoad( relay: RelayClient, @@ -53,27 +51,27 @@ export async function fencedEnumerationLoad( }; // ── 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. + // subscribeFenced() resolves `established` ONLY on EOSE — no fallback timer. + // Events are delivered synchronously to `fenceEvents`; no drain timer needed. const fenceEvents: RelayEvent[] = []; - let lapsed = false; - const startGeneration = relay.getConnectionGeneration(); - let unsubscribeFence: (() => Promise) | null = null; - const unsubscribeReconnect = relay.subscribeToReconnects(() => { - lapsed = true; - }); - + let fence: Awaited>; try { - unsubscribeFence = await relay.subscribeLive(baseFilter, (ev) => { + fence = await relay.subscribeFenced(baseFilter, (ev) => { fenceEvents.push(ev); }); - // Verify the subscribe itself didn't trigger a reconnect. - if (relay.getConnectionGeneration() !== startGeneration) lapsed = true; } catch { - unsubscribeReconnect(); + return emptyIncomplete(); + } + + if (fence.lapsed) { + return emptyIncomplete(); + } + + // Wait for EOSE — resolves only when the relay confirms this subscription. + await fence.established; + + if (fence.lapsed) { + await fence.unsubscribe(); return emptyIncomplete(); } @@ -83,23 +81,17 @@ export async function fencedEnumerationLoad( const bandEvents: RelayEvent[] = []; let complete = false; - while (true) { - if (lapsed) break; - + while (!fence.lapsed) { 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 (fence.lapsed) break; if (band.length === 0) { - // Empty continuation — load proven complete. complete = true; break; } @@ -112,17 +104,14 @@ export async function fencedEnumerationLoad( 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(); + if (fence.lapsed) break; let pinned: RelayEvent[]; try { pinned = await relay.fetchEvents({ ...baseFilter, since: T, until: T }); } catch { break; } - if (relay.getConnectionGeneration() !== genPinned || lapsed) { - break; - } + if (fence.lapsed) break; for (const ev of pinned) bandEvents.push(ev); if (pinned.length > C) C = pinned.length; @@ -133,17 +122,16 @@ export async function fencedEnumerationLoad( } if (T === 0) { - // T=0 exhausted: require a strictly-older empty continuation (spec :352). - const genCont = relay.getConnectionGeneration(); + // 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 (relay.getConnectionGeneration() !== genCont || lapsed) { - break; - } + if (fence.lapsed) break; if (cont.length === 0) { complete = true; } else { @@ -155,24 +143,19 @@ export async function fencedEnumerationLoad( 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(); + // ── Unsubscribe fence ──────────────────────────────────────────────────── + // Events delivered synchronously — no timer drain needed. + await fence.unsubscribe(); - const allEvents = [...bandEvents, ...fenceEvents]; - - if (lapsed && !complete) { - return buildResult(false, allEvents, pubkey); + // A lapse at ANY point (including after `complete` was tentatively set) + // forces complete:false. + if (fence.lapsed) { + const all = [...bandEvents, ...fenceEvents]; + return buildResult(false, all, pubkey); } - // ── Coordinate deduplicate before merge (NIP-RS.md:348) ───────────────── - const deduped = deduplicateByCoordinate(allEvents); - const merged = await mergeReadStateEventsStructured(deduped, pubkey); - return { complete, merged }; + const all = [...bandEvents, ...fenceEvents]; + return buildResult(complete, all, pubkey); } /** @@ -199,10 +182,7 @@ export function deduplicateByCoordinate(events: RelayEvent[]): RelayEvent[] { } function emptyIncomplete(): FencedLoadResult { - return { - complete: false, - merged: { frontiers: new Map(), overrides: new Map() }, - }; + return { complete: false, events: [] }; } async function buildResult( @@ -211,6 +191,10 @@ async function buildResult( pubkey: string, ): Promise { const deduped = deduplicateByCoordinate(events); - const merged = await mergeReadStateEventsStructured(deduped, pubkey); - return { complete, merged }; + const parsed: ParsedReadStateEvent[] = []; + for (const ev of deduped) { + const p = await parseReadStateEvent(ev, pubkey); + if (p) parsed.push(p); + } + return { complete, events: parsed }; } diff --git a/desktop/src/features/channels/readState/readStateStorage.ts b/desktop/src/features/channels/readState/readStateStorage.ts index 81f3843a4f..ea2fcba610 100644 --- a/desktop/src/features/channels/readState/readStateStorage.ts +++ b/desktop/src/features/channels/readState/readStateStorage.ts @@ -16,7 +16,7 @@ export type StoredReadState = { contexts: Map; publishableContextIds: Set; contextSourceCreatedAt: Map; - overrideRegisters: Map; + overrideRegisters: Map; }; function mergeLocalStorageKey( @@ -102,7 +102,7 @@ export function readStoredReadState(pubkey: string): StoredReadState { contexts, publishableContextIds: readPublishableContextIds(pubkey), contextSourceCreatedAt: readContextSourceCreatedAt(pubkey), - overrideRegisters: readOverrideRegisters(pubkey), + overrideRegisters: readOverrideState(pubkey), }; } @@ -112,23 +112,65 @@ function isPrunableContextKey(contextId: string): boolean { ); } -// Key for persisted override registers (raw context ID → OverrideRegister triple). -function localOverrideRegistersKey(pubkey: string): string { +// Key for persisted override state: registers + their frontier timestamps in one atomic write. +// v2 stores {s, c, b, f} per context where f = the last known effective frontier. +function localOverrideStateKey(pubkey: string): string { + return `buzz.nip-rs.override-state.v2:${pubkey}`; +} + +// Legacy key retained for migration reads only — never written to after v2 migration. +function localOverrideRegistersKeyLegacy(pubkey: string): string { return `buzz.nip-rs.override-registers.v1:${pubkey}`; } -function readOverrideRegisters(pubkey: string): Map { - const result = new Map(); - const raw = localStorage.getItem(localOverrideRegistersKey(pubkey)); - if (!raw) return result; +export type StoredOverrideEntry = OverrideRegister & { f: number }; + +function readOverrideState(pubkey: string): Map { + const result = new Map(); + const raw = localStorage.getItem(localOverrideStateKey(pubkey)); + if (raw) { + try { + const parsed = JSON.parse(raw); + if (isPlainRecord(parsed)) { + for (const [rawCtx, value] of Object.entries(parsed)) { + if (!isPlainRecord(value)) continue; + const { s, c, b, f } = value; + if ( + typeof s === "number" && + Number.isInteger(s) && + s >= 0 && + s <= 0xffffffff && + typeof c === "number" && + Number.isInteger(c) && + c >= 0 && + c <= 0xffffffff && + typeof b === "number" && + Number.isInteger(b) && + b >= 0 && + typeof f === "number" && + Number.isInteger(f) && + f >= 0 + ) { + result.set(rawCtx, { s, c, b, f }); + } + } + } + } catch { + // Corrupt storage — return empty; repopulated on next ingest. + } + return result; + } + // Migration: read from legacy v1 key if v2 key absent. + const legacyRaw = localStorage.getItem( + localOverrideRegistersKeyLegacy(pubkey), + ); + if (!legacyRaw) return result; try { - const parsed = JSON.parse(raw); + const parsed = JSON.parse(legacyRaw); if (!isPlainRecord(parsed)) return result; for (const [rawCtx, value] of Object.entries(parsed)) { if (!isPlainRecord(value)) continue; - const s = value.s; - const c = value.c; - const b = value.b; + const { s, c, b } = value; if ( typeof s === "number" && Number.isInteger(s) && @@ -142,11 +184,11 @@ function readOverrideRegisters(pubkey: string): Map { Number.isInteger(b) && b >= 0 ) { - result.set(rawCtx, { s, c, b }); + result.set(rawCtx, { s, c, b, f: 0 }); // frontier unknown from legacy key } } } catch { - // Corrupt storage — return empty map; will be repopulated on next ingest. + // Corrupt legacy storage — ignore. } return result; } @@ -189,7 +231,7 @@ export function writeStoredReadState( contexts: ReadonlyMap, publishableContextIds: ReadonlySet, contextSourceCreatedAt: ReadonlyMap, - overrideRegisters: ReadonlyMap, + overrideRegisters: ReadonlyMap, ): boolean { const pruned = pruneStaleContexts(contexts, Math.floor(Date.now() / 1_000)); @@ -218,14 +260,19 @@ export function writeStoredReadState( JSON.stringify(sourceState), ); - // Persist override registers atomically with frontier state. - const regState: Record = {}; - for (const [rawCtx, reg] of overrideRegisters) { - regState[rawCtx] = { s: reg.s, c: reg.c, b: reg.b }; + // 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. + const overrideState: Record< + string, + { s: number; c: number; b: number; f: number } + > = {}; + for (const [rawCtx, entry] of overrideRegisters) { + overrideState[rawCtx] = { s: entry.s, c: entry.c, b: entry.b, f: entry.f }; } const ok4 = setLocalStorageItemWithRecovery( - localOverrideRegistersKey(pubkey), - JSON.stringify(regState), + localOverrideStateKey(pubkey), + JSON.stringify(overrideState), ); return ok1 && ok2 && ok3 && ok4; diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 685980896c..57afb425e6 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -14,7 +14,9 @@ import { } from "@/shared/constants/kinds"; import { getTextPayload, + createFencedSubscription, type ConnectionState, + type FenceHandle, type PendingEvent, type RelaySubscription, type RelaySubscriptionFilter, @@ -145,7 +147,10 @@ export class RelayClient { } for (const [subId, sub] of this.subscriptions) { - if (sub.mode !== "live") { + if (sub.mode === "fenced") { + sub.lapsed = true; + sub.resolveEstablished(); + } else if (sub.mode !== "live") { window.clearTimeout(sub.timeout); sub.reject(error); } else { @@ -413,6 +418,25 @@ export class RelayClient { return this.subscribe(filter, onEvent); } + /** @see `createFencedSubscription` in relayClientShared for contract. */ + async subscribeFenced( + filter: RelaySubscriptionFilter, + onEvent: (event: RelayEvent) => void, + ): Promise { + await this.ensureConnected(); + const deps = { + connectionGeneration: () => this.connectionGeneration, + subscriptions: this.subscriptions, + sendReq: (id: string, f: RelaySubscriptionFilter) => + this.sendRawWithReconnectRetry( + ["REQ", id, f], + "Failed to establish fenced subscription.", + ), + closeSub: (id: string) => this.closeSubscription(id), + }; + return createFencedSubscription(deps, filter, onEvent); + } + async subscribeToChannelMentionEvents( channelId: string, pubkey: string, @@ -425,10 +449,7 @@ export class RelayClient { } async preconnect() { - // Explicit re-engagement. If the session went terminal (auth rejection) - // the caller is asking us to try again, so clear the latch. A manual - // reconnect also bypasses the current delay once; ordinary operations do - // not, so background traffic cannot continuously defeat backoff. + // Re-engage after terminal state; bypasses current delay once. this.terminal = false; this.keepAliveRequested = true; if (this.reconnectTimeout !== null) { @@ -459,22 +480,14 @@ export class RelayClient { getConnectionState(): ConnectionState { return this.connectionStateEmitter.get(); } - /** - * Subscribe to connection-state transitions. The listener is invoked - * immediately with the current state so callers don't need a separate - * `getConnectionState()` call to seed their UI. - */ + /** Subscribe to connection-state transitions; listener fires immediately with current state. */ subscribeToConnectionState(listener: (state: ConnectionState) => void) { return this.connectionStateEmitter.subscribe(listener); } private async ensureConnected() { if (shouldRefuseConnect({ terminal: this.terminal })) { - // Session is terminal (e.g. relay rejected auth). Refuse to connect - // until an explicit re-engagement (disconnect()/preconnect()) clears - // the flag. Without this, the reconnect timer's catch handler — and - // the retry wrappers in publishEvent / sendRawWithReconnectRetry — - // would race the terminal "disconnected" state back to "reconnecting". + // Terminal after auth rejection — refuse until preconnect() re-engages. throw new Error("Relay session is terminal; cannot reconnect."); } @@ -491,9 +504,7 @@ export class RelayClient { hasPendingReconnect: this.reconnectTimeout !== null, }) ) { - // The reconnect coordinator owns outage pacing. Query, publish, and - // subscription callers must wait for its scheduled attempt instead of - // clearing the timer and creating an immediate reconnect storm. + // Reconnect coordinator owns pacing; wait rather than creating a storm. return this.waitForScheduledReconnect(); } @@ -633,23 +644,16 @@ export class RelayClient { } private async sendRaw(payload: unknown[]) { - if (this.wsId === null) { - throw new Error("Relay socket is not connected."); - } - + if (this.wsId === null) throw new Error("Relay socket is not connected."); await invoke("plugin:websocket|send", { id: this.wsId, - message: { - type: "Text", - data: JSON.stringify(payload), - }, + message: { type: "Text", data: JSON.stringify(payload) }, }); } private normalizeRelayError(error: unknown, fallbackMessage: string) { return error instanceof Error ? error : new Error(fallbackMessage); } - private recoverFromSocketFailure( error: unknown, fallbackMessage: string, @@ -670,7 +674,6 @@ export class RelayClient { error, fallbackMessage, ); - try { await this.ensureConnected(); await this.sendRaw(payload); @@ -684,11 +687,7 @@ export class RelayClient { } private async closeSubscription(subId: string) { - if (this.wsId === null) { - return; - } - - await this.sendRaw(["CLOSE", subId]); + if (this.wsId !== null) await this.sendRaw(["CLOSE", subId]); } async publishEvent( @@ -819,8 +818,7 @@ export class RelayClient { if (type === "NOTICE" && typeof rest[0] === "string") { const notice: string = rest[0]; - // Relay back-pressure signal — activate the gate so pending operations - // back off until the window expires. + // Relay rate-limit notice — activate backoff gate. if (notice.startsWith("rate-limited:")) { activateRateLimit(parseRateLimitHint(notice)); } @@ -968,9 +966,7 @@ export class RelayClient { return; } - // Apply ±25% jitter so a fleet of clients reconnecting simultaneously - // spreads their AUTH storms across a 50% window instead of all hitting - // the relay at the same instant. + // Apply ±25% jitter to spread reconnect storms. const jitter = this.reconnectDelayMs * (0.75 + Math.random() * 0.5); const delay = Math.min(jitter, RECONNECT_MAX_DELAY_MS); this.reconnectDelayMs = Math.min( @@ -1011,12 +1007,7 @@ export class RelayClient { } } - private resetConnection( - error: Error, - options?: { - reconnect?: boolean; - }, - ) { + private resetConnection(error: Error, options?: { reconnect?: boolean }) { this.onMessageChannel = null; this.stallWatchdog.stop(); this.connectionGeneration++; @@ -1062,6 +1053,12 @@ export class RelayClient { } for (const [subId, subscription] of this.subscriptions) { + if (subscription.mode === "fenced") { + subscription.lapsed = true; + subscription.resolveEstablished(); + this.subscriptions.delete(subId); + continue; + } if (subscription.mode !== "live") { window.clearTimeout(subscription.timeout); subscription.reject(error); diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 9108f7b6d7..b1aa0ef6b1 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -64,6 +64,117 @@ type LiveSubscription = { closedRetryTimeout?: number; }; +/** + * Fenced subscription for NIP-RS full-state loads. + * + * Unlike `live`, a fenced subscription: + * - Resolves `established` ONLY on this subscription's own EOSE — never via + * a fallback timer or terminal CLOSED. + * - Delivers events synchronously (bypasses the EVENT_BATCH_MS buffer) so + * the loader's synchronous drain barrier needs no timer. + * - On CLOSED or any connection lapse, sets `lapsed = true` and resolves + * `established` (in case the loader is still awaiting it), then removes + * itself without retrying. + */ +export type FencedSubscription = { + mode: "fenced"; + filter: RelaySubscriptionFilter; + onEvent: (event: RelayEvent) => void; + /** Connection generation at subscribe time — mismatch = lapsed. */ + generation: number; + /** Resolves on EOSE or lapse (check `lapsed` after awaiting). */ + resolveEstablished: () => void; + /** True after CLOSED, reconnect, or generation change. */ + lapsed: boolean; +}; + +/** + * Handle returned by `RelayClient.subscribeFenced()`. + * + * - `established` resolves ONLY on the subscription's own EOSE, or resolves + * after setting `lapsed = true` if the connection lapses before EOSE. + * - `lapsed` is synchronously readable — check it after awaiting `established` + * and at any point during enumeration. + * - `unsubscribe()` closes the relay subscription when the load is complete. + */ +export type FenceHandle = { + /** Resolves on EOSE (check `lapsed` after awaiting). */ + established: Promise; + /** True if the connection lapsed before or after EOSE. */ + readonly lapsed: boolean; + /** Close the relay subscription. */ + unsubscribe(): Promise; +}; + +export function buildFenceHandle( + sub: FencedSubscription, + established: Promise, + unsubscribe: () => Promise, +): FenceHandle { + return { + established, + get lapsed() { + return sub.lapsed; + }, + unsubscribe, + }; +} + +/** + * 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. + */ +export async function createFencedSubscription( + deps: { + connectionGeneration: () => number; + subscriptions: Map; + sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise; + closeSub: (subId: string) => Promise; + }, + filter: RelaySubscriptionFilter, + onEvent: (event: RelayEvent) => void, +): Promise { + const generation = deps.connectionGeneration(); + let resolveEstablished = () => {}; + const established = new Promise((r) => { + resolveEstablished = r; + }); + const subId = `fenced-${crypto.randomUUID()}`; + const fencedSub: FencedSubscription = { + mode: "fenced", + filter, + onEvent, + generation, + resolveEstablished, + lapsed: false, + }; + const lapseAndReturn = (): FenceHandle => { + fencedSub.lapsed = true; + resolveEstablished(); + deps.subscriptions.delete(subId); + return buildFenceHandle(fencedSub, established, async () => {}); + }; + if (deps.connectionGeneration() !== generation) return lapseAndReturn(); + deps.subscriptions.set(subId, fencedSub); + try { + await deps.sendReq(subId, filter); + } catch { + return lapseAndReturn(); + } + if (deps.connectionGeneration() !== generation || fencedSub.lapsed) + return lapseAndReturn(); + return buildFenceHandle(fencedSub, established, async () => { + if (deps.subscriptions.get(subId) !== fencedSub) return; + deps.subscriptions.delete(subId); + await deps.closeSub(subId); + }); +} + export type PendingEvent = { event: RelayEvent; resolve: (event: RelayEvent) => void; @@ -74,7 +185,8 @@ export type PendingEvent = { export type RelaySubscription = | HistorySubscription | FirstEventSubscription - | LiveSubscription; + | LiveSubscription + | FencedSubscription; export function sortEvents(events: RelayEvent[]) { return [...events].sort((left, right) => { diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts index fd366671fd..edb360966c 100644 --- a/desktop/src/shared/api/relayClosedRecovery.ts +++ b/desktop/src/shared/api/relayClosedRecovery.ts @@ -36,6 +36,13 @@ export function handleRelayClosed({ const subscription = subscriptions.get(subId); if (!subscription) return; if (subscription.mode !== "live") { + if (subscription.mode === "fenced") { + // CLOSED lapses the fence — mark and resolve (unblocks any awaiter). + subscription.lapsed = true; + subscription.resolveEstablished(); + subscriptions.delete(subId); + return; + } // Classify before rejecting so a `rate-limited:` history CLOSED arms the // gate for concurrent ops. A history sub can't be retried (the caller holds // the promise), so we still reject immediately after arming. @@ -136,6 +143,11 @@ export function prepareSubscriptionEvent( if (subscription.mode === "first") { return false; } + if (subscription.mode === "fenced") { + // Fenced events are delivered synchronously — caller delivers directly. + subscription.onEvent(event); + return false; // handled here; do NOT enqueue in the batch buffer + } subscription.closedRetryAttempt = 0; clearClosedRetry(subscription); subscription.lastSeenCreatedAt = Math.max( @@ -156,6 +168,11 @@ export function handleSubscriptionEose({ }) { const subscription = subscriptions.get(subId); if (!subscription) return; + if (subscription.mode === "fenced") { + // EOSE proves this subscription's fence is established. + subscription.resolveEstablished(); + return; + } if (subscription.mode === "live") { subscription.resolveReady?.(); subscription.resolveReady = undefined;