fix(server): PR sync reads share GitHub requests again on busy servers

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<void>();
+        const release = yield* Deferred.make<void>();
+        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<number>) => ({
+          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<void>();
+        const synced1 = yield* Deferred.make<void>();
+        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* () {

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<LinkEntry>) {
       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<LinkEntry>,
+      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 });

apps/server/src/pullRequest/PullRequestService.ts

@@ -200,6 +200,8 @@ export class PullRequestService extends Context.Service<
       input: PullRequestRef,
       options?: { readonly recoverTransientFailure?: boolean },
     ) => Effect.Effect<PullRequestSummary, PullRequestError>;
+    /** Whether the reference's host batches summary reads; false when it cannot be resolved. */
+    readonly batchesReads: (input: PullRequestRef) => Effect.Effect<boolean>;
     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)),

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.

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"))),