Skip to content
Open
88 changes: 88 additions & 0 deletions apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"],
Expand Down Expand Up @@ -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",
Expand Down
19 changes: 18 additions & 1 deletion apps/server/src/orchestration-v2/PullRequestSyncReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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<string>();
const failedReads = new Map<string, { readonly retryAtMs: number; readonly backoffMs: number }>();
// Pause per project and host because credentials are project-specific; leave paused PRs due until reset.
const pausedUntil = new Map<string, number>();

const isDue = (key: string, entries: ReadonlyArray<LinkEntry>, 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));
Expand Down Expand Up @@ -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<string>();
Expand Down Expand Up @@ -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,
Expand All @@ -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 {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After a non-rate-limit failure this always records failedReads (1m, then doubling to the 15m slow interval). The PR body says interruptions are excluded from backoff, but the branch visible here only special-cases rateLimitRetryAt.

Does an Effect interruption (for example stopping the reactor mid-sweep, or a fiber interrupt while glab is still running) get filtered before this else, or can it leave the link backed off until a later successful read clears the map? If interruptions already bail out earlier this is fine — asking because a restart would otherwise make a flaky host look "paused" for up to 15 minutes with no host pause recorded.

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);
Expand Down
Loading