fix(nip-rs): add fence primitive, parsed-record loader, transactional storage (contracts 1-3)

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

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<void>) | null = null;
-  const unsubscribeReconnect = relay.subscribeToReconnects(() => {
-    lapsed = true;
-  });
-
+  let fence: Awaited<ReturnType<typeof relay.subscribeFenced>>;
   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<void>((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<FencedLoadResult> {
   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 };
 }

desktop/src/features/channels/readState/readStateStorage.ts

@@ -16,7 +16,7 @@ export type StoredReadState = {
   contexts: Map<string, number>;
   publishableContextIds: Set<string>;
   contextSourceCreatedAt: Map<string, number>;
-  overrideRegisters: Map<string, OverrideRegister>;
+  overrideRegisters: Map<string, StoredOverrideEntry>;
 };
 
 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<string, OverrideRegister> {
-  const result = new Map<string, OverrideRegister>();
-  const raw = localStorage.getItem(localOverrideRegistersKey(pubkey));
-  if (!raw) return result;
+export type StoredOverrideEntry = OverrideRegister & { f: number };
+
+function readOverrideState(pubkey: string): Map<string, StoredOverrideEntry> {
+  const result = new Map<string, StoredOverrideEntry>();
+  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<string, OverrideRegister> {
         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<string, number>,
   publishableContextIds: ReadonlySet<string>,
   contextSourceCreatedAt: ReadonlyMap<string, number>,
-  overrideRegisters: ReadonlyMap<string, OverrideRegister>,
+  overrideRegisters: ReadonlyMap<string, StoredOverrideEntry>,
 ): 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<string, { s: number; c: number; b: number }> = {};
-  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;

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<FenceHandle> {
+    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);

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<void>;
+  /** True if the connection lapsed before or after EOSE. */
+  readonly lapsed: boolean;
+  /** Close the relay subscription. */
+  unsubscribe(): Promise<void>;
+};
+
+export function buildFenceHandle(
+  sub: FencedSubscription,
+  established: Promise<void>,
+  unsubscribe: () => Promise<void>,
+): 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<string, RelaySubscription>;
+    sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise<void>;
+    closeSub: (subId: string) => Promise<void>;
+  },
+  filter: RelaySubscriptionFilter,
+  onEvent: (event: RelayEvent) => void,
+): Promise<FenceHandle> {
+  const generation = deps.connectionGeneration();
+  let resolveEstablished = () => {};
+  const established = new Promise<void>((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) => {

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;