diff --git a/apps/server/src/git/GitManager.ts b/apps/server/src/git/GitManager.ts index dfd13a7c06609..23a79020c10f6 100644 --- a/apps/server/src/git/GitManager.ts +++ b/apps/server/src/git/GitManager.ts @@ -18,6 +18,7 @@ import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { GitActionProgressEvent, @@ -189,6 +190,9 @@ const PR_LOOKUP_FAILURE_MAX_TTL = Duration.minutes(15); const PR_LOOKUP_CACHE_CAPACITY = 2_048; const PR_LOOKUP_GIT_STATE_TTL = Duration.minutes(5); const SETTLED_PR_LOOKUP_RECHECK_INTERVAL = Duration.minutes(15); +// Shared by every background caller (backfill, settlement, cleanup), so startup discovery +// cannot fill the git and forge process slots that foreground work also needs. +const BRANCH_PR_LOOKUP_CONCURRENCY = 4; const PR_THREAD_CONTEXT_BYTES = 16_000; const PR_THREAD_COMMAND_OUTPUT_CHARS = 2_000; const isSourceControlProviderError = Schema.is(SourceControlProviderError); @@ -2579,6 +2583,7 @@ export const make = Effect.gen(function* () { prLookupGitStateByKey.set(entryKey, { readAtMs, value }); return value; }); + const branchPrLookups = yield* Semaphore.make(BRANCH_PR_LOOKUP_CONCURRENCY); const settledPrLookupByBranchKey = new Map< string, { readonly url: string; readonly fingerprint: string; readonly confirmedAtMs: number } @@ -2879,7 +2884,7 @@ export const make = Effect.gen(function* () { }); } return confirmed; - }); + }, branchPrLookups.withPermits(1)); const checkedOutPullRequest: GitManager["Service"]["checkedOutPullRequest"] = Effect.fn( "checkedOutPullRequest", diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 098c76184bbc9..ad134daab0f32 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -8,6 +8,7 @@ import { ProjectId, ProviderInstanceId, PullRequestOperationError, + PullRequestUnavailableError, RunId, ThreadId, TurnItemId, @@ -180,7 +181,7 @@ interface HarnessOptions { readonly snapshot: TestShellSnapshot; readonly summary?: ( input: PullRequestRef, - ) => Effect.Effect; + ) => Effect.Effect; readonly stack?: ( input: PullRequestRef, ) => Effect.Effect; @@ -1674,6 +1675,45 @@ describe("PullRequestSyncReactor", () => { }), ); + it.effect("stops reading a pull request on a host it cannot browse each sweep", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("unsupported", { pullRequests: [makeLink(9, { state: "open" })] }), + makeThread("fine", { pullRequests: [makeLink(8, { state: "open" })] }), + ]), + summary: (input) => + input.number === 9 + ? Effect.fail(new PullRequestUnavailableError({ reason: "provider-unsupported" })) + : Effect.succeed(makeSummary(input)), + }); + const unsupportedReads = Ref.get(fixture.summaryCalls).pipe( + Effect.map((calls) => calls.filter((call) => call.number === 9).length), + ); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + for (let index = 0; index < 5; index += 1) yield* sweepAgain(fixture, reactor); + const afterSweeps = yield* unsupportedReads; + + // A refresh another link asks for must not re-read the unsupported one as well. + yield* reactor.requestSync(makeLink(9)); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + const afterItsRefresh = yield* unsupportedReads; + for (let index = 0; index < 3; index += 1) { + yield* reactor.requestSync(makeLink(8)); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + } + assert.deepStrictEqual( + { afterSweeps, afterItsRefresh, afterOtherRefreshes: yield* unsupportedReads }, + { afterSweeps: 1, afterItsRefresh: 2, afterOtherRefreshes: 2 }, + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ); + it.effect( "holds missing links through sibling and link updates but honors a reader refresh", () => diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index d21b607f1bf58..46648e13b4b57 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -366,7 +366,8 @@ export const make = Effect.gen(function* () { // linking dozens of pull requests) is read together and shares the summary batches. let requestedSweepQueued = false; const retryStacks = new Set(); - // Group keys the host answered not-found for; they are read on the slow cadence until found. + // Group keys the host answered not-found for, or this server cannot read (no CLI, signed out, + // unsupported host); they are read on the slow cadence until a read succeeds. const missing = new Set(); // Pause per project and host because credentials are project-specific; leave paused PRs due until reset. const pausedUntil = new Map(); @@ -510,9 +511,10 @@ export const make = Effect.gen(function* () { if (invalidated) yield* pullRequests.invalidate({ reference: ref }); const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }).pipe( Effect.tapError((error) => - error.reason === "not-found" + error.reason === "not-found" || error._tag === "PullRequestUnavailableError" ? Effect.sync(() => { - // Not-found is a host answer: it consumes the refresh and starts the backoff. + // Not-found, or a host this server cannot read, won't change by asking again + // next sweep: it consumes the refresh and starts the backoff. missing.add(key); lastSyncedAt.set(key, nowMs); requested.delete(key);