authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-05 15:07:38-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 15:07:38-07:00
log54b6b667f340c03132e9fc2cb2c3d657571442a2
tree09b0ba23179efcb108ae5d95b00b2814d17a8172
parent3d2a98f99be10b7621a368f4e05f36a853ac48d1
signature Signed by PGP key B5690EEEBB952194

fix(server): PR sync waits out a GitHub rate limit pause instead of failing every PR (#16203)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

2 files changed, 206 insertions(+), 11 deletions(-)

apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts+131
...@@ -19,15 +19,18 @@ import {...@@ -19,15 +19,18 @@ import {
19 type ThreadPullRequestSnapshot,19 type ThreadPullRequestSnapshot,
20} from "@t3tools/contracts";20} from "@t3tools/contracts";
21import { assert, describe, it } from "@effect/vitest";21import { assert, describe, it } from "@effect/vitest";
22import * as Clock from "effect/Clock";
22import * as Crypto from "effect/Crypto";23import * as Crypto from "effect/Crypto";
23import * as DateTime from "effect/DateTime";24import * as DateTime from "effect/DateTime";
24import * as Deferred from "effect/Deferred";25import * as Deferred from "effect/Deferred";
25import * as Effect from "effect/Effect";26import * as Effect from "effect/Effect";
26import * as Layer from "effect/Layer";27import * as Layer from "effect/Layer";
28import * as Logger from "effect/Logger";
27import * as Queue from "effect/Queue";29import * as Queue from "effect/Queue";
28import * as Ref from "effect/Ref";30import * as Ref from "effect/Ref";
29import { TestClock } from "effect/testing";31import { TestClock } from "effect/testing";
3032
33import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts";
31import * as PullRequestService from "../pullRequest/PullRequestService.ts";34import * as PullRequestService from "../pullRequest/PullRequestService.ts";
32import * as ServerActivation from "../serverActivation.ts";35import * as ServerActivation from "../serverActivation.ts";
33import * as Orchestrator from "./Orchestrator.ts";36import * as Orchestrator from "./Orchestrator.ts";
...@@ -908,6 +911,134 @@ describe("PullRequestSyncReactor", () => {...@@ -908,6 +911,134 @@ describe("PullRequestSyncReactor", () => {
908 ),911 ),
909 );912 );
910913
914 it.effect("leaves a rate limited host unread until its pause ends", () => {
915 const skips: Array<ReadonlyArray<unknown>> = [];
916 const logger = Logger.make(({ logLevel, message }) => {
917 const parts = Array.isArray(message) ? message : [message];
918 if (logLevel === "Warn" && parts[0] === "pull request sync skipped") skips.push(parts);
919 });
920 const retryAt = Date.parse(NOW) + 3 * 60_000;
921 return Effect.scoped(
922 Effect.gen(function* () {
923 yield* TestClock.setTime(Date.parse(NOW));
924 const fixture = yield* makeHarness({
925 snapshot: makeSnapshot([
926 makeThread("first", { pullRequests: [makeLink(7, { state: "open" })] }),
927 makeThread("second", { pullRequests: [makeLink(8, { state: "open" })] }),
928 makeThread("third", {
929 pullRequests: [
930 makeLink(
931 9,
932 { state: "open" },
933 { host: "forge.example", url: "https://forge.example/owner/repository/pulls/9" },
934 ),
935 ],
936 }),
937 ]),
938 summary: (input) =>
939 Effect.gen(function* () {
940 if (input.host === "github.com" && (yield* Clock.currentTimeMillis) < retryAt) {
941 const paused = new PullRequestProviderError({
942 provider: "github",
943 operation: "getChangeRequestSummary",
944 reason: "rate-limited",
945 detail: "paused",
946 retryAt,
947 });
948 return yield* new PullRequestOperationError({
949 operation: "summary",
950 detail: "paused",
951 cause: paused,
952 });
953 }
954 return makeSummary(input, { state: "open" });
955 }),
956 });
957 const githubReads = Ref.get(fixture.summaryCalls).pipe(
958 Effect.map((calls) =>
959 calls.filter((call) => call.host === "github.com").map((call) => call.number),
960 ),
961 );
962
963 yield* Effect.gen(function* () {
964 // The first refused read pauses the host, so the sweep does not try the other.
965 const reactor = yield* startAndSweep(fixture);
966 assert.deepStrictEqual(yield* githubReads, [7]);
967 assert.strictEqual(skips.length, 1);
968 assert.deepInclude(skips[0]![1], { count: 1 });
969
970 // A requested refresh waits for the pause like the sweep does.
971 yield* reactor.requestSync({
972 host: "github.com",
973 repository: "owner/repository",
974 number: 7,
975 });
976 yield* Queue.take(fixture.snapshotReads);
977 yield* reactor.drain;
978 yield* sweepAgain(fixture, reactor);
979 yield* sweepAgain(fixture, reactor);
980 assert.deepStrictEqual(yield* githubReads, [7]);
981 assert.strictEqual(skips.length, 1);
982 // Other hosts keep their cadence.
983 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 4);
984
985 yield* sweepAgain(fixture, reactor);
986 assert.sameMembers((yield* githubReads).slice(1), [7, 8]);
987 }).pipe(
988 // The reactor forks its worker while its layer builds, so the logger must reach it there.
989 Effect.provide(
990 fixture.layer.pipe(Layer.provide(Logger.layer([logger], { mergeWithExisting: false }))),
991 ),
992 );
993 }),
994 );
995 });
996
997 it.effect("pauses the host when only its stack read is rate limited", () =>
998 Effect.scoped(
999 Effect.gen(function* () {
1000 yield* TestClock.setTime(Date.parse(NOW));
1001 const retryAt = Date.parse(NOW) + 3 * 60_000;
1002 const fixture = yield* makeHarness({
1003 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7, null)] })]),
1004 summary: (input) => Effect.succeed(makeSummary(input, { state: "open" })),
1005 stack: () =>
1006 Effect.gen(function* () {
1007 if ((yield* Clock.currentTimeMillis) >= retryAt) return null;
1008 return yield* new PullRequestOperationError({
1009 operation: "stack",
1010 detail: "paused",
1011 cause: new PullRequestProviderError({
1012 provider: "github",
1013 operation: "getChangeRequestStack",
1014 reason: "rate-limited",
1015 detail: "paused",
1016 retryAt,
1017 }),
1018 });
1019 }),
1020 });
1021 const reads = Effect.all([
1022 Ref.get(fixture.summaryCalls).pipe(Effect.map((calls) => calls.length)),
1023 Ref.get(fixture.stackCalls).pipe(Effect.map((calls) => calls.length)),
1024 ]);
1025
1026 yield* Effect.gen(function* () {
1027 const reactor = yield* startAndSweep(fixture);
1028 assert.deepStrictEqual(yield* reads, [1, 1]);
1029 yield* sweepAgain(fixture, reactor);
1030 yield* sweepAgain(fixture, reactor);
1031 assert.deepStrictEqual(yield* reads, [1, 1]);
1032
1033 // At retryAt the pull request is read again, stack included.
1034 yield* sweepAgain(fixture, reactor);
1035 assert.deepStrictEqual(yield* reads, [2, 2]);
1036 assert.strictEqual((yield* Ref.get(fixture.syncCommands)).length, 1);
1037 }).pipe(Effect.provide(fixture.layer));
1038 }),
1039 ),
1040 );
1041
911 it.effect("reads open links fresh only when a run that ran a merge command ends", () =>1042 it.effect("reads open links fresh only when a run that ran a merge command ends", () =>
912 Effect.scoped(1043 Effect.scoped(
913 Effect.gen(function* () {1044 Effect.gen(function* () {
apps/server/src/orchestration-v2/PullRequestSyncReactor.ts+75-11
...@@ -16,16 +16,19 @@ import {...@@ -16,16 +16,19 @@ import {
16 visibleThreadPullRequests,16 visibleThreadPullRequests,
17} from "@t3tools/shared/threadPullRequests";17} from "@t3tools/shared/threadPullRequests";
18import * as Cause from "effect/Cause";18import * as Cause from "effect/Cause";
19import * as Clock from "effect/Clock";
19import * as Context from "effect/Context";20import * as Context from "effect/Context";
20import * as Crypto from "effect/Crypto";21import * as Crypto from "effect/Crypto";
21import * as DateTime from "effect/DateTime";22import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";23import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";24import * as Layer from "effect/Layer";
25import * as Schema from "effect/Schema";
24import * as Schedule from "effect/Schedule";26import * as Schedule from "effect/Schedule";
25import type * as Scope from "effect/Scope";27import type * as Scope from "effect/Scope";
26import * as Semaphore from "effect/Semaphore";28import * as Semaphore from "effect/Semaphore";
27import * as Stream from "effect/Stream";29import * as Stream from "effect/Stream";
2830
31import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts";
29import * as PullRequestService from "../pullRequest/PullRequestService.ts";32import * as PullRequestService from "../pullRequest/PullRequestService.ts";
30import { forkParked } from "../serverActivation.ts";33import { forkParked } from "../serverActivation.ts";
31import * as Orchestrator from "./Orchestrator.ts";34import * as Orchestrator from "./Orchestrator.ts";
...@@ -36,6 +39,8 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;...@@ -36,6 +39,8 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
36/** Shell commands that can merge or close a pull request without a merge notification. */39/** Shell commands that can merge or close a pull request without a merge notification. */
37const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u;40const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u;
3841
42const isPullRequestProviderError = Schema.is(PullRequestProviderError);
43
39type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;44type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;
4045
41interface LinkEntry {46interface LinkEntry {
...@@ -107,6 +112,21 @@ function stacksEqual(...@@ -107,6 +112,21 @@ function stacksEqual(
107 );112 );
108}113}
109114
115function skipReason(cause: Cause.Cause<unknown>): string {
116 const error = Cause.squash(cause);
117 return error instanceof Error ? error.message : String(error);
118}
119
120/** When a host read failed because the host is rate limited, the time that pause ends. */
121function rateLimitRetryAt(cause: Cause.Cause<unknown>): number | undefined {
122 let error: unknown = Cause.squash(cause);
123 while (error instanceof Error) {
124 if (isPullRequestProviderError(error) && error.reason === "rate-limited") return error.retryAt;
125 error = error.cause;
126 }
127 return undefined;
128}
129
110function isUnsettled(thread: ProjectionStore.ProjectionThreadPullRequests): boolean {130function isUnsettled(thread: ProjectionStore.ProjectionThreadPullRequests): boolean {
111 return thread.settledOverride !== "settled" && thread.settledAt === null;131 return thread.settledOverride !== "settled" && thread.settledAt === null;
112}132}
...@@ -122,7 +142,10 @@ export class PullRequestSyncReactor extends Context.Service<...@@ -122,7 +142,10 @@ export class PullRequestSyncReactor extends Context.Service<
122 {142 {
123 readonly start: () => Effect.Effect<void, never, Scope.Scope>;143 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
124 readonly drain: Effect.Effect<void>;144 readonly drain: Effect.Effect<void>;
125 /** Force the next sweep to re-read this pull request, even when its snapshot is terminal. */145 /**
146 * Force the next sweep to re-read this pull request, even when its snapshot is terminal.
147 * While its host is rate limited, the read waits for the first sweep after the pause.
148 */
126 readonly requestSync: (key: ThreadPullRequestKey) => Effect.Effect<void>;149 readonly requestSync: (key: ThreadPullRequestKey) => Effect.Effect<void>;
127 }150 }
128>()("t3/orchestration-v2/PullRequestSyncReactor") {}151>()("t3/orchestration-v2/PullRequestSyncReactor") {}
...@@ -141,6 +164,10 @@ export const make = Effect.gen(function* () {...@@ -141,6 +164,10 @@ export const make = Effect.gen(function* () {
141 // linking dozens of pull requests) is read together and shares the summary batches.164 // linking dozens of pull requests) is read together and shares the summary batches.
142 let requestedSweepQueued = false;165 let requestedSweepQueued = false;
143 const retryStacks = new Set<string>();166 const retryStacks = new Set<string>();
167 // Rate limit pauses by project and host, since each project reads with its own credential.
168 // A paused host refuses every read without asking it, so the sweep leaves its pull requests
169 // due until the pause ends rather than failing each of them every minute.
170 const pausedUntil = new Map<string, number>();
144171
145 const isDue = (key: string, entries: ReadonlyArray<LinkEntry>, nowMs: number): boolean => {172 const isDue = (key: string, entries: ReadonlyArray<LinkEntry>, nowMs: number): boolean => {
146 if (requested.has(key) || retryStacks.has(key)) return true;173 if (requested.has(key) || retryStacks.has(key)) return true;
...@@ -279,10 +306,15 @@ export const make = Effect.gen(function* () {...@@ -279,10 +306,15 @@ export const make = Effect.gen(function* () {
279 })),306 })),
280 Effect.catchCauseIf(307 Effect.catchCauseIf(
281 (cause) => !Cause.hasInterruptsOnly(cause),308 (cause) => !Cause.hasInterruptsOnly(cause),
282 () =>309 (cause) =>
283 Effect.logWarning("pull request stack lookup failed", {310 rateLimitRetryAt(cause) !== undefined
284 key,311 ? // The sweep records the pause and holds the host's other reads until it ends.
285 }).pipe(Effect.as(null)),312 Effect.sync(() => retryStacks.add(key)).pipe(
313 Effect.andThen(Effect.failCause(cause)),
314 )
315 : Effect.logWarning("pull request stack lookup failed", {
316 key,
317 }).pipe(Effect.as(null)),
286 ),318 ),
287 );319 );
288 if (needsStack) {320 if (needsStack) {
...@@ -312,18 +344,50 @@ export const make = Effect.gen(function* () {...@@ -312,18 +344,50 @@ export const make = Effect.gen(function* () {
312 );344 );
313 });345 });
314346
347 // Failed host reads by reason: how many, and the first key that failed that way.
348 const skips = new Map<string, { count: number; readonly key: string }>();
349 const readGroup = (key: string, entries: ReadonlyArray<LinkEntry>, pauseKey: string) =>
350 syncGroup(key, entries).pipe(
351 Effect.catchCause((cause) => {
352 if (Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause);
353 const retryAt = rateLimitRetryAt(cause);
354 if (retryAt !== undefined) {
355 pausedUntil.set(pauseKey, Math.max(retryAt, pausedUntil.get(pauseKey) ?? 0));
356 }
357 const reason = skipReason(cause);
358 const skip = skips.get(reason);
359 if (skip === undefined) skips.set(reason, { count: 1, key });
360 else skip.count += 1;
361 return Effect.void;
362 }),
363 );
315 yield* Effect.forEach(364 yield* Effect.forEach(
316 groups,365 groups,
317 ([key, entries]) =>366 ([key, entries]) => {
318 (scope === "all" || requested.has(key)) && isDue(key, entries, nowMs)367 if (!((scope === "all" || requested.has(key)) && isDue(key, entries, nowMs))) {
319 ? syncGroup(key, entries).pipe(368 return Effect.void;
320 Effect.catchCause(logSkipped("pull request sync skipped", { key })),369 }
321 )370 const first = entries[0]!;
322 : Effect.void,371 const pauseKey = `${first.thread.projectId}\0${normalizeThreadPullRequestKey(first.link).host}`;
372 // Checked against the clock as each read starts, so a pause found earlier in this sweep
373 // holds the rest, and one that ends during the sweep lets the rest through.
374 return Clock.currentTimeMillis.pipe(
375 Effect.flatMap((startedAtMs) =>
376 (pausedUntil.get(pauseKey) ?? 0) > startedAtMs
377 ? Effect.void
378 : readGroup(key, entries, pauseKey),
379 ),
380 );
381 },
323 // As wide as one batched summary read, so the sweep's reads on a host arrive together and382 // As wide as one batched summary read, so the sweep's reads on a host arrive together and
324 // GitHub answers them in one request rather than one `gh pr view` apiece.383 // GitHub answers them in one request rather than one `gh pr view` apiece.
325 { concurrency: 25, discard: true },384 { concurrency: 25, discard: true },
326 );385 );
386 // A host failure such as a signed-out CLI fails every due pull request the same way, so a
387 // sweep reports one line per reason rather than one per pull request.
388 for (const [reason, { count, key }] of skips) {
389 yield* Effect.logWarning("pull request sync skipped", { count, key, reason });
390 }
327 });391 });
328392
329 const worker = yield* makeDrainableWorker((scope: "all" | "requested") =>393 const worker = yield* makeDrainableWorker((scope: "all" | "requested") =>