fix(desktop): recover relay-closed subscriptions (#2060)

Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>

desktop/src/shared/api/relayClientSession.ts

@@ -1,5 +1,4 @@
 import { Channel, invoke } from "@tauri-apps/api/core";
-
 import {
   createAuthEvent,
   getRelayWsUrl,
@@ -15,7 +14,6 @@ import {
 } from "@/shared/constants/kinds";
 import {
   getTextPayload,
-  sortEvents,
   type ConnectionState,
   type PendingEvent,
   type RelaySubscription,
@@ -30,6 +28,12 @@ import {
   buildGlobalStreamFilter,
 } from "@/shared/api/relayChannelFilters";
 import { collectWithConcurrency } from "@/shared/api/concurrency";
+import {
+  clearClosedRetry,
+  handleRelayClosed,
+  handleSubscriptionEose,
+  prepareSubscriptionEvent,
+} from "@/shared/api/relayClosedRecovery";
 import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
 import { RelayConnectionStateEmitter } from "@/shared/api/relayConnectionStateEmitter";
 import {
@@ -39,7 +43,6 @@ import {
 import { RelayStallWatchdog } from "@/shared/api/relayStallWatchdog";
 import { closeWebSocket } from "@/shared/api/relayWebSocketClose";
 import { buildThreadReferenceTags } from "@/features/messages/lib/threading";
-
 const RECONNECT_BASE_DELAY_MS = 1_000,
   RECONNECT_MAX_DELAY_MS = 30_000,
   EVENT_BATCH_MS = 16,
@@ -135,6 +138,8 @@ export class RelayClient {
       if (sub.mode === "history") {
         window.clearTimeout(sub.timeout);
         sub.reject(error);
+      } else {
+        clearClosedRetry(sub);
       }
       this.subscriptions.delete(subId);
     }
@@ -596,6 +601,7 @@ export class RelayClient {
       }
 
       this.subscriptions.delete(subId);
+      clearClosedRetry(active);
       await this.closeSubscription(subId);
     };
   }
@@ -769,6 +775,20 @@ export class RelayClient {
 
     if (type === "EOSE" && typeof rest[0] === "string") {
       this.handleEose(rest[0]);
+      return;
+    }
+
+    if (type === "CLOSED" && typeof rest[0] === "string") {
+      handleRelayClosed({
+        subscriptions: this.subscriptions,
+        subId: rest[0],
+        message: typeof rest[1] === "string" ? rest[1] : "",
+        sendReq: (subId, filter) =>
+          this.sendRawWithReconnectRetry(
+            ["REQ", subId, filter],
+            "Failed to restore relay subscription after CLOSED.",
+          ),
+      });
     }
   }
 
@@ -796,16 +816,7 @@ export class RelayClient {
       return;
     }
 
-    if (subscription.mode === "history") {
-      subscription.events.push(event);
-      return;
-    }
-
-    subscription.lastSeenCreatedAt = Math.max(
-      subscription.lastSeenCreatedAt ?? 0,
-      event.created_at,
-    );
-
+    if (!prepareSubscriptionEvent(subscription, event)) return;
     this.eventBuffer.push({ subId, event });
     this.flushTimeout ??= window.setTimeout(
       () => this.flushEventBuffer(),
@@ -828,21 +839,11 @@ export class RelayClient {
   }
 
   private handleEose(subId: string) {
-    const subscription = this.subscriptions.get(subId);
-    if (!subscription) {
-      return;
-    }
-
-    if (subscription.mode === "live") {
-      subscription.resolveReady?.();
-      subscription.resolveReady = undefined;
-      return;
-    }
-
-    window.clearTimeout(subscription.timeout);
-    this.subscriptions.delete(subId);
-    void this.closeSubscription(subId);
-    subscription.resolve(sortEvents(subscription.events));
+    handleSubscriptionEose({
+      subscriptions: this.subscriptions,
+      subId,
+      closeSubscription: (id) => this.closeSubscription(id),
+    });
   }
 
   private handleOk(eventId: string, success: boolean, message: string) {
@@ -1004,6 +1005,7 @@ export class RelayClient {
 
       subscription.resolveReady?.();
       subscription.resolveReady = undefined;
+      clearClosedRetry(subscription);
     }
 
     for (const [eventId, pendingEvent] of this.pendingEvents) {

desktop/src/shared/api/relayClientShared.ts

@@ -52,6 +52,8 @@ type LiveSubscription = {
   onEvent: (event: RelayEvent) => void;
   resolveReady?: () => void;
   lastSeenCreatedAt?: number;
+  closedRetryAttempt?: number;
+  closedRetryTimeout?: number;
 };
 
 export type PendingEvent = {

desktop/src/shared/api/relayClosedPolicy.test.mjs

@@ -0,0 +1,32 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { isRetryableRelayClosed } from "./relayClosedPolicy.ts";
+
+test("retries transient CLOSED responses", () => {
+  for (const message of [
+    "rate-limited: slow down",
+    "error: database error",
+    "server shutting down",
+    "",
+  ]) {
+    assert.equal(isRetryableRelayClosed(message), true, message);
+  }
+});
+
+test("does not retry permanent CLOSED responses", () => {
+  for (const message of [
+    "restricted: not a channel member",
+    "restricted: channel access revoked",
+    "auth-required: not authenticated",
+    "blocked: banned",
+    "invalid: malformed filter",
+    "pow: difficulty too low",
+    "duplicate: subscription exists",
+    "unsupported: filter",
+    "error: mixed search and non-search filters not supported",
+    "error: too many subscriptions",
+  ]) {
+    assert.equal(isRetryableRelayClosed(message), false, message);
+  }
+});

desktop/src/shared/api/relayClosedPolicy.ts

@@ -0,0 +1,19 @@
+/**
+ * CLOSED ends a NIP-01 subscription. Retry failures that may recover without
+ * changing the request; authorization and malformed-filter failures require a
+ * caller/state change and would otherwise loop forever.
+ */
+export function isRetryableRelayClosed(message: string) {
+  const normalized = message.trim().toLowerCase();
+  return !(
+    normalized.startsWith("restricted:") ||
+    normalized.startsWith("auth-required:") ||
+    normalized.startsWith("blocked:") ||
+    normalized.startsWith("invalid:") ||
+    normalized.startsWith("pow:") ||
+    normalized.startsWith("duplicate:") ||
+    normalized.startsWith("unsupported:") ||
+    normalized.startsWith("error: mixed search") ||
+    normalized.startsWith("error: too many subscriptions")
+  );
+}

desktop/src/shared/api/relayClosedRecovery.ts

@@ -0,0 +1,133 @@
+import { isRetryableRelayClosed } from "@/shared/api/relayClosedPolicy";
+import {
+  sortEvents,
+  type RelaySubscription,
+  type RelaySubscriptionFilter,
+} from "@/shared/api/relayClientShared";
+import type { RelayEvent } from "@/shared/api/types";
+
+const RETRY_BASE_DELAY_MS = 1_000;
+const RETRY_MAX_DELAY_MS = 30_000;
+
+type LiveSubscription = Extract<RelaySubscription, { mode: "live" }>;
+
+export function clearClosedRetry(subscription: LiveSubscription) {
+  if (subscription.closedRetryTimeout === undefined) return;
+  window.clearTimeout(subscription.closedRetryTimeout);
+  subscription.closedRetryTimeout = undefined;
+}
+
+export function handleRelayClosed({
+  subscriptions,
+  subId,
+  message,
+  sendReq,
+}: {
+  subscriptions: Map<string, RelaySubscription>;
+  subId: string;
+  message: string;
+  sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise<void>;
+}) {
+  const subscription = subscriptions.get(subId);
+  if (!subscription) return;
+  if (subscription.mode === "history") {
+    window.clearTimeout(subscription.timeout);
+    subscriptions.delete(subId);
+    subscription.reject(
+      new Error(message || "Relay closed the history subscription."),
+    );
+    return;
+  }
+  recoverLiveSubscriptionFromClosed({
+    subscriptions,
+    subId,
+    subscription,
+    message,
+    sendReq,
+  });
+}
+
+function recoverLiveSubscriptionFromClosed({
+  subscriptions,
+  subId,
+  subscription,
+  message,
+  sendReq,
+}: {
+  subscriptions: Map<string, RelaySubscription>;
+  subId: string;
+  subscription: LiveSubscription;
+  message: string;
+  sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise<void>;
+}) {
+  subscription.resolveReady?.();
+  subscription.resolveReady = undefined;
+  if (!isRetryableRelayClosed(message)) {
+    subscriptions.delete(subId);
+    return;
+  }
+  if (subscription.closedRetryTimeout !== undefined) return;
+
+  const attempt = subscription.closedRetryAttempt ?? 0;
+  const delayMs = Math.min(
+    RETRY_BASE_DELAY_MS * 2 ** attempt,
+    RETRY_MAX_DELAY_MS,
+  );
+  subscription.closedRetryAttempt = attempt + 1;
+  subscription.closedRetryTimeout = window.setTimeout(() => {
+    subscription.closedRetryTimeout = undefined;
+    if (subscriptions.get(subId) !== subscription) return;
+    void sendReq(subId, subscription.filter).catch((error) => {
+      if (subscriptions.get(subId) !== subscription) return;
+      console.error("Failed to restore closed relay subscription", error);
+      recoverLiveSubscriptionFromClosed({
+        subscriptions,
+        subId,
+        subscription,
+        message,
+        sendReq,
+      });
+    });
+  }, delayMs);
+}
+
+export function prepareSubscriptionEvent(
+  subscription: RelaySubscription,
+  event: RelayEvent,
+) {
+  if (subscription.mode === "history") {
+    subscription.events.push(event);
+    return false;
+  }
+  subscription.closedRetryAttempt = 0;
+  clearClosedRetry(subscription);
+  subscription.lastSeenCreatedAt = Math.max(
+    subscription.lastSeenCreatedAt ?? 0,
+    event.created_at,
+  );
+  return true;
+}
+
+export function handleSubscriptionEose({
+  subscriptions,
+  subId,
+  closeSubscription,
+}: {
+  subscriptions: Map<string, RelaySubscription>;
+  subId: string;
+  closeSubscription: (subId: string) => Promise<void>;
+}) {
+  const subscription = subscriptions.get(subId);
+  if (!subscription) return;
+  if (subscription.mode === "live") {
+    subscription.resolveReady?.();
+    subscription.resolveReady = undefined;
+    subscription.closedRetryAttempt = 0;
+    clearClosedRetry(subscription);
+    return;
+  }
+  window.clearTimeout(subscription.timeout);
+  subscriptions.delete(subId);
+  void closeSubscription(subId);
+  subscription.resolve(sortEvents(subscription.events));
+}

desktop/src/testing/e2eBridge.ts

@@ -164,6 +164,8 @@ type E2eConfig = {
     applyCommunityDelayMs?: number;
     openDmDelayMs?: number;
     sendMessageDelayMs?: number;
+    /** Close the first channel-window live REQ; its retry is accepted. */
+    closeChannelLiveSubscriptionOnce?: boolean;
     /** Reject successive kind-9 sends with these messages, then resume. */
     sendMessageErrors?: string[];
     /** Reject successive managed-agent starts, then resume. */
@@ -2563,6 +2565,7 @@ const mockReminderEvents: RelayEvent[] = [];
 let mockRelayMembers: RawRelayMember[] = [];
 const mockSockets = new Map<number, MockSocket>();
 let mockWebsocketSendMutexWedged = false;
+let mockClosedChannelLiveSubscription = false;
 const realSockets = new Map<number, WebSocket>();
 let mockManagedAgents: MockManagedAgent[] = [];
 
@@ -8234,6 +8237,16 @@ function sendToMockSocket(args: {
         channelIds.size === 1
           ? (channelIds.values().next().value as string)
           : undefined;
+      if (
+        getConfig()?.mock?.closeChannelLiveSubscriptionOnce &&
+        !mockClosedChannelLiveSubscription &&
+        onlyChannelId &&
+        kinds.has(KIND_CHANNEL_THREAD_SUMMARY)
+      ) {
+        mockClosedChannelLiveSubscription = true;
+        sendWsText(socket.handler, ["CLOSED", subId, "rate-limited"]);
+        return;
+      }
       socket.subscriptions.set(subId, {
         channelId: onlyChannelId ?? GLOBAL_MOCK_SUBSCRIPTION,
         kinds: kinds.size > 0 ? [...kinds] : null,
@@ -8451,6 +8464,7 @@ export function maybeInstallE2eTauriMocks() {
     return;
   }
 
+  mockClosedChannelLiveSubscription = false;
   mockGlobalAgentConfig = config.mock?.globalAgentConfig
     ? { ...config.mock.globalAgentConfig }
     : null;

desktop/tests/e2e/messaging.spec.ts

@@ -1158,6 +1158,35 @@ test("thread refetch preserves a live reply and reaction received in flight", as
   await expect(replyRow.getByLabel("Toggle 👍 reaction")).toBeVisible();
 });
 
+test("thread reply appears after relay closes and restores its live subscription", async ({
+  page,
+}) => {
+  await installMockBridge(page, { closeChannelLiveSubscriptionOnce: true });
+  await page.goto("/");
+  await page.getByTestId("channel-general").click();
+  await expect(page.getByTestId("chat-title")).toHaveText("general");
+
+  const seed = `Thread CLOSED seed ${Date.now()}`;
+  await page.getByTestId("message-input").fill(seed);
+  await page.getByTestId("send-message").click();
+  await expect(page.getByTestId("message-timeline")).toContainText(seed);
+
+  const rootMessage = page
+    .getByTestId("message-timeline")
+    .getByTestId("message-row")
+    .last();
+  await rootMessage.hover();
+  await rootMessage.getByRole("button", { name: "Reply" }).click();
+
+  const threadPanel = page.getByTestId("message-thread-panel");
+  const reply = `Thread reply after CLOSED ${Date.now()}`;
+  await threadPanel.getByTestId("message-input").fill(reply);
+  await page.waitForTimeout(1_100);
+  await threadPanel.getByTestId("send-message").click();
+
+  await expect(threadPanel).toContainText(reply);
+});
+
 test("thread composer keeps focus after sending a thread reply", async ({
   page,
 }) => {

desktop/tests/helpers/bridge.ts

@@ -184,6 +184,8 @@ type MockBridgeOptions = {
   applyCommunityDelayMs?: number;
   openDmDelayMs?: number;
   sendMessageDelayMs?: number;
+  /** Close the first channel-window live REQ; its retry is accepted. */
+  closeChannelLiveSubscriptionOnce?: boolean;
   /** Reject successive kind-9 sends with these messages, then resume. */
   sendMessageErrors?: string[];
   /** Reject successive managed-agent starts, then resume. */