From 9e5a9518e7f3fb62cdf14dc735d1092ed5844a85 Mon Sep 17 00:00:00 2001 From: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz> Date: Sun, 20 Sep 2026 10:48:46 -0600 Subject: [PATCH] perf(desktop): share identical in-flight reconnect repair pages Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz> --- .../scripts/benchmark-reconnect-sharing.mjs | 149 ++++++ .../api/relayClientReconnectSharing.test.mjs | 133 +++++ .../src/shared/api/relayReconnectReplay.ts | 23 +- .../shared/api/relayReconnectSharing.test.mjs | 321 ++++++++++++ desktop/tests/e2e/relay-reconnect.spec.ts | 118 +++++ perf/RECONNECT_SHARING.md | 118 +++++ .../RECONNECT_SHARING_RESULTS_2026_09_20.json | 462 ++++++++++++++++++ 7 files changed, 1323 insertions(+), 1 deletion(-) create mode 100644 desktop/scripts/benchmark-reconnect-sharing.mjs create mode 100644 desktop/src/shared/api/relayClientReconnectSharing.test.mjs create mode 100644 desktop/src/shared/api/relayReconnectSharing.test.mjs create mode 100644 perf/RECONNECT_SHARING.md create mode 100644 perf/RECONNECT_SHARING_RESULTS_2026_09_20.json diff --git a/desktop/scripts/benchmark-reconnect-sharing.mjs b/desktop/scripts/benchmark-reconnect-sharing.mjs new file mode 100644 index 0000000000..ac8fc2719e --- /dev/null +++ b/desktop/scripts/benchmark-reconnect-sharing.mjs @@ -0,0 +1,149 @@ +import assert from "node:assert/strict"; +import { performance } from "node:perf_hooks"; +import { setTimeout as sleep } from "node:timers/promises"; +import { pathToFileURL } from "node:url"; +import { resolve } from "node:path"; +// Run from desktop after extracting the exact pre-change TS source to a file: +// node --import ./test-loader.mjs scripts/benchmark-reconnect-sharing.mjs /path/to/baseline.ts +// Synthetic transport only: timings are callback completion, not native/UI paint. +const baselinePath = process.argv[2]; +if (!baselinePath) + throw new Error("Provide a pristine relayReconnectReplay.ts baseline path"); +const { replayLiveSubscriptions: baseline } = await import( + pathToFileURL(resolve(baselinePath)).href +); +import { replayLiveSubscriptions as candidate } from "../src/shared/api/relayReconnectReplay.ts"; +import { CHANNEL_EVENT_KINDS } from "../src/shared/constants/kinds.ts"; +import { resetRateLimitGate } from "../src/shared/api/relayRateLimitGate.ts"; + +globalThis.window = globalThis; +const latency = 20; +const id = (channel, n) => + `${channel.toString(16).padStart(8, "0")}${n.toString(16).padStart(56, "0")}`; +const event = (channel, n) => ({ + id: id(channel, n), + pubkey: "a".repeat(64), + kind: 9, + created_at: 2000 + Math.floor(n / 3), + tags: [["h", `channel-${channel}`]], + content: `row ${n}`, + sig: "b".repeat(128), +}); +async function run(replay, channels, overlap, slots) { + resetRateLimitGate(); + const rows = Array.from({ length: channels }, (_, c) => + Array.from({ length: 520 }, (_, n) => event(c, n)).sort( + (a, b) => b.created_at - a.created_at || a.id.localeCompare(b.id), + ), + ); + const subscriptions = new Map(); + const delivered = new Map(); + let foregroundCompleteMs = null, + readCount = 0, + active = 0, + peak = 0, + sent = 0; + const waiting = []; + const start = performance.now(); + function add(key, c) { + const received = new Set(); + delivered.set(key, received); + subscriptions.set(key, { + mode: "live", + filter: { + kinds: [...CHANNEL_EVENT_KINDS], + "#h": [`channel-${c}`], + since: 1000, + limit: 1000, + }, + lastSeenCreatedAt: 2200, + onEvent(e) { + assert(!received.has(e.id)); + received.add(e.id); + if (key === "visible" && received.size === 520) + foregroundCompleteMs = performance.now() - start; + }, + }); + } + add("visible", 0); + for (let c = overlap ? 0 : 1; c < channels; c++) add(`background-${c}`, c); + await replay({ + subscriptions, + visibleChannelId: "channel-0", + generation: 1, + sendRaw: async () => { + sent++; + }, + requestRepair: async ({ channelId, since, until, beforeId, limit }) => { + readCount++; + if (active >= slots) await new Promise((r) => waiting.push(r)); + active++; + peak = Math.max(peak, active); + try { + await sleep(latency); + const c = Number(channelId.split("-")[1]); + return rows[c] + .filter( + (e) => + e.created_at >= since && + (until === undefined || + e.created_at < until || + (e.created_at === until && (!beforeId || e.id > beforeId))), + ) + .slice(0, limit); + } finally { + active--; + waiting.shift()?.(); + } + }, + }); + const total = performance.now() - start; + for (const [key, sub] of subscriptions) { + const channel = Number(sub.filter["#h"][0].split("-")[1]); + assert.deepEqual( + [...delivered.get(key)], + rows[channel].map((row) => row.id), + "each subscriber must receive every expected row in relay order", + ); + assert.equal(sub.pendingReplaySince, undefined); + assert.equal(sub.reconnectReplay.repairDone, true); + } + return { + foregroundCompleteMs, + totalMs: total, + reads: readCount, + delivered: [...delivered.values()].reduce((sum, v) => sum + v.size, 0), + reqs: sent, + peak, + }; +} +const results = []; +for (const [channels, overlap, slots] of [ + [1, true, 2], + [12, true, 2], + [32, true, 2], + [12, false, 2], + [12, true, 4], +]) { + const samples = { before: [], after: [] }; + for (let pair = 0; pair < 5; pair++) { + for (const arm of pair % 2 ? ["after", "before"] : ["before", "after"]) { + samples[arm].push( + await run( + arm === "before" ? baseline : candidate, + channels, + overlap, + slots, + ), + ); + } + } + results.push({ + channels, + overlap, + upstreamSlots: slots, + modeledReadMs: latency, + samples, + }); +} +console.log(JSON.stringify(results, null, 2)); diff --git a/desktop/src/shared/api/relayClientReconnectSharing.test.mjs b/desktop/src/shared/api/relayClientReconnectSharing.test.mjs new file mode 100644 index 0000000000..b430d9092a --- /dev/null +++ b/desktop/src/shared/api/relayClientReconnectSharing.test.mjs @@ -0,0 +1,133 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { CHANNEL_EVENT_KINDS } from "../constants/kinds.ts"; + +const timers = new Map(); +let timerId = 0; +let invoke; +globalThis.window = { + setTimeout: (fn) => { + timers.set(++timerId, fn); + return timerId; + }, + clearTimeout: (id) => timers.delete(id), + __TAURI_INTERNALS__: { invoke: (...args) => invoke(...args) }, +}; +const { RelayClient } = await import("./relayClientSession.ts"); + +async function flush() { + for (let i = 0; i < 80; i++) await Promise.resolve(); +} + +function event(id, createdAt) { + return Object.freeze({ + id: String(id).padStart(64, "0"), + pubkey: "a".repeat(64), + kind: 9, + created_at: createdAt, + tags: Object.freeze([Object.freeze(["h", "channel-1"])]), + content: "", + sig: "b".repeat(128), + }); +} + +function fixture() { + timers.clear(); + const client = new RelayClient(); + client.wsId = 7; + const reads = []; + const frames = []; + const deliver = (frame) => + client.handleWsMessage( + { type: "Text", data: JSON.stringify(frame) }, + client.connectionGeneration, + ); + invoke = async (command, args) => { + if (command === "plugin:websocket|send") { + const frame = JSON.parse(args.message.data); + frames.push(frame); + if (frame[0] === "REQ") await deliver(["EOSE", frame[1]]); + return; + } + if (command === "get_channel_reconnect_repair") { + return new Promise((resolve, reject) => + reads.push({ args, resolve, reject }), + ); + } + assert.fail(`unexpected native command: ${command}`); + }; + return { client, reads, frames, deliver }; +} + +for (const [name, backgroundStart, foregroundStart, cursor, expectedReads] of [ + ["same startup second", 2_000, 2_000, 2_100, 1], + ["different subscription start floors", 2_000, 2_100, 2_200, 2], + ["older subscriptions with converged replay floors", 2_000, 2_100, 5_000, 1], +]) { + test(`real RelayClient native repair: ${name}`, async (t) => { + const f = fixture(); + let now = backgroundStart; + t.mock.method(Date, "now", () => now * 1_000); + const background = []; + const foreground = []; + // The background filter is exactly the one in useLiveChannelUpdates. + const disposeBackground = await f.client.subscribeLive( + { + kinds: [...CHANNEL_EVENT_KINDS], + "#h": ["channel-1"], + since: Math.floor(Date.now() / 1_000), + limit: 1_000, + }, + (row) => background.push(row.id), + ); + now = foregroundStart; + const disposeForeground = await f.client.subscribeToChannelLive( + "channel-1", + (row) => foreground.push(row.id), + ); + const subIds = f.frames + .filter((frame) => frame[0] === "REQ") + .map((frame) => frame[1]); + // Seed the cursor through the real inbound EVENT and EOSE dispatch. + for (const subId of subIds) { + await f.deliver(["EVENT", subId, event(1, cursor)]); + await f.deliver(["EOSE", subId]); + } + background.length = 0; + foreground.length = 0; + f.frames.length = 0; + f.client.setVisibleChannelId("channel-1"); + const running = f.client.replayLiveSubscriptions(); + await flush(); + assert.equal(f.frames.filter((frame) => frame[0] === "REQ").length, 2); + assert.equal(f.reads.length, expectedReads); + for (const read of f.reads) { + assert.deepEqual(read.args, { + channelId: "channel-1", + since: Math.max( + cursor - 1_865, + read === f.reads[0] ? backgroundStart : foregroundStart, + ), + limit: 500, + until: null, + beforeId: null, + }); + read.resolve(Object.freeze([event(2, cursor + 1)])); + } + await running; + assert.deepEqual(background, [event(2, 0).id]); + assert.deepEqual(foreground, background); + assert.equal(f.client.wsId, 7, "repair must preserve the healthy socket"); + for (const sub of f.client.subscriptions.values()) { + assert.equal(sub.pendingReplaySince, undefined); + assert.equal( + sub.reconnectReplay, + undefined, + "EOSE plus repair retires dedupe", + ); + } + await disposeForeground(); + await disposeBackground(); + assert.equal(f.client.subscriptions.size, 0); + }); +} diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 60a57f1a0b..8fce219750 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -337,6 +337,27 @@ export async function replayLiveSubscriptions({ } } + // Overlapping foreground/background subscriptions can repair the same page. + // Share only an identical outstanding read within this connection's pass: + // each subscriber still owns delivery, cursor progress, and failure recovery. + const inFlightPages = new Map>(); + const requestSharedRepair = (request: ChannelReconnectRepairRequest) => { + const key = JSON.stringify([ + request.channelId, + request.since, + request.limit, + request.until, + request.beforeId, + ]); + const existing = inFlightPages.get(key); + if (existing) return existing; + const pending = requestRepair(request).finally(() => { + inFlightPages.delete(key); + }); + inFlightPages.set(key, pending); + return pending; + }; + await runWithConcurrency( replayRequests.filter( ( @@ -381,7 +402,7 @@ export async function replayLiveSubscriptions({ // floor the superseding connection needs. isActive: () => isActive() && subscriptions.get(subId) === subscription, - requestRepair, + requestRepair: requestSharedRepair, }); // A stale-connection abort is NOT completion: the superseding // connection shares this subscription object and still needs the diff --git a/desktop/src/shared/api/relayReconnectSharing.test.mjs b/desktop/src/shared/api/relayReconnectSharing.test.mjs new file mode 100644 index 0000000000..da4bc8d718 --- /dev/null +++ b/desktop/src/shared/api/relayReconnectSharing.test.mjs @@ -0,0 +1,321 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + PAGE_REPLAY_MAX_ATTEMPTS, + RECONNECT_REPLAY_PAGE_LIMIT, + replayLiveSubscriptions, +} from "./relayReconnectReplay.ts"; +import { buildChannelFilter } from "./relayChannelFilters.ts"; +import { shouldDispatchSubscriptionEvent } from "./relayClosedRecovery.ts"; + +function deferred() { + let resolve; + let reject; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} + +async function flush() { + // Drain continuations without clock-based assertions or transport sleeps. + for (let i = 0; i < 60; i++) await Promise.resolve(); +} + +function event(id) { + return Object.freeze({ + id: String(id).padStart(64, "0"), + pubkey: "a".repeat(64), + created_at: 3_000, + kind: 9, + tags: Object.freeze([]), + content: "", + sig: "b".repeat(128), + }); +} + +function subscriber(channelId = "channel-1", since = 2_000) { + const received = []; + return { + mode: "live", + filter: { ...buildChannelFilter(channelId, 1_000), since }, + lastSeenCreatedAt: 3_100, + onEvent: (row) => received.push(row.id), + received, + }; +} + +function fixture(entries = [subscriber(), subscriber()], options = {}) { + const subscriptions = new Map(entries.map((sub, i) => [String(i), sub])); + const reads = []; + const frames = []; + const requestRepair = (request) => { + const pending = deferred(); + reads.push({ request, ...pending }); + return pending.promise; + }; + const replay = (overrides = {}) => + replayLiveSubscriptions({ + subscriptions, + sendRaw: async (frame) => { + frames.push(frame); + }, + requestRepair, + generation: 1, + ...options, + ...overrides, + }); + return { entries, subscriptions, reads, frames, replay }; +} + +test("overlapping repairs share each composite-cursor page, not subscriber delivery", async () => { + const f = fixture(); + const running = f.replay(); + await flush(); + assert.equal(f.frames.length, 2, "both live REQs must precede repair"); + assert.equal( + f.reads.length, + 1, + "one outstanding native read for equal pages", + ); + // A live frame has already reached only one subscriber. Shared history must + // dedupe for it, but still deliver that row to the other subscriber. + const first = Object.freeze( + Array.from({ length: RECONNECT_REPLAY_PAGE_LIMIT }, (_, i) => event(i + 1)), + ); + const live = first[0]; + assert.equal(shouldDispatchSubscriptionEvent(f.entries[0], live), true); + f.entries[0].onEvent(live); + f.reads[0].resolve(first); + await flush(); + assert.equal(f.reads.length, 2); + assert.deepEqual(f.reads[1].request, { + channelId: "channel-1", + since: 2_000, + limit: 500, + until: 3_000, + beforeId: first.at(-1).id, + }); + const last = event(501); + f.reads[1].resolve(Object.freeze([last])); + await running; + const expected = [...first, last].map((row) => row.id); + for (const sub of f.entries) { + assert.deepEqual(sub.received, expected); + assert.equal(sub.pendingReplaySince, undefined); + assert.equal(sub.reconnectReplay.repairDone, true); + } +}); + +for (const [label, second] of [ + ["channel", () => subscriber("channel-2")], + ["history floor", () => subscriber("channel-1", 2_100)], +]) { + test(`different ${label} requests never share`, async () => { + const f = fixture([subscriber(), second()]); + const running = f.replay(); + await flush(); + assert.equal(f.reads.length, 2); + f.reads[0].resolve([event(1)]); + f.reads[1].resolve([event(2)]); + await running; + assert.deepEqual(f.entries[0].received, [event(1).id]); + assert.deepEqual(f.entries[1].received, [event(2).id]); + }); +} + +test("completed pages are not cached for a later subscriber", async () => { + const f = fixture(undefined, { pageReplayConcurrency: 1 }); + const running = f.replay(); + await flush(); + f.reads[0].resolve([event(1)]); + await flush(); + assert.equal(f.reads.length, 2); + assert.deepEqual(f.reads[1].request, f.reads[0].request); + f.reads[1].resolve([event(2)]); + await running; + assert.deepEqual(f.entries[0].received, [event(1).id]); + assert.deepEqual(f.entries[1].received, [event(2).id]); +}); + +test("a rejected shared page is evicted and both subscribers can retry", async (t) => { + t.mock.method(console, "warn", () => {}); + const f = fixture(); + const running = f.replay(); + await flush(); + assert.equal(f.reads.length, 1); + f.reads[0].reject(new Error("temporary native read failure")); + await flush(); + assert.equal(f.reads.length, 2, "retry must not reuse a rejected promise"); + assert.deepEqual(f.reads[0].request, f.reads[1].request); + f.reads[1].resolve([event(1)]); + await running; + for (const sub of f.entries) { + assert.deepEqual(sub.received, [event(1).id]); + assert.equal(sub.pendingReplaySince, undefined); + } +}); + +test("exhausted shared reads preserve each floor without rejecting the replay", async (t) => { + t.mock.method(console, "warn", () => {}); + const f = fixture(); + const running = f.replay(); + for (let attempt = 0; attempt < PAGE_REPLAY_MAX_ATTEMPTS; attempt++) { + await flush(); + assert.equal(f.reads.length, attempt + 1); + f.reads[attempt].reject(new Error("unavailable")); + } + await running; + for (const sub of f.entries) { + assert.deepEqual(sub.received, []); + assert.equal(sub.pendingReplaySince, 2_000); + } + const retry = f.replay({ generation: 2 }); + await flush(); + assert.equal(f.reads.length, PAGE_REPLAY_MAX_ATTEMPTS + 1); + f.reads.at(-1).resolve([event(1)]); + await retry; + for (const sub of f.entries) assert.equal(sub.pendingReplaySince, undefined); +}); + +test("synchronous request failures retain the existing bounded retry contract", async (t) => { + t.mock.method(console, "warn", () => {}); + const f = fixture(); + let attempts = 0; + await f.replay({ + requestRepair: () => { + attempts++; + throw new Error("synchronous boundary failure"); + }, + }); + assert.equal(attempts, 2 * PAGE_REPLAY_MAX_ATTEMPTS); + for (const sub of f.entries) assert.equal(sub.pendingReplaySince, 2_000); +}); + +for (const action of ["dispose", "replace"]) { + test(`${action} during a shared read leaves the other subscriber intact`, async () => { + const f = fixture(); + const running = f.replay(); + await flush(); + assert.equal(f.reads.length, 1); + const replacement = subscriber(); + if (action === "dispose") f.subscriptions.delete("0"); + else f.subscriptions.set("0", replacement); + f.reads[0].resolve([event(1)]); + await running; + assert.deepEqual(f.entries[0].received, []); + assert.deepEqual(replacement.received, []); + assert.deepEqual(f.entries[1].received, [event(1).id]); + assert.equal(f.entries[1].pendingReplaySince, undefined); + }); +} + +test("superseding reconnect does not share stale reads or release its retry floor", async () => { + let generation = 1; + const f = fixture(); + const stale = f.replay({ isActive: () => generation === 1 }); + await flush(); + assert.equal(f.reads.length, 1); + generation = 2; + const current = f.replay({ generation, isActive: () => generation === 2 }); + await flush(); + assert.equal(f.reads.length, 2, "new connection needs a new native read"); + f.reads[0].resolve([event(1)]); + await stale; + for (const sub of f.entries) { + assert.deepEqual(sub.received, []); + assert.equal(sub.pendingReplaySince, 2_000); + assert.equal(sub.reconnectReplay.generation, 2); + assert.equal(sub.reconnectReplay.repairDone, false); + } + f.reads[1].resolve([event(2)]); + await current; + for (const sub of f.entries) { + assert.deepEqual(sub.received, [event(2).id]); + assert.equal(sub.pendingReplaySince, undefined); + } +}); + +test("concurrent replay owners never share pages across sessions", async () => { + const left = fixture(); + const right = fixture(); + const leftRun = left.replay(); + const rightRun = right.replay(); + await flush(); + assert.equal(left.reads.length, 1); + assert.equal(right.reads.length, 1); + left.reads[0].resolve([event(1)]); + right.reads[0].resolve([event(2)]); + await Promise.all([leftRun, rightRun]); + for (const sub of left.entries) assert.deepEqual(sub.received, [event(1).id]); + for (const sub of right.entries) + assert.deepEqual(sub.received, [event(2).id]); +}); + +for (const boundary of ["timestamp", "event id"]) { + test(`in-flight pages with a different ${boundary} cursor cannot share`, async () => { + // The third worker starts after the first reader has paged: its page one + // may see a newer snapshot, so equal floors do not imply equal cursors. + const f = fixture([subscriber(), subscriber("blocker"), subscriber()], { + pageReplayConcurrency: 2, + }); + const running = f.replay(); + await flush(); + const page = Array.from({ length: 500 }, (_, i) => event(i + 1)); + f.reads[0].resolve(page); + await flush(); + assert.equal(f.reads.length, 3); + f.reads[1].resolve([]); + await flush(); + assert.equal( + f.reads.length, + 4, + "initial and cursor pages must be separate", + ); + const alternate = [...page]; + alternate[499] = + boundary === "timestamp" + ? { ...page[499], created_at: 2_999 } + : event(501); + f.reads[3].resolve(alternate); + await flush(); + assert.equal( + f.reads.length, + 5, + "different composite cursors must not share", + ); + assert.equal(f.reads[2].request.since, f.reads[4].request.since); + if (boundary === "timestamp") { + assert.equal(f.reads[2].request.beforeId, f.reads[4].request.beforeId); + assert.notEqual(f.reads[2].request.until, f.reads[4].request.until); + } else { + assert.equal(f.reads[2].request.until, f.reads[4].request.until); + assert.notEqual(f.reads[2].request.beforeId, f.reads[4].request.beforeId); + } + f.reads[2].resolve([event(600)]); + f.reads[4].resolve([event(700)]); + await running; + assert.equal(f.entries[0].received.at(-1), event(600).id); + assert.equal(f.entries[2].received.at(-1), event(700).id); + }); +} + +test("one subscriber callback failure does not poison another's shared page", async (t) => { + t.mock.method(console, "warn", () => {}); + const broken = subscriber(); + broken.onEvent = () => { + throw new Error("consumer failure"); + }; + const f = fixture([broken, subscriber()]); + const running = f.replay(); + await flush(); + f.reads[0].resolve([event(1)]); + await flush(); + assert.equal(f.reads.length, 2, "failed consumer owns its own retry"); + assert.deepEqual(f.entries[1].received, [event(1).id]); + assert.equal(f.entries[1].pendingReplaySince, undefined); + f.reads[1].resolve([event(1)]); + await running; + assert.deepEqual(f.entries[1].received, [event(1).id]); +}); diff --git a/desktop/tests/e2e/relay-reconnect.spec.ts b/desktop/tests/e2e/relay-reconnect.spec.ts index 7868e577da..f3bf33f84c 100644 --- a/desktop/tests/e2e/relay-reconnect.spec.ts +++ b/desktop/tests/e2e/relay-reconnect.spec.ts @@ -708,3 +708,121 @@ test("reconnect backfills more missed channel messages than the live subscriptio ) .toBe(true); }); + +test("reconnect shares overlapping native repair pages and recovers offline messages", async ({ + page, +}) => { + // Same initial second gives foreground/background identical history floors. + // Different-floor behavior is covered at the RelayClient native-command seam. + const now = new Date("2026-09-20T17:00:00Z"); + await page.clock.setFixedTime(now); + await page.goto("/"); + await page.getByTestId("channel-general").click(); + await expect(page.getByTestId("chat-title")).toHaveText("general"); + await emitMockMessages(page, [ + { content: "sharing cursor seed", createdAt: now.getTime() / 1_000 }, + ]); + await expect(page.getByTestId("message-timeline")).toContainText( + "sharing cursor seed", + ); + + // Observe the real renderer -> native boundary; only transport is modeled. + await page.evaluate(() => { + const internals = ( + window as unknown as { + __TAURI_INTERNALS__: { + invoke: ( + command: string, + args?: unknown, + options?: unknown, + ) => Promise; + }; + } + ).__TAURI_INTERNALS__; + const original = internals.invoke.bind(internals); + const reads: unknown[] = []; + const returned: string[] = []; + (window as unknown as { repairReads: unknown[] }).repairReads = reads; + (window as unknown as { repairedIds: string[] }).repairedIds = returned; + internals.invoke = async (command, args, options) => { + if (command === "get_channel_reconnect_repair") { + reads.push(args); + await new Promise((resolve) => setTimeout(resolve, 20)); + const rows = (await original(command, args, options)) as Array<{ + id: string; + }>; + returned.push(...rows.map((row) => row.id)); + return rows; + } + return original(command, args, options); + }; + }); + await setMockWebsocketUnavailable(page, true); + await disconnectMockWebsockets(page); + const offline = Array.from({ length: 520 }, (_, index) => ({ + content: `offline shared repair ${index}`, + createdAt: now.getTime() / 1_000 + 1, + })); + await emitMockMessages(page, offline); + await emitMockMessages(page, [ + { + content: "newest offline shared repair", + createdAt: now.getTime() / 1_000 + 2, + }, + ]); + await setMockWebsocketUnavailable(page, false); + await expect + .poll(() => + page.evaluate(() => window.__BUZZ_E2E_GET_RELAY_CONNECTION_STATE__?.()), + ) + .toBe("connected"); + await expect + .poll(() => + page.evaluate( + () => + (window as unknown as { repairReads: unknown[] }).repairReads.length, + ), + ) + .toBe(2); + await expect(page.getByTestId("message-timeline")).toContainText( + "newest offline shared repair", + ); + // Seed plus 521 offline events: no gaps or duplicates across the dense + // same-second page boundary. UI visibility alone would not prove this. + await expect + .poll(() => + page.evaluate(() => { + const ids = (window as unknown as { repairedIds: string[] }) + .repairedIds; + return [ids.length, new Set(ids).size]; + }), + ) + .toEqual([522, 522]); + const reads = await page.evaluate( + () => + ( + window as unknown as { + repairReads: Array<{ + channelId: string; + since: number; + limit: number; + until: number | null; + beforeId: string | null; + }>; + } + ).repairReads, + ); + expect(reads[0]).toMatchObject({ + since: now.getTime() / 1_000, + limit: 500, + until: null, + beforeId: null, + }); + expect(reads[1]).toMatchObject({ + channelId: reads[0].channelId, + since: reads[0].since, + limit: 500, + until: now.getTime() / 1_000 + 1, + }); + expect(reads[1].beforeId).toBeTruthy(); +}); diff --git a/perf/RECONNECT_SHARING.md b/perf/RECONNECT_SHARING.md new file mode 100644 index 0000000000..ffa22a9d2a --- /dev/null +++ b/perf/RECONNECT_SHARING.md @@ -0,0 +1,118 @@ +# Reconnect repair sharing: measured first slice + +## Scope and recommendation + +Share **only identical outstanding native repair pages within one reconnect +pass**. Keep this as a small work-reduction change, not an app-wide latency claim. +Production diff: 22 added lines, one replacement in +`desktop/src/shared/api/relayReconnectReplay.ts`. No changed timers, concurrency, +transport choice, native code, server policy, or retained response cache. + +The active channel's window subscription and the background unread subscription +can request the same fixed-kind channel repair page. The key includes channel, +lower bound, limit, upper bound, and event-id cursor. Entries disappear on +success or rejection. Every subscriber retains delivery, dedupe, pinned retry +floor, cancellation, and generation checks. Memory is scoped to outstanding +reads in this replay pass, bounded by the existing four repair workers. + +## Matched measurements + +Baseline: `ef2aa1ae38fadcc0bc22b8bf6ed96b35933146be`. +The production source hashes are: + +- Before: `930e8f214ecf0000be0c14081e0100f75f32e8517b36d1a81557d5d0574262e4` +- After: `9a4962c36f5c10c50155833f83a19b130b2c24fa73fd68482251a24214b174e1` + +Node v24.15.0, macOS arm64; five alternating before/after pairs per case, +medians below. Final measurement ran without concurrent build/test jobs. Real-clock runs of +the production replay module; **synthetic 20 ms upstream reads** with a modeled +capacity of two or four. The replay worker count stays at its production default +of four in every case. Upstream slots are a workload model, not a claim that the +native HTTP client has a two-slot semaphore. Existing live-REQ batch delays remain. + +Each channel contains 520 rows in relay order (`created_at DESC, id ASC`), +requiring two pages. Every subscriber's exact delivered ID sequence, uniqueness, +and completed repair state are asserted. One overlapping foreground subscription +adds no new channel, but it independently receives all 520 rows. + +| Channels | Overlap | Upstream slots | Repair reads | Foreground callbacks complete, ms | All repair complete, ms | Deliveries, both arms | +|---:|:---:|---:|---:|---:|---:|---:| +| 1 | yes | 2 | 4 → 2 | 42.36 → 42.90 | 42.40 → 42.94 | 1040 | +| 12 | yes | 2 | 26 → 24 | 114.54 → 93.91 | 346.57 → 324.26 | 6760 | +| 32 | yes | 2 | 66 → 64 | 267.86 → 246.92 | 918.31 → 896.38 | 17160 | +| 12 | no | 2 | 24 → 24 | 114.12 → 114.32 | 302.13 → 302.38 | 6240 | +| 12 | yes | 4 | 26 → 24 | 93.10 → 93.12 | 220.15 → 220.45 | 6760 | + +Raw samples: [RECONNECT_SHARING_RESULTS_2026_09_20.json](RECONNECT_SHARING_RESULTS_2026_09_20.json). +The earlier scratch fixture used the wrong event-id tie direction; this report +and the checked-in harness use the relay's actual order and supersede that run. + +**Interpretation:** 26 → 24 reads is a 7.7% work reduction for the 12-channel +fixture. Roughly one modeled 20 ms service interval is saved when two upstream +slots are contended. With four upstream slots or just one channel, timing is +essentially unchanged. The negative control (no overlap) saves no requests. +These results do not establish native app paint latency, a deployed speedup, +or how often the exact-page overlap occurs in real sessions. + +## Why exact matching matters + +Production callers are `useLiveChannelUpdates.ts` (background) and +`relayClientSession.ts::subscribeToChannelLive` via `messages/hooks.ts` +(foreground). Both constrain history to their own subscription-start second. +If those floors differ, this optimization intentionally does not combine the +reads. Older subscriptions can converge once their common last-seen cursor +minus the 1,865-second replay lookback exceeds both initial floors. + +The native command is `get_channel_reconnect_repair`, whose fixed kind list +already includes channel events independent of the live filter; differing live +filter kinds are therefore not part of the native repair key. No cross-session +or cross-community map is introduced. + +## Regression evidence + +- 17 new Node tests: actual replay and actual RelayClient → native invoke seam; + dense-second pagination, per-subscriber live dedupe, different channel/floor/ + upper-bound/event-id cursor, settled-entry eviction, async failures and retry + exhaustion, synchronous throws, disposal/replacement, stale generations, + separate sessions, and one consumer callback failure. +- Exact baseline source substituted through an isolated Node module loader: + **10 tests fail / 7 controls pass**; patched source: **17 pass**. Production + files were never overwritten for the baseline check. This also falsifies the + real RelayClient wiring, not merely an exported helper. +- Desktop pre-push hook lanes passed: full desktop Node suite (**6,525/6,525**), lint/static guards, and TypeScript. +- Browser reconnect file: **17/17 passed**, including a new real UI workflow + with a fixed clock and delayed mocked native repair command. The test seeds a + cursor, disconnects, inserts 521 offline events, reconnects, and asserts two + repair reads, all 522 distinct returned IDs including the cursor seed across + the dense-second boundary, and newest-message visibility. +- Production frontend/protected-artifact build and E2E build passed. Build emits + chunk-size and mixed static/dynamic-import warnings; no bundling changes here. + +The browser uses mocked transport; native Rust, a live relay, and native frame +paint were **not** measured. The existing browser history-CLOSED injection does +not intercept the newer native repair path, so its green result is not counted +as native repair-rejection evidence; the new Node failures exercise the current +native boundary/replay path instead. Full repository `just ci` was not run; +repository-wide Rust/mobile/server checks are not certified by desktop checks. + +## Reproduce + +```bash +. ./bin/activate-hermit +mkdir -p .scratch/reconnect-sharing +git show ef2aa1ae38fadcc0bc22b8bf6ed96b35933146be:desktop/src/shared/api/relayReconnectReplay.ts \ + > .scratch/reconnect-sharing/baseline.ts +cd desktop +node --import ./test-loader.mjs scripts/benchmark-reconnect-sharing.mjs \ + ../.scratch/reconnect-sharing/baseline.ts > ../.scratch/reconnect-sharing/results.json +node --import ./test-loader.mjs --experimental-strip-types --test \ + src/shared/api/relayReconnectSharing.test.mjs \ + src/shared/api/relayClientReconnectSharing.test.mjs +pnpm build:e2e +pnpm exec playwright test --project=smoke tests/e2e/relay-reconnect.spec.ts +``` + +Do not reuse another worktree's browser server or build output. No native or +relay deployment is necessary for these fixtures. A broader performance phase +should start with real session traces, not extrapolate this modeled gain into +ordinary sends, navigation, startup, or paint. diff --git a/perf/RECONNECT_SHARING_RESULTS_2026_09_20.json b/perf/RECONNECT_SHARING_RESULTS_2026_09_20.json new file mode 100644 index 0000000000..222b036884 --- /dev/null +++ b/perf/RECONNECT_SHARING_RESULTS_2026_09_20.json @@ -0,0 +1,462 @@ +[ + { + "channels": 1, + "overlap": true, + "upstreamSlots": 2, + "modeledReadMs": 20, + "samples": { + "before": [ + { + "foregroundCompleteMs": 42.357208000000014, + "totalMs": 42.400458000000015, + "reads": 4, + "delivered": 1040, + "reqs": 2, + "peak": 2 + }, + { + "foregroundCompleteMs": 42.38529100000005, + "totalMs": 42.42029100000002, + "reads": 4, + "delivered": 1040, + "reqs": 2, + "peak": 2 + }, + { + "foregroundCompleteMs": 41.409249999999986, + "totalMs": 41.51608299999998, + "reads": 4, + "delivered": 1040, + "reqs": 2, + "peak": 2 + }, + { + "foregroundCompleteMs": 41.797582999999975, + "totalMs": 41.847916, + "reads": 4, + "delivered": 1040, + "reqs": 2, + "peak": 2 + }, + { + "foregroundCompleteMs": 43.14429200000001, + "totalMs": 43.23145800000003, + "reads": 4, + "delivered": 1040, + "reqs": 2, + "peak": 2 + } + ], + "after": [ + { + "foregroundCompleteMs": 41.27362499999998, + "totalMs": 41.291709, + "reads": 2, + "delivered": 1040, + "reqs": 2, + "peak": 1 + }, + { + "foregroundCompleteMs": 42.89704199999997, + "totalMs": 42.90854199999998, + "reads": 2, + "delivered": 1040, + "reqs": 2, + "peak": 1 + }, + { + "foregroundCompleteMs": 43.530917000000045, + "totalMs": 43.61708300000004, + "reads": 2, + "delivered": 1040, + "reqs": 2, + "peak": 1 + }, + { + "foregroundCompleteMs": 43.371916999999996, + "totalMs": 43.404832999999996, + "reads": 2, + "delivered": 1040, + "reqs": 2, + "peak": 1 + }, + { + "foregroundCompleteMs": 42.88708299999996, + "totalMs": 42.93799999999999, + "reads": 2, + "delivered": 1040, + "reqs": 2, + "peak": 1 + } + ] + } + }, + { + "channels": 12, + "overlap": true, + "upstreamSlots": 2, + "modeledReadMs": 20, + "samples": { + "before": [ + { + "foregroundCompleteMs": 114.93866600000001, + "totalMs": 349.415666, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.54520900000011, + "totalMs": 346.253375, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.54433300000005, + "totalMs": 346.96474999999987, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.11275000000023, + "totalMs": 346.56816700000036, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.49504200000001, + "totalMs": 346.0308749999999, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 2 + } + ], + "after": [ + { + "foregroundCompleteMs": 93.92358300000001, + "totalMs": 322.61412500000006, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 93.90395799999988, + "totalMs": 327.888917, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 95.18108299999994, + "totalMs": 326.14874999999984, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 93.90625, + "totalMs": 324.2624169999999, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 2 + }, + { + "foregroundCompleteMs": 93.48470799999996, + "totalMs": 323.89516600000024, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 2 + } + ] + } + }, + { + "channels": 32, + "overlap": true, + "upstreamSlots": 2, + "modeledReadMs": 20, + "samples": { + "before": [ + { + "foregroundCompleteMs": 265.68708400000014, + "totalMs": 919.4678750000003, + "reads": 66, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 268.0442499999999, + "totalMs": 921.8637500000004, + "reads": 66, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 268.6330419999995, + "totalMs": 916.8857500000004, + "reads": 66, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 267.8592090000002, + "totalMs": 914.6216669999994, + "reads": 66, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 267.5680830000001, + "totalMs": 918.3133749999997, + "reads": 66, + "delivered": 17160, + "reqs": 33, + "peak": 2 + } + ], + "after": [ + { + "foregroundCompleteMs": 247.74966700000004, + "totalMs": 897.8374999999996, + "reads": 64, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 246.9157499999992, + "totalMs": 895.1426249999995, + "reads": 64, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 246.45674999999937, + "totalMs": 898.4444579999999, + "reads": 64, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 244.8017920000002, + "totalMs": 896.3845000000001, + "reads": 64, + "delivered": 17160, + "reqs": 33, + "peak": 2 + }, + { + "foregroundCompleteMs": 248.34179200000108, + "totalMs": 895.069875000001, + "reads": 64, + "delivered": 17160, + "reqs": 33, + "peak": 2 + } + ] + } + }, + { + "channels": 12, + "overlap": false, + "upstreamSlots": 2, + "modeledReadMs": 20, + "samples": { + "before": [ + { + "foregroundCompleteMs": 114.53399999999965, + "totalMs": 304.1634159999994, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 113.72308300000077, + "totalMs": 301.1756250000017, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.11912499999926, + "totalMs": 301.3702080000003, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 112.84720900000138, + "totalMs": 302.1286250000012, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.26287500000035, + "totalMs": 303.65229200000067, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + } + ], + "after": [ + { + "foregroundCompleteMs": 114.48599999999897, + "totalMs": 302.38054199999897, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 113.953125, + "totalMs": 301.2017080000005, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.3387500000008, + "totalMs": 302.71120900000096, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 113.40599999999904, + "totalMs": 301.54983399999946, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + }, + { + "foregroundCompleteMs": 114.31720900000073, + "totalMs": 302.4698750000007, + "reads": 24, + "delivered": 6240, + "reqs": 12, + "peak": 2 + } + ] + } + }, + { + "channels": 12, + "overlap": true, + "upstreamSlots": 4, + "modeledReadMs": 20, + "samples": { + "before": [ + { + "foregroundCompleteMs": 93.09925000000112, + "totalMs": 220.59691700000076, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 93.86495800000193, + "totalMs": 220.3563749999994, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 92.94799999999668, + "totalMs": 220.02841699999772, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 93.9084589999984, + "totalMs": 220.14979200000016, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 92.23720799999865, + "totalMs": 219.22141600000032, + "reads": 26, + "delivered": 6760, + "reqs": 13, + "peak": 4 + } + ], + "after": [ + { + "foregroundCompleteMs": 92.62316699999792, + "totalMs": 220.4742499999993, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 93.06379200000083, + "totalMs": 218.34933400000227, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 93.11562500000218, + "totalMs": 220.46287500000108, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 94.55674999999974, + "totalMs": 220.4484999999986, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 4 + }, + { + "foregroundCompleteMs": 93.74179200000071, + "totalMs": 220.27625000000262, + "reads": 24, + "delivered": 6760, + "reqs": 13, + "peak": 4 + } + ] + } + } +]