fix(desktop): stop random timeline message loss + page reconnect replay (#1105)

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

desktop/src/features/messages/hooks.ts

@@ -4,6 +4,7 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
 import {
   channelMessagesKey,
   dedupeMessagesById,
+  mergeTimelineHistoryMessages,
   normalizeTimelineMessages,
   sortMessages,
 } from "@/features/messages/lib/messageQueryKeys";
@@ -180,10 +181,10 @@ export function useChannelMessagesQuery(channel: Channel | null) {
       );
       const currentMessages =
         queryClient.getQueryData<RelayEvent[]>(queryKey) ?? [];
-      const mergedHistory = normalizeTimelineMessages([
-        ...currentMessages,
-        ...history,
-      ]);
+      const mergedHistory = mergeTimelineHistoryMessages(
+        currentMessages,
+        history,
+      );
 
       return mergedHistory;
     },
@@ -208,14 +209,7 @@ export function useChannelSubscription(channel: Channel | null) {
 
     queryClient.setQueryData<RelayEvent[]>(
       channelMessagesKey(channelId),
-      (current = []) => {
-        const mergedHistory = normalizeTimelineMessages([
-          ...current,
-          ...history,
-        ]);
-
-        return mergedHistory;
-      },
+      (current = []) => mergeTimelineHistoryMessages(current, history),
     );
   });
 

desktop/src/features/messages/lib/messageQueryKeys.test.mjs

