diff --git a/apps/mobile/src/features/board/useBoardVcsStatuses.ts b/apps/mobile/src/features/board/useBoardVcsStatuses.ts index 90c985c09dc1..5d0a195aa343 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 (shared budgeted refresh) — + * 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 a3440cd4848e..49e34eeacaf3 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 (shared budgeted remote refresh). + * Deduped per (environmentId, cwd); use full `status` 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 bc5d21e3b3e9..0cee37d71138 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -554,6 +554,62 @@ describe("VcsStatusBroadcaster", () => { ); }); + it.effect("list mode uses a shared budgeted refresher, not a per-cwd 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 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); + 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; + 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()))); + }); + 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 591ba77c1c2f..e28f0b407d32 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,22 +519,121 @@ 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* () { 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); - const initialRemote = cachedStatus?.remote?.value ?? null; - yield* retainRemotePoller( - cwd, - options?.automaticRemoteRefreshInterval ?? - Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL), - cachedStatus?.remote === null || cachedStatus?.remote === undefined, - ); + let cachedStatus = yield* getCachedStatus(cwd); + + // 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 === "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 ?? + Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL), + cachedStatus?.remote === null || cachedStatus?.remote === undefined, + ); + release = releaseRemotePoller(cwd).pipe(Effect.ignore, Effect.asVoid); + } - const 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 diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 1bc0db276e1e..a700780e149d 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 634b5f6d676a..ef1bf4be866b 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 3c3efbe487fb..66c8a4615da4 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 b774880b6519..23026da065d8 100644 --- a/apps/web/src/components/board/useBoardVcsStatuses.ts +++ b/apps/web/src/components/board/useBoardVcsStatuses.ts @@ -17,12 +17,8 @@ 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: shared budgeted remote refresh + * (PR/git stay fresh; no per-row poller). Dedupe by (environmentId, cwd). */ export function useBoardVcsStatuses( targets: ReadonlyArray, @@ -62,7 +58,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 ac85762a5433..a75521f6aa9a 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -272,20 +272,40 @@ 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: 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", + 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 0ac5c5fee2d5..5741b2419156 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: "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/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 57c5380fb318..4e4634a045be 100644 --- a/packages/contracts/src/git.ts +++ b/packages/contracts/src/git.ts @@ -105,8 +105,22 @@ export type GitResolvedPullRequest = typeof GitResolvedPullRequest.Type; // RPC Inputs +/** + * How the server should refresh remote VCS status for a subscription. + * + * - `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; + 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;