diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 73093af6c750b..302e12dfc05a7 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -80,7 +80,7 @@ function layerTestFor(state: { remoteStatusError?: GitManagerError; backgroundWorkEnabled?: boolean; automaticFetchInterval?: Effect.Effect; - beforeRemoteStatus?: Effect.Effect; + beforeRemoteStatus?: Effect.Effect; beforeLocalStatus?: Effect.Effect; beforeCurrentBranch?: Effect.Effect; beforeInvalidate?: Effect.Effect; @@ -1724,6 +1724,92 @@ 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))); + }); + + 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.remoteStatusCalls, 1); + yield* broadcaster.refreshStatus("/repo"); + assert.equal(state.remoteStatusCalls, 2); + }).pipe(Effect.provide(layerTestFor(state))); + }); }); const newState = (): Parameters[0] => ({ diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 28daef7d790ca..ace2289e49aed 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -799,6 +799,57 @@ export const make = Effect.gen(function* () { } }); + // 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()); + + 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, 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( + 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); + }); + // Serialize branch publication and cache invalidation with detailed reads and refreshes. const refreshBranchStatusCore = Effect.fn("VcsStatusBroadcaster.refreshBranchStatusCore")( (cwd: string) => @@ -1028,7 +1079,7 @@ export const make = Effect.gen(function* () { ), getStatus, refreshLocalStatus, - refreshStatus, + refreshStatus: shareQueuedRefreshStatus, refreshPullRequestStatus, streamStatus, });