@@ -0,0 +1,204 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  mergeTimelineHistoryMessages,
+  normalizeTimelineMessages,
+} from "./messageQueryKeys.ts";
+import { mergeTimelineCacheMessages } from "../hooks.ts";
+
+const CHANNEL_ID = "timeline-window-test";
+const PUBKEY = "a".repeat(64);
+
+function event({ id, kind = 9, createdAt, tags, content = "" }) {
+  return {
+    id,
+    pubkey: PUBKEY,
+    created_at: createdAt,
+    kind,
+    tags: tags ?? [["h", CHANNEL_ID]],
+    content,
+    sig: "mocksig".repeat(20).slice(0, 128),
+  };
+}
+
+function id(prefix, index) {
+  return `${prefix}${String(index).padStart(64 - prefix.length, "0")}`;
+}
+
+test("normalizeTimelineMessages caps visible content, not unrelated auxiliary events", () => {
+  const june12Roots = [
+    "1f86d0450b3c2c376691e7a8232fbd8a5b8408ecad2f5eb0209e7bfcfdf9af80",
+    "3b745de3d9ff91c464b0cbf26e3e628a5a8c05dccf3f9781b1fbd99a0f6f5e7b",
+    "cb404eeae1517bb2ed2e9975dfd2efd12d0a7ef17b87c3a6573b251b9865f7a4",
+    "337de49c712fcf84f5689a6c11ce36018817197d578468b8132c6dc3d1a13131",
+    "ac683f35cda2e8c1e9ae609e0b9f5dc23a7d434637b4f0505495c3a2b6f52aae",
+  ];
+  const messages = [];
+
+  for (let index = 0; index < 500; index += 1) {
+    messages.push(event({ id: id("old", index), createdAt: 1_000 + index }));
+  }
+  for (const [index, rootId] of june12Roots.entries()) {
+    messages.push(
+      event({ id: rootId, createdAt: 2_000 + index, content: "June 12 root" }),
+    );
+  }
+  for (let index = 0; index < 1_303; index += 1) {
+    messages.push(
+      event({
+        id: id("del", index),
+        kind: 5,
+        createdAt: 3_000 + index,
+        tags: [
+          ["h", CHANNEL_ID],
+          ["e", id("zzz", index)],
+        ],
+      }),
+    );
+  }
+  for (let index = 0; index < 231; index += 1) {
+    messages.push(
+      event({
+        id: id("rea", index),
+        kind: 7,
+        createdAt: 5_000 + index,
+        tags: [
+          ["h", CHANNEL_ID],
+          ["e", id("yyy", index)],
+        ],
+        content: "+",
+      }),
+    );
+  }
+  for (let index = 0; index < 1_495; index += 1) {
+    messages.push(event({ id: id("new", index), createdAt: 6_000 + index }));
+  }
+
+  const normalized = normalizeTimelineMessages(messages);
+
+  assert.equal(normalized.filter((item) => item.kind === 9).length, 2_000);
+  assert.deepEqual(
+    june12Roots.map((rootId) => normalized.some((item) => item.id === rootId)),
+    [true, true, true, true, true],
+  );
+  assert.equal(normalized.filter((item) => item.kind === 5).length, 1_303);
+  assert.equal(normalized.filter((item) => item.kind === 7).length, 231);
+});
+
+test("normalizeTimelineMessages still caps old visible content", () => {
+  const retainedRoot = `${"a".repeat(63)}1`;
+  const reaction = `${"b".repeat(63)}1`;
+  const reactionDeletion = `${"c".repeat(63)}1`;
+  const messages = [];
+
+  for (let index = 0; index < 2_000; index += 1) {
+    messages.push(event({ id: id("old", index), createdAt: 1_000 + index }));
+  }
+  messages.push(event({ id: retainedRoot, createdAt: 4_000 }));
+  messages.push(
+    event({
+      id: reaction,
+      kind: 7,
+      createdAt: 4_001,
+      tags: [
+        ["h", CHANNEL_ID],
+        ["e", retainedRoot],
+      ],
+      content: "+",
+    }),
+  );
+  messages.push(
+    event({
+      id: reactionDeletion,
+      kind: 5,
+      createdAt: 4_002,
+      tags: [
+        ["h", CHANNEL_ID],
+        ["e", reaction],
+      ],
+    }),
+  );
+
+  const normalized = normalizeTimelineMessages(messages);
+
+  assert.equal(
+    normalized.some((item) => item.id === id("old", 0)),
+    false,
+  );
+  assert.equal(
+    normalized.some((item) => item.id === retainedRoot),
+    true,
+  );
+  assert.equal(
+    normalized.some((item) => item.id === reaction),
+    true,
+  );
+  assert.equal(
+    normalized.some((item) => item.id === reactionDeletion),
+    true,
+  );
+});
+
+test("timeline history and live cache merges retain the same visible content regardless of order", () => {
+  const seedMessages = [];
+  const olderPage = [];
+  const liveMessage = event({ id: id("liv", 0), createdAt: 20_000 });
+
+  for (let index = 0; index < 700; index += 1) {
+    seedMessages.push(
+      event({ id: id("new", index), createdAt: 10_000 + index }),
+    );
+  }
+  for (let index = 0; index < 1_303; index += 1) {
+    seedMessages.push(
+      event({
+        id: id("del", index),
+        kind: 5,
+        createdAt: 11_000 + index,
+        tags: [
+          ["h", CHANNEL_ID],
+          ["e", id("zzz", index)],
+        ],
+      }),
+    );
+  }
+  for (let index = 0; index < 231; index += 1) {
+    seedMessages.push(
+      event({
+        id: id("rea", index),
+        kind: 7,
+        createdAt: 13_000 + index,
+        tags: [
+          ["h", CHANNEL_ID],
+          ["e", id("yyy", index)],
+        ],
+        content: "+",
+      }),
+    );
+  }
+  for (let index = 0; index < 1_500; index += 1) {
+    olderPage.push(event({ id: id("old", index), createdAt: 1_000 + index }));
+  }
+
+  const historyThenLive = mergeTimelineCacheMessages(
+    mergeTimelineHistoryMessages(seedMessages, olderPage),
+    liveMessage,
+  );
+  const liveThenHistory = mergeTimelineHistoryMessages(
+    mergeTimelineCacheMessages(seedMessages, liveMessage),
+    olderPage,
+  );
+  const historyThenLiveContent = historyThenLive
+    .filter((item) => item.kind === 9)
+    .map((item) => item.id);
+  const liveThenHistoryContent = liveThenHistory
+    .filter((item) => item.kind === 9)
+    .map((item) => item.id);
+
+  assert.equal(historyThenLiveContent.length, 2_000);
+  assert.equal(liveThenHistoryContent.length, 2_000);
+  assert.deepEqual(liveThenHistoryContent, historyThenLiveContent);
+  assert.equal(historyThenLiveContent[0], id("old", 201));
+  assert.equal(historyThenLiveContent.at(-1), liveMessage.id);
+});

desktop/src/features/messages/lib/messageQueryKeys.ts

@@ -1,4 +1,16 @@
 import type { RelayEvent } from "@/shared/api/types";
+import {
+  KIND_JOB_ACCEPTED,
+  KIND_JOB_CANCEL,
+  KIND_JOB_ERROR,
+  KIND_JOB_PROGRESS,
+  KIND_JOB_REQUEST,
+  KIND_JOB_RESULT,
+  KIND_STREAM_MESSAGE,
+  KIND_STREAM_MESSAGE_DIFF,
+  KIND_STREAM_MESSAGE_V2,
+  KIND_SYSTEM_MESSAGE,
+} from "@/shared/constants/kinds";
 
 const MAX_TIMELINE_MESSAGES = 2_000;
 
@@ -30,17 +42,51 @@ export function sortMessages(messages: RelayEvent[]) {
   );
 }
 
