fix(desktop): page reconnect replay windows

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

Author: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta

desktop/src/shared/api/relayClientSession.ts

@@ -21,6 +21,7 @@ import {
   type RelaySubscription,
   type RelaySubscriptionFilter,
 } from "@/shared/api/relayClientShared";
+import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
 import { RelayConnectionStateEmitter } from "@/shared/api/relayConnectionStateEmitter";
 import {
   shouldRefuseConnect,
@@ -31,7 +32,6 @@ import { buildThreadReferenceTags } from "@/features/messages/lib/threading";
 
 const RECONNECT_BASE_DELAY_MS = 1_000,
   RECONNECT_MAX_DELAY_MS = 30_000,
-  RECONNECT_REPLAY_SKEW_SECS = 5,
   EVENT_BATCH_MS = 16;
 
 /**
@@ -165,7 +165,10 @@ export class RelayClient {
 
   private async fetchHistory(filter: RelaySubscriptionFilter) {
     await this.ensureConnected();
+    return this.requestHistory(filter);
+  }
 
+  private requestHistory(filter: RelaySubscriptionFilter) {
     return new Promise<RelayEvent[]>((resolve, reject) => {
       const subId = `history-${crypto.randomUUID()}`;
       const timeout = window.setTimeout(() => {
@@ -860,45 +863,20 @@ export class RelayClient {
     return false;
   }
 
-  private buildReplayFilter(filter: RelaySubscriptionFilter, since?: number) {
-    if (since === undefined) {
-      return filter;
-    }
-
-    return {
-      ...filter,
-      since: filter.since === undefined ? since : Math.max(filter.since, since),
-    };
-  }
-
   private async replayLiveSubscriptions() {
-    for (const [subId, subscription] of this.subscriptions) {
-      if (subscription.mode !== "live") {
-        continue;
-      }
-
-      const replaySince =
-        subscription.lastSeenCreatedAt === undefined
-          ? undefined
-          : Math.max(
-              0,
-              subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS,
-            );
-
-      try {
-        await this.sendRaw([
-          "REQ",
-          subId,
-          this.buildReplayFilter(subscription.filter, replaySince),
-        ]);
-      } catch (error) {
-        const reconnectError =
-          error instanceof Error
-            ? error
-            : new Error("Failed to restore relay subscriptions.");
-        this.resetConnection(reconnectError);
-        throw reconnectError;
-      }
+    try {
+      await replayLiveSubscriptions({
+        subscriptions: this.subscriptions,
+        sendRaw: (payload) => this.sendRaw(payload),
+        requestHistory: (filter) => this.requestHistory(filter),
+      });
+    } catch (error) {
+      const reconnectError =
+        error instanceof Error
+          ? error
+          : new Error("Failed to restore relay subscriptions.");
+      this.resetConnection(reconnectError);
+      throw reconnectError;
     }
   }
 

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

@@ -0,0 +1,176 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  buildReconnectReplayFilter,
+  replayLiveSubscriptions,
+} from "./relayReconnectReplay.ts";
+import { RelayClient } from "./relayClientSession.ts";
+
+function replayFilter(filter, since, until) {
+  return buildReconnectReplayFilter(filter, since, until);
+}
+
+function event(id, createdAt) {
+  return {
+    id,
+    pubkey: "pubkey",
+    created_at: createdAt,
+    kind: 9,
+    tags: [],
+    content: "",
+    sig: "sig",
+  };
+}
+
+function eventRange(prefix, start, count) {
+  return Array.from({ length: count }, (_, index) =>
+    event(`${prefix}-${index}`, start + index),
+  );
+}
+
+test("reconnect replay preserves small steady-state limits when adding since", () => {
+  const filter = {
+    kinds: [9, 40002],
+    "#h": ["channel-1"],
+    limit: 50,
+  };
+
+  assert.deepEqual(replayFilter(filter, 123), {
+    kinds: [9, 40002],
+    "#h": ["channel-1"],
+    limit: 50,
+    since: 123,
+  });
+});
+
+test("reconnect replay caps large steady-state limits", () => {
+  const filter = {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 1000,
+  };
+
+  assert.deepEqual(replayFilter(filter, 123), {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 500,
+    since: 123,
+  });
+});
+
+test("reconnect replay keeps the stricter existing since window", () => {
+  const filter = {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 50,
+    since: 200,
+  };
+
+  assert.deepEqual(replayFilter(filter, 123), {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 50,
+    since: 200,
+  });
+});
+
+test("reconnect replay applies the stricter until window", () => {
+  const filter = {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 50,
+    until: 300,
+  };
+
+  assert.deepEqual(replayFilter(filter, 123, 400), {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 50,
+    since: 123,
+    until: 300,
+  });
+});
+
+test("initial subscription replay preserves the original filter", () => {
+  const filter = {
+    kinds: [9],
+    "#h": ["channel-1"],
+    limit: 50,
+  };
+
+  assert.equal(replayFilter(filter, undefined), filter);
+});
+
+test("channel reconnect replay pages the missed window until a short page", async () => {
+  const delivered = [];
+  const historyFilters = [];
+  const sentPayloads = [];
+  const pages = [
+    eventRange("newest", 1501, 500),
+    eventRange("middle", 1002, 500),
+    eventRange("oldest", 995, 8),
+  ];
+  const client = new RelayClient();
+  const filter = client.buildChannelFilter("channel-1", 50);
+  const subscriptions = new Map([
+    [
+      "live-1",
+      {
+        mode: "live",
+        filter,
+        onEvent: (event) => delivered.push(event),
+        lastSeenCreatedAt: 1000,
+      },
+    ],
+  ]);
+
+  await replayLiveSubscriptions({
+    subscriptions,
+    now: 2000,
+    sendRaw: async (payload) => {
+      sentPayloads.push(payload);
+    },
+    requestHistory: async (filter) => {
+      historyFilters.push(filter);
+      return pages.shift() ?? [];
+    },
+  });
+
+  assert.deepEqual(sentPayloads, [
+    [
+      "REQ",
+      "live-1",
+      {
+        kinds: filter.kinds,
+        "#h": ["channel-1"],
+        limit: 50,
+        since: 1995,
+      },
+    ],
+  ]);
+  assert.deepEqual(historyFilters, [
+    {
+      kinds: filter.kinds,
+      "#h": ["channel-1"],
+      limit: 500,
+      since: 995,
+      until: 2000,
+    },
+    {
+      kinds: filter.kinds,
+      "#h": ["channel-1"],
+      limit: 500,
+      since: 995,
+      until: 1501,
+    },
+    {
+      kinds: filter.kinds,
+      "#h": ["channel-1"],
+      limit: 500,
+      since: 995,
+      until: 1002,
+    },
+  ]);
+  assert.equal(delivered.length, 1008);
+});

desktop/src/shared/api/relayReconnectReplay.ts

@@ -0,0 +1,126 @@
+import { CHANNEL_EVENT_KINDS } from "@/shared/constants/kinds";
+import type {
+  RelaySubscription,
+  RelaySubscriptionFilter,
+} from "@/shared/api/relayClientShared";
+import type { RelayEvent } from "@/shared/api/types";
+
+const RECONNECT_REPLAY_SKEW_SECS = 5;
+export const RECONNECT_REPLAY_PAGE_LIMIT = 500;
+
+export function buildReconnectReplayFilter(
+  filter: RelaySubscriptionFilter,
+  since?: number,
+  until?: number,
+  limit = Math.min(filter.limit, RECONNECT_REPLAY_PAGE_LIMIT),
+) {
+  if (since === undefined) return filter;
+
+  const replayFilter: RelaySubscriptionFilter = {
+    ...filter,
+    limit,
+    since: filter.since === undefined ? since : Math.max(filter.since, since),
+  };
+
+  if (until !== undefined) {
+    replayFilter.until =
+      filter.until === undefined ? until : Math.min(filter.until, until);
+  }
+
+  return replayFilter;
+}
+
+export function shouldPageReconnectReplay(filter: RelaySubscriptionFilter) {
+  return (
+    filter.limit > 0 &&
+    Array.isArray(filter["#h"]) &&
+    filter["#h"].length === 1 &&
+    CHANNEL_EVENT_KINDS.every((kind) => filter.kinds.includes(kind))
+  );
+}
+
+export async function replayReconnectHistoryPages({
+  subscription,
+  since,
+  until,
+  isActive,
+  requestHistory,
+}: {
+  subscription: Extract<RelaySubscription, { mode: "live" }>;
+  since: number;
+  until: number;
+  isActive: () => boolean;
+  requestHistory: (filter: RelaySubscriptionFilter) => Promise<RelayEvent[]>;
+}) {
+  let pageUntil = until;
+
+  while (pageUntil >= since) {
+    if (!isActive()) return;
+
+    const events = await requestHistory(
+      buildReconnectReplayFilter(
+        subscription.filter,
+        since,
+        pageUntil,
+        RECONNECT_REPLAY_PAGE_LIMIT,
+      ),
+    );
+
+    if (!isActive()) return;
+
+    for (const event of events) subscription.onEvent(event);
+    if (events.length < RECONNECT_REPLAY_PAGE_LIMIT) return;
+
+    const oldestCreatedAt = events[0]?.created_at;
+    if (oldestCreatedAt === undefined || oldestCreatedAt <= since) return;
+
+    pageUntil =
+      oldestCreatedAt < pageUntil ? oldestCreatedAt : oldestCreatedAt - 1;
+  }
+}
+
+export async function replayLiveSubscriptions({
+  subscriptions,
+  sendRaw,
+  requestHistory,
+  now = Math.floor(Date.now() / 1_000),
+}: {
+  subscriptions: Map<string, RelaySubscription>;
+  sendRaw: (payload: unknown[]) => Promise<void>;
+  requestHistory: (filter: RelaySubscriptionFilter) => Promise<RelayEvent[]>;
+  now?: number;
+}) {
+  for (const [subId, subscription] of subscriptions) {
+    if (subscription.mode !== "live") continue;
+
+    const replaySince =
+      subscription.lastSeenCreatedAt === undefined
+        ? undefined
+        : Math.max(
+            0,
+            subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS,
+          );
+    const shouldPageReplay =
+      replaySince !== undefined &&
+      shouldPageReconnectReplay(subscription.filter);
+    const liveReplaySince = shouldPageReplay
+      ? Math.max(0, now - RECONNECT_REPLAY_SKEW_SECS)
+      : replaySince;
+
+    await sendRaw([
+      "REQ",
+      subId,
+      buildReconnectReplayFilter(subscription.filter, liveReplaySince),
+    ]);
+
+    if (shouldPageReplay) {
+      await replayReconnectHistoryPages({
+        subscription,
+        since: replaySince,
+        until: now,
+        isActive: () => subscriptions.get(subId) === subscription,
+        requestHistory,
+      });
+    }
+  }
+}