diff --git a/apps/web/src/components/Sidebar.motion.test.ts b/apps/web/src/components/Sidebar.motion.test.ts index f8570553058a..95ae6422e2b8 100644 --- a/apps/web/src/components/Sidebar.motion.test.ts +++ b/apps/web/src/components/Sidebar.motion.test.ts @@ -321,6 +321,31 @@ describe("sidebar list motion", () => { expect(clone.animations[0]!.cancel).toHaveBeenCalledOnce(); }); + 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); + layout([]); + 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", () => { 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..ac4deb45a0a0 100644 --- a/apps/web/src/components/Sidebar.motion.ts +++ b/apps/web/src/components/Sidebar.motion.ts @@ -1,4 +1,10 @@ const motionTiming = { duration: 150, easing: "ease-out" }; +// 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 }; @@ -104,7 +110,20 @@ export function createSidebarListMotion(parent: HTMLUListElement) { }, ]), ); - const shouldAnimate = animate && positions !== null && !reducedMotion?.matches; + 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 63c7deca4c5e..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 diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 1c0d838026fb..0d933c39f8ba 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,94 @@ describe("environment shell synchronization", () => { }), ); + 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.bounded(bufferSize); + 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, + ); + + // 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 === 50, + ), + Stream.runCollect, + Effect.forkScoped({ startImmediately: true }), + ); + yield* Queue.offerAll( + events, + 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); + 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), + ); + 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..95d90f9b36f2 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -135,47 +135,54 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ), ); - const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( - item: OrchestrationShellStreamItem, + // 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, ) { - 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 +227,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 +257,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) => {