From 8a7d541b2cb184985aa6e30d4aa30dc130c2d32e Mon Sep 17 00:00:00 2001 From: T3 Code PR Stack <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:25:36 +0200 Subject: [PATCH 1/2] fix(vcs): list-mode status without remote pollers for row UIs Break the O(N) always-on VCS poller model for sidebar/board/thread lists: subscribe with mode=list loads local (+ one coalesced remote) but does not retain the automatic remote poller. Active git chrome keeps full mode. --- .../src/features/board/useBoardVcsStatuses.ts | 6 +-- apps/mobile/src/state/use-thread-pr.ts | 8 ++-- .../src/vcs/VcsStatusBroadcaster.test.ts | 46 +++++++++++++++++++ apps/server/src/vcs/VcsStatusBroadcaster.ts | 33 +++++++++---- apps/web/src/components/Sidebar.tsx | 4 +- apps/web/src/components/SidebarV2.tsx | 2 +- .../src/components/ThreadStatusIndicators.tsx | 2 +- .../components/board/useBoardVcsStatuses.ts | 12 ++--- packages/client-runtime/src/state/vcs.ts | 37 +++++++++++---- packages/contracts/src/git.test.ts | 15 ++++++ packages/contracts/src/git.ts | 13 ++++++ 11 files changed, 142 insertions(+), 36 deletions(-) diff --git a/apps/mobile/src/features/board/useBoardVcsStatuses.ts b/apps/mobile/src/features/board/useBoardVcsStatuses.ts index 90c985c09dc..c8460f17276 100644 --- a/apps/mobile/src/features/board/useBoardVcsStatuses.ts +++ b/apps/mobile/src/features/board/useBoardVcsStatuses.ts @@ -17,8 +17,8 @@ const EMPTY_STATUSES_ATOM = Atom.make( ).pipe(Atom.withLabel("mobile:board-vcs-statuses:empty")); /** - * Aggregated VCS status for the board — one derived atom over the per-cwd - * status family (same shape as web `useBoardVcsStatuses`). + * Aggregated list-mode VCS status for the board (no remote poller) — same + * shape as web `useBoardVcsStatuses`. */ export function useBoardVcsStatuses( targets: ReadonlyArray, @@ -55,7 +55,7 @@ export function useBoardVcsStatuses( Option.getOrNull( AsyncResult.value( get( - vcsEnvironment.status({ + vcsEnvironment.listStatus({ environmentId: target.environmentId, input: { cwd: target.cwd }, }), diff --git a/apps/mobile/src/state/use-thread-pr.ts b/apps/mobile/src/state/use-thread-pr.ts index a3440cd4848..cec860688a0 100644 --- a/apps/mobile/src/state/use-thread-pr.ts +++ b/apps/mobile/src/state/use-thread-pr.ts @@ -11,10 +11,8 @@ export { } from "./thread-pr-presentation"; /** - * Live PR status for a thread's branch. Subscriptions are deduplicated per - * (environmentId, cwd) by the atom family, so many rows on the same worktree - * or project root share one stream — and virtualization means only visible - * rows subscribe at all. + * List-mode PR status for a thread's branch (no remote poller). Deduped per + * (environmentId, cwd); use full `vcsEnvironment.status` only for active git chrome. */ export function useThreadPr( thread: EnvironmentThreadShell, @@ -23,7 +21,7 @@ export function useThreadPr( const cwd = thread.worktreePath ?? projectCwd; const gitStatus = useEnvironmentQuery( thread.branch !== null && cwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.listStatus({ environmentId: thread.environmentId, input: { cwd }, }) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index bc5d21e3b3e..4b05aac2caf 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -554,6 +554,52 @@ describe("VcsStatusBroadcaster", () => { ); }); + it.effect("list mode does not start a remote poller", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: remoteStatusWithPr, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + remoteStatusRefreshUpstreamValues: [] as Array, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const scope = yield* Scope.make(); + const snapshotDeferred = yield* Deferred.make(); + yield* Stream.runForEach( + broadcaster.streamStatus( + { cwd: "/repo", mode: "list" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.seconds(30)) }, + ), + (event) => + event._tag === "snapshot" + ? Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore) + : Effect.void, + ).pipe(Effect.forkIn(scope)); + + const snapshot = yield* Deferred.await(snapshotDeferred); + assert.deepStrictEqual(snapshot, { + _tag: "snapshot", + local: baseLocalStatus, + remote: remoteStatusWithPr, + } satisfies VcsStatusStreamEvent); + // One coalesced remote load for the initial badge fill — no force invalidate. + assert.equal(state.remoteStatusCalls, 1); + assert.equal(state.remoteInvalidationCalls, 0); + assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues, [false]); + + // Advance well past the full-mode poll interval; list mode must not re-poll. + yield* TestClock.adjust(Duration.minutes(5)); + yield* Effect.yieldNow; + assert.equal(state.remoteStatusCalls, 1); + + yield* Scope.close(scope, Exit.void); + }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); + }); + it.effect("delays automatic refresh when a cached remote snapshot is available", () => { const state = { currentLocalStatus: baseLocalStatus, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 591ba77c1c2..106d8873fc8 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -511,18 +511,35 @@ export const make = Effect.gen(function* () { Stream.unwrap( Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + const mode = input.mode ?? "full"; const subscription = yield* PubSub.subscribe(changesPubSub); const initialLocal = yield* getOrLoadLocalStatus(cwd); - const cachedStatus = yield* getCachedStatus(cwd); + let cachedStatus = yield* getCachedStatus(cwd); + + // List mode must not start a remote poller (O(N) storm root for sidebar/board). + // One coalesced remote load when uncached so PR badges can populate once; later + // updates arrive via pubsub when a full subscriber or explicit refresh runs. + if ( + mode === "list" && + (cachedStatus?.remote === null || cachedStatus?.remote === undefined) + ) { + yield* refreshRemoteStatus(cwd, { refreshUpstream: false }).pipe(Effect.ignore); + cachedStatus = yield* getCachedStatus(cwd); + } + const initialRemote = cachedStatus?.remote?.value ?? null; - yield* retainRemotePoller( - cwd, - options?.automaticRemoteRefreshInterval ?? - Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL), - cachedStatus?.remote === null || cachedStatus?.remote === undefined, - ); - const release = releaseRemotePoller(cwd).pipe(Effect.ignore, Effect.asVoid); + // Full mode only: long-lived remote poller for the active thread / git chrome. + let release: Effect.Effect = Effect.void; + if (mode === "full") { + yield* retainRemotePoller( + cwd, + options?.automaticRemoteRefreshInterval ?? + Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL), + cachedStatus?.remote === null || cachedStatus?.remote === undefined, + ); + release = releaseRemotePoller(cwd).pipe(Effect.ignore, Effect.asVoid); + } // When remote is not cached yet, emit localUpdated only — never a snapshot that // fabricates remote defaults (pr:null). Downstream clients treat that fake null diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 1bc0db276e1..a700780e149 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -524,7 +524,7 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr const gitCwd = thread.worktreePath ?? threadProjectCwd ?? props.projectCwd; const gitStatus = useEnvironmentQuery( thread.branch != null && gitCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.listStatus({ environmentId: thread.environmentId, input: { cwd: gitCwd }, }) @@ -3055,7 +3055,7 @@ const SidebarRecentThreadRow = memo(function SidebarRecentThreadRow(props: { const gitCwd = thread.worktreePath ?? project.workspaceRoot; const gitStatus = useEnvironmentQuery( (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.listStatus({ environmentId: thread.environmentId, input: { cwd: gitCwd }, }) diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index 634b5f6d676..ef1bf4be866 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -566,7 +566,7 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { const gitCwd = thread.worktreePath ?? props.projectCwd; const gitStatus = useEnvironmentQuery( (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.listStatus({ environmentId: thread.environmentId, input: { cwd: gitCwd }, }) diff --git a/apps/web/src/components/ThreadStatusIndicators.tsx b/apps/web/src/components/ThreadStatusIndicators.tsx index 3c3efbe487f..66c8a4615da 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -363,7 +363,7 @@ export function ThreadRowLeadingStatus({ thread }: { thread: SidebarThreadSummar const gitCwd = thread.worktreePath ?? threadProjectCwd; const gitStatus = useEnvironmentQuery( (thread.branch != null || thread.worktreePath !== null) && gitCwd !== null - ? vcsEnvironment.status({ + ? vcsEnvironment.listStatus({ environmentId: thread.environmentId, input: { cwd: gitCwd }, }) diff --git a/apps/web/src/components/board/useBoardVcsStatuses.ts b/apps/web/src/components/board/useBoardVcsStatuses.ts index b774880b651..3a69c4ffff0 100644 --- a/apps/web/src/components/board/useBoardVcsStatuses.ts +++ b/apps/web/src/components/board/useBoardVcsStatuses.ts @@ -17,12 +17,10 @@ const EMPTY_STATUSES_ATOM = Atom.make( ).pipe(Atom.withLabel("web:board-vcs-statuses:empty")); /** - * Aggregated VCS status subscription for the board: one derived atom over the - * per-cwd status subscription family, read with a single useAtomValue. The - * family dedupes identical (environmentId, cwd) keys into one WS subscription - * and keeps entries warm for 5 minutes after last use, so filter toggles - * don't churn subscriptions. Entries are `null` until the first snapshot - * streams in. + * Aggregated list-mode VCS status for the board: one derived atom over the + * per-cwd listStatus family (no server remote poller). Dedupe by + * (environmentId, cwd); shorter idle TTL than full status. Entries are `null` + * until the first snapshot streams in. */ export function useBoardVcsStatuses( targets: ReadonlyArray, @@ -62,7 +60,7 @@ export function useBoardVcsStatuses( Option.getOrNull( AsyncResult.value( get( - vcsEnvironment.status({ + vcsEnvironment.listStatus({ environmentId: target.environmentId, input: { cwd: target.cwd }, }), diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index ac85762a543..7acd5631628 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -272,20 +272,39 @@ export function createVcsEnvironmentAtoms( cwd: target.input.cwd, }); + const statusStream = (input: EnvironmentRpcInput) => + subscribe(WS_METHODS.subscribeVcsStatus, input).pipe( + Stream.mapAccum( + () => null as VcsStatusResult | null, + (current, event) => { + const next = applyGitStatusStreamEvent(current, event); + return [next, [next]] as const; + }, + ), + ); + return { listRefs, + /** + * Full VCS status (includes server remote poller). Use for the active thread / + * git chrome only — not for high-cardinality lists. + */ status: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:vcs:status", + subscribe: statusStream, + }), + /** + * List/badge VCS status: no remote poller on the server. Shorter idle TTL so + * off-screen rows drop streams. Prefer this for sidebar/board/thread rows. + */ + listStatus: createEnvironmentSubscriptionAtomFamily(runtime, { + label: "environment-data:vcs:status-list", + idleTtlMs: 60_000, subscribe: (input: EnvironmentRpcInput) => - subscribe(WS_METHODS.subscribeVcsStatus, input).pipe( - Stream.mapAccum( - () => null as VcsStatusResult | null, - (current, event) => { - const next = applyGitStatusStreamEvent(current, event); - return [next, [next]] as const; - }, - ), - ), + statusStream({ + ...input, + mode: "list", + }), }), pull: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:pull", diff --git a/packages/contracts/src/git.test.ts b/packages/contracts/src/git.test.ts index 0ac5c5fee2d..5741b241915 100644 --- a/packages/contracts/src/git.test.ts +++ b/packages/contracts/src/git.test.ts @@ -3,6 +3,7 @@ import * as Schema from "effect/Schema"; import { VcsCreateWorktreeInput, + VcsStatusInput, GitPreparePullRequestThreadInput, GitRunStackedActionResult, GitRunStackedActionInput, @@ -12,6 +13,7 @@ import { } from "./git.ts"; const decodeCreateWorktreeInput = Schema.decodeUnknownSync(VcsCreateWorktreeInput); +const decodeVcsStatusInput = Schema.decodeUnknownSync(VcsStatusInput); const decodePreparePullRequestThreadInput = Schema.decodeUnknownSync( GitPreparePullRequestThreadInput, ); @@ -21,6 +23,19 @@ const decodeActionProgressEvent = Schema.decodeUnknownSync(GitActionProgressEven const decodeGitCommandError = Schema.decodeUnknownSync(GitCommandError); const decodeResolvePullRequestResult = Schema.decodeUnknownSync(GitResolvePullRequestResult); +describe("VcsStatusInput", () => { + it("accepts cwd-only input as full-mode compatible", () => { + const parsed = decodeVcsStatusInput({ cwd: "/repo" }); + expect(parsed.cwd).toBe("/repo"); + expect(parsed.mode).toBeUndefined(); + }); + + it("accepts list mode for high-cardinality list subscriptions", () => { + const parsed = decodeVcsStatusInput({ cwd: "/repo/worktree", mode: "list" }); + expect(parsed.mode).toBe("list"); + }); +}); + describe("VcsCreateWorktreeInput", () => { it("accepts omitted newRefName for existing-refName worktrees", () => { const parsed = decodeCreateWorktreeInput({ diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 57c5380fb31..8ed8c95df7a 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -105,8 +105,21 @@ export type GitResolvedPullRequest = typeof GitResolvedPullRequest.Type; // RPC Inputs +/** + * How aggressively the server should refresh remote VCS status for a subscription. + * + * - `full` (default): long-lived remote poller (automatic git fetch interval) — for the + * active thread / git chrome. + * - `list`: local status + best-effort remote snapshot, **no** remote poller — for + * sidebar/board/list PR badges so N loaded worktrees do not create N pollers. + */ +export const VcsStatusSubscribeMode = Schema.Literals(["full", "list"]); +export type VcsStatusSubscribeMode = typeof VcsStatusSubscribeMode.Type; + export const VcsStatusInput = Schema.Struct({ cwd: TrimmedNonEmptyStringSchema, + /** Omit or `full` for active surfaces; use `list` for high-cardinality list UIs. */ + mode: Schema.optionalKey(VcsStatusSubscribeMode), }); export type VcsStatusInput = typeof VcsStatusInput.Type; From 825db396620c7c72ed46132b6c7d4ff4eb29f0d6 Mon Sep 17 00:00:00 2001 From: T3 Code PR Stack <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:30:07 +0200 Subject: [PATCH 2/2] fix(vcs): keep list PR state fresh via shared budgeted refresher List mode no longer drops live remote/PR updates. A single shared sweep refreshes all list-interested worktrees on a 60s cadence with concurrency 2, skipping cwds that already have a full-mode poller. Per-row 30s pollers stay gone so O(N) storms do not return. --- .../src/features/board/useBoardVcsStatuses.ts | 4 +- apps/mobile/src/state/use-thread-pr.ts | 4 +- .../src/vcs/VcsStatusBroadcaster.test.ts | 62 +++++---- apps/server/src/vcs/VcsStatusBroadcaster.ts | 124 +++++++++++++++--- .../components/board/useBoardVcsStatuses.ts | 6 +- packages/client-runtime/src/state/vcs.ts | 5 +- packages/contracts/src/git.ts | 11 +- 7 files changed, 160 insertions(+), 56 deletions(-) diff --git a/apps/mobile/src/features/board/useBoardVcsStatuses.ts b/apps/mobile/src/features/board/useBoardVcsStatuses.ts index c8460f17276..5d0a195aa34 100644 --- a/apps/mobile/src/features/board/useBoardVcsStatuses.ts +++ b/apps/mobile/src/features/board/useBoardVcsStatuses.ts @@ -17,8 +17,8 @@ const EMPTY_STATUSES_ATOM = Atom.make( ).pipe(Atom.withLabel("mobile:board-vcs-statuses:empty")); /** - * Aggregated list-mode VCS status for the board (no remote poller) — same - * shape as web `useBoardVcsStatuses`. + * Aggregated list-mode VCS status for the board (shared budgeted refresh) — + * same shape as web `useBoardVcsStatuses`. */ export function useBoardVcsStatuses( targets: ReadonlyArray, diff --git a/apps/mobile/src/state/use-thread-pr.ts b/apps/mobile/src/state/use-thread-pr.ts index cec860688a0..49e34eeacaf 100644 --- a/apps/mobile/src/state/use-thread-pr.ts +++ b/apps/mobile/src/state/use-thread-pr.ts @@ -11,8 +11,8 @@ export { } from "./thread-pr-presentation"; /** - * List-mode PR status for a thread's branch (no remote poller). Deduped per - * (environmentId, cwd); use full `vcsEnvironment.status` only for active git chrome. + * List-mode PR status for a thread's branch (shared budgeted remote refresh). + * Deduped per (environmentId, cwd); use full `status` for active git chrome. */ export function useThreadPr( thread: EnvironmentThreadShell, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 4b05aac2caf..0cee37d7113 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -554,7 +554,7 @@ describe("VcsStatusBroadcaster", () => { ); }); - it.effect("list mode does not start a remote poller", () => { + it.effect("list mode uses a shared budgeted refresher, not a per-cwd poller", () => { const state = { currentLocalStatus: baseLocalStatus, currentRemoteStatus: remoteStatusWithPr, @@ -568,33 +568,43 @@ describe("VcsStatusBroadcaster", () => { return Effect.gen(function* () { const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; const scope = yield* Scope.make(); - const snapshotDeferred = yield* Deferred.make(); - yield* Stream.runForEach( - broadcaster.streamStatus( - { cwd: "/repo", mode: "list" }, - { automaticRemoteRefreshInterval: Effect.succeed(Duration.seconds(30)) }, - ), - (event) => - event._tag === "snapshot" - ? Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore) - : Effect.void, - ).pipe(Effect.forkIn(scope)); - - const snapshot = yield* Deferred.await(snapshotDeferred); - assert.deepStrictEqual(snapshot, { - _tag: "snapshot", - local: baseLocalStatus, - remote: remoteStatusWithPr, - } satisfies VcsStatusStreamEvent); - // One coalesced remote load for the initial badge fill — no force invalidate. - assert.equal(state.remoteStatusCalls, 1); + const repoAReady = yield* Deferred.make(); + const repoBReady = yield* Deferred.make(); + + const trackReady = (cwd: string, ready: Deferred.Deferred) => + Stream.runForEach( + broadcaster.streamStatus( + { cwd, mode: "list" }, + { automaticRemoteRefreshInterval: Effect.succeed(Duration.seconds(30)) }, + ), + (event) => + event._tag === "snapshot" || event._tag === "remoteUpdated" + ? Deferred.succeed(ready, undefined).pipe(Effect.ignore) + : Effect.void, + ).pipe(Effect.forkIn(scope, { startImmediately: true })); + + // Two list worktrees — shared refresher, not two independent 30s pollers. + yield* trackReady("/repo-a", repoAReady); + yield* trackReady("/repo-b", repoBReady); + yield* Deferred.await(repoAReady); + yield* Deferred.await(repoBReady); + + // Initial fill: one remote load per list cwd (subscribe fill and/or shared sweep). + assert.isAtLeast(state.remoteStatusCalls, 2); + assert.isAtMost(state.remoteStatusCalls, 4); assert.equal(state.remoteInvalidationCalls, 0); - assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues, [false]); - - // Advance well past the full-mode poll interval; list mode must not re-poll. - yield* TestClock.adjust(Duration.minutes(5)); + for (const flag of state.remoteStatusRefreshUpstreamValues) { + assert.equal(flag, false); + } + const afterInitial = state.remoteStatusCalls; + + // Over ~90s: shared list cadence (~60s) should add about one sweep for both + // cwds (+2), not two independent 30s full pollers (≈ +6). + yield* TestClock.adjust(Duration.seconds(90)); yield* Effect.yieldNow; - assert.equal(state.remoteStatusCalls, 1); + const delta = state.remoteStatusCalls - afterInitial; + assert.isAtLeast(delta, 2); + assert.isAtMost(delta, 4); yield* Scope.close(scope, Exit.void); }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer()))); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 106d8873fc8..e28f0b407d3 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -25,11 +25,20 @@ import { mergeGitStatusParts } from "@t3tools/shared/git"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; const DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30); +/** + * Shared list-mode remote refresh: one loop for all list-interested worktrees, + * not one fiber per cwd. Keeps PR badges fresh without O(N) independent pollers. + */ +const LIST_REMOTE_REFRESH_INTERVAL = Duration.seconds(60); +const LIST_REMOTE_REFRESH_CONCURRENCY = 2; const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30); const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15); const MAX_FAILURE_DIAGNOSTIC_VALUES = 8; const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128; +/** Exported for tests — list subscriptions share this cadence. */ +export const LIST_MODE_REMOTE_REFRESH_INTERVAL = LIST_REMOTE_REFRESH_INTERVAL; + function boundedDiagnosticValue(value: string): string { return value.slice(0, MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH); } @@ -190,6 +199,9 @@ export const make = Effect.gen(function* () { ); const cacheRef = yield* Ref.make(new Map()); const pollersRef = yield* SynchronizedRef.make(new Map()); + /** cwd → list-mode subscriber count (high-cardinality sidebar/board rows). */ + const listInterestRef = yield* SynchronizedRef.make(new Map()); + const listRefreshFiberRef = yield* SynchronizedRef.make | null>(null); const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* ( cwd: string, @@ -507,6 +519,84 @@ export const make = Effect.gen(function* () { } }); + /** + * Budgeted remote refresh for every list-interested cwd that does not already + * have a full-mode poller. One fiber total, concurrency-capped — not O(N) fibers. + */ + const runListRemoteRefreshSweep = Effect.fn("VcsStatusBroadcaster.runListRemoteRefreshSweep")( + function* () { + const interests = yield* SynchronizedRef.get(listInterestRef); + const fullPollers = yield* SynchronizedRef.get(pollersRef); + const cwds = [...interests.keys()].filter((cwd) => !fullPollers.has(cwd)); + if (cwds.length === 0) { + return; + } + yield* Effect.forEach( + cwds, + (cwd) => refreshRemoteStatus(cwd, { refreshUpstream: false }).pipe(Effect.ignore), + { concurrency: LIST_REMOTE_REFRESH_CONCURRENCY }, + ); + }, + ); + + const makeListRemoteRefreshLoop = () => + Effect.gen(function* () { + // Initial sweep so list badges do not wait a full interval after first open. + yield* runListRemoteRefreshSweep(); + return yield* Effect.forever( + Effect.sleep(LIST_REMOTE_REFRESH_INTERVAL).pipe(Effect.andThen(runListRemoteRefreshSweep)), + ); + }); + + const retainListInterest = Effect.fn("VcsStatusBroadcaster.retainListInterest")(function* ( + cwd: string, + ) { + yield* SynchronizedRef.modifyEffect(listInterestRef, (interests) => { + const next = new Map(interests); + next.set(cwd, (next.get(cwd) ?? 0) + 1); + return Effect.succeed([undefined, next] as const); + }); + + yield* SynchronizedRef.modifyEffect(listRefreshFiberRef, (existing) => { + if (existing) { + return Effect.succeed([undefined, existing] as const); + } + return makeListRemoteRefreshLoop().pipe( + Effect.forkIn(broadcasterScope), + Effect.map((fiber) => [undefined, fiber] as const), + ); + }); + }); + + const releaseListInterest = Effect.fn("VcsStatusBroadcaster.releaseListInterest")(function* ( + cwd: string, + ) { + const shouldStopLoop = yield* SynchronizedRef.modifyEffect(listInterestRef, (interests) => { + const current = interests.get(cwd) ?? 0; + if (current <= 0) { + return Effect.succeed([false, interests] as const); + } + const next = new Map(interests); + if (current === 1) { + next.delete(cwd); + } else { + next.set(cwd, current - 1); + } + return Effect.succeed([next.size === 0, next] as const); + }); + + if (!shouldStopLoop) { + return; + } + + const fiber = yield* SynchronizedRef.modifyEffect(listRefreshFiberRef, (existing) => + Effect.succeed([existing, null] as const), + ); + if (fiber) { + yield* Fiber.interrupt(fiber).pipe(Effect.ignore); + } + }); + const streamStatus: VcsStatusBroadcaster["Service"]["streamStatus"] = (input, options) => Stream.unwrap( Effect.gen(function* () { @@ -516,22 +606,24 @@ export const make = Effect.gen(function* () { const initialLocal = yield* getOrLoadLocalStatus(cwd); let cachedStatus = yield* getCachedStatus(cwd); - // List mode must not start a remote poller (O(N) storm root for sidebar/board). - // One coalesced remote load when uncached so PR badges can populate once; later - // updates arrive via pubsub when a full subscriber or explicit refresh runs. - if ( - mode === "list" && - (cachedStatus?.remote === null || cachedStatus?.remote === undefined) - ) { - yield* refreshRemoteStatus(cwd, { refreshUpstream: false }).pipe(Effect.ignore); - cachedStatus = yield* getCachedStatus(cwd); - } - - const initialRemote = cachedStatus?.remote?.value ?? null; - - // Full mode only: long-lived remote poller for the active thread / git chrome. + // List mode: shared budgeted refresher keeps remote/PR state fresh for all + // list-interested cwds without one 30s poller fiber per worktree (storm root). + // Full mode: dedicated poller (may include git fetch) for active git chrome. let release: Effect.Effect = Effect.void; - if (mode === "full") { + if (mode === "list") { + // Registers cwd with the shared budgeted refresher (keeps PR/remote fresh). + // No per-cwd poller — that was the O(N) storm root. + yield* retainListInterest(cwd); + cachedStatus = yield* getCachedStatus(cwd); + // New cwd (or cache cold): fill once now so badges are not empty until the + // next shared sweep. The shared loop continues periodic updates for all + // list-interested cwds with bounded concurrency. + if (cachedStatus?.remote === null || cachedStatus?.remote === undefined) { + yield* refreshRemoteStatus(cwd, { refreshUpstream: false }).pipe(Effect.ignore); + cachedStatus = yield* getCachedStatus(cwd); + } + release = releaseListInterest(cwd).pipe(Effect.ignore, Effect.asVoid); + } else { yield* retainRemotePoller( cwd, options?.automaticRemoteRefreshInterval ?? @@ -541,6 +633,8 @@ export const make = Effect.gen(function* () { release = releaseRemotePoller(cwd).pipe(Effect.ignore, Effect.asVoid); } + const initialRemote = cachedStatus?.remote?.value ?? null; + // When remote is not cached yet, emit localUpdated only — never a snapshot that // fabricates remote defaults (pr:null). Downstream clients treat that fake null // PR as "no PR" and thrash badges (Discord ▫️⇄❌🔀 on every rehydrate). diff --git a/apps/web/src/components/board/useBoardVcsStatuses.ts b/apps/web/src/components/board/useBoardVcsStatuses.ts index 3a69c4ffff0..23026da065d 100644 --- a/apps/web/src/components/board/useBoardVcsStatuses.ts +++ b/apps/web/src/components/board/useBoardVcsStatuses.ts @@ -17,10 +17,8 @@ const EMPTY_STATUSES_ATOM = Atom.make( ).pipe(Atom.withLabel("web:board-vcs-statuses:empty")); /** - * Aggregated list-mode VCS status for the board: one derived atom over the - * per-cwd listStatus family (no server remote poller). Dedupe by - * (environmentId, cwd); shorter idle TTL than full status. Entries are `null` - * until the first snapshot streams in. + * Aggregated list-mode VCS status for the board: shared budgeted remote refresh + * (PR/git stay fresh; no per-row poller). Dedupe by (environmentId, cwd). */ export function useBoardVcsStatuses( targets: ReadonlyArray, diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index 7acd5631628..a75521f6aa9 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -294,8 +294,9 @@ export function createVcsEnvironmentAtoms( subscribe: statusStream, }), /** - * List/badge VCS status: no remote poller on the server. Shorter idle TTL so - * off-screen rows drop streams. Prefer this for sidebar/board/thread rows. + * List/badge VCS status: shared budgeted remote refresh on the server (keeps PR + * state fresh without per-row pollers). Shorter idle TTL so off-screen rows drop. + * Prefer for sidebar/board/thread rows; use `status` for active git chrome. */ listStatus: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:vcs:status-list", diff --git a/packages/contracts/src/git.ts b/packages/contracts/src/git.ts index 8ed8c95df7a..4e4634a045b 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -106,12 +106,13 @@ export type GitResolvedPullRequest = typeof GitResolvedPullRequest.Type; // RPC Inputs /** - * How aggressively the server should refresh remote VCS status for a subscription. + * How the server should refresh remote VCS status for a subscription. * - * - `full` (default): long-lived remote poller (automatic git fetch interval) — for the - * active thread / git chrome. - * - `list`: local status + best-effort remote snapshot, **no** remote poller — for - * sidebar/board/list PR badges so N loaded worktrees do not create N pollers. + * - `full` (default): dedicated per-cwd remote poller (automatic git fetch interval) — + * for the active thread / git chrome. + * - `list`: still keeps remote/PR state **up to date** via a **shared budgeted** + * refresher for all list-interested worktrees (not one poller fiber per row). + * Use for sidebar/board/list PR badges. */ export const VcsStatusSubscribeMode = Schema.Literals(["full", "list"]); export type VcsStatusSubscribeMode = typeof VcsStatusSubscribeMode.Type;