perf(desktop): GUI performance sweep — async offload, poll reduction, render stabilization (#1641)

Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
Co-authored-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc <f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1cc3ha7z055mu0rwwu7806t2wt8mj3pvu0uv5mfp2c50dahaqhczshdalg6 <c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1t2tgm7d8f995uqvmnm8h88sg3wnpp9a5xysjf6dg3tjmgt3ltulqdp8ehr <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1jmc9dt2lyvzu3h0kxlwxt5zg4fxp9476awyxw6gwxn72g6cw7exqs64whm <96f056ad5f2305c8ddf637dc65d048aa4c12d7daeb8867690e34fca46b0ef64c@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>

desktop/src/shared/api/concurrency.ts

@@ -0,0 +1,20 @@
+export async function collectWithConcurrency<T, R>(
+  items: T[],
+  concurrency: number,
+  worker: (item: T) => Promise<R>,
+): Promise<R[]> {
+  const workerCount = Math.min(Math.max(1, concurrency), items.length);
+  const results = new Array<R>(items.length);
+  let nextIndex = 0;
+
+  await Promise.all(
+    Array.from({ length: workerCount }, async () => {
+      while (nextIndex < items.length) {
+        const currentIndex = nextIndex++;
+        results[currentIndex] = await worker(items[currentIndex]);
+      }
+    }),
+  );
+
+  return results;
+}

desktop/src/shared/api/relayClientSession.ts

@@ -29,6 +29,7 @@ import {
   buildChannelMentionFilter,
   buildGlobalStreamFilter,
 } from "@/shared/api/relayChannelFilters";
+import { collectWithConcurrency } from "@/shared/api/concurrency";
 import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
 import { RelayConnectionStateEmitter } from "@/shared/api/relayConnectionStateEmitter";
 import {
@@ -41,7 +42,8 @@ import { buildThreadReferenceTags } from "@/features/messages/lib/threading";
 
 const RECONNECT_BASE_DELAY_MS = 1_000,
   RECONNECT_MAX_DELAY_MS = 30_000,
-  EVENT_BATCH_MS = 16;
+  EVENT_BATCH_MS = 16,
+  AUX_BACKFILL_CONCURRENCY = 4;
 
 /**
  * Passive liveness check. The relay sends heartbeat pings every 30s; if no
@@ -217,10 +219,11 @@ export class RelayClient {
       chunks.push(eventIds.slice(i, i + AUX_BACKFILL_CHUNK_SIZE));
     }
 
-    const batches: RelayEvent[][] = [];
-    for (const ids of chunks) {
-      batches.push(await this.requestHistory(buildFilter(channelId, ids)));
-    }
+    const batches = await collectWithConcurrency(
+      chunks,
+      AUX_BACKFILL_CONCURRENCY,
+      (ids) => this.requestHistory(buildFilter(channelId, ids)),
+    );
 
     return batches.flat();
   }

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

@@ -0,0 +1,81 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { isRelayDependentQueryKey } from "./relayQueryInvalidation.ts";
+
+test("relay invalidation includes relay-backed channel and profile queries", () => {
+  for (const queryKey of [
+    ["channels"],
+    ["channels", "channel-1", "members"],
+    ["channel-messages", "channel-1"],
+    ["thread-replies", "channel-1", "root-1"],
+    ["forum-posts", "channel-1"],
+    ["home-feed"],
+    ["users-batch", "alice"],
+    ["presence", "alice"],
+    ["user-status", "alice"],
+    ["relay-agents"],
+    ["relayMembers"],
+    ["archivedIdentities"],
+    ["oaOwner", "alice"],
+  ]) {
+    assert.equal(isRelayDependentQueryKey(queryKey), true, queryKey.join("/"));
+  }
+});
+
+test("relay invalidation includes social/workflow relay queries", () => {
+  for (const queryKey of [
+    ["global-notes"],
+    ["liked-notes", "alice"],
+    ["pulse-reactions", "note-1"],
+    ["workflows", "channel-1"],
+    ["workflows-all", "channel-1"],
+    ["workflow-runs", "workflow-1"],
+    ["run-approvals", "workflow-1", "run-1"],
+    ["reminders", "alice"],
+    ["custom-emoji"],
+  ]) {
+    assert.equal(isRelayDependentQueryKey(queryKey), true, queryKey.join("/"));
+  }
+});
+
+test("relay invalidation excludes local Tauri and disk-only query roots", () => {
+  for (const queryKey of [
+    ["identity"],
+    ["managed-agents"],
+    ["personas"],
+    ["teams"],
+    ["acp-runtimes"],
+    ["backend-providers"],
+    ["managed-agent-log", "agent-1", 200],
+    ["workspace-icon", "wss://relay.example"],
+    ["agent-memory", "agent-1"],
+  ]) {
+    assert.equal(isRelayDependentQueryKey(queryKey), false, queryKey.join("/"));
+  }
+});
+
+test("relay invalidation separates relay project queries from local repo work", () => {
+  for (const queryKey of [
+    ["projects"],
+    ["project", "project-1"],
+    ["project", "project-1", "issues"],
+    ["project", "project-1", "pull-requests"],
+    ["projects", "issues", ["project-1"]],
+    ["projects", "activity-summaries", ["addr-1"]],
+  ]) {
+    assert.equal(isRelayDependentQueryKey(queryKey), true, queryKey.join("/"));
+  }
+
+  for (const queryKey of [
+    ["project", "project-1", "repo-state"],
+    ["project", "project-1", "repo-snapshot", "main"],
+    ["project", "project-1", "repo-diff", "main"],
+    ["project", "project-1", "local-repo-diff"],
+    ["project", "project-1", "commit-diff", "remote"],
+    ["projects", "local-repositories", "default"],
+    ["projects", "repo-snapshots", "default", ["project-1"]],
+  ]) {
+    assert.equal(isRelayDependentQueryKey(queryKey), false, queryKey.join("/"));
+  }
+});

desktop/src/shared/api/relayQueryInvalidation.ts

@@ -0,0 +1,84 @@
+const RELAY_QUERY_ROOTS = new Set<string>([
+  "archivedIdentities",
+  "channel-canvas",
+  "channel-messages",
+  "channels",
+  "contact-list",
+  "custom-emoji",
+  "custom-emoji-own",
+  "forum-posts",
+  "forum-thread",
+  "global-notes",
+  "home-feed",
+  "liked-notes",
+  "my-notes",
+  "myRelayMembership",
+  "oaOwner",
+  "presence",
+  "profile",
+  "pulse-note",
+  "pulse-reactions",
+  "pulse-timeline",
+  "relay-agents",
+  "relayMembers",
+  "reminders",
+  "run-approvals",
+  "search-messages",
+  "thread-replies",
+  "user-profile",
+  "user-search",
+  "user-status",
+  "users-batch",
+  "workflow",
+  "workflow-runs",
+  "workflows",
+  "workflows-all",
+]);
+
+const RELAY_PROJECT_QUERY_PARTS = new Set<string>([
+  "activity-summaries",
+  "issues",
+  "pull-requests",
+]);
+
+const LOCAL_PROJECT_QUERY_PARTS = new Set<string>([
+  "commit-diff",
+  "local-repo-diff",
+  "local-repo-snapshot",
+  "local-repositories",
+  "repo-diff",
+  "repo-snapshot",
+  "repo-state",
+  "repo-sync-status",
+]);
+
+function isRelayDependentProjectQueryKey(queryKey: readonly unknown[]) {
+  if (queryKey[0] === "projects") {
+    const scope = queryKey[1];
+    if (scope === undefined) return true;
+    if (typeof scope !== "string") return false;
+    if (LOCAL_PROJECT_QUERY_PARTS.has(scope)) return false;
+    return RELAY_PROJECT_QUERY_PARTS.has(scope);
+  }
+
+  if (queryKey[0] === "project") {
+    const scope = queryKey[2];
+    if (scope === undefined) return true;
+    if (typeof scope !== "string") return false;
+    if (LOCAL_PROJECT_QUERY_PARTS.has(scope)) return false;
+    return RELAY_PROJECT_QUERY_PARTS.has(scope);
+  }
+
+  return false;
+}
+
+export function isRelayDependentQueryKey(queryKey: readonly unknown[]) {
+  const root = queryKey[0];
+  if (typeof root !== "string") return false;
+  if (RELAY_QUERY_ROOTS.has(root)) return true;
+  return isRelayDependentProjectQueryKey(queryKey);
+}
+
+export function isRelayDependentQuery(query: { queryKey: readonly unknown[] }) {
+  return isRelayDependentQueryKey(query.queryKey);
+}

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

@@ -172,3 +172,81 @@ test("channel reconnect replay pages the missed window until a short page", asyn
   ]);
   assert.equal(delivered.length, 1008);
 });
+
+test("reconnect replay starts live REQs in parallel and preserves per-sub page order", async () => {
+  const sentPayloads = [];
+  const sendResolvers = [];
+  const historyFiltersByChannel = {
+    "channel-1": [],
+    "channel-2": [],
+  };
+  const pagesByChannel = {
+    "channel-1": [
+      eventRange("c1-full", 1501, 500),
+      eventRange("c1-short", 1490, 2),
+    ],
+    "channel-2": [
+      eventRange("c2-full", 1701, 500),
+      eventRange("c2-short", 1690, 2),
+    ],
+  };
+  const subscriptions = new Map([
+    [
+      "live-1",
+      {
+        mode: "live",
+        filter: buildChannelFilter("channel-1", 50),
+        onEvent: () => {},
+        lastSeenCreatedAt: 1000,
+      },
+    ],
+    [
+      "live-2",
+      {
+        mode: "live",
+        filter: buildChannelFilter("channel-2", 50),
+        onEvent: () => {},
+        lastSeenCreatedAt: 1000,
+      },
+    ],
+  ]);
+
+  const replayPromise = replayLiveSubscriptions({
+    subscriptions,
+    now: 2000,
+    pageReplayConcurrency: 2,
+    sendRaw: (payload) => {
+      sentPayloads.push(payload);
+      return new Promise((resolve) => {
+        sendResolvers.push(resolve);
+      });
+    },
+    requestHistory: async (filter) => {
+      const channelId = filter["#h"]?.[0];
+      historyFiltersByChannel[channelId].push(filter.until);
+      return pagesByChannel[channelId].shift() ?? [];
+    },
+  });
+
+  await Promise.resolve();
+
+  assert.deepEqual(
+    sentPayloads.map((payload) => payload[1]),
+    ["live-1", "live-2"],
+  );
+  assert.equal(sendResolvers.length, 2);
+  assert.deepEqual(historyFiltersByChannel, {
+    "channel-1": [],
+    "channel-2": [],
+  });
+
+  for (const resolve of sendResolvers) {
+    resolve();
+  }
+  await replayPromise;
+
+  assert.deepEqual(historyFiltersByChannel, {
+    "channel-1": [2000, 1501],
+    "channel-2": [2000, 1701],
+  });
+});

desktop/src/shared/api/relayReconnectReplay.ts

@@ -7,6 +7,25 @@ import type { RelayEvent } from "@/shared/api/types";
 
 const RECONNECT_REPLAY_SKEW_SECS = 5;
 export const RECONNECT_REPLAY_PAGE_LIMIT = 500;
+export const RECONNECT_REPLAY_PAGE_CONCURRENCY = 4;
+
+async function runWithConcurrency<T>(
+  items: T[],
+  concurrency: number,
+  worker: (item: T) => Promise<void>,
+) {
+  const workerCount = Math.min(Math.max(1, concurrency), items.length);
+  let nextIndex = 0;
+
+  await Promise.all(
+    Array.from({ length: workerCount }, async () => {
+      while (nextIndex < items.length) {
+        const item = items[nextIndex++];
+        await worker(item);
+      }
+    }),
+  );
+}
 
 export function buildReconnectReplayFilter(
   filter: RelaySubscriptionFilter,
@@ -84,34 +103,60 @@ export async function replayLiveSubscriptions({
   sendRaw,
   requestHistory,
   now = Math.floor(Date.now() / 1_000),
+  pageReplayConcurrency = RECONNECT_REPLAY_PAGE_CONCURRENCY,
 }: {
   subscriptions: Map<string, RelaySubscription>;
   sendRaw: (payload: unknown[]) => Promise<void>;
   requestHistory: (filter: RelaySubscriptionFilter) => Promise<RelayEvent[]>;
   now?: number;
+  pageReplayConcurrency?: number;
 }) {
-  for (const [subId, subscription] of subscriptions) {
-    if (subscription.mode !== "live") continue;
+  const replayRequests = Array.from(subscriptions.entries())
+    .filter(
+      (
+        entry,
+      ): entry is [string, Extract<RelaySubscription, { mode: "live" }>] =>
+        entry[1].mode === "live",
+    )
+    .map(([subId, subscription]) => {
+      const replaySince =
+        subscription.lastSeenCreatedAt === undefined
+          ? undefined
+          : Math.max(
+              0,
+              subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS,
+            );
+      const shouldPageReplay =
+        replaySince !== undefined &&
+        shouldPageReconnectReplay(subscription.filter);
 
-    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),
-    ]);
+      return { subId, subscription, replaySince, shouldPageReplay };
+    });
 
-    if (shouldPageReplay) {
+  await Promise.all(
+    replayRequests.map(
+      ({ subId, subscription, replaySince, shouldPageReplay }) =>
+        sendRaw([
+          "REQ",
+          subId,
+          shouldPageReplay
+            ? subscription.filter
+            : buildReconnectReplayFilter(subscription.filter, replaySince),
+        ]),
+    ),
+  );
+
+  await runWithConcurrency(
+    replayRequests.filter(
+      (
+        request,
+      ): request is typeof request & {
+        replaySince: number;
+        shouldPageReplay: true;
+      } => request.shouldPageReplay && request.replaySince !== undefined,
+    ),
+    pageReplayConcurrency,
+    async ({ subId, subscription, replaySince }) => {
       await replayReconnectHistoryPages({
         subscription,
         since: replaySince,
@@ -119,6 +164,6 @@ export async function replayLiveSubscriptions({
         isActive: () => subscriptions.get(subId) === subscription,
         requestHistory,
       });
-    }
-  }
+    },
+  );
 }

desktop/src/shared/api/tauriWorkflows.ts

@@ -169,6 +169,22 @@ export async function getChannelWorkflows(
   return raw.map(fromRawWorkflow);
 }
 
+/**
+ * Fetch workflows across many channels in a single relay round-trip.
+ *
+ * Replaces the per-channel `Promise.all(getChannelWorkflows)` fanout on the
+ * Workflows overview: the backend `#h` filter matches any listed channel, and
+ * each returned workflow carries its own `channelId` so callers can group.
+ */
+export async function getChannelsWorkflows(
+  channelIds: string[],
+): Promise<Workflow[]> {
+  const raw = await invokeTauri<RawWorkflow[]>("get_channels_workflows", {
+    channelIds,
+  });
+  return raw.map(fromRawWorkflow);
+}
+
 export async function getWorkflow(workflowId: string): Promise<Workflow> {
   const raw = await invokeTauri<RawWorkflow>("get_workflow", { workflowId });
   return fromRawWorkflow(raw);

desktop/src/shared/api/useReconnectRelay.ts

@@ -16,6 +16,7 @@ import { invoke } from "@tauri-apps/api/core";
 import { toast } from "sonner";
 
 import { relayClient } from "@/shared/api/relayClient";
+import { isRelayDependentQuery } from "@/shared/api/relayQueryInvalidation";
 import { relayReconnectController } from "@/shared/api/relayReconnectController";
 
 function buildDeps(onSuccess: () => void, onBackstop: () => void) {
@@ -59,12 +60,14 @@ export function useReconnectRelay(): {
   onSuccessRef.current = React.useCallback(() => {
     // Defer query invalidation so callers render the recovered state first.
     window.setTimeout(() => {
-      void queryClient.invalidateQueries().catch((err) => {
-        console.error(
-          "[useReconnectRelay] failed to refresh queries after reconnect:",
-          err,
-        );
-      });
+      void queryClient
+        .invalidateQueries({ predicate: isRelayDependentQuery })
+        .catch((err) => {
+          console.error(
+            "[useReconnectRelay] failed to refresh queries after reconnect:",
+            err,
+          );
+        });
     }, 0);
   }, [queryClient]);
 

desktop/src/shared/api/useRelayAutoHeal.ts

@@ -3,6 +3,7 @@ import * as React from "react";
 import { useQueryClient } from "@tanstack/react-query";
 
 import type { ConnectionState } from "@/shared/api/relayClientShared";
+import { isRelayDependentQuery } from "@/shared/api/relayQueryInvalidation";
 import {
   isRelayConnectionDegraded,
   useRelayConnection,
@@ -82,12 +83,12 @@ export class RelayAutoHealScheduler {
 
 /**
  * Auto-heal: when the connection recovers from a degraded state, invalidate
- * all queries so errored queries (e.g. messages, which don't poll) refetch
- * automatically without requiring a manual reconnect action.
+ * relay-dependent queries so errored queries (e.g. messages, which don't poll)
+ * refetch automatically without requiring a manual reconnect action.
  *
  * Rate-limited to prevent a flappy connection (e.g. VPN toggling) from
- * firing an unfiltered invalidation — ~20-40 requests across active queries
- * with retry:1 — every time the relay briefly recovers.
+ * firing a relay-wide invalidation across active queries with retry:1 every
+ * time the relay briefly recovers.
  *
  * When a recovery is suppressed by the rate limiter (an earlier flap consumed
  * the budget), a deferred heal is scheduled for the remaining window so the
@@ -101,7 +102,10 @@ export function useRelayAutoHeal(): void {
 
   if (schedulerRef.current === null) {
     schedulerRef.current = new RelayAutoHealScheduler(
-      () => void queryClient.invalidateQueries(),
+      () =>
+        void queryClient.invalidateQueries({
+          predicate: isRelayDependentQuery,
+        }),
       AUTO_HEAL_MIN_INTERVAL_MS,
       window.setTimeout.bind(window),
       window.clearTimeout.bind(window),