From 3e348b96b99edcf41388799557274113449baa3b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:17:37 -0500 Subject: [PATCH] perf(ui): cut live-updates churn that inflates tab memory (#9624) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Its web UI keeps live views (issue threads, run transcripts, dashboard) fresh via React Query polling plus a live-events websocket, coordinated across tabs with a `BroadcastChannel` layer > - Long-lived tabs viewing live agent runs grew to multi-GB memory footprints while their JS heap stayed ~60–200 MB — so the memory is off-heap (Blink/native + committed allocator arenas), not a classic JS leak > - Live profiling of a reproduced 2.7 GB / 66 MB tab showed 15–30% idle CPU, ~3 fetches/sec across overlapping poll loops, and ~8 `setInterval` create/clear cycles per second whose rate grew ~7× as the tab aged — relentless allocation churn that inflates committed memory the OS never reclaims, amplified across tabs by the cross-tab fan-out > - This pull request cuts that churn at its four largest sources (invalidation storm, per-instance 1 s timers, redundant polling, unbounded streamed-run set) > - The benefit is that idle tabs do far less periodic work, so their off-heap footprint stops ballooning over a long session ## Linked Issues or Issue Description No public GitHub issue exists; describing inline per CONTRIBUTING.md → "Link Issues or Describe Them In-PR", following the bug report template. **What happened?** Browser tabs viewing live agent runs grew to 8–16 GB memory footprint over a long session (multiple tabs open), while each tab's "live" JS heap stayed only ~150–250 MB. Every idle tab also burned 15–30% CPU. Tabs eventually approached the ~4 GB V8 heap ceiling / OS pressure and could crash. **Expected behavior** Tabs viewing live runs should hold a bounded footprint and do minimal work while idle, regardless of how long they stay open or how many tabs are open. **Steps to reproduce** Open several issue/run tabs that have agents actively streaming and leave them open for a while. Watch Chrome's Task Manager: Memory Footprint climbs into the GBs while "JavaScript Memory" stays small, and CPU stays high on idle tabs. Reproduced in ~90 minutes: a tab reached 2.7 GB footprint on a 66 MB JS heap, and the per-second timer-churn rate was ~7× higher on a 90-minute-old tab than a fresh one. **Paperclip version or commit** Branch `fix/live-updates-churn`, off `master`. **Deployment mode** Local dev (`pnpm dev`), web UI. Not adapter-specific — core UI live-updates plumbing (observed with `claude_local` / `codex_local` runs). ## What Changed - **`ui/src/lib/query-invalidation-batcher.ts` (new)** — `createInvalidationBatcher` throttles + de-dupes React Query invalidations into one flush per ~300 ms, and `createCoalescingQueryClient` wraps the client via a `Proxy` so only `invalidateQueries` is batched (optimistic `setQueryData` writes stay immediate). Wired into `LiveUpdatesProvider`, which previously invalidated synchronously on every websocket event. - **`ui/src/hooks/useSecondTick.ts` (new)** — one shared, ref-counted, page-wide 1 s ticker. `useLiveElapsed` in `IssueChatThread` now uses it instead of a per-instance `setInterval` that forced a full-thread re-render every second per live element. - **`ui/src/components/transcript/useLiveRunTranscripts.ts`** — when the realtime websocket is enabled, the recurring log poll backs off to a 30 s safety-net cadence instead of polling every 2 s on top of the live stream. Added a marker for the durable poll→push rearchitecture. - **`ui/src/lib/issueChatTranscriptRuns.ts`** — `resolveIssueChatTranscriptRuns` now caps the streamed run set (live/active runs always kept; most-recent linked runs fill up to 20) so a large run history can't open a live-transcript poll per historical run. - **`ui/src/main.tsx`** — explicit `gcTime` so cross-tab-published cache entries for unobserved resources are collected promptly. - Tests for the batcher, shared ticker, and run cap. ## Verification - `vitest`: new suites `query-invalidation-batcher.test.ts` (batcher collapses 20 invalidations → 1 flush; keeps distinct keys/variants; dispose cancels; proxy passes non-invalidate methods through), `useSecondTick.test.tsx` (single ref-counted timer, stops when idle), `issueChatTranscriptRuns.test.ts` (cap keeps newest + live). All pass. - Existing affected suites pass: `LiveUpdatesProvider` (23), `IssueChatThread` (), `useLiveRunTranscripts`, `AgentDetail.instructions` — 109 tests across affected files. - `tsc -b` clean. - Behavior confirmed by live profiling before the change (2.7 GB / 66 MB tab, ~8 interval churns/sec growing 7× with age). Runtime churn reduction should be re-measured against a rebuilt bundle with the same instrumentation. ## Risks Low-to-moderate; all changes reduce work rather than add features. - **Invalidation batching** delays live-driven refetches by up to ~300 ms. Optimistic `setQueryData` writes (e.g. the visible issue's new comment) remain immediate, so foreground updates still feel instant; only the safety-net refetch is throttled. Non-live invalidations (user actions, mutations) are unaffected — they use the real client. - **Poll back-off** relies on the websocket as the live source when realtime is enabled; a 30 s fallback poll still covers gaps/reconnects (both the transcript hook and `LiveUpdatesProvider` also auto-reconnect). - **Run cap (20)** means an issue with a very large run history streams live transcripts only for its live/active + 20 most-recent runs; older runs still open normally via their run pages. - Downstream test fallout (timing-sensitive tests around invalidation/polling) may need adjustment — flagged intentionally for follow-up. Durable follow-up (out of scope, marked in code): replace transcript/run polling with server push (SSE/websocket deltas) so idle tabs do no periodic work at all. ## Model Used - **Provider:** Anthropic, via the Claude Code CLI. - **Model:** Claude Opus 4.8 (`claude-opus-4-8`). - **Reasoning mode:** Extended thinking enabled. - **Capabilities used:** tool use (shell, file editing), sub-agent fan-out for codebase analysis, and the Chrome DevTools MCP to reproduce and profile the memory/CPU churn on a live instance. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work (a bug/perf fix, not planned core feature work) - [x] I have searched GitHub for duplicate or related PRs and linked them above (none found) - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have considered and documented any risks above - [ ] I have updated relevant documentation to reflect my changes (N/A — no user-facing docs; rationale documented inline) - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.8 --- ui/src/components/IssueChatThread.tsx | 11 +- .../transcript/useLiveRunTranscripts.ts | 20 ++- ui/src/context/LiveUpdatesProvider.tsx | 17 ++- ui/src/hooks/useSecondTick.test.tsx | 45 ++++++ ui/src/hooks/useSecondTick.ts | 54 +++++++ ui/src/lib/issueChatTranscriptRuns.test.ts | 59 ++++++- ui/src/lib/issueChatTranscriptRuns.ts | 31 +++- ui/src/lib/query-invalidation-batcher.test.ts | 141 +++++++++++++++++ ui/src/lib/query-invalidation-batcher.ts | 144 ++++++++++++++++++ ui/src/main.tsx | 4 + 10 files changed, 510 insertions(+), 16 deletions(-) create mode 100644 ui/src/hooks/useSecondTick.test.tsx create mode 100644 ui/src/hooks/useSecondTick.ts create mode 100644 ui/src/lib/query-invalidation-batcher.test.ts create mode 100644 ui/src/lib/query-invalidation-batcher.ts diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index aed1de750d..a23f16d90d 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -45,6 +45,7 @@ import type { } from "@paperclipai/shared"; import type { ActiveRunForIssue, LiveRunForIssue } from "../api/heartbeats"; import { useLiveRunTranscripts } from "./transcript/useLiveRunTranscripts"; +import { useSecondTick } from "../hooks/useSecondTick"; import { usePaperclipIssueRuntime, type PaperclipIssueRuntimeReassignment } from "../hooks/usePaperclipIssueRuntime"; import { useOptionalToastActions } from "../context/ToastContext"; import { copyTextToClipboard } from "../lib/clipboard"; @@ -305,12 +306,10 @@ function findCoTSegmentIndex( } function useLiveElapsed(startMs: number | null | undefined, active: boolean): string | null { - const [, rerender] = useState(0); - useEffect(() => { - if (!active || !startMs) return; - const interval = setInterval(() => rerender((n) => n + 1), 1000); - return () => clearInterval(interval); - }, [active, startMs]); + // Drive the 1s refresh from the shared page-wide ticker instead of a + // per-instance setInterval, so a thread with many live elements uses one + // timer rather than one per element. + useSecondTick(Boolean(active && startMs)); if (!active || !startMs) return null; return formatDurationWords(Date.now() - startMs); } diff --git a/ui/src/components/transcript/useLiveRunTranscripts.ts b/ui/src/components/transcript/useLiveRunTranscripts.ts index 5b8e4edb9e..1d272954ee 100644 --- a/ui/src/components/transcript/useLiveRunTranscripts.ts +++ b/ui/src/components/transcript/useLiveRunTranscripts.ts @@ -8,8 +8,16 @@ import { buildTranscript, getUIAdapter, onAdapterChange, type RunLogChunk, type import { queryKeys } from "../../lib/queryKeys"; import { buildSameOriginWebSocketUrl } from "../../lib/websocket-url"; +// TODO(perf): this whole hook polls the log/runs endpoints on an interval. The +// durable fix is server push (SSE/websocket) for transcript deltas so idle tabs +// do no periodic work at all; the constants below only reduce the churn of the +// current polling approach. const LOG_POLL_INTERVAL_MS = 2000; const LOG_READ_LIMIT_BYTES = 256_000; +// When realtime websocket updates are enabled, the frequent log poll is +// redundant with the live stream; keep only a slow safety-net poll to cover +// gaps and reconnects instead of polling every couple of seconds. +const REALTIME_FALLBACK_POLL_INTERVAL_MS = 30_000; const EMPTY_RUN_LOG_CHUNKS: RunLogChunk[] = []; export interface RunTranscriptSource { @@ -315,17 +323,23 @@ export function useLiveRunTranscripts({ void readAll(); const activeRuns = normalizedRuns.filter((run) => !isTerminalStatus(run.status)); - const interval = activeRuns.length > 0 && logPollIntervalMs > 0 + // The realtime websocket is the primary live source when enabled, so the + // recurring poll only needs to run as a slow fallback rather than doubling + // the live update work every couple of seconds. + const effectivePollMs = enableRealtimeUpdates + ? Math.max(logPollIntervalMs, REALTIME_FALLBACK_POLL_INTERVAL_MS) + : logPollIntervalMs; + const interval = activeRuns.length > 0 && effectivePollMs > 0 ? window.setInterval(() => { void Promise.all(activeRuns.map((run) => readRunLog(run))); - }, logPollIntervalMs) + }, effectivePollMs) : null; return () => { cancelled = true; if (interval !== null) window.clearInterval(interval); }; - }, [logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey]); + }, [enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey]); useEffect(() => { if (!enableRealtimeUpdates) return; diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index 19eb9ce7f7..f864e45a6c 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -1,5 +1,6 @@ -import { useEffect, useRef, type ReactNode } from "react"; +import { useEffect, useMemo, useRef, type ReactNode } from "react"; import { useQuery, useQueryClient, type InfiniteData, type QueryClient } from "@tanstack/react-query"; +import { createCoalescingQueryClient, createInvalidationBatcher } from "../lib/query-invalidation-batcher"; import type { Agent, Issue, IssueComment, LiveEvent } from "@paperclipai/shared"; import type { RunForIssue } from "../api/activity"; import type { ActiveRunForIssue, LiveRunForIssue } from "../api/heartbeats"; @@ -1197,6 +1198,16 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { agentId: null, }); + // Coalesce the per-event invalidation storm. Optimistic setQueryData writes + // still pass straight through (immediate); only invalidateQueries is batched + // and flushed at most a few times per second. + const invalidationBatcher = useMemo(() => createInvalidationBatcher(queryClient), [queryClient]); + const coalescingClient = useMemo( + () => createCoalescingQueryClient(queryClient, invalidationBatcher), + [queryClient, invalidationBatcher], + ); + useEffect(() => () => invalidationBatcher.dispose(), [invalidationBatcher]); + useEffect(() => { pathnameRef.current = location.pathname; }, [location.pathname]); @@ -1258,7 +1269,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { try { const parsed = JSON.parse(raw) as LiveEvent; - handleLiveEvent(queryClient, liveCompanyId, pathnameRef.current, parsed, pushToast, gateRef.current, { + handleLiveEvent(coalescingClient, liveCompanyId, pathnameRef.current, parsed, pushToast, gateRef.current, { userId: currentActorRef.current.userId, agentId: currentActorRef.current.agentId, }); @@ -1293,7 +1304,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { socket = null; closeSocketQuietly(activeSocket, "provider_unmount"); }; - }, [queryClient, liveCompanyId, pushToast, canConnectSocket, socketAuthKey]); + }, [coalescingClient, liveCompanyId, pushToast, canConnectSocket, socketAuthKey]); return <>{children}; } diff --git a/ui/src/hooks/useSecondTick.test.tsx b/ui/src/hooks/useSecondTick.test.tsx new file mode 100644 index 0000000000..e9daa23ead --- /dev/null +++ b/ui/src/hooks/useSecondTick.test.tsx @@ -0,0 +1,45 @@ +// @vitest-environment jsdom + +import { flushSync } from "react-dom"; +import { createRoot } from "react-dom/client"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { __secondTickInternals, useSecondTick } from "./useSecondTick"; + +function mount(active: boolean) { + const container = document.createElement("div"); + const root = createRoot(container); + function Probe() { + useSecondTick(active); + return null; + } + flushSync(() => root.render()); + return { unmount: () => flushSync(() => root.unmount()) }; +} + +describe("useSecondTick", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("shares a single timer across multiple active subscribers and stops when idle", () => { + const a = mount(true); + const b = mount(true); + + expect(__secondTickInternals.subscriberCount()).toBe(2); + expect(__secondTickInternals.isRunning()).toBe(true); + + a.unmount(); + expect(__secondTickInternals.subscriberCount()).toBe(1); + expect(__secondTickInternals.isRunning()).toBe(true); // still one subscriber + + b.unmount(); + expect(__secondTickInternals.subscriberCount()).toBe(0); + expect(__secondTickInternals.isRunning()).toBe(false); // timer stopped when idle + }); + + it("does not subscribe or start a timer when inactive", () => { + const c = mount(false); + expect(__secondTickInternals.subscriberCount()).toBe(0); + expect(__secondTickInternals.isRunning()).toBe(false); + c.unmount(); + }); +}); diff --git a/ui/src/hooks/useSecondTick.ts b/ui/src/hooks/useSecondTick.ts new file mode 100644 index 0000000000..8065b6bf9a --- /dev/null +++ b/ui/src/hooks/useSecondTick.ts @@ -0,0 +1,54 @@ +import { useEffect, useState } from "react"; + +/** + * A single, process-wide 1-second ticker shared by every subscriber. + * + * Live "elapsed time" displays (e.g. "2m ago", running-agent timers) used to + * each create their own `setInterval(..., 1000)`. On a busy issue thread that + * meant dozens of independent 1s timers, each forcing a component re-render + * every second — a major driver of steady-state CPU churn (and, compounded + * across a long-lived tab, off-heap allocation growth). This collapses all of + * them onto one interval that only runs while at least one subscriber is active. + */ +const subscribers = new Set<() => void>(); +let intervalId: ReturnType | null = null; + +function ensureRunning(): void { + if (intervalId !== null) return; + intervalId = setInterval(() => { + for (const notify of subscribers) notify(); + }, 1000); +} + +function stopIfIdle(): void { + if (intervalId !== null && subscribers.size === 0) { + clearInterval(intervalId); + intervalId = null; + } +} + +/** + * Re-render the calling component once per second while `active` is true, driven + * by the shared ticker. Returns a monotonically increasing tick count in case a + * caller wants it as a dependency. + */ +export function useSecondTick(active: boolean): number { + const [tick, setTick] = useState(0); + useEffect(() => { + if (!active) return; + const notify = () => setTick((n) => n + 1); + subscribers.add(notify); + ensureRunning(); + return () => { + subscribers.delete(notify); + stopIfIdle(); + }; + }, [active]); + return tick; +} + +/** Test-only hook to observe the shared ticker's internal state. */ +export const __secondTickInternals = { + subscriberCount: () => subscribers.size, + isRunning: () => intervalId !== null, +}; diff --git a/ui/src/lib/issueChatTranscriptRuns.test.ts b/ui/src/lib/issueChatTranscriptRuns.test.ts index 3f45465923..cc4216cef8 100644 --- a/ui/src/lib/issueChatTranscriptRuns.test.ts +++ b/ui/src/lib/issueChatTranscriptRuns.test.ts @@ -1,5 +1,20 @@ import { describe, expect, it } from "vitest"; -import { resolveIssueChatTranscriptRuns } from "./issueChatTranscriptRuns"; +import type { LiveRunForIssue } from "../api/heartbeats"; +import type { IssueChatLinkedRun } from "./issue-chat-messages"; +import { MAX_ISSUE_CHAT_TRANSCRIPT_RUNS, resolveIssueChatTranscriptRuns } from "./issueChatTranscriptRuns"; + +function linkedRun(n: number, isoDate: string): IssueChatLinkedRun { + return { + runId: `run-${n}`, + status: "succeeded", + agentId: "agent-1", + adapterType: "codex_local", + createdAt: isoDate, + startedAt: isoDate, + finishedAt: isoDate, + hasStoredOutput: true, + } as IssueChatLinkedRun; +} describe("resolveIssueChatTranscriptRuns", () => { it("uses adapterType from linked runs without requiring agent metadata", () => { @@ -27,4 +42,46 @@ describe("resolveIssueChatTranscriptRuns", () => { }, ]); }); + + it("caps linked runs to the limit, keeping the most recent by createdAt", () => { + // 30 runs, run-0 oldest … run-29 newest. + const linkedRuns = Array.from({ length: 30 }, (_, i) => + linkedRun(i, new Date(Date.UTC(2026, 0, 1, 0, i)).toISOString()), + ); + + const runs = resolveIssueChatTranscriptRuns({ linkedRuns }); + + expect(runs.length).toBe(MAX_ISSUE_CHAT_TRANSCRIPT_RUNS); + // Newest run retained, oldest dropped. + const ids = runs.map((r) => r.id); + expect(ids).toContain("run-29"); + expect(ids).not.toContain("run-0"); + }); + + it("respects a custom limit", () => { + const linkedRuns = Array.from({ length: 10 }, (_, i) => + linkedRun(i, new Date(Date.UTC(2026, 0, 1, 0, i)).toISOString()), + ); + + const runs = resolveIssueChatTranscriptRuns({ linkedRuns, limit: 3 }); + + expect(runs.length).toBe(3); + expect(runs.map((r) => r.id)).toEqual(["run-9", "run-8", "run-7"]); + }); + + it("always retains live/active runs even beyond the limit", () => { + const linkedRuns = Array.from({ length: 25 }, (_, i) => + linkedRun(i, new Date(Date.UTC(2026, 0, 1, 0, i)).toISOString()), + ); + const liveRuns = [ + { id: "live-1", status: "running", adapterType: "claude_local", logBytes: null, lastOutputBytes: 10 }, + ] as unknown as LiveRunForIssue[]; + + const runs = resolveIssueChatTranscriptRuns({ linkedRuns, liveRuns, limit: 5 }); + + const ids = runs.map((r) => r.id); + // The live run is always present; linked runs fill the remaining slots. + expect(ids).toContain("live-1"); + expect(runs.length).toBe(5); + }); }); diff --git a/ui/src/lib/issueChatTranscriptRuns.ts b/ui/src/lib/issueChatTranscriptRuns.ts index a2b5c1e754..3b67fc286a 100644 --- a/ui/src/lib/issueChatTranscriptRuns.ts +++ b/ui/src/lib/issueChatTranscriptRuns.ts @@ -2,12 +2,31 @@ import type { ActiveRunForIssue, LiveRunForIssue } from "../api/heartbeats"; import type { RunTranscriptSource } from "../components/transcript/useLiveRunTranscripts"; import type { IssueChatLinkedRun } from "./issue-chat-messages"; +/** + * Upper bound on how many runs an issue thread streams live transcripts for. + * + * Live/active runs are always included (they are the ones actually streaming); + * older linked runs are historical and only fill the remaining slots, most + * recent first. Without this cap a long-lived issue with a large run history + * would open a live-transcript poll/subscription for every run it ever had — + * multiplying the steady-state polling and re-render churn (and off-heap + * growth) that this change set is fixing. + */ +export const MAX_ISSUE_CHAT_TRANSCRIPT_RUNS = 20; + +function toTimestamp(value: Date | string | null | undefined): number { + if (!value) return 0; + const ms = value instanceof Date ? value.getTime() : new Date(value).getTime(); + return Number.isFinite(ms) ? ms : 0; +} + export function resolveIssueChatTranscriptRuns(args: { linkedRuns?: readonly IssueChatLinkedRun[]; liveRuns?: readonly LiveRunForIssue[]; activeRun?: ActiveRunForIssue | null; + limit?: number; }): RunTranscriptSource[] { - const { linkedRuns = [], liveRuns = [], activeRun = null } = args; + const { linkedRuns = [], liveRuns = [], activeRun = null, limit = MAX_ISSUE_CHAT_TRANSCRIPT_RUNS } = args; const combined = new Map(); for (const run of liveRuns) { @@ -30,8 +49,14 @@ export function resolveIssueChatTranscriptRuns(args: { }); } - for (const run of linkedRuns) { - if (combined.has(run.runId)) continue; + // Live/active runs above are always retained; fill the remaining slots with + // the most recently created linked runs so the retained set is bounded. + const remainingLinked = [...linkedRuns] + .filter((run) => !combined.has(run.runId) && run.adapterType) + .sort((a, b) => toTimestamp(b.createdAt) - toTimestamp(a.createdAt)); + + for (const run of remainingLinked) { + if (combined.size >= limit) break; const adapterType = run.adapterType; if (!adapterType) continue; combined.set(run.runId, { diff --git a/ui/src/lib/query-invalidation-batcher.test.ts b/ui/src/lib/query-invalidation-batcher.test.ts new file mode 100644 index 0000000000..12ce094ac7 --- /dev/null +++ b/ui/src/lib/query-invalidation-batcher.test.ts @@ -0,0 +1,141 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { QueryClient } from "@tanstack/react-query"; +import { + createCoalescingQueryClient, + createInvalidationBatcher, +} from "./query-invalidation-batcher"; + +function fakeClient() { + const calls: unknown[] = []; + const client = { + invalidateQueries: vi.fn((filters?: unknown) => { + calls.push(filters); + return Promise.resolve(); + }), + }; + return { client, calls }; +} + +describe("createInvalidationBatcher", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("coalesces repeated invalidations of the same key into one call per window", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + for (let i = 0; i < 20; i++) batcher.schedule({ queryKey: ["dashboard", "c1"] }); + expect(client.invalidateQueries).not.toHaveBeenCalled(); // nothing until flush + + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(1); + expect(client.invalidateQueries).toHaveBeenCalledWith({ queryKey: ["dashboard", "c1"] }); + }); + + it("keeps distinct keys and distinct refetchType variants separate", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + batcher.schedule({ queryKey: ["a"] }); + batcher.schedule({ queryKey: ["b"] }); + batcher.schedule({ queryKey: ["a"] }); // dup of first + batcher.schedule({ queryKey: ["a"], refetchType: "inactive" }); // distinct variant + + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(3); + }); + + it("never coalesces predicate-based invalidations", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + // Two distinct predicate filters must both run — they can't be proven equal. + batcher.schedule({ predicate: () => true }); + batcher.schedule({ predicate: () => false }); + + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(2); + }); + + it("schedule() resolves only after the flush has invalidated", async () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + let resolved = false; + const p = batcher.schedule({ queryKey: ["a"] }).then(() => { + resolved = true; + }); + + await Promise.resolve(); + expect(resolved).toBe(false); // not yet — window still open + expect(client.invalidateQueries).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(300); + await p; + expect(resolved).toBe(true); + expect(client.invalidateQueries).toHaveBeenCalledTimes(1); + }); + + it("starts a fresh window after flushing", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + batcher.schedule({ queryKey: ["a"] }); + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(1); + + batcher.schedule({ queryKey: ["a"] }); + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(2); + }); + + it("dispose cancels a pending flush", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + batcher.schedule({ queryKey: ["a"] }); + batcher.dispose(); + vi.advanceTimersByTime(1000); + expect(client.invalidateQueries).not.toHaveBeenCalled(); + }); + + it("flush() invalidates immediately", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300); + + batcher.schedule({ queryKey: ["a"] }); + batcher.flush(); + expect(client.invalidateQueries).toHaveBeenCalledTimes(1); + }); +}); + +describe("createCoalescingQueryClient", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("batches invalidateQueries but passes other methods straight through", () => { + const setQueryData = vi.fn(); + const getQueryData = vi.fn(() => ({ some: "data" })); + const realInvalidate = vi.fn(() => Promise.resolve()); + const real = { invalidateQueries: realInvalidate, setQueryData, getQueryData } as unknown as QueryClient; + + const batcher = createInvalidationBatcher( + { invalidateQueries: realInvalidate } as unknown as QueryClient, + 300, + ); + const proxied = createCoalescingQueryClient(real, batcher); + + // setQueryData / getQueryData pass through immediately. + proxied.setQueryData(["k"], 1); + expect(setQueryData).toHaveBeenCalledWith(["k"], 1); + expect(proxied.getQueryData(["k"])).toEqual({ some: "data" }); + + // invalidateQueries is deferred through the batcher. + void proxied.invalidateQueries({ queryKey: ["k"] }); + void proxied.invalidateQueries({ queryKey: ["k"] }); + expect(realInvalidate).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(300); + expect(realInvalidate).toHaveBeenCalledTimes(1); + }); +}); diff --git a/ui/src/lib/query-invalidation-batcher.ts b/ui/src/lib/query-invalidation-batcher.ts new file mode 100644 index 0000000000..d5967aff5c --- /dev/null +++ b/ui/src/lib/query-invalidation-batcher.ts @@ -0,0 +1,144 @@ +import type { InvalidateQueryFilters, QueryClient } from "@tanstack/react-query"; + +/** + * Coalesces React Query invalidations triggered by the live-events stream. + * + * `LiveUpdatesProvider` used to call `queryClient.invalidateQueries(...)` + * synchronously for every websocket event. During an active agent run these + * fire many times per second, and each invalidation cascades into refetches + * and re-renders — the dominant source of steady-state CPU churn (and, over a + * long-lived tab, off-heap allocation growth). + * + * The batcher collects invalidation filters over a short window, de-duplicates + * identical ones, and flushes them in a single pass at most once per interval. + * A trailing-throttle (not a pure debounce) is used deliberately: during a + * continuous event stream a pure debounce would never flush, so UI updates + * would stall; throttling guarantees the buffered invalidations flush every + * `intervalMs`. + */ +export interface InvalidationBatcher { + /** Schedule an invalidation; resolves once the batched flush has run. */ + schedule: (filters: InvalidateQueryFilters) => Promise; + /** Flush any buffered invalidations immediately (e.g. before teardown). */ + flush: () => Promise; + dispose: () => void; +} + +export const DEFAULT_INVALIDATION_INTERVAL_MS = 300; + +type Deferred = { promise: Promise; resolve: () => void }; + +function createDeferred(): Deferred { + let resolve!: () => void; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; +} + +export function createInvalidationBatcher( + queryClient: Pick, + intervalMs: number = DEFAULT_INVALIDATION_INTERVAL_MS, +): InvalidationBatcher { + // Keyed by a stable serialization of the filters so repeated invalidations of + // the same key (the common case) collapse to one entry. + const pending = new Map(); + let timer: ReturnType | null = null; + // Resolves when the currently-buffered window has been flushed. `schedule` + // hands this back so callers that await it observe the real invalidation. + let windowDeferred: Deferred | null = null; + + const flush = async (): Promise => { + if (timer !== null) { + clearTimeout(timer); + timer = null; + } + const deferred = windowDeferred; + windowDeferred = null; + if (pending.size === 0) { + deferred?.resolve(); + return; + } + const filtersList = [...pending.values()]; + pending.clear(); + try { + await Promise.all(filtersList.map((filters) => queryClient.invalidateQueries(filters))); + } finally { + deferred?.resolve(); + } + }; + + const schedule = (filters: InvalidateQueryFilters): Promise => { + pending.set(serializeFilters(filters), filters); + if (windowDeferred === null) { + windowDeferred = createDeferred(); + } + if (timer === null) { + timer = setTimeout(() => void flush(), intervalMs); + } + return windowDeferred.promise; + }; + + const dispose = () => { + if (timer !== null) { + clearTimeout(timer); + timer = null; + } + pending.clear(); + // Release any awaiters so they don't hang forever after teardown. + windowDeferred?.resolve(); + windowDeferred = null; + }; + + return { schedule, flush, dispose }; +} + +let uniqueFilterCounter = 0; + +function serializeFilters(filters: InvalidateQueryFilters): string { + // A predicate is an opaque function — two predicate-based filters can never be + // proven equal, so never coalesce them (that would silently drop one). Give + // each its own key. + if (typeof filters.predicate === "function") { + return `__predicate__:${(uniqueFilterCounter += 1)}`; + } + // queryKey drives the identity; refetchType/exact/type change the behavior, so + // keep them distinct while still collapsing exact repeats. + try { + return JSON.stringify([ + filters.queryKey ?? null, + filters.refetchType ?? null, + filters.exact ?? null, + filters.type ?? null, + ]); + } catch { + // Non-serializable filter (shouldn't happen for our query keys) — fall back + // to a unique key so it is never dropped. + return `__nonserializable__:${(uniqueFilterCounter += 1)}`; + } +} + +/** + * Wrap a QueryClient so `invalidateQueries` is routed through `batcher` while + * every other method (reads, `setQueryData`, …) passes straight through to the + * real client. Returned value is a `QueryClient` and can be used anywhere one + * is expected. Private class fields keep working because methods are bound to + * the real client via `Reflect.get(target, prop, target)`. + * + * The batched `invalidateQueries` returns a promise that resolves only after the + * flush actually runs, so callers that await it still observe completion. + */ +export function createCoalescingQueryClient( + queryClient: QueryClient, + batcher: InvalidationBatcher, +): QueryClient { + return new Proxy(queryClient, { + get(target, prop) { + if (prop === "invalidateQueries") { + return (filters?: InvalidateQueryFilters) => batcher.schedule(filters ?? {}); + } + const value = Reflect.get(target, prop, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} diff --git a/ui/src/main.tsx b/ui/src/main.tsx index b37c5a0824..39a37bd5c5 100644 --- a/ui/src/main.tsx +++ b/ui/src/main.tsx @@ -32,6 +32,10 @@ const queryClient = new QueryClient({ defaultOptions: { queries: { staleTime: 30_000, + // Explicit so cross-tab-published cache entries for resources this tab + // isn't observing get collected promptly rather than lingering. Single + // tuning point if we need to trim the cache footprint further. + gcTime: 5 * 60_000, refetchOnWindowFocus: true, }, },