diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index 394764fa96f47..a47a5a2eefb4a 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -102,6 +102,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; } @@ -132,6 +137,7 @@ export class GitManager extends Context.Service< ) => Effect.Effect; readonly localStatus: ( input: VcsStatusInput, + options?: GitLocalStatusOptions, ) => Effect.Effect; readonly remoteStatus: ( input: VcsStatusInput, @@ -1202,9 +1208,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)), ); @@ -1223,13 +1232,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; @@ -2554,9 +2572,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 3cffe66441365..287207840ad6a 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 upstreamDivergence: (cwd: string) => Effect.Effect< @@ -341,11 +342,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 852c9f9ba0631..08fad4ae56aff 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -78,6 +78,7 @@ function layerTestFor(state: { localInvalidationCalls: number; remoteInvalidationCalls: number; remoteStatusRefreshUpstreamValues?: Array; + localStatusBranchChanges?: Array; refSnapshot?: string | null; remoteStatusError?: GitManagerError; backgroundWorkEnabled?: boolean; @@ -109,9 +110,13 @@ 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; if (state.failLocalStatus) { return (state.onLocalStatusFailure ?? Effect.void).pipe( Effect.andThen( @@ -126,7 +131,7 @@ function layerTestFor(state: { ); } return (state.beforeLocalStatus ?? Effect.void).pipe( - Effect.as(state.currentLocalStatus), + Effect.as(includeBranchChanges ? state.currentLocalStatus : withoutBranchChanges), ); }), repositoryRoot: () => Effect.sync(() => (state.currentLocalStatus.isRepo ? "/repo" : null)), @@ -1158,6 +1163,116 @@ describe("VcsStatusBroadcaster", () => { }).pipe(Effect.provide(layerTestFor(state))); }); + 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 paletteEvents = yield* Queue.unbounded(); + yield* broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }).pipe( + Stream.runForEach((event) => Queue.offer(paletteEvents, event)), + Effect.forkScoped, + ); + const paletteSnapshot = yield* Queue.take(paletteEvents); + assert.strictEqual(paletteSnapshot._tag, "snapshot"); + if (paletteSnapshot._tag === "snapshot") { + assert.isUndefined(paletteSnapshot.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 paletteUpdate = yield* Queue.take(paletteEvents); + assert.strictEqual(paletteUpdate._tag, "localUpdated"); + if (paletteUpdate._tag === "localUpdated") { + assert.deepStrictEqual(paletteUpdate.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: [], + fileSystem: FileSystem.makeNoop({ realPath: (cwd) => Effect.succeed(cwd) }), + }; + + 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, { startImmediately: true }), + ); + 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); + // Starting immediately registers the header's demand before it waits behind the read. + const secondEvents = yield* subscribeHeader(yield* Scope.make()); + yield* Deferred.succeed(finishRead, undefined); + const snapshot = yield* Queue.take(secondEvents); + assert.strictEqual(snapshot._tag, "snapshot"); + if (snapshot._tag === "snapshot") { + assert.deepStrictEqual(snapshot.local.branchChanges, branchChanges); + } + + 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(); @@ -1233,7 +1348,7 @@ describe("VcsStatusBroadcaster", () => { event._tag === "snapshot" ? Deferred.succeed(subscribed, undefined) : Effect.void, ), Stream.filter((event) => event._tag !== "snapshot"), - Stream.take(3), + Stream.take(4), Stream.runCollect, Effect.forkScoped, ); @@ -1254,7 +1369,8 @@ describe("VcsStatusBroadcaster", () => { assert.deepStrictEqual(moved.branchChanges, totals(0).branchChanges); assert.deepStrictEqual( [...(yield* Fiber.join(events))].map((event) => event._tag), - ["remoteUpdated", "remoteUpdated", "localUpdated"], + // The first refresh adds the totals this palette-style subscription skipped. + ["localUpdated", "remoteUpdated", "remoteUpdated", "localUpdated"], ); state.localStatusCalls = 0; @@ -2097,8 +2213,10 @@ it.effect("serializes branch reads with detailed refresh publication", () => { yield* Fiber.join(older); yield* Fiber.join(newer); state.branchRead = () => Effect.succeed("new"); - const result = yield* b.getStatus({ cwd: "/repo" }); - assert.equal(result.refName, "new"); + const result = yield* b + .streamStatus({ cwd: "/repo", includeRemote: false }) + .pipe(Stream.runHead, Effect.flatMap(Effect.fromOption)); + assert.equal(result._tag === "snapshot" && result.local.refName, "new"); assert.equal(state.localStatusCalls, 1); }).pipe(Effect.provide(layerTestFor(state))); }); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 8e69679d50638..b691b9d2978a2 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -143,8 +143,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; } @@ -315,14 +319,51 @@ 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. Command palette 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, + read: { readonly local: VcsStatusLocalResult; readonly includesBranchChanges: boolean }, observation: number, options?: { publish?: boolean }, ) { - const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local }; + const { local, includesBranchChanges } = read; + const nextLocal = { + fingerprint: fingerprintStatusPart(local), + value: local, + includesBranchChanges, + }; const [localChanged, 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; @@ -373,16 +414,17 @@ export const make = Effect.gen(function* () { options?: { publish?: boolean }, ) { const observation = ++branchObservation; - const local = yield* workflow.localStatus({ cwd }); - return yield* updateCachedLocalStatus(cwd, local, observation, options); + return yield* updateCachedLocalStatus(cwd, yield* readLocalStatus(cwd), observation, options); }); const getOrLoadLocalStatusUnlocked = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")( function* (cwd: string) { 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; + } yield* workflow.invalidateLocalStatus(cwd); return yield* loadLocalStatus(cwd, { publish: true }); } @@ -407,6 +449,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) { @@ -423,14 +466,18 @@ 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. 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) { @@ -440,13 +487,19 @@ 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 observation = ++branchObservation; - const local = yield* workflow.localStatus({ cwd }); - localRefreshResults.set(cwd, { startedAt, local }); - return yield* updateCachedLocalStatus(cwd, local, observation, { publish: true }); + const read = yield* readLocalStatus(cwd); + localRefreshResults.set(cwd, { startedAt, ...read }); + return yield* updateCachedLocalStatus(cwd, read, observation, { publish: true }); }), ); }, @@ -623,6 +676,7 @@ export const make = Effect.gen(function* () { "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd, options) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); + yield* acquireBranchChangesDemand(cwd); const refsBeforeLocalRead = options?.localOnly ? null : yield* workflow.readRefSnapshot(cwd); yield* refreshLocalStatusCore(cwd); if (options?.localOnly) { @@ -631,7 +685,7 @@ export const make = Effect.gen(function* () { return mergeGitStatusParts(parts.local, parts.remote); } return yield* refreshRemoteAndLocalTotals(cwd, rawCwd, refsBeforeLocalRead); - }); + }, Effect.scoped); const refreshAfterCheckout: VcsStatusBroadcaster["Service"]["refreshAfterCheckout"] = Effect.fn( "VcsStatusBroadcaster.refreshAfterCheckout", @@ -1055,6 +1109,7 @@ export const make = Effect.gen(function* () { else detailedSubscribers.set(cwd, remaining); }), ); + if (input.includeRemote !== false) yield* acquireBranchChangesDemand(cwd); const subscription = yield* PubSub.subscribe(changesPubSub); // Watch before reading so a checkout in between still publishes. yield* Effect.acquireRelease(retainHeadWatcher(cwd), () => releaseHeadWatcher(cwd));