diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 0161054781567..9b5174fdc6fdb 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -302,6 +302,24 @@ const sweepAgain = Effect.fn("sweepPullRequestSyncHarness")(function* ( yield* reactor.drain; }); +const sweepFor = Effect.fn("sweepPullRequestSyncHarnessFor")(function* ( + fixture: Harness, + reactor: PullRequestSyncReactor.PullRequestSyncReactor["Service"], + minutes: number, +) { + for (let minute = 0; minute < minutes; minute++) yield* sweepAgain(fixture, reactor); +}); + +const failingGitLabLink = makeLink(13, null, { + host: "gitlab.com", + url: "https://gitlab.com/owner/repository/-/merge_requests/13", +}); + +const gitLabCliFailure = new PullRequestOperationError({ + operation: "summary", + detail: "GitLab CLI command failed", +}); + function runUpdated( threadId: ThreadId, status: OrchestrationV2Run["status"], @@ -1519,6 +1537,76 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect( + "caps the failed-read backoff at the slow cadence and restarts it after a success", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const failing = yield* Ref.make(true); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("one", { pullRequests: [failingGitLabLink] })]), + summary: (input) => + Ref.get(failing).pipe( + Effect.flatMap((fails) => + fails ? Effect.fail(gitLabCliFailure) : Effect.succeed(makeSummary(input)), + ), + ), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + const readMinutes = [0]; + const sweepMinutes = Effect.fnUntraced(function* (from: number, to: number) { + for (let minute = from; minute <= to; minute++) { + const before = (yield* Ref.get(fixture.summaryCalls)).length; + yield* sweepAgain(fixture, reactor); + if ((yield* Ref.get(fixture.summaryCalls)).length > before) + readMinutes.push(minute); + } + }); + + yield* sweepMinutes(1, 60); + assert.deepStrictEqual(readMinutes, [0, 1, 3, 7, 15, 30, 45, 60]); + + yield* Ref.set(failing, false); + yield* sweepMinutes(61, 75); + yield* Ref.set(failing, true); + readMinutes.length = 0; + yield* sweepMinutes(76, 80); + assert.deepStrictEqual(readMinutes, [76, 77, 79]); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("reads a backed-off link at once when a refresh asks, then keeps backing off", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("one", { pullRequests: [failingGitLabLink] })]), + summary: () => Effect.fail(gitLabCliFailure), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* sweepFor(fixture, reactor, 4); + const backedOffReads = (yield* Ref.get(fixture.summaryCalls)).length; + + yield* reactor.requestSync(failingGitLabLink); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, backedOffReads + 1); + + // The refresh failed at minute 4 with the backoff doubled to 8 minutes. + yield* sweepFor(fixture, reactor, 7); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, backedOffReads + 1); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, backedOffReads + 2); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect.each([ "gh pr merge 7 --squash 2>&1 | tail -3", "cafe pr merge 7 --squash", diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 9dffe4babb424..19ce7de1f8ddb 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -47,6 +47,8 @@ const QUIET_SYNC_INTERVAL_MS = 5 * 60 * 1_000; // A pull request closed this long ago is rarely reopened, so the sweep stops reading it, as it // does merged ones. A reader that sees it open, or a requested refresh, still reads it at once. const CLOSED_FINAL_AFTER_MS = 24 * 60 * 60 * 1_000; +// A link whose host read failed waits a sweep, then twice as long after each further failure. +const FAILED_READ_BACKOFF_MS = 60 * 1_000; const isChangeRequestCloseCommand = (command: string) => sourceControlDefinitions.definitions.some( @@ -346,10 +348,12 @@ 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(); + const failedReads = new Map(); // Pause per project and host because credentials are project-specific; leave paused PRs due until reset. const pausedUntil = new Map(); const isDue = (key: string, entries: ReadonlyArray, nowMs: number): boolean => { + if ((failedReads.get(key)?.retryAtMs ?? 0) > nowMs) return false; if (requested.has(key) || retryStacks.has(key)) return true; if (entries.some((entry) => entry.link.snapshot === null)) return true; const active = entries.filter((entry) => isUnsettled(entry.thread)); @@ -397,12 +401,17 @@ export const make = Effect.gen(function* () { // A host-level refresh reaches every project, each retaining its own pending retry. for (const [key, entries] of groups) { - if (pendingRequests.has(threadPullRequestKeyOf(entries[0]!.link))) requested.add(key); + if (!pendingRequests.has(threadPullRequestKeyOf(entries[0]!.link))) continue; + requested.add(key); + // A new request reads at once; a failure after it backs off further, not from the start. + const failed = failedReads.get(key); + if (failed !== undefined) failedReads.set(key, { ...failed, retryAtMs: nowMs }); } pendingRequests.clear(); for (const key of lastSyncedAt.keys()) if (!groups.has(key)) lastSyncedAt.delete(key); for (const key of retryStacks) if (!groups.has(key)) retryStacks.delete(key); + for (const key of failedReads.keys()) if (!groups.has(key)) failedReads.delete(key); for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key); const linkedThisSweep = new Set(); @@ -562,6 +571,7 @@ export const make = Effect.gen(function* () { } // The host answered, so the cadence clock ticks even if a dispatch below is rejected. lastSyncedAt.set(key, nowMs); + failedReads.delete(key); requested.delete(key); yield* Effect.forEach( entries, @@ -587,6 +597,13 @@ export const make = Effect.gen(function* () { const retryAt = rateLimitRetryAt(cause); if (retryAt !== undefined) { pausedUntil.set(pauseKey, Math.max(retryAt, pausedUntil.get(pauseKey) ?? 0)); + } else { + const previous = failedReads.get(key); + const backoffMs = + previous === undefined + ? FAILED_READ_BACKOFF_MS + : Math.min(previous.backoffMs * 2, SLOW_SYNC_INTERVAL_MS); + failedReads.set(key, { retryAtMs: nowMs + backoffMs, backoffMs }); } const reason = skipReason(cause); const skip = skips.get(reason);