diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 06b1e3d9f9f69..02a8e43c1d0e2 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -171,6 +171,7 @@ const missingSummary = Effect.fail( interface HarnessOptions { readonly invalidate?: PullRequestService.PullRequestService["Service"]["invalidate"]; + readonly batchesReads?: (input: PullRequestRef) => boolean; readonly snapshot: TestShellSnapshot; readonly summary?: ( input: PullRequestRef, @@ -228,6 +229,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: const layerDependencies = Layer.mergeAll( Layer.mock(PullRequestService.PullRequestService)({ withSharedProjectReads: (effect) => effect, + batchesReads: (input) => Effect.succeed(options.batchesReads?.(input) ?? true), summary, stack, invalidate: options.invalidate ?? (() => Effect.void), @@ -1224,6 +1226,110 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("reads hosts that batch all at once and other hosts at the old sweep width", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const numbers = Array.from({ length: 60 }, (_, index) => index + 1); + const hostOf = (number: number) => (number <= 30 ? "github.com" : "gitlab.example.com"); + const batches = (input: PullRequestRef) => input.host === "github.com"; + const batched = yield* Ref.make(0); + const unbatched = yield* Ref.make(0); + const started = yield* Queue.unbounded(); + const release = yield* Deferred.make(); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot( + numbers.map((number) => + makeThread(`thread-${number}`, { + pullRequests: [makeLink(number, null, { host: hostOf(number) })], + }), + ), + ), + batchesReads: batches, + summary: (input) => { + const inFlight = batches(input) ? batched : unbatched; + return Ref.update(inFlight, (count) => count + 1).pipe( + Effect.andThen(Queue.offer(started, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.andThen(Ref.update(inFlight, (count) => count - 1)), + Effect.as(makeSummary(input)), + ); + }, + }); + yield* Effect.gen(function* () { + const reactor = yield* PullRequestSyncReactor.PullRequestSyncReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + yield* Queue.takeN(started, 55); + assert.strictEqual(yield* Ref.get(batched), 30); + assert.strictEqual(yield* Ref.get(unbatched), 25); + yield* Deferred.succeed(release, undefined); + yield* reactor.drain; + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, numbers.length); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("reads a member a sibling flags while the sweep is full", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const stackOf = (layers: ReadonlyArray) => ({ + id: "9", + number: 9, + url: "https://github.com/owner/repository/stack/9", + base: "main", + layers: layers.map(stackLayer), + }); + // #50 is merged and not due until #1's stack read shows it a different layer list. + const merged = makeLink( + 50, + { state: "merged", mergedAt: NOW }, + { stack: { kind: "native", ...stackOf([50]) } }, + ); + const started = yield* Ref.make(0); + const full = yield* Deferred.make(); + const synced1 = yield* Deferred.make(); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + // Another host takes 24 of the 25 shared slots and #1 the last one. + ...Array.from({ length: 24 }, (_, index) => + makeThread(`thread-${index + 2}`, { + pullRequests: [makeLink(index + 2, null, { host: "gitlab.example.com" })], + }), + ), + makeThread("thread-1", { pullRequests: [makeLink(1)] }), + makeThread("thread-50", { pullRequests: [merged] }), + ]), + batchesReads: () => false, + // #1 reads once all 25 slots are taken; the other 24 hold theirs until #1 has synced. + summary: (input) => + Ref.updateAndGet(started, (count) => count + 1).pipe( + // Forked, so #1 resumes only after the sweep has handed out its next slot. + Effect.tap((count) => + count === 25 ? Effect.forkChild(Deferred.succeed(full, undefined)) : Effect.void, + ), + Effect.andThen( + input.number === 50 + ? Effect.void + : Deferred.await(input.number === 1 ? full : synced1), + ), + Effect.as(makeSummary(input)), + ), + stack: (input) => Effect.succeed(input.number === 1 ? stackOf([1, 50]) : null), + onSync: (command) => + command.number === 1 ? Deferred.succeed(synced1, undefined) : Effect.void, + }); + yield* Effect.gen(function* () { + yield* startAndSweep(fixture); + const read = (yield* Ref.get(fixture.summaryCalls)).map((ref) => ref.number); + assert.include(read, 50); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("skips the stack read for a pull request the summary places in no stack", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 5590a20ce0198..7573a95bb4362 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -183,6 +183,19 @@ interface LinkEntry { readonly link: ThreadPullRequestLink; } +const referenceOf = ({ thread, link }: LinkEntry) => ({ + projectId: thread.projectId, + host: normalizeThreadPullRequestKey(link).host, + repository: link.repository, + number: link.number, +}); + +/** Pauses and batching both belong to a project's host. */ +const pauseKeyOf = (entry: LinkEntry) => { + const { projectId, host } = referenceOf(entry); + return `${projectId}\0${host}`; +}; + /** Another layer of the stack `url` belongs to, in its host's canonical shape. */ function siblingChangeRequestUrl(url: string, number: number): string | null { if (!Number.isSafeInteger(number) || number < 1) return null; @@ -478,12 +491,7 @@ export const make = Effect.gen(function* () { const syncGroup = Effect.fnUntraced(function* (key: string, entries: ReadonlyArray) { const first = entries[0]!; - const ref = { - projectId: first.thread.projectId, - host: normalizeThreadPullRequestKey(first.link).host, - repository: first.link.repository, - number: first.link.number, - }; + const ref = referenceOf(first); const refreshRequested = requested.has(key); // A held merged summary is served without asking the host, so a snapshot still missing its // merge time has to drop it or the slow-cadence re-read would hand back the same gap. @@ -612,25 +620,63 @@ export const make = Effect.gen(function* () { return Effect.void; }), ); + const pools = { + // Wider than one summary batch, so reads that finish their cache checks at different + // times still share batches. + batched: { width: 100, reads: yield* Semaphore.make(100) }, + // One read per pull request, often a CLI process, so these keep the narrower width. + unbatched: { width: 25, reads: yield* Semaphore.make(25) }, + }; + const isPaused = (pauseKey: string, atMs: number) => (pausedUntil.get(pauseKey) ?? 0) > atMs; + // Judged once a slot is free, so a stack retry a sibling flags meanwhile still counts, and + // pauses set during the sweep apply at once. + const readIfDue = ( + key: string, + entries: ReadonlyArray, + pauseKey: string, + reads: Semaphore.Semaphore, + ) => + reads.withPermits(1)( + Effect.suspend(() => + isDue(key, entries, nowMs) + ? Clock.currentTimeMillis.pipe( + Effect.flatMap((startedAtMs) => + isPaused(pauseKey, startedAtMs) ? Effect.void : readGroup(key, entries, pauseKey), + ), + ) + : Effect.void, + ), + ); + const groupsByHost = Map.groupBy( + Array.from(groups).filter(([key]) => scope === "all" || requested.has(key)), + ([, entries]) => pauseKeyOf(entries[0]!), + ); yield* Effect.forEach( - groups, - ([key, entries]) => { - if (!((scope === "all" || requested.has(key)) && isDue(key, entries, nowMs))) { + groupsByHost, + ([pauseKey, hostGroups]) => { + // A host paused or with nothing due skips the lookup. Stack retries only flag groups on + // the flagging group's own host, so nothing on it becomes due this sweep. + if ( + isPaused(pauseKey, nowMs) || + !hostGroups.some(([key, entries]) => isDue(key, entries, nowMs)) + ) { return Effect.void; } - const first = entries[0]!; - const pauseKey = `${first.thread.projectId}\0${normalizeThreadPullRequestKey(first.link).host}`; - // Check pauses at each read so changes during the sweep apply immediately. - return Clock.currentTimeMillis.pipe( - Effect.flatMap((startedAtMs) => - (pausedUntil.get(pauseKey) ?? 0) > startedAtMs - ? Effect.void - : readGroup(key, entries, pauseKey), - ), + return pullRequests.batchesReads(referenceOf(hostGroups[0]![1][0]!)).pipe( + Effect.flatMap((batches) => { + const { width, reads } = batches ? pools.batched : pools.unbatched; + return Effect.forEach( + hostGroups, + ([key, entries]) => readIfDue(key, entries, pauseKey, reads), + { + concurrency: width, + discard: true, + }, + ); + }), ); }, - // Match summary batch width so GitHub reads coalesce into one request. - { concurrency: 25, discard: true }, + { concurrency: "unbounded", discard: true }, ); for (const [reason, { count, key }] of skips) { yield* Effect.logWarning("pull request sync skipped", { count, key, reason }); diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 91ee91b9ab083..b1aaa7e63e87d 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -200,6 +200,8 @@ export class PullRequestService extends Context.Service< input: PullRequestRef, options?: { readonly recoverTransientFailure?: boolean }, ) => Effect.Effect; + /** Whether the reference's host batches summary reads; false when it cannot be resolved. */ + readonly batchesReads: (input: PullRequestRef) => Effect.Effect; readonly stack: ( input: PullRequestRef, options?: { readonly includeDetails?: boolean }, @@ -536,6 +538,7 @@ function withRateLimitBackoff( ...(api.readsPublicHostWithoutCheckout === undefined ? {} : { readsPublicHostWithoutCheckout: api.readsPublicHostWithoutCheckout }), + ...(api.batchesReads === undefined ? {} : { batchesReads: api.batchesReads }), getViewer: options?.viewerAllowsPause === true ? interactive("getViewer", api.getViewer) @@ -3074,6 +3077,16 @@ export const make = Effect.gen(function* () { ); }; + const batchesReads: PullRequestService["Service"]["batchesReads"] = Effect.fn( + "PullRequestService.batchesReads", + )( + function* (input) { + const project = yield* requireProject(input); + return project.api.batchesReads === true; + }, + Effect.orElseSucceed(() => false), + ); + const stack: PullRequestService["Service"]["stack"] = (input, options) => persistedRead( input, @@ -3850,6 +3863,7 @@ export const make = Effect.gen(function* () { return yield* listStats({ ...input, refs: refs.flatMap(Option.toArray) }); }), summary: credentialCached(summary), + batchesReads, stack: credentialCached(stack), subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe( Effect.map((subscription) => Stream.fromSubscription(subscription)), diff --git a/packages/source-control-core/src/server/PullRequestProvider.ts b/packages/source-control-core/src/server/PullRequestProvider.ts index d5aab49d2cc0a..8418abda840d6 100644 --- a/packages/source-control-core/src/server/PullRequestProvider.ts +++ b/packages/source-control-core/src/server/PullRequestProvider.ts @@ -318,6 +318,8 @@ export interface PullRequestProviderApi { * definition's `publicHost` can be read from a project with no checkout of its repository. */ readonly readsPublicHostWithoutCheckout?: boolean; + /** Concurrent summary reads share requests, so callers may keep many in flight. */ + readonly batchesReads?: boolean; /** * Absent means every host detected as this kind is read. A host it refuses is listed as one that * cannot be browsed here, and nothing is sent for it. diff --git a/packages/source-control-github/src/server/GitHubPullRequestProvider.ts b/packages/source-control-github/src/server/GitHubPullRequestProvider.ts index 4809efc68998b..a680f8e0ae125 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestProvider.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestProvider.ts @@ -399,6 +399,7 @@ export const make = Effect.gen(function* () { capabilities: CAPABILITIES, unavailableRequirement: UNAVAILABLE_REQUIREMENT, readsPublicHostWithoutCheckout: true, + batchesReads: true, mergeMessageRewrite: removeAgentCredits, getRoutingIdentity: (input) => cli.getRoutingIdentity(input).pipe(Effect.mapError(fail("routeIdentity"))),