From aeb18a1a8ec63d2569a8a13d53660e9d495e2faf Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 08:52:53 -0500 Subject: [PATCH 1/4] fix(server): overlapping VCS status refreshes share one follow-up refresh Each vcs.refreshStatus call cleared the status caches and ran its own local scan, remote status read and PR lookup, one after another behind the remote write lock. A burst of focus events from several windows queued one full refresh per event. Calls that arrive while a refresh for the same cwd runs now wait for one shared follow-up refresh. They cannot reuse the running one, which may have read Git before their change. The broadcaster scope owns the work, so an interrupted caller does not cancel it for the others. Fixes #7076 Co-Authored-By: Claude Opus 5.5 --- .../src/vcs/VcsStatusBroadcaster.test.ts | 41 +++++++++++++ apps/server/src/vcs/VcsStatusBroadcaster.ts | 57 ++++++++++++++++++- 2 files changed, 97 insertions(+), 1 deletion(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 334c0d4747bce..7ddaa0d0b6524 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -600,4 +600,45 @@ describe("VcsStatusBroadcaster", () => { assert.isTrue(Option.isSome(yield* Deferred.poll(remoteInterrupted))); }).pipe(Effect.provide(layerTest)); }); + + it.effect("refreshes requested during a refresh share one follow-up refresh", () => { + const state: Parameters[0] = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const fetchStarted = yield* Deferred.make(); + const finishFetch = yield* Deferred.make(); + state.beforeRemoteStatus = Deferred.succeed(fetchStarted, undefined).pipe( + Effect.andThen(Deferred.await(finishFetch)), + ); + const first = yield* broadcaster.refreshStatus("/repo").pipe(Effect.forkScoped); + yield* Deferred.await(fetchStarted); + // Every read after the first sees a PR, so a caller that reused the + // first refresh would get a status from before its own request. + state.beforeRemoteStatus = Effect.sync(() => { + state.currentRemoteStatus = remoteStatusWithPr; + }); + const later = yield* Effect.all( + [broadcaster.refreshStatus("/repo"), broadcaster.refreshStatus("/repo")], + { concurrency: "unbounded" }, + ).pipe(Effect.forkScoped({ startImmediately: true })); + yield* Deferred.succeed(finishFetch, undefined); + + assert.isNull((yield* Fiber.join(first)).pr); + const [second, third] = yield* Fiber.join(later); + assert.deepStrictEqual(second.pr, remoteStatusWithPr.pr); + assert.deepStrictEqual(third.pr, remoteStatusWithPr.pr); + assert.equal(state.remoteStatusCalls, 2); + + yield* broadcaster.refreshStatus("/repo"); + assert.equal(state.remoteStatusCalls, 3); + }).pipe(Effect.provide(layerTestFor(state))); + }); }); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 71427713e5b88..d27ba4818dcb5 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -1,5 +1,6 @@ import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -756,10 +757,64 @@ export const make = Effect.gen(function* () { }), ); + // A caller that arrives while a refresh for its cwd runs cannot reuse that + // refresh: it may have read Git before the caller's change, such as a PR an + // agent just opened. Those callers share one follow-up refresh instead of + // each queueing another full status read and PR lookup behind the remote + // lock. Keyed by the raw cwd so each refresh keeps its caller's auto-pull + // policy input. + // No entry means idle; `null` means a refresh runs with no follow-up queued. + type SharedRefresh = Deferred.Deferred; + const queuedRefreshesRef = yield* Ref.make(new Map()); + + const runQueuedRefreshes = Effect.fn("VcsStatusBroadcaster.runQueuedRefreshes")(function* ( + rawCwd: string, + first: SharedRefresh, + ) { + let next: SharedRefresh | null = first; + while (next !== null) { + yield* Deferred.into(refreshStatus(rawCwd), next); + next = yield* Ref.modify(queuedRefreshesRef, (queued) => { + const followUp = queued.get(rawCwd) ?? null; + const nextQueued = new Map(queued); + if (followUp === null) { + nextQueued.delete(rawCwd); + } else { + nextQueued.set(rawCwd, null); + } + return [followUp, nextQueued] as const; + }); + } + }); + + const shareQueuedRefreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( + "VcsStatusBroadcaster.shareQueuedRefreshStatus", + )(function* (rawCwd) { + const requested = yield* Deferred.make(); + // Registering and forking together keeps an interrupt from leaving an entry with no worker. + const shared = yield* Effect.uninterruptible( + Effect.gen(function* () { + const followUp = yield* Ref.modify( + queuedRefreshesRef, + (queued): readonly [SharedRefresh | undefined, Map] => { + if (!queued.has(rawCwd)) return [undefined, new Map(queued).set(rawCwd, null)]; + const followUp = queued.get(rawCwd) ?? requested; + return [followUp, new Map(queued).set(rawCwd, followUp)]; + }, + ); + if (followUp !== undefined) return followUp; + // The broadcaster scope owns the work, so one caller's interrupt does not cancel it for the rest. + yield* runQueuedRefreshes(rawCwd, requested).pipe(Effect.forkIn(broadcasterScope)); + return requested; + }), + ); + return yield* Deferred.await(shared); + }); + return VcsStatusBroadcaster.of({ getStatus, refreshLocalStatus, - refreshStatus, + refreshStatus: shareQueuedRefreshStatus, refreshPullRequestStatus, streamStatus, }); From bca39c0aa01ed3f00fbeede5a58eeaeebb496011 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 13:17:36 +0000 Subject: [PATCH 2/4] fix(server): preserve local-only status refreshes when sharing work --- .../src/vcs/VcsStatusBroadcaster.test.ts | 46 ++++++++++++++++++- apps/server/src/vcs/VcsStatusBroadcaster.ts | 12 ++--- 2 files changed, 49 insertions(+), 9 deletions(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 60287064fb4ed..513e1fe284460 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -78,7 +78,7 @@ function layerTestFor(state: { remoteStatusRefreshUpstreamValues?: Array; backgroundWorkEnabled?: boolean; automaticFetchInterval?: Effect.Effect; - beforeRemoteStatus?: Effect.Effect; + beforeRemoteStatus?: Effect.Effect; }) { return VcsStatusBroadcaster.layer.pipe( Layer.provideMerge(NodeServices.layer), @@ -754,4 +754,48 @@ describe("VcsStatusBroadcaster", () => { assert.equal(state.remoteStatusCalls, 3); }).pipe(Effect.provide(layerTestFor(state))); }); + it.effect("a failed refresh releases interrupted waiters and runs the shared follow-up", () => { + const state: Parameters[0] = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: remoteStatusWithPr, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + state.beforeRemoteStatus = Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.andThen( + Effect.fail( + new GitManagerError({ + operation: "VcsStatusBroadcaster.test", + cwd: "/repo", + detail: "remote status failed", + }), + ), + ), + ); + const first = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(started); + const interrupted = yield* broadcaster + .refreshStatus("/repo") + .pipe(Effect.forkScoped({ startImmediately: true })); + const remaining = yield* broadcaster + .refreshStatus("/repo") + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Fiber.interrupt(interrupted); + state.beforeRemoteStatus = Effect.void; + yield* Deferred.succeed(release, undefined); + assert.isTrue(Exit.isFailure(yield* Fiber.join(first))); + assert.deepStrictEqual((yield* Fiber.join(remaining)).pr, remoteStatusWithPr.pr); + assert.equal(state.remoteInvalidationCalls, 2); + yield* broadcaster.refreshStatus("/repo"); + assert.equal(state.remoteInvalidationCalls, 3); + }).pipe(Effect.provide(layerTestFor(state))); + }); }); diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 5917e6a59e062..6f84f4058ccaa 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -773,13 +773,8 @@ export const make = Effect.gen(function* () { }), ); - // A caller that arrives while a refresh for its cwd runs cannot reuse that - // refresh: it may have read Git before the caller's change, such as a PR an - // agent just opened. Those callers share one follow-up refresh instead of - // each queueing another full status read and PR lookup behind the remote - // lock. Keyed by the raw cwd so each refresh keeps its caller's auto-pull - // policy input. - // No entry means idle; `null` means a refresh runs with no follow-up queued. + // Requests during a refresh need a later read; share one follow-up per raw cwd. + // Raw paths preserve auto-pull policy. Missing means idle; null means running without a follow-up. type SharedRefresh = Deferred.Deferred; const queuedRefreshesRef = yield* Ref.make(new Map()); @@ -805,7 +800,8 @@ export const make = Effect.gen(function* () { const shareQueuedRefreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( "VcsStatusBroadcaster.shareQueuedRefreshStatus", - )(function* (rawCwd) { + )(function* (rawCwd, options) { + if (options?.localOnly) return yield* refreshStatus(rawCwd, options); const requested = yield* Deferred.make(); // Registering and forking together keeps an interrupt from leaving an entry with no worker. const shared = yield* Effect.uninterruptible( From 14b81ef61a9aa5e653557888cd410ffe0a9ff82b Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 13:52:10 +0000 Subject: [PATCH 3/4] test(server): count remote reads, not invalidations, in the failed-refresh test Main's refresh now invalidates remote status twice (invalidateStatus and readRemoteStatus), so counting invalidations no longer counts refreshes. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/vcs/VcsStatusBroadcaster.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 3b8840b0707dc..6f19c373f8a01 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -1191,9 +1191,9 @@ describe("VcsStatusBroadcaster", () => { yield* Deferred.succeed(release, undefined); assert.isTrue(Exit.isFailure(yield* Fiber.join(first))); assert.deepStrictEqual((yield* Fiber.join(remaining)).pr, remoteStatusWithPr.pr); - assert.equal(state.remoteInvalidationCalls, 2); + assert.equal(state.remoteStatusCalls, 1); yield* broadcaster.refreshStatus("/repo"); - assert.equal(state.remoteInvalidationCalls, 3); + assert.equal(state.remoteStatusCalls, 2); }).pipe(Effect.provide(layerTestFor(state))); }); }); From 22756233aa40f1f361d8019772d559fdba55fada Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 06:58:05 +0000 Subject: [PATCH 4/4] refactor(server): clarify shared VCS refresh queue comment Co-Authored-By: Claude Opus 5.5 --- apps/server/src/vcs/VcsStatusBroadcaster.test.ts | 1 + apps/server/src/vcs/VcsStatusBroadcaster.ts | 5 +++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 02fba620a06fe..302e12dfc05a7 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -1765,6 +1765,7 @@ describe("VcsStatusBroadcaster", () => { assert.equal(state.remoteStatusCalls, 3); }).pipe(Effect.provide(layerTestFor(state))); }); + it.effect("a failed refresh releases interrupted waiters and runs the shared follow-up", () => { const state: Parameters[0] = { currentLocalStatus: baseLocalStatus, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index b734b970c4d58..ace2289e49aed 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -799,8 +799,9 @@ export const make = Effect.gen(function* () { } }); - // Requests during a refresh need a later read; share one follow-up per raw cwd. - // Raw paths preserve auto-pull policy. Missing means idle; null means running without a follow-up. + // A request made during a refresh needs a read that starts after it, so overlapping requests + // share one follow-up. Keyed by raw cwd because auto-pull policy depends on it. + // Missing means idle; null means running with no follow-up queued. type SharedRefresh = Deferred.Deferred; const queuedRefreshesRef = yield* Ref.make(new Map());