diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index a944922d530b5..e1af44f4b0057 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -100,6 +100,11 @@ export interface GitRunStackedActionOptions { readonly progressReporter?: GitActionProgressReporter; } +export interface GitLocalStatusOptions { + /** Defaults to true. The branch-wide totals cost about half of a local status read. */ + readonly includeBranchChanges?: boolean; +} + export interface GitRemoteStatusOptions extends GitVcsDriver.GitRemoteStatusOptions { readonly refreshMissingPullRequest?: boolean; } @@ -129,6 +134,7 @@ export class GitManager extends Context.Service< ) => Effect.Effect; readonly localStatus: ( input: VcsStatusInput, + options?: GitLocalStatusOptions, ) => Effect.Effect; readonly remoteStatus: ( input: VcsStatusInput, @@ -1125,9 +1131,12 @@ export const make = Effect.gen(function* () { behindCount: 0, aheadOfDefaultCount: 0, }; - const readLocalStatus = Effect.fn("readLocalStatus")(function* (cwd: string) { + const readLocalStatus = Effect.fn("readLocalStatus")(function* ( + cwd: string, + includeBranchChanges: boolean, + ) { const details = yield* gitCore - .statusDetailsLocal(cwd, { includeDivergence: false, includeBranchChanges: true }) + .statusDetailsLocal(cwd, { includeDivergence: false, includeBranchChanges }) .pipe( Effect.catchIf(isNotGitRepositoryError, () => Effect.succeed(nonRepositoryStatusDetails)), ); @@ -1146,13 +1155,22 @@ export const make = Effect.gen(function* () { ...(details.branchChanges ? { branchChanges: details.branchChanges } : {}), } satisfies VcsStatusLocalResult; }); - const localStatusResultCache = yield* Cache.makeWith(readLocalStatus, { - capacity: STATUS_RESULT_CACHE_CAPACITY, - timeToLive: (exit) => (Exit.isSuccess(exit) ? STATUS_RESULT_CACHE_TTL : Duration.zero), - }); + const makeLocalStatusResultCache = (includeBranchChanges: boolean) => + Cache.makeWith((cwd: string) => readLocalStatus(cwd, includeBranchChanges), { + capacity: STATUS_RESULT_CACHE_CAPACITY, + timeToLive: (exit) => (Exit.isSuccess(exit) ? STATUS_RESULT_CACHE_TTL : Duration.zero), + }); + // Separate entries, so a read without totals is never served to a caller that shows them. + const localStatusResultCache = yield* makeLocalStatusResultCache(true); + const localStatusWithoutBranchChangesCache = yield* makeLocalStatusResultCache(false); const invalidateLocalStatusResultCache = (cwd: string) => normalizeStatusCacheKey(cwd).pipe( - Effect.flatMap((cacheKey) => Cache.invalidate(localStatusResultCache, cacheKey)), + Effect.flatMap((cacheKey) => + Effect.all([ + Cache.invalidate(localStatusResultCache, cacheKey), + Cache.invalidate(localStatusWithoutBranchChangesCache, cacheKey), + ]), + ), ); const prLookupEpochByCwd = new Map(); const prLookupEpoch = (cwd: string) => prLookupEpochByCwd.get(cwd) ?? 0; @@ -2409,9 +2427,14 @@ export const make = Effect.gen(function* () { }); const localStatus: GitManager["Service"]["localStatus"] = Effect.fn("localStatus")( - function* (input) { + function* (input, options) { const cacheKey = yield* normalizeStatusCacheKey(input.cwd); - return yield* Cache.get(localStatusResultCache, cacheKey); + return yield* Cache.get( + options?.includeBranchChanges === false + ? localStatusWithoutBranchChangesCache + : localStatusResultCache, + cacheKey, + ); }, ); const remoteStatus: GitManager["Service"]["remoteStatus"] = Effect.fn("remoteStatus")( diff --git a/apps/server/src/git/GitWorkflowService.ts b/apps/server/src/git/GitWorkflowService.ts index bf89c7cc1baaa..4fe895654a1da 100644 --- a/apps/server/src/git/GitWorkflowService.ts +++ b/apps/server/src/git/GitWorkflowService.ts @@ -51,6 +51,7 @@ export class GitWorkflowService extends Context.Service< ) => Effect.Effect; readonly localStatus: ( input: VcsStatusInput, + options?: GitManager.GitLocalStatusOptions, ) => Effect.Effect; readonly currentBranch: (cwd: string) => Effect.Effect; readonly remoteStatus: ( @@ -325,11 +326,11 @@ export const make = Effect.gen(function* () { ), ); }), - localStatus: Effect.fn("GitWorkflowService.localStatus")(function (input) { + localStatus: Effect.fn("GitWorkflowService.localStatus")(function (input, options) { return detectGitRepositoryForStatus("GitWorkflowService.localStatus", input.cwd).pipe( Effect.flatMap((isGitRepository) => isGitRepository - ? gitManager.localStatus(input) + ? gitManager.localStatus(input, options) : Effect.succeed(nonRepositoryLocalStatus()), ), ); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 60f39b592d6de..11e0677e94723 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -11,6 +11,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Scope from "effect/Scope"; import * as Scheduler from "effect/Scheduler"; import * as Stream from "effect/Stream"; @@ -72,6 +73,7 @@ function layerTestFor(state: { localInvalidationCalls: number; remoteInvalidationCalls: number; remoteStatusRefreshUpstreamValues?: Array; + localStatusBranchChanges?: Array; remoteStatusError?: GitManagerError; backgroundWorkEnabled?: boolean; automaticFetchInterval?: Effect.Effect; @@ -95,11 +97,15 @@ function layerTestFor(state: { ), Layer.provide( Layer.mock(GitWorkflowService.GitWorkflowService)({ - localStatus: () => + localStatus: (_input, options) => Effect.suspend(() => { state.localStatusCalls += 1; + const includeBranchChanges = options?.includeBranchChanges !== false; + state.localStatusBranchChanges?.push(includeBranchChanges); + const { branchChanges: _branchChanges, ...withoutBranchChanges } = + state.currentLocalStatus; return (state.beforeLocalStatus ?? Effect.void).pipe( - Effect.as(state.currentLocalStatus), + Effect.as(includeBranchChanges ? state.currentLocalStatus : withoutBranchChanges), ); }), currentBranch: () => Effect.sync(() => state.currentLocalStatus.refName), @@ -717,6 +723,111 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); }); + it.effect("reads Changes totals only while a subscriber shows them", () => { + const branchChanges = { baseRef: "origin/main", insertions: 3, deletions: 1 }; + const state: Parameters[0] = { + currentLocalStatus: { ...baseLocalStatus, branchChanges }, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + localStatusBranchChanges: [], + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const sidebarEvents = yield* Queue.unbounded(); + yield* broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }).pipe( + Stream.runForEach((event) => Queue.offer(sidebarEvents, event)), + Effect.forkScoped, + ); + const sidebarSnapshot = yield* Queue.take(sidebarEvents); + assert.strictEqual(sidebarSnapshot._tag, "snapshot"); + if (sidebarSnapshot._tag === "snapshot") { + assert.isUndefined(sidebarSnapshot.local.branchChanges); + } + yield* broadcaster.refreshLocalStatus("/repo"); + assert.deepStrictEqual(state.localStatusBranchChanges, [false, false]); + + const headerScope = yield* Scope.make(); + const headerEvents = yield* Queue.unbounded(); + yield* broadcaster.streamStatus({ cwd: "/repo" }).pipe( + Stream.runForEach((event) => Queue.offer(headerEvents, event)), + Effect.forkIn(headerScope), + ); + const headerSnapshot = yield* Queue.take(headerEvents); + assert.strictEqual(headerSnapshot._tag, "snapshot"); + if (headerSnapshot._tag === "snapshot") { + assert.deepStrictEqual(headerSnapshot.local.branchChanges, branchChanges); + } + const sidebarUpdate = yield* Queue.take(sidebarEvents); + assert.strictEqual(sidebarUpdate._tag, "localUpdated"); + if (sidebarUpdate._tag === "localUpdated") { + assert.deepStrictEqual(sidebarUpdate.local.branchChanges, branchChanges); + } + // The header's first remote poll also rereads local status. + while ((yield* Queue.take(headerEvents))._tag !== "remoteUpdated") {} + const turnEnd = yield* broadcaster.refreshLocalStatus("/repo"); + assert.deepStrictEqual(turnEnd.branchChanges, branchChanges); + + yield* Scope.close(headerScope, Exit.void); + yield* broadcaster.refreshLocalStatus("/repo"); + assert.deepStrictEqual(state.localStatusBranchChanges, [ + false, + false, + true, + true, + true, + false, + ]); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }); + + it.effect("a light refresh that a header joins mid-read publishes Changes totals", () => { + const branchChanges = { baseRef: "origin/main", insertions: 3, deletions: 1 }; + const state: Parameters[0] = { + currentLocalStatus: { ...baseLocalStatus, branchChanges }, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + localStatusBranchChanges: [], + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const subscribeHeader = Effect.fn("subscribeHeader")(function* (scope: Scope.Scope) { + const events = yield* Queue.unbounded(); + yield* broadcaster.streamStatus({ cwd: "/repo" }).pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkIn(scope), + ); + return events; + }); + const firstHeader = yield* Scope.make(); + const firstEvents = yield* subscribeHeader(firstHeader); + while ((yield* Queue.take(firstEvents))._tag !== "remoteUpdated") {} + yield* Scope.close(firstHeader, Exit.void); + + const readStarted = yield* Deferred.make(); + const finishRead = yield* Deferred.make(); + state.beforeLocalStatus = Deferred.succeed(readStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishRead)), + ); + const turnEnd = yield* broadcaster.refreshLocalStatus("/repo").pipe(Effect.forkScoped); + yield* Deferred.await(readStarted); + const secondEvents = yield* subscribeHeader(yield* Scope.make()); + const snapshot = yield* Queue.take(secondEvents); + assert.strictEqual(snapshot._tag, "snapshot"); + yield* Deferred.succeed(finishRead, undefined); + + assert.deepStrictEqual((yield* Fiber.join(turnEnd)).branchChanges, branchChanges); + assert.deepStrictEqual(state.localStatusBranchChanges?.slice(-2), [false, true]); + }).pipe(Effect.provide(layerTestFor(state)), Effect.scoped); + }); + it.effect("an explicit refresh reads local totals after the fetch", () => { return Effect.gen(function* () { const fetchStarted = yield* Deferred.make(); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 3921077ebbb42..bec9b1deb3981 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -134,8 +134,12 @@ interface CachedRemoteStatus extends CachedValue { readonly branch: string | null; } +interface CachedLocalStatus extends CachedValue { + readonly includesBranchChanges: boolean; +} + interface CachedVcsStatus { - readonly local: CachedValue | null; + readonly local: CachedLocalStatus | null; readonly remote: CachedRemoteStatus | null; } @@ -271,9 +275,50 @@ export const make = Effect.gen(function* () { return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null)); }); + // Only full subscribers and direct reads show Changes totals. Sidebar rows subscribe without + // remote status and skip them, which halves the git processes a burst of rows spawns. + const branchChangesDemand = new Map(); + const wantsBranchChanges = (cwd: string) => branchChangesDemand.has(cwd); + const acquireBranchChangesDemand = (cwd: string) => + Effect.acquireRelease( + Effect.sync(() => { + branchChangesDemand.set(cwd, (branchChangesDemand.get(cwd) ?? 0) + 1); + }), + () => + Effect.sync(() => { + const count = (branchChangesDemand.get(cwd) ?? 1) - 1; + if (count > 0) branchChangesDemand.set(cwd, count); + else branchChangesDemand.delete(cwd); + }), + ); + + const readLocalStatus = Effect.fn("VcsStatusBroadcaster.readLocalStatus")(function* ( + cwd: string, + ) { + const includeBranchChanges = wantsBranchChanges(cwd); + const local = yield* workflow.localStatus({ cwd }, { includeBranchChanges }); + // A subscriber that shows totals arrived during the read and must not receive it without them. + if (!includeBranchChanges && wantsBranchChanges(cwd)) { + return { + local: yield* workflow.localStatus({ cwd }, { includeBranchChanges: true }), + includesBranchChanges: true, + }; + } + return { local, includesBranchChanges: includeBranchChanges }; + }); + const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")( - function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) { - const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local }; + function* ( + cwd: string, + read: { readonly local: VcsStatusLocalResult; readonly includesBranchChanges: boolean }, + options?: { publish?: boolean }, + ) { + const { local, includesBranchChanges } = read; + const nextLocal = { + fingerprint: fingerprintStatusPart(local), + value: local, + includesBranchChanges, + }; const [localChanged, branchChanged, remoteCleared] = yield* Ref.modify(cacheRef, (cache) => { const previous = cache.get(cwd) ?? { local: null, remote: null }; const remote = previous.remote?.branch === local.refName ? previous.remote : null; @@ -328,8 +373,7 @@ export const make = Effect.gen(function* () { const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* ( cwd: string, ) { - const local = yield* workflow.localStatus({ cwd }); - return yield* updateCachedLocalStatus(cwd, local); + return yield* updateCachedLocalStatus(cwd, yield* readLocalStatus(cwd)); }); const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* ( @@ -337,8 +381,10 @@ export const make = Effect.gen(function* () { ) { const cached = yield* getCachedStatus(cwd); if (cached?.local) { - const branch = yield* workflow.currentBranch(cwd); - if (branch === cached.local.value.refName) return cached.local.value; + if (cached.local.includesBranchChanges || !wantsBranchChanges(cwd)) { + const branch = yield* workflow.currentBranch(cwd); + if (branch === cached.local.value.refName) return cached.local.value; + } return yield* refreshLocalStatusCore(cwd); } return yield* loadLocalStatus(cwd); @@ -358,6 +404,7 @@ export const make = Effect.gen(function* () { "VcsStatusBroadcaster.getStatus", )(function* (input) { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + yield* acquireBranchChangesDemand(cwd); yield* getOrLoadLocalStatus(cwd); const cached = yield* getCachedStatus(cwd); if (cached?.local && cached.remote) { @@ -374,7 +421,7 @@ export const make = Effect.gen(function* () { return mergeGitStatusParts(parts.local, parts.remote); }), ); - }); + }, Effect.scoped); // A refresh that started after a request saw every change that request could mean, so queued // callers share it instead of each running git status again. @@ -382,7 +429,11 @@ export const make = Effect.gen(function* () { const localRefreshRequests = new Map(); const localRefreshResults = new Map< string, - { readonly startedAt: number; readonly local: VcsStatusLocalResult } + { + readonly startedAt: number; + readonly local: VcsStatusLocalResult; + readonly includesBranchChanges: boolean; + } >(); const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")( function* (cwd: string) { @@ -392,12 +443,18 @@ export const make = Effect.gen(function* () { cwd, Effect.gen(function* () { const shared = localRefreshResults.get(cwd); - if (shared !== undefined && shared.startedAt >= requestedAt) return shared.local; + if ( + shared !== undefined && + shared.startedAt >= requestedAt && + (shared.includesBranchChanges || !wantsBranchChanges(cwd)) + ) { + return shared.local; + } const startedAt = localRefreshRequests.get(cwd) ?? requestedAt; yield* workflow.invalidateLocalStatus(cwd); - const local = yield* workflow.localStatus({ cwd }); - localRefreshResults.set(cwd, { startedAt, local }); - return yield* updateCachedLocalStatus(cwd, local, { publish: true }); + const read = yield* readLocalStatus(cwd); + localRefreshResults.set(cwd, { startedAt, ...read }); + return yield* updateCachedLocalStatus(cwd, read, { publish: true }); }), ); }, @@ -511,6 +568,7 @@ export const make = Effect.gen(function* () { "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd, options) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); + yield* acquireBranchChangesDemand(cwd); const previous = options?.localOnly ? yield* getCachedStatus(cwd) : null; const local = yield* refreshLocalStatusCore(cwd); if (options?.localOnly) { @@ -547,7 +605,7 @@ export const make = Effect.gen(function* () { return mergeGitStatusParts(parts.local, parts.remote); }), ); - }); + }, Effect.scoped); const refreshPullRequestStatus: VcsStatusBroadcaster["Service"]["refreshPullRequestStatus"] = Effect.fn("VcsStatusBroadcaster.refreshPullRequestStatus")(function* (rawCwd) { @@ -738,6 +796,7 @@ export const make = Effect.gen(function* () { Stream.unwrap( Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + if (input.includeRemote !== false) yield* acquireBranchChangesDemand(cwd); const subscription = yield* PubSub.subscribe(changesPubSub); const initialLocal = yield* getOrLoadLocalStatus(cwd); const cachedStatus = yield* getCachedStatus(cwd);