From 64759ec9891af3794b6ce356668aca93d6530fb9 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 31 Oct 2026 23:18:18 +0000 Subject: [PATCH 1/6] fix(server): PR sync reads batch again on a busy server The sweep ran 25 groups at a time to match the 25-PR summary batch. Each read reaches the batch resolver only after its cache check, which a busy server spreads over hundreds of milliseconds, so batches split. Every slot then refilled on its own, and later reads went out about one PR per request. Run 100 groups at once so a sweep's reads still meet in a batch; the cap keeps hosts that read each PR separately bounded. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 39 +++++++++++++++++++ .../PullRequestSyncReactor.ts | 5 ++- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index a0a284b5db5d4..b07c3621f7156 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1101,6 +1101,45 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("has every due read of a sweep wider than one summary batch in flight at once", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const numbers = Array.from({ length: 30 }, (_, index) => index + 1); + const inFlight = yield* Ref.make({ active: 0, peak: 0 }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot( + numbers.map((number) => + makeThread(`thread-${number}`, { pullRequests: [makeLink(number)] }), + ), + ), + summary: (input) => + Ref.update(inFlight, ({ active, peak }) => ({ + active: active + 1, + peak: Math.max(peak, active + 1), + })).pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.andThen( + Ref.update(inFlight, (state) => ({ ...state, active: state.active - 1 })), + ), + Effect.as(makeSummary(input)), + ), + }); + yield* Effect.gen(function* () { + const reactor = yield* PullRequestSyncReactor.PullRequestSyncReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + yield* TestClock.adjust("1 second"); + // A narrower sweep reads the rest in a second wave. + yield* TestClock.adjust("1 second"); + yield* reactor.drain; + assert.strictEqual((yield* Ref.get(inFlight)).peak, numbers.length); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, numbers.length); + }).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 2c97b1867ef48..ecc6f20339789 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -588,8 +588,9 @@ export const make = Effect.gen(function* () { ), ); }, - // Match summary batch width so GitHub reads coalesce into one request. - { concurrency: 25, discard: true }, + // Wider than one summary batch, so reads that finish their cache checks at different times + // still share batches; the cap bounds hosts that read each pull request on its own. + { concurrency: 100, discard: true }, ); for (const [reason, { count, key }] of skips) { yield* Effect.logWarning("pull request sync skipped", { count, key, reason }); From 4c7e6cb6cbce658c64e24fba9aaa8d2bb895127b Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 00:00:25 +0000 Subject: [PATCH 2/6] fix(server): PR sync keeps hosts without batched reads at the old width The wider sweep let hosts that read each pull request on its own run up to 100 reads at once. Only GitHub batches summary reads, so providers now say whether they batch, and the sweep reads the rest in a 25-wide pool beside the 100-wide batched one. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 46 ++++++++++++++ .../PullRequestSyncReactor.ts | 63 +++++++++++-------- .../src/pullRequest/PullRequestService.ts | 10 +++ .../src/server/PullRequestProvider.ts | 2 + .../src/server/GitHubPullRequestProvider.ts | 1 + 5 files changed, 96 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index b07c3621f7156..8e4d5d4b124bd 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -166,6 +166,7 @@ function makeSummary( interface HarnessOptions { readonly invalidate?: PullRequestService.PullRequestService["Service"]["invalidate"]; + readonly batchesReads?: (input: PullRequestRef) => boolean; readonly snapshot: TestShellSnapshot; readonly summary?: ( input: PullRequestRef, @@ -223,6 +224,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), @@ -1140,6 +1142,50 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("keeps hosts that read each pull request on its own to 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 batches = (input: PullRequestRef) => input.number <= 30; + const batched = yield* Ref.make({ active: 0, peak: 0 }); + const unbatched = yield* Ref.make({ active: 0, peak: 0 }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot( + numbers.map((number) => + makeThread(`thread-${number}`, { pullRequests: [makeLink(number)] }), + ), + ), + batchesReads: batches, + summary: (input) => { + const inFlight = batches(input) ? batched : unbatched; + return Ref.update(inFlight, ({ active, peak }) => ({ + active: active + 1, + peak: Math.max(peak, active + 1), + })).pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.andThen( + Ref.update(inFlight, (state) => ({ ...state, active: state.active - 1 })), + ), + Effect.as(makeSummary(input)), + ); + }, + }); + yield* Effect.gen(function* () { + const reactor = yield* PullRequestSyncReactor.PullRequestSyncReactor; + yield* reactor.start(); + yield* Deferred.succeed(fixture.activation, undefined); + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("1 second"); + yield* reactor.drain; + assert.strictEqual((yield* Ref.get(batched)).peak, 30); + assert.strictEqual((yield* Ref.get(unbatched)).peak, 25); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, numbers.length); + }).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 ecc6f20339789..e054f4b3f29e5 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -180,6 +180,13 @@ interface LinkEntry { readonly link: ThreadPullRequestLink; } +const referenceOf = ({ thread, link }: LinkEntry) => ({ + projectId: thread.projectId, + host: normalizeThreadPullRequestKey(link).host, + repository: link.repository, + number: link.number, +}); + /** 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; @@ -452,12 +459,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. @@ -571,26 +573,35 @@ export const make = Effect.gen(function* () { return Effect.void; }), ); - yield* Effect.forEach( - groups, - ([key, entries]) => { - if (!((scope === "all" || requested.has(key)) && 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), - ), - ); - }, - // Wider than one summary batch, so reads that finish their cache checks at different times - // still share batches; the cap bounds hosts that read each pull request on its own. - { concurrency: 100, discard: true }, + const readIfDue = ([key, entries]: readonly [string, ReadonlyArray]) => { + if (!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), + ), + ); + }; + const batched: Array]> = []; + const unbatched: Array]> = []; + for (const [key, entries] of groups) { + if (scope === "requested" && !requested.has(key)) continue; + const batches = yield* pullRequests.batchesReads(referenceOf(entries[0]!)); + (batches ? batched : unbatched).push([key, entries]); + } + yield* Effect.all( + [ + // Wider than one summary batch, so reads that finish their cache checks at different + // times still share batches. + Effect.forEach(batched, readIfDue, { concurrency: 100, discard: true }), + // One read per pull request, often a CLI process, so these keep the narrower width. + Effect.forEach(unbatched, readIfDue, { concurrency: 25, discard: true }), + ], + { concurrency: 2, 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 e77434c1697b3..ba9bcbfcde182 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -196,6 +196,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 }, @@ -519,6 +521,7 @@ function withRateLimitBackoff( ...(api.readsPublicHostWithoutCheckout === undefined ? {} : { readsPublicHostWithoutCheckout: api.readsPublicHostWithoutCheckout }), + ...(api.batchesReads === undefined ? {} : { batchesReads: api.batchesReads }), getViewer: options?.viewerAllowsPause === true ? interactive("getViewer", api.getViewer) @@ -2946,6 +2949,12 @@ export const make = Effect.gen(function* () { ); }; + const batchesReads: PullRequestService["Service"]["batchesReads"] = (input) => + requireProject(input).pipe( + Effect.map((project) => project.api.batchesReads === true), + Effect.orElseSucceed(() => false), + ); + const stack: PullRequestService["Service"]["stack"] = (input, options) => persistedRead( input, @@ -3714,6 +3723,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 a175c365d080e..d90cc7a133a28 100644 --- a/packages/source-control-core/src/server/PullRequestProvider.ts +++ b/packages/source-control-core/src/server/PullRequestProvider.ts @@ -308,6 +308,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 a99aa87c9652d..b9cd658ba5a9a 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestProvider.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestProvider.ts @@ -346,6 +346,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"))), From d3d91b320b27eb1f2feafb3b3d77ddfc9a7c7f27 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 00:05:05 +0000 Subject: [PATCH 3/6] test(server): gate the PR sync width test instead of sleeping Folds the batched and unbatched width checks into one test that holds reads at a Deferred gate and asserts the in-flight counts there, and resolves batching once per project host per sweep. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 74 +++++-------------- .../PullRequestSyncReactor.ts | 27 +++++-- 2 files changed, 38 insertions(+), 63 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 8e4d5d4b124bd..3e4c72bc6701f 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1103,70 +1103,32 @@ describe("PullRequestSyncReactor", () => { ), ); - it.effect("has every due read of a sweep wider than one summary batch in flight at once", () => - Effect.scoped( - Effect.gen(function* () { - yield* TestClock.setTime(Date.parse(NOW)); - const numbers = Array.from({ length: 30 }, (_, index) => index + 1); - const inFlight = yield* Ref.make({ active: 0, peak: 0 }); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot( - numbers.map((number) => - makeThread(`thread-${number}`, { pullRequests: [makeLink(number)] }), - ), - ), - summary: (input) => - Ref.update(inFlight, ({ active, peak }) => ({ - active: active + 1, - peak: Math.max(peak, active + 1), - })).pipe( - Effect.andThen(Effect.sleep("1 second")), - Effect.andThen( - Ref.update(inFlight, (state) => ({ ...state, active: state.active - 1 })), - ), - Effect.as(makeSummary(input)), - ), - }); - yield* Effect.gen(function* () { - const reactor = yield* PullRequestSyncReactor.PullRequestSyncReactor; - yield* reactor.start(); - yield* Deferred.succeed(fixture.activation, undefined); - yield* TestClock.adjust("1 second"); - // A narrower sweep reads the rest in a second wave. - yield* TestClock.adjust("1 second"); - yield* reactor.drain; - assert.strictEqual((yield* Ref.get(inFlight)).peak, numbers.length); - assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, numbers.length); - }).pipe(Effect.provide(fixture.layer)); - }), - ), - ); - - it.effect("keeps hosts that read each pull request on its own to the old sweep width", () => + 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 batches = (input: PullRequestRef) => input.number <= 30; - const batched = yield* Ref.make({ active: 0, peak: 0 }); - const unbatched = yield* Ref.make({ active: 0, peak: 0 }); + 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)] }), + makeThread(`thread-${number}`, { + pullRequests: [makeLink(number, null, { host: hostOf(number) })], + }), ), ), batchesReads: batches, summary: (input) => { const inFlight = batches(input) ? batched : unbatched; - return Ref.update(inFlight, ({ active, peak }) => ({ - active: active + 1, - peak: Math.max(peak, active + 1), - })).pipe( - Effect.andThen(Effect.sleep("1 second")), - Effect.andThen( - Ref.update(inFlight, (state) => ({ ...state, active: state.active - 1 })), - ), + 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)), ); }, @@ -1175,11 +1137,11 @@ describe("PullRequestSyncReactor", () => { const reactor = yield* PullRequestSyncReactor.PullRequestSyncReactor; yield* reactor.start(); yield* Deferred.succeed(fixture.activation, undefined); - yield* TestClock.adjust("1 second"); - yield* TestClock.adjust("1 second"); + 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(batched)).peak, 30); - assert.strictEqual((yield* Ref.get(unbatched)).peak, 25); assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, numbers.length); }).pipe(Effect.provide(fixture.layer)); }), diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index e054f4b3f29e5..1912d7cb0f987 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -187,6 +187,12 @@ const referenceOf = ({ thread, link }: LinkEntry) => ({ 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; @@ -573,10 +579,10 @@ export const make = Effect.gen(function* () { return Effect.void; }), ); - const readIfDue = ([key, entries]: readonly [string, ReadonlyArray]) => { + const readIfDue = (key: string) => { + const entries = groups.get(key)!; if (!isDue(key, entries, nowMs)) return Effect.void; - const first = entries[0]!; - const pauseKey = `${first.thread.projectId}\0${normalizeThreadPullRequestKey(first.link).host}`; + const pauseKey = pauseKeyOf(entries[0]!); // Check pauses at each read so changes during the sweep apply immediately. return Clock.currentTimeMillis.pipe( Effect.flatMap((startedAtMs) => @@ -586,12 +592,19 @@ export const make = Effect.gen(function* () { ), ); }; - const batched: Array]> = []; - const unbatched: Array]> = []; + const batchesByHost = new Map(); + const batched: Array = []; + const unbatched: Array = []; for (const [key, entries] of groups) { if (scope === "requested" && !requested.has(key)) continue; - const batches = yield* pullRequests.batchesReads(referenceOf(entries[0]!)); - (batches ? batched : unbatched).push([key, entries]); + const first = entries[0]!; + const pauseKey = pauseKeyOf(first); + let batches = batchesByHost.get(pauseKey); + if (batches === undefined) { + batches = yield* pullRequests.batchesReads(referenceOf(first)); + batchesByHost.set(pauseKey, batches); + } + (batches ? batched : unbatched).push(key); } yield* Effect.all( [ From 42d34b984e2e5f0f769353bb56ddcc6807c74cc7 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 00:13:56 +0000 Subject: [PATCH 4/6] fix(server): resolve each host's batching beside the others A slow or unresolvable project no longer holds back every other host's reads: each host resolves concurrently and feeds its due groups into the 100-wide or 25-wide permit pool. A host with nothing due skips the lookup. The service operation gets a named span. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.ts | 49 +++++++++++-------- .../src/pullRequest/PullRequestService.ts | 14 ++++-- 2 files changed, 38 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 1912d7cb0f987..4b0fa89515275 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -592,29 +592,38 @@ export const make = Effect.gen(function* () { ), ); }; - const batchesByHost = new Map(); - const batched: Array = []; - const unbatched: Array = []; + const keysByHost = new Map>(); for (const [key, entries] of groups) { if (scope === "requested" && !requested.has(key)) continue; - const first = entries[0]!; - const pauseKey = pauseKeyOf(first); - let batches = batchesByHost.get(pauseKey); - if (batches === undefined) { - batches = yield* pullRequests.batchesReads(referenceOf(first)); - batchesByHost.set(pauseKey, batches); - } - (batches ? batched : unbatched).push(key); + const pauseKey = pauseKeyOf(entries[0]!); + const keys = keysByHost.get(pauseKey) ?? []; + keys.push(key); + keysByHost.set(pauseKey, keys); } - yield* Effect.all( - [ - // Wider than one summary batch, so reads that finish their cache checks at different - // times still share batches. - Effect.forEach(batched, readIfDue, { concurrency: 100, discard: true }), - // One read per pull request, often a CLI process, so these keep the narrower width. - Effect.forEach(unbatched, readIfDue, { concurrency: 25, discard: true }), - ], - { concurrency: 2, discard: true }, + // Wider than one summary batch, so reads that finish their cache checks at different times + // still share batches. + const batchedReads = yield* Semaphore.make(100); + // One read per pull request, often a CLI process, so these keep the narrower width. + const unbatchedReads = yield* Semaphore.make(25); + yield* Effect.forEach( + keysByHost.values(), + (keys) => { + // Stack retries only flag groups on the flagging group's host, so a host with nothing + // due yet stays that way this sweep. + if (!keys.some((key) => isDue(key, groups.get(key)!, nowMs))) return Effect.void; + return pullRequests + .batchesReads(referenceOf(groups.get(keys[0]!)![0]!)) + .pipe( + Effect.flatMap((batches) => + Effect.forEach( + keys, + (key) => (batches ? batchedReads : unbatchedReads).withPermits(1)(readIfDue(key)), + { concurrency: "unbounded", 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 ba9bcbfcde182..cd9366f77e1ef 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -2949,11 +2949,15 @@ export const make = Effect.gen(function* () { ); }; - const batchesReads: PullRequestService["Service"]["batchesReads"] = (input) => - requireProject(input).pipe( - Effect.map((project) => project.api.batchesReads === true), - Effect.orElseSucceed(() => false), - ); + 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( From 30d132a5aa67e1bea8cd15f95cfdffab746b57bd Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 00:23:45 +0000 Subject: [PATCH 5/6] fix(server): PR sync checks each group's due state when its read can start Each host's groups now run in a loop bounded by that host's width, with the shared permit taken inside the read, so a group a sibling flags for a stack retry mid-sweep is still read in the same sweep. Hosts paused at sweep start skip the batching lookup. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 59 ++++++++++++++ .../PullRequestSyncReactor.ts | 78 ++++++++++--------- 2 files changed, 101 insertions(+), 36 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 3e4c72bc6701f..7a9abb20b43ac 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1148,6 +1148,65 @@ describe("PullRequestSyncReactor", () => { ), ); + 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((number) => ({ + number, + headBranch: `layer-${number}`, + state: "open" as const, + })), + }); + // #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([ + ...Array.from({ length: 25 }, (_, index) => + makeThread(`thread-${index + 1}`, { pullRequests: [makeLink(index + 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 4b0fa89515275..34f120feba2a9 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -579,49 +579,55 @@ export const make = Effect.gen(function* () { return Effect.void; }), ); - const readIfDue = (key: string) => { - const entries = groups.get(key)!; + 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 readIfDue = ( + key: string, + entries: ReadonlyArray, + reads: Semaphore.Semaphore, + ) => { if (!isDue(key, entries, nowMs)) return Effect.void; const pauseKey = pauseKeyOf(entries[0]!); // 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 reads.withPermits(1)( + Clock.currentTimeMillis.pipe( + Effect.flatMap((startedAtMs) => + (pausedUntil.get(pauseKey) ?? 0) > startedAtMs + ? Effect.void + : readGroup(key, entries, pauseKey), + ), ), ); }; - const keysByHost = new Map>(); - for (const [key, entries] of groups) { - if (scope === "requested" && !requested.has(key)) continue; - const pauseKey = pauseKeyOf(entries[0]!); - const keys = keysByHost.get(pauseKey) ?? []; - keys.push(key); - keysByHost.set(pauseKey, keys); - } - // Wider than one summary batch, so reads that finish their cache checks at different times - // still share batches. - const batchedReads = yield* Semaphore.make(100); - // One read per pull request, often a CLI process, so these keep the narrower width. - const unbatchedReads = yield* Semaphore.make(25); + const groupsByHost = Map.groupBy( + Array.from(groups).filter(([key]) => scope === "all" || requested.has(key)), + ([, entries]) => pauseKeyOf(entries[0]!), + ); yield* Effect.forEach( - keysByHost.values(), - (keys) => { - // Stack retries only flag groups on the flagging group's host, so a host with nothing - // due yet stays that way this sweep. - if (!keys.some((key) => isDue(key, groups.get(key)!, nowMs))) return Effect.void; - return pullRequests - .batchesReads(referenceOf(groups.get(keys[0]!)![0]!)) - .pipe( - Effect.flatMap((batches) => - Effect.forEach( - keys, - (key) => (batches ? batchedReads : unbatchedReads).withPermits(1)(readIfDue(key)), - { concurrency: "unbounded", discard: true }, - ), - ), - ); + groupsByHost, + ([pauseKey, hostGroups]) => { + // Stack retries only flag groups on the flagging group's own host, so a host with + // nothing due now reads nothing this sweep. + if ( + (pausedUntil.get(pauseKey) ?? 0) > nowMs || + !hostGroups.some(([key, entries]) => isDue(key, entries, nowMs)) + ) { + return Effect.void; + } + 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, reads), { + concurrency: width, + discard: true, + }); + }), + ); }, { concurrency: "unbounded", discard: true }, ); From a79ca6246e1eeeb0f0225028e439f896ee521297 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 00:29:38 +0000 Subject: [PATCH 6/6] fix(server): PR sync judges a group due once it holds a shared slot A host's own loop can have room while the shared pool is full, so the due check now runs inside the permit. A stack retry a sibling flags while another host holds the slots still counts this sweep. Co-Authored-By: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 14 +++--- .../PullRequestSyncReactor.ts | 44 +++++++++++-------- 2 files changed, 32 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 7a9abb20b43ac..18c1811d72c49 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1157,11 +1157,7 @@ describe("PullRequestSyncReactor", () => { number: 9, url: "https://github.com/owner/repository/stack/9", base: "main", - layers: layers.map((number) => ({ - number, - headBranch: `layer-${number}`, - state: "open" as const, - })), + layers: layers.map(stackLayer), }); // #50 is merged and not due until #1's stack read shows it a different layer list. const merged = makeLink( @@ -1174,9 +1170,13 @@ describe("PullRequestSyncReactor", () => { const synced1 = yield* Deferred.make(); const fixture = yield* makeHarness({ snapshot: makeSnapshot([ - ...Array.from({ length: 25 }, (_, index) => - makeThread(`thread-${index + 1}`, { pullRequests: [makeLink(index + 1)] }), + // 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, diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 34f120feba2a9..b78e56f7d8d89 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -586,24 +586,26 @@ export const make = Effect.gen(function* () { // 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, - ) => { - if (!isDue(key, entries, nowMs)) return Effect.void; - const pauseKey = pauseKeyOf(entries[0]!); - // Check pauses at each read so changes during the sweep apply immediately. - return reads.withPermits(1)( - Clock.currentTimeMillis.pipe( - Effect.flatMap((startedAtMs) => - (pausedUntil.get(pauseKey) ?? 0) > startedAtMs - ? Effect.void - : readGroup(key, entries, pauseKey), - ), + ) => + 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]!), @@ -611,10 +613,10 @@ export const make = Effect.gen(function* () { yield* Effect.forEach( groupsByHost, ([pauseKey, hostGroups]) => { - // Stack retries only flag groups on the flagging group's own host, so a host with - // nothing due now reads nothing this sweep. + // 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 ( - (pausedUntil.get(pauseKey) ?? 0) > nowMs || + isPaused(pauseKey, nowMs) || !hostGroups.some(([key, entries]) => isDue(key, entries, nowMs)) ) { return Effect.void; @@ -622,10 +624,14 @@ export const make = Effect.gen(function* () { 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, reads), { - concurrency: width, - discard: true, - }); + return Effect.forEach( + hostGroups, + ([key, entries]) => readIfDue(key, entries, pauseKey, reads), + { + concurrency: width, + discard: true, + }, + ); }), ); },