+function isTimelineWindowContentEvent(event: RelayEvent) {
+  return (
+    event.kind === KIND_STREAM_MESSAGE ||
+    event.kind === KIND_STREAM_MESSAGE_V2 ||
+    event.kind === KIND_STREAM_MESSAGE_DIFF ||
+    event.kind === KIND_SYSTEM_MESSAGE ||
+    event.kind === KIND_JOB_REQUEST ||
+    event.kind === KIND_JOB_ACCEPTED ||
+    event.kind === KIND_JOB_PROGRESS ||
+    event.kind === KIND_JOB_RESULT ||
+    event.kind === KIND_JOB_CANCEL ||
+    event.kind === KIND_JOB_ERROR
+  );
+}
+
 /**
- * Sort, dedupe, and cap the timeline at {@link MAX_TIMELINE_MESSAGES} so
- * de-virtualized rendering does not grow into an unbounded DOM during
- * long-lived channel sessions.
+ * Sort, dedupe, and cap the timeline at {@link MAX_TIMELINE_MESSAGES} visible
+ * content events so de-virtualized rendering does not grow into an unbounded
+ * DOM during long-lived channel sessions.
+ *
+ * Auxiliary events (reactions, edits, tombstones) are kept in cache so they can
+ * still apply to retained or later-loaded content, but they must not consume the
+ * visible message window and evict older loaded roots.
  */
 export function normalizeTimelineMessages(messages: RelayEvent[]) {
   const normalized = sortMessages(messages);
+  const contentEvents = normalized.filter(isTimelineWindowContentEvent);
 
-  if (normalized.length <= MAX_TIMELINE_MESSAGES) {
+  if (contentEvents.length <= MAX_TIMELINE_MESSAGES) {
     return normalized;
   }
 
-  return normalized.slice(-MAX_TIMELINE_MESSAGES);
+  const retainedContentIds = new Set(
+    contentEvents.slice(-MAX_TIMELINE_MESSAGES).map((event) => event.id),
+  );
+
+  return normalized.filter(
+    (event) =>
+      !isTimelineWindowContentEvent(event) || retainedContentIds.has(event.id),
+  );
+}
+
+export function mergeTimelineHistoryMessages(
+  current: RelayEvent[],
+  history: RelayEvent[],
+) {
+  return normalizeTimelineMessages([...current, ...history]);
 }

desktop/src/features/messages/useFetchOlderMessages.ts

@@ -3,7 +3,7 @@ import { useQueryClient } from "@tanstack/react-query";
 
 import {
   channelMessagesKey,
-  sortMessages,
+  mergeTimelineHistoryMessages,
 } from "@/features/messages/lib/messageQueryKeys";
 import { relayClient } from "@/shared/api/relayClient";
 import type { Channel, RelayEvent } from "@/shared/api/types";
@@ -64,7 +64,7 @@ export function useFetchOlderMessages(channel: Channel | null) {
 
       if (olderMessages.length > 0) {
         queryClient.setQueryData<RelayEvent[]>(queryKey, (current = []) =>
-          sortMessages([...current, ...olderMessages]),
+          mergeTimelineHistoryMessages(current, olderMessages),
         );
 
         const updatedMessages =

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,175 @@
+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,
+      },
+    ],
+  ]);
+  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,124 @@
+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);
+    await sendRaw([
+      "REQ",
+      subId,
+      shouldPageReplay
+        ? subscription.filter
+        : buildReconnectReplayFilter(subscription.filter, replaySince),
+    ]);
+
+    if (shouldPageReplay) {
+      await replayReconnectHistoryPages({
+        subscription,
+        since: replaySince,
+        until: now,
+        isActive: () => subscriptions.get(subId) === subscription,
+        requestHistory,
+      });
+    }
+  }
+}

desktop/src/testing/e2eBridge.ts

@@ -466,6 +466,9 @@ type MockFilter = {
   "#h"?: string[];
   authors?: string[];
   kinds?: number[];
+  limit?: number;
+  since?: number;
+  until?: number;
 };
 
 type MockSocket = {
@@ -606,6 +609,7 @@ declare global {
     }>;
     __BUZZ_E2E_SET_RELAY_CONNECTION_STATE__?: (state: ConnectionState) => void;
     __BUZZ_E2E_SET_STALL_WEBSOCKET_SENDS__?: (stall: boolean) => void;
+    __BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__?: () => number;
     __BUZZ_E2E_SET_MESH__?: (mesh: {
       admitted?: boolean;
       models?: Array<{ id: string; name: string | null }>;
@@ -2263,8 +2267,29 @@ function getMockMessageStore(channelId: string): RelayEvent[] {
   return seeded;
 }
 
-function emitMockHistory(socket: MockSocket, subId: string, channelId: string) {
-  const events = getMockMessageStore(channelId);
+function emitMockHistory(
+  socket: MockSocket,
+  subId: string,
+  channelId: string,
+  filter: MockFilter,
+) {
+  const events = getMockMessageStore(channelId)
+    .filter((event) => {
+      if (filter.kinds && !filter.kinds.includes(event.kind)) {
+        return false;
+      }
+      if (filter.since !== undefined && event.created_at < filter.since) {
+        return false;
+      }
+      if (filter.until !== undefined && event.created_at > filter.until) {
+        return false;
+      }
+      return true;
+    })
+    .sort((left, right) => right.created_at - left.created_at)
+    .slice(0, filter.limit ?? 50)
+    .sort((left, right) => left.created_at - right.created_at);
+
   for (const event of events) {
     sendWsText(socket.handler, ["EVENT", subId, event]);
   }
@@ -5847,7 +5872,7 @@ function sendToMockSocket(args: {
       return;
     }
 
-    emitMockHistory(socket, subId, channelId);
+    emitMockHistory(socket, subId, channelId, filter);
     return;
   }
 
@@ -6100,6 +6125,11 @@ export function maybeInstallE2eTauriMocks() {
     config.mock.stallWebsocketSends = stall;
     if (!stall) mockWebsocketSendMutexWedged = false;
   };
+  window.__BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__ = () => {
+    const socketIds = [...mockSockets.keys()];
+    for (const socketId of socketIds) disconnectMockSocket(socketId);
+    return socketIds.length;
+  };
   // Tests flip `admitted` to exercise the denial path: mesh_ensure_client_node
   // rejects when not admitted, which proves relay membership is the gate and
   // that the create flow surfaces denial copy without spawning the agent.

desktop/tests/e2e/relay-reconnect.spec.ts

@@ -19,6 +19,46 @@ async function setMockWebsocketSendsStalled(
   }, stall);
 }
 
+async function disconnectMockWebsockets(page: import("@playwright/test").Page) {
+  const disconnected = await page.evaluate(() => {
+    const disconnect = (
+      window as Window & {
+        __BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__?: () => number;
+      }
+    ).__BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__;
+    if (!disconnect) {
+      throw new Error("E2E mock websocket disconnect seam is not installed.");
+    }
+    return disconnect();
+  });
+
+  expect(disconnected).toBeGreaterThan(0);
+}
+
+async function emitMockMessages(
+  page: import("@playwright/test").Page,
+  messages: Array<{ content: string; createdAt: number }>,
+) {
+  await page.evaluate((items) => {
+    const emit = (
+      window as Window & {
+        __BUZZ_E2E_EMIT_MOCK_MESSAGE__?: (input: {
+          channelName: string;
+          content: string;
+          createdAt: number;
+        }) => unknown;
+      }
+    ).__BUZZ_E2E_EMIT_MOCK_MESSAGE__;
+    if (!emit) {
+      throw new Error("E2E mock message emitter is not installed.");
+    }
+
+    for (const item of items) {
+      emit({ channelName: "general", ...item });
+    }
+  }, messages);
+}
+
 async function driveConnectionDegraded(
   page: import("@playwright/test").Page,
   state: "reconnecting" | "stalled" | "disconnected",
@@ -110,3 +150,36 @@ test("profile popover does not show relay reconnect controls", async ({
   });
   await expect(page.getByTestId("profile-popover-reconnect")).toHaveCount(0);
 });
+
+test("reconnect backfills more missed channel messages than the live subscription limit", async ({
+  page,
+}) => {
+  await page.goto("/");
+  await page.getByTestId("channel-general").click();
+  await expect(page.getByTestId("chat-title")).toHaveText("general");
+
+  const baseCreatedAt = Math.floor(Date.now() / 1_000) - 300;
+  const seenBeforeDisconnect = "reconnect e2e seen before disconnect";
+  await emitMockMessages(page, [
+    { content: seenBeforeDisconnect, createdAt: baseCreatedAt },
+  ]);
+  await expect(page.getByTestId("message-timeline")).toContainText(
+    seenBeforeDisconnect,
+  );
+
+  await disconnectMockWebsockets(page);
+
+  const missedMessages = Array.from({ length: 260 }, (_, index) => ({
+    content: `reconnect e2e missed ${String(index + 1).padStart(3, "0")}`,
+    createdAt: baseCreatedAt + index + 1,
+  }));
+  await emitMockMessages(page, missedMessages);
+
+  await expect(page.getByTestId("message-timeline")).toContainText(
+    "reconnect e2e missed 001",
+    { timeout: 15_000 },
+  );
+  await expect(page.getByTestId("message-timeline")).toContainText(
+    "reconnect e2e missed 260",
+  );
+});