From 1d2d1a583ddb8df312e0bcee1f17077d398873e6 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 6 Sep 2026 15:46:12 -0700 Subject: [PATCH 1/3] perf(web): reduce sidebar rerenders during remote updates --- .../web/src/components/Sidebar.motion.test.ts | 10 +++ apps/web/src/components/Sidebar.motion.ts | 14 +++- apps/web/src/components/Sidebar.tsx | 71 ++++++++++++++++++- 3 files changed, 91 insertions(+), 4 deletions(-) diff --git a/apps/web/src/components/Sidebar.motion.test.ts b/apps/web/src/components/Sidebar.motion.test.ts index f8570553058a..b5268ed64bda 100644 --- a/apps/web/src/components/Sidebar.motion.test.ts +++ b/apps/web/src/components/Sidebar.motion.test.ts @@ -321,6 +321,16 @@ describe("sidebar list motion", () => { expect(clone.animations[0]!.cancel).toHaveBeenCalledOnce(); }); + it("skips expensive motion artifacts for a large list change", () => { + const rows = Array.from({ length: 41 }, (_, index) => new TestRow(`row-${index}`)); + const { motion, layout, parent } = fixture(rows); + motion.update(true); + layout([]); + motion.update(true); + expect(rows.every((row) => row.clones.length === 0)).toBe(true); + expect(parent.children).toHaveLength(0); + }); + it("respects reduced motion while keeping the next baseline fresh", () => { const a = new TestRow("a"); const b = new TestRow("b"); diff --git a/apps/web/src/components/Sidebar.motion.ts b/apps/web/src/components/Sidebar.motion.ts index 065e22be1b30..07af59d4ea45 100644 --- a/apps/web/src/components/Sidebar.motion.ts +++ b/apps/web/src/components/Sidebar.motion.ts @@ -1,4 +1,9 @@ const motionTiming = { duration: 150, easing: "ease-out" }; +// A scope change can remove a large part of the list at once. Cloning every +// removed row for a fade and starting one animation per displaced row costs +// more than the transition is worth, especially while several environments +// stream shell updates. Keep motion for small, local changes only. +const MAX_ANIMATED_ROWS_PER_UPDATE = 40; type RowPosition = { top: number; left: number; width: number; height: number }; @@ -104,7 +109,14 @@ export function createSidebarListMotion(parent: HTMLUListElement) { }, ]), ); - const shouldAnimate = animate && positions !== null && !reducedMotion?.matches; + const canAnimate = animate && positions !== null && !reducedMotion?.matches; + const removedCount = canAnimate + ? [...positions!.keys()].filter((node) => !next.has(node)).length + : 0; + const movedCount = canAnimate + ? [...next].filter(([node, position]) => positions!.get(node)?.top !== position.top).length + : 0; + const shouldAnimate = canAnimate && removedCount + movedCount <= MAX_ANIMATED_ROWS_PER_UPDATE; if (!shouldAnimate) clearFades(); else { for (const [node, position] of positions!) { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 63c7deca4c5e..6f2d61eed9c9 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -887,6 +887,8 @@ const SidebarDraftBlock = memo(function SidebarDraftBlock(props: { ); }); +// dnd-kit returns a fresh bag from each sortable wrapper render. Compare its +// fields instead of the bag object so shell updates can skip unchanged rows. const SidebarThreadRow = memo(function SidebarThreadRow(props: { thread: SidebarThreadSummary; variant: "card" | "slim"; @@ -1870,7 +1872,55 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { ); -}); +}, areSidebarThreadRowPropsEqual); + +function areSidebarThreadRowPropsEqual( + previous: SidebarThreadRowProps, + next: SidebarThreadRowProps, +): boolean { + if (!areSortableThreadRowBagsEqual(previous.sortable, next.sortable)) return false; + + const previousValues = previous as Record; + const nextValues = next as Record; + for (const key in previousValues) { + if (key !== "sortable" && !Object.is(previousValues[key], nextValues[key])) return false; + } + return true; +} + +type SidebarThreadRowProps = { + readonly sortable?: SortableThreadRowBag | undefined; + readonly [key: string]: unknown; +}; + +function areSortableThreadRowBagsEqual( + previous: SortableThreadRowBag | undefined, + next: SortableThreadRowBag | undefined, +): boolean { + if (previous === next) return true; + if (previous === undefined || next === undefined) return false; + return ( + areSortableListenersEqual(previous.listeners, next.listeners) && + previous.setNodeRef === next.setNodeRef && + previous.transform === next.transform && + previous.transition === next.transition && + previous.isDragging === next.isDragging + ); +} + +function areSortableListenersEqual( + previous: SortableThreadRowBag["listeners"], + next: SortableThreadRowBag["listeners"], +): boolean { + if (previous === next) return true; + if (previous === undefined || next === undefined) return false; + const previousEntries = Object.entries(previous); + const nextValues = next as Record; + return ( + previousEntries.length === Object.keys(next).length && + previousEntries.every(([key, value]) => Object.is(value, nextValues[key])) + ); +} function latestTurnDiff( thread: SidebarThreadSummary, @@ -3181,15 +3231,30 @@ export default function Sidebar() { snoozedThreads.length, visibleSnoozedThreads, ]); + // Shell updates replace one row object at a time, which gives the derived + // list a new identity even when its order is unchanged. Motion only needs + // to measure after a structural list change, so use the item ids as the + // dependency instead of the transient array identity. + const sidebarListItemOrderKey = useMemo( + () => sidebarListItems.map(sidebarListItemId).join("\0"), + [sidebarListItems], + ); + const sidebarListItemCount = sidebarListItems.length; const listMotionPaused = dragState !== null; useLayoutEffect(() => { // Drag release clears the baseline, so its commit cannot replay the // sortable preview. Later thread actions can animate while writes settle. // Draft navigation can reveal a frozen row without changing the draft count. listMotionRef.current?.update( - !listMotionPaused && sidebarListItems.length + visibleDraftSessionCount > 0, + !listMotionPaused && sidebarListItemCount + visibleDraftSessionCount > 0, ); - }, [listMotionPaused, routeDraftIdForRows, sidebarListItems, visibleDraftSessionCount]); + }, [ + listMotionPaused, + routeDraftIdForRows, + sidebarListItemCount, + sidebarListItemOrderKey, + visibleDraftSessionCount, + ]); const handleThreadDragOver = useCallback( (event: DragOverEvent) => { const target = event.over From 230061f55a461d3c016645b1c09cf12642dc6d65 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 6 Sep 2026 16:37:21 -0700 Subject: [PATCH 2/3] perf(web): apply each shell chunk as one sidebar update Bulk actions like snoozing 50 threads reach the client as one server coalesced chunk, but the client applied each event as its own state write. Every write rendered the whole sidebar again. Fold each chunk into one state change so a bulk action costs one render. Simplify the row memo: memoize the dnd-kit bag inside the sortable wrapper instead of a custom props comparator. Restore per-update motion measurement so rows still slide when a row changes height. Gate motion on the number of fades, not displaced rows, since clones are the expensive part and a single removal near the top of a long list should still slide the rest. Co-Authored-By: Claude Fable 5.1 --- .../web/src/components/Sidebar.motion.test.ts | 17 +++- apps/web/src/components/Sidebar.motion.ts | 33 +++++--- apps/web/src/components/Sidebar.tsx | 79 +++--------------- .../src/state/shell-sync.test.ts | 83 +++++++++++++++++++ packages/client-runtime/src/state/shell.ts | 80 ++++++++++-------- 5 files changed, 173 insertions(+), 119 deletions(-) diff --git a/apps/web/src/components/Sidebar.motion.test.ts b/apps/web/src/components/Sidebar.motion.test.ts index b5268ed64bda..95ae6422e2b8 100644 --- a/apps/web/src/components/Sidebar.motion.test.ts +++ b/apps/web/src/components/Sidebar.motion.test.ts @@ -321,7 +321,7 @@ describe("sidebar list motion", () => { expect(clone.animations[0]!.cancel).toHaveBeenCalledOnce(); }); - it("skips expensive motion artifacts for a large list change", () => { + it("skips fades when a large list change would clone too many rows", () => { const rows = Array.from({ length: 41 }, (_, index) => new TestRow(`row-${index}`)); const { motion, layout, parent } = fixture(rows); motion.update(true); @@ -329,6 +329,21 @@ describe("sidebar list motion", () => { motion.update(true); expect(rows.every((row) => row.clones.length === 0)).toBe(true); expect(parent.children).toHaveLength(0); + const entering = Array.from({ length: 41 }, (_, index) => new TestRow(`new-${index}`)); + layout(entering); + motion.update(true); + expect(entering.every((row) => row.animate.mock.calls.length === 0)).toBe(true); + }); + + it("still slides rows when one removal displaces a large list", () => { + const rows = Array.from({ length: 60 }, (_, index) => new TestRow(`row-${index}`)); + const { motion, layout } = fixture(rows); + motion.update(true); + const [removed, ...rest] = rows; + layout(rest); + motion.update(true); + expect(removed!.clones).toHaveLength(1); + expect(rest.every((row) => row.animations.length === 1)).toBe(true); }); it("respects reduced motion while keeping the next baseline fresh", () => { diff --git a/apps/web/src/components/Sidebar.motion.ts b/apps/web/src/components/Sidebar.motion.ts index 07af59d4ea45..ac4deb45a0a0 100644 --- a/apps/web/src/components/Sidebar.motion.ts +++ b/apps/web/src/components/Sidebar.motion.ts @@ -1,9 +1,10 @@ const motionTiming = { duration: 150, easing: "ease-out" }; -// A scope change can remove a large part of the list at once. Cloning every -// removed row for a fade and starting one animation per displaced row costs -// more than the transition is worth, especially while several environments -// stream shell updates. Keep motion for small, local changes only. -const MAX_ANIMATED_ROWS_PER_UPDATE = 40; +// A project filter change or a bulk snooze swaps a large part of the list at +// once. Fades are the expensive part: every removed row gets a deep clone and +// every clone and entering row gets its own animation, and the layout reads +// in between force synchronous reflows. Translating displaced rows is cheap, +// so only the fade count decides whether an update animates. +const MAX_FADED_ROWS_PER_UPDATE = 40; type RowPosition = { top: number; left: number; width: number; height: number }; @@ -109,14 +110,20 @@ export function createSidebarListMotion(parent: HTMLUListElement) { }, ]), ); - const canAnimate = animate && positions !== null && !reducedMotion?.matches; - const removedCount = canAnimate - ? [...positions!.keys()].filter((node) => !next.has(node)).length - : 0; - const movedCount = canAnimate - ? [...next].filter(([node, position]) => positions!.get(node)?.top !== position.top).length - : 0; - const shouldAnimate = canAnimate && removedCount + movedCount <= MAX_ANIMATED_ROWS_PER_UPDATE; + let fadeCount = 0; + if (positions !== null) { + for (const [node, position] of positions) { + if (!next.has(node) && position.height > 0) fadeCount++; + } + for (const [node, position] of next) { + if (!positions.has(node) && position.height > 0) fadeCount++; + } + } + const shouldAnimate = + animate && + positions !== null && + !reducedMotion?.matches && + fadeCount <= MAX_FADED_ROWS_PER_UPDATE; if (!shouldAnimate) clearFades(); else { for (const [node, position] of positions!) { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 6f2d61eed9c9..4bbdd68b0e6a 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -499,7 +499,13 @@ function SortableThreadRow(props: { disabled: { draggable: props.disabled }, animateLayoutChanges: animateSidebarLayoutChanges, }); - return props.children({ listeners, setNodeRef, transform, transition, isDragging }); + // dnd-kit memoizes each field but not the bag, so the memoized row would + // rerender on every shell update without this. + const bag = useMemo( + () => ({ listeners, setNodeRef, transform, transition, isDragging }), + [listeners, setNodeRef, transform, transition, isDragging], + ); + return props.children(bag); } // Unsent work shares one look: the new-thread draft rows and thread rows @@ -887,8 +893,6 @@ const SidebarDraftBlock = memo(function SidebarDraftBlock(props: { ); }); -// dnd-kit returns a fresh bag from each sortable wrapper render. Compare its -// fields instead of the bag object so shell updates can skip unchanged rows. const SidebarThreadRow = memo(function SidebarThreadRow(props: { thread: SidebarThreadSummary; variant: "card" | "slim"; @@ -1872,55 +1876,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { ); -}, areSidebarThreadRowPropsEqual); - -function areSidebarThreadRowPropsEqual( - previous: SidebarThreadRowProps, - next: SidebarThreadRowProps, -): boolean { - if (!areSortableThreadRowBagsEqual(previous.sortable, next.sortable)) return false; - - const previousValues = previous as Record; - const nextValues = next as Record; - for (const key in previousValues) { - if (key !== "sortable" && !Object.is(previousValues[key], nextValues[key])) return false; - } - return true; -} - -type SidebarThreadRowProps = { - readonly sortable?: SortableThreadRowBag | undefined; - readonly [key: string]: unknown; -}; - -function areSortableThreadRowBagsEqual( - previous: SortableThreadRowBag | undefined, - next: SortableThreadRowBag | undefined, -): boolean { - if (previous === next) return true; - if (previous === undefined || next === undefined) return false; - return ( - areSortableListenersEqual(previous.listeners, next.listeners) && - previous.setNodeRef === next.setNodeRef && - previous.transform === next.transform && - previous.transition === next.transition && - previous.isDragging === next.isDragging - ); -} - -function areSortableListenersEqual( - previous: SortableThreadRowBag["listeners"], - next: SortableThreadRowBag["listeners"], -): boolean { - if (previous === next) return true; - if (previous === undefined || next === undefined) return false; - const previousEntries = Object.entries(previous); - const nextValues = next as Record; - return ( - previousEntries.length === Object.keys(next).length && - previousEntries.every(([key, value]) => Object.is(value, nextValues[key])) - ); -} +}); function latestTurnDiff( thread: SidebarThreadSummary, @@ -3231,30 +3187,15 @@ export default function Sidebar() { snoozedThreads.length, visibleSnoozedThreads, ]); - // Shell updates replace one row object at a time, which gives the derived - // list a new identity even when its order is unchanged. Motion only needs - // to measure after a structural list change, so use the item ids as the - // dependency instead of the transient array identity. - const sidebarListItemOrderKey = useMemo( - () => sidebarListItems.map(sidebarListItemId).join("\0"), - [sidebarListItems], - ); - const sidebarListItemCount = sidebarListItems.length; const listMotionPaused = dragState !== null; useLayoutEffect(() => { // Drag release clears the baseline, so its commit cannot replay the // sortable preview. Later thread actions can animate while writes settle. // Draft navigation can reveal a frozen row without changing the draft count. listMotionRef.current?.update( - !listMotionPaused && sidebarListItemCount + visibleDraftSessionCount > 0, + !listMotionPaused && sidebarListItems.length + visibleDraftSessionCount > 0, ); - }, [ - listMotionPaused, - routeDraftIdForRows, - sidebarListItemCount, - sidebarListItemOrderKey, - visibleDraftSessionCount, - ]); + }, [listMotionPaused, routeDraftIdForRows, sidebarListItems, visibleDraftSessionCount]); const handleThreadDragOver = useCallback( (event: DragOverEvent) => { const target = event.over diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 1c0d838026fb..994d10ca56fe 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -6,6 +6,7 @@ import { } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -151,6 +152,88 @@ describe("environment shell synchronization", () => { }), ); + it.live("applies one chunk of live events as one state change", () => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService( + ShellSnapshotLoader, + ShellSnapshotLoader.of({ load: () => Effect.succeed(Option.none()) }), + ), + ); + yield* SubscriptionRef.set(supervisorState, { + desired: true, + network: "online", + phase: "connected", + stage: null, + attempt: 1, + generation: 1, + lastFailure: null, + retryAt: null, + }); + yield* Queue.offer(events, { kind: "snapshot", snapshot: LIVE_SHELL_SNAPSHOT }); + yield* Queue.offer(events, { kind: "synchronized" }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => state.status === "live"), + Stream.runHead, + ); + + // A bulk action (snooze 30 threads) reaches the client as one server + // coalesced chunk. The shell state must change once for it, not 30 times. + const observed = yield* SubscriptionRef.changes(shellState).pipe( + Stream.drop(1), + Stream.takeUntil( + (state) => Option.isSome(state.snapshot) && state.snapshot.value.threads.length === 30, + ), + Stream.runCollect, + Effect.forkScoped, + ); + yield* Queue.offerAll( + events, + Array.from({ length: 30 }, (_, index) => ({ + kind: "thread-upserted" as const, + sequence: 2 + index, + thread: { id: `thread-${index}` } as never, + })), + ); + const states = yield* Fiber.join(observed); + expect(states).toHaveLength(1); + expect(Option.getOrThrow(states[0]!.snapshot).snapshotSequence).toBe(31); + }).pipe(Effect.scoped), + ); + it.effect("requests a full socket snapshot when the HTTP refresh fails", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationShellSnapshot = { diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 69799bbd1168..b87a35265ef5 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -135,47 +135,55 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ), ); - const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( - item: OrchestrationShellStreamItem, + // The server coalesces shell events into one chunk per window, and each + // chunk arrives here as one array. Folding the whole chunk into a single + // state write keeps one bulk action (snoozing 50 threads) at one sidebar + // render instead of one per thread. + const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( + items: ReadonlyArray, ) { - if (item.kind === "synchronized") { - yield* Ref.set(awaitingCompletion, false); - yield* SubscriptionRef.update(state, (current) => - Option.isSome(current.snapshot) - ? { ...current, status: "live" as const, error: Option.none() } - : current, - ); - return; - } - - const current = yield* SubscriptionRef.get(state); - const nextSnapshot = - item.kind === "snapshot" - ? item.snapshot - : Option.match(current.snapshot, { - onNone: () => null, - onSome: (snapshot) => - item.sequence > snapshot.snapshotSequence - ? applyShellStreamEvent(snapshot, item) - : snapshot, - }); - if (nextSnapshot === null) { - return; + const initial = yield* SubscriptionRef.get(state); + let waiting = yield* Ref.get(awaitingCompletion); + let next = initial; + let receivedSnapshot = false; + for (const item of items) { + if (item.kind === "synchronized") { + waiting = false; + if (Option.isSome(next.snapshot)) { + next = { ...next, status: "live", error: Option.none() }; + } + continue; + } + const nextSnapshot = + item.kind === "snapshot" + ? item.snapshot + : Option.match(next.snapshot, { + onNone: () => null, + onSome: (snapshot) => + item.sequence > snapshot.snapshotSequence + ? applyShellStreamEvent(snapshot, item) + : snapshot, + }); + if (nextSnapshot === null) continue; + receivedSnapshot ||= item.kind === "snapshot"; + next = { + snapshot: Option.some(nextSnapshot), + status: waiting ? "synchronizing" : "live", + error: Option.none(), + }; } - - const waiting = yield* Ref.get(awaitingCompletion); - yield* SubscriptionRef.set(state, { - snapshot: Option.some(nextSnapshot), - status: waiting ? "synchronizing" : "live", - error: Option.none(), - }); - if (item.kind === "snapshot") { + yield* Ref.set(awaitingCompletion, waiting); + if (next === initial) return; + yield* SubscriptionRef.set(state, next); + if (receivedSnapshot) { const session = yield* Ref.get(activeSubscriptionSession); if (session !== null) { yield* Ref.set(lastAuthoritativeSession, session); } } - yield* Queue.offer(persistence, nextSnapshot); + if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) { + yield* Queue.offer(persistence, next.snapshot.value); + } }); const foregroundResubscriptions = Option.match(wakeups, { @@ -220,7 +228,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const httpSnapshot = yield* snapshotLoader.load(prepared); if (Option.isSome(httpSnapshot)) { - yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + yield* applyItems([{ kind: "snapshot", snapshot: httpSnapshot.value }]); canResume = true; current = yield* SubscriptionRef.get(state); } @@ -250,7 +258,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEach(applyItem)), + ).pipe(Stream.runForEachArray(applyItems)), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { From 90cd35c3648bcb4529fcc0ec5279a26d03ff6a81 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 6 Sep 2026 17:00:32 -0700 Subject: [PATCH 3/3] test(client-runtime): cover bounded shell event batches --- .../src/state/shell-sync.test.ts | 24 ++++++++++++------- packages/client-runtime/src/state/shell.ts | 7 +++--- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 994d10ca56fe..0d933c39f8ba 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -152,9 +152,13 @@ describe("environment shell synchronization", () => { }), ); - it.live("applies one chunk of live events as one state change", () => + it.live.each([ + { bufferSize: Infinity, expectedSequences: [51] }, + // RpcClient defaults to a 16-event buffer, which splits larger server chunks. + { bufferSize: 16, expectedSequences: [17, 33, 49, 51] }, + ])("batches live events with a $bufferSize event buffer", ({ bufferSize, expectedSequences }) => Effect.gen(function* () { - const events = yield* Queue.unbounded(); + const events = yield* Queue.bounded(bufferSize); const client = { [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), } as unknown as WsRpcProtocolClient; @@ -210,27 +214,29 @@ describe("environment shell synchronization", () => { Stream.runHead, ); - // A bulk action (snooze 30 threads) reaches the client as one server - // coalesced chunk. The shell state must change once for it, not 30 times. + // Observe before publishing so no batch can arrive before the subscription. const observed = yield* SubscriptionRef.changes(shellState).pipe( Stream.drop(1), Stream.takeUntil( - (state) => Option.isSome(state.snapshot) && state.snapshot.value.threads.length === 30, + (state) => Option.isSome(state.snapshot) && state.snapshot.value.threads.length === 50, ), Stream.runCollect, - Effect.forkScoped, + Effect.forkScoped({ startImmediately: true }), ); yield* Queue.offerAll( events, - Array.from({ length: 30 }, (_, index) => ({ + Array.from({ length: 50 }, (_, index) => ({ kind: "thread-upserted" as const, sequence: 2 + index, thread: { id: `thread-${index}` } as never, })), ); const states = yield* Fiber.join(observed); - expect(states).toHaveLength(1); - expect(Option.getOrThrow(states[0]!.snapshot).snapshotSequence).toBe(31); + const snapshots = states.map((state) => Option.getOrThrow(state.snapshot)); + expect(snapshots.map((snapshot) => snapshot.snapshotSequence)).toEqual(expectedSequences); + expect(snapshots.at(-1)!.threads.map((thread) => thread.id)).toEqual( + Array.from({ length: 50 }, (_, index) => `thread-${index}`), + ); }).pipe(Effect.scoped), ); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index b87a35265ef5..95d90f9b36f2 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -135,10 +135,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ), ); - // The server coalesces shell events into one chunk per window, and each - // chunk arrives here as one array. Folding the whole chunk into a single - // state write keeps one bulk action (snoozing 50 threads) at one sidebar - // render instead of one per thread. + // Apply each received batch with one state write. The RPC client's bounded + // buffer can split a server chunk, so a bulk action can still need several + // writes, but each write includes every event in that batch. const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( items: ReadonlyArray, ) {