From f3821b7f117d062258c27ee82c6964899800c7c0 Mon Sep 17 00:00:00 2001 From: abi <36452265+abiassi@users.noreply.github.com> Date: Sat, 10 Oct 2026 12:24:32 +0100 Subject: [PATCH 1/5] test(server): show PR sync re-reading pull requests the host cannot find Co-authored-by: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 183 +++++++++++++++++- 1 file changed, 179 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 19e67f949b611..f634f5dc49a7c 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -38,7 +38,7 @@ import * as ServerActivation from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts"; -import type { PullRequestTestThread } from "./testkit/pullRequestFixtures.ts"; +import { type PullRequestTestThread, v2PullRequestThread } from "./testkit/pullRequestFixtures.ts"; const NOW = "2026-08-28T12:00:00.000Z"; const PROJECT_ID = ProjectId.make("sync-project"); @@ -162,6 +162,17 @@ function makeSummary( }; } +/** How a host answers for a pull request it cannot find, such as one in a deleted repository. */ +function notFound(input: PullRequestRef) { + return Effect.fail( + new PullRequestOperationError({ + operation: "summary", + detail: `GitHub could not find ${input.repository}#${input.number}.`, + reason: "not-found", + }), + ); +} + interface HarnessOptions { readonly invalidate?: PullRequestService.PullRequestService["Service"]["invalidate"]; readonly snapshot: TestShellSnapshot; @@ -363,6 +374,16 @@ function commandRan(threadId: ThreadId, input: string): OrchestrationV2DomainEve }; } +function pullRequestsSynced(thread: PullRequestTestThread): OrchestrationV2DomainEvent { + return { + type: "thread.pull-request-synced", + id: EventId.make(`event:${thread.id}:pull-requests`), + threadId: thread.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: { ...v2PullRequestThread(thread), lastVisitedAt: null }, + }; +} + /** What the reactor would have persisted, so the next sweep sees its own writes. */ function applySync( snapshot: TestShellSnapshot, @@ -862,7 +883,11 @@ describe("PullRequestSyncReactor", () => { makeThread("settled", { settledOverride: "settled", settledAt: "2026-08-21T00:00:00.000Z", - pullRequests: [makeLink(5, { state: "open" }), makeLink(6, { state: "closed" })], + pullRequests: [ + makeLink(5, { state: "open" }), + makeLink(6, { state: "closed" }), + makeLink(7), + ], }), ]), }); @@ -883,7 +908,7 @@ describe("PullRequestSyncReactor", () => { yield* sweepAgain(fixture, reactor); assert.deepStrictEqual( (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number).toSorted(), - [5, 6], + [5, 6, 7], ); }).pipe(Effect.provide(fixture.layer)); }), @@ -970,12 +995,162 @@ describe("PullRequestSyncReactor", () => { }); yield* Effect.gen(function* () { - yield* startAndSweep(fixture); + const reactor = yield* startAndSweep(fixture); assert.deepStrictEqual( (yield* Ref.get(fixture.syncCommands)).map((command) => command.number), [8], ); + yield* sweepAgain(fixture, reactor); + assert.strictEqual( + (yield* Ref.get(fixture.summaryCalls)).filter((call) => call.number === 7).length, + 2, + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect( + "asks every fifteen minutes for pull requests the host cannot find, until it finds them", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const found = yield* Ref.make(false); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("renamed", { + pullRequests: [makeLink(9), makeLink(10, { state: "open" })], + }), + ]), + summary: (input) => + Ref.get(found).pipe( + Effect.flatMap((isFound) => + isFound ? Effect.succeed(makeSummary(input)) : notFound(input), + ), + ), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + + // The repository is back, so its open pull requests return to the per-minute reads. + yield* Ref.set(found, true); + yield* sweepAgain(fixture, reactor); + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)) + .map((call) => call.number) + .toSorted((left, right) => left - right), + [9, 9, 10, 10], + ); + const commands = yield* Ref.get(fixture.syncCommands); + yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands)); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 6); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("keeps the slow cadence for a missing pull request whose stack read failed", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const found = yield* Ref.make(true); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("renamed", { pullRequests: [makeLink(4)] })]), + summary: (input) => + Ref.get(found).pipe( + Effect.flatMap((isFound) => + isFound ? Effect.succeed(makeSummary(input)) : notFound(input), + ), + ), + stack: () => + Effect.fail(new PullRequestOperationError({ operation: "stack", detail: "host down" })), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + // The next sweep retries the failed stack read, and finds the repository gone. + yield* Ref.set(found, false); + for (let index = 0; index < 15; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("stops asking for a requested pull request once the host cannot find it", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("settled", { + settledOverride: "settled", + settledAt: "2026-08-21T00:00:00.000Z", + pullRequests: [makeLink(5, { state: "open" })], + }), + ]), + summary: notFound, + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* Queue.offer(fixture.stateChanges, { + host: "github.com", + repository: "owner/repository", + number: 5, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1); + + for (let index = 0; index < 15; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("reads a missing pull request again for a reader, not for another link's sync", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("renamed", { pullRequests: [makeLink(9)] })]), + summary: (input) => + input.number === 9 ? notFound(input) : Effect.succeed(makeSummary(input)), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + // Linking another pull request syncs the thread's links, the unreadable one included. + const linked = makeThread("renamed", { pullRequests: [makeLink(9), makeLink(8)] }); + yield* Ref.set(fixture.snapshots, makeSnapshot([linked], 2)); + yield* Queue.offer(fixture.domainEvents, pullRequestsSynced(linked)); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number), + [9, 8], + ); + + yield* Queue.offer(fixture.stateChanges, { + host: "github.com", + repository: "owner/repository", + number: 9, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number), + [9, 8, 9], + ); }).pipe(Effect.provide(fixture.layer)); }), ), From 58fd2755d198096b02b42149e2dfe80f6010f156 Mon Sep 17 00:00:00 2001 From: abi <36452265+abiassi@users.noreply.github.com> Date: Sat, 10 Oct 2026 12:24:33 +0100 Subject: [PATCH 2/5] fix(server): PR sync stops reading missing pull requests every minute Co-authored-by: Claude Opus 5.5 --- .../PullRequestSyncReactor.ts | 33 +++++++++++++++---- 1 file changed, 26 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index a966cbd954561..6c0c753813464 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -166,22 +166,27 @@ 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(); + // Pull requests the host last answered it cannot find, such as one in a deleted repository. + const missing = new Set(); // Rate limit pauses by project and host, since each project reads with its own credential. // A paused host refuses every read without asking it, so the sweep leaves its pull requests // due until the pause ends rather than failing each of them every minute. const pausedUntil = new Map(); const isDue = (key: string, entries: ReadonlyArray, nowMs: number): boolean => { - if (requested.has(key) || retryStacks.has(key)) return true; - if (entries.some((entry) => entry.link.snapshot === null)) return true; + if (requested.has(key) || (retryStacks.has(key) && !missing.has(key))) return true; // Settled threads stop watching their pull requests. Unsettling one makes its links due on // the next sweep, since the cadence clock below kept running while it was settled. const active = entries.filter((entry) => isUnsettled(entry.thread)); + if (active.length === 0) return false; + // Closed requests can reopen on the host, and a missing one can come back. + const last = lastSyncedAt.get(key); + const slowDue = last === undefined || nowMs - last >= SLOW_SYNC_INTERVAL_MS; + if (missing.has(key)) return slowDue; + if (active.some((entry) => entry.link.snapshot === null)) return true; if (active.every((entry) => entry.link.snapshot?.state === "merged")) return false; if (active.some((entry) => entry.link.snapshot?.state === "open")) return true; - // Closed requests can reopen on the host. - const last = lastSyncedAt.get(key); - return last === undefined || nowMs - last >= SLOW_SYNC_INTERVAL_MS; + return slowDue; }; const logSkipped = @@ -208,6 +213,7 @@ export const make = Effect.gen(function* () { 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 missing) if (!groups.has(key)) missing.delete(key); for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key); // Layers auto-linked this sweep, so two links of one thread that share a @@ -286,7 +292,19 @@ export const make = Effect.gen(function* () { }; const generation = requested.get(key); if (generation !== undefined) yield* pullRequests.invalidate({ reference: ref }); - const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }); + const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }).pipe( + Effect.tapError((error) => + error._tag === "PullRequestOperationError" && error.reason === "not-found" + ? // The host answered, so the request is done and the cadence clock ticks. + Effect.sync(() => { + missing.add(key); + lastSyncedAt.set(key, nowMs); + if (requested.get(key) === generation) requested.delete(key); + }) + : Effect.void, + ), + ); + missing.delete(key); const fields = snapshotFieldsOf(summary); const needsStack = generation !== undefined || @@ -440,10 +458,11 @@ export const make = Effect.gen(function* () { yield* forkParked( Stream.runForEach(events, (event) => { switch (event.type) { + // Another link's sync is no reason to ask again for one the host cannot find. case "thread.pull-request-synced": return Effect.forEach( visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( - (link) => link.snapshot === null, + (link) => link.snapshot === null && !missing.has(threadPullRequestKeyOf(link)), ), requestSync, { discard: true }, From 67a7e7722695f44bdf18381e6a165141e190db3f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 06:24:59 +0000 Subject: [PATCH 3/5] test(server): cover missing PR cadence and refresh triggers --- .../PullRequestSyncReactor.test.ts | 163 +++++++++++++++++- 1 file changed, 160 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index ea8080f115870..12a66a6864584 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -39,7 +39,7 @@ import * as ServerActivation from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts"; -import type { PullRequestTestThread } from "./testkit/pullRequestFixtures.ts"; +import { type PullRequestTestThread, v2PullRequestThread } from "./testkit/pullRequestFixtures.ts"; const NOW = "2026-08-28T12:00:00.000Z"; const PROJECT_ID = ProjectId.make("sync-project"); @@ -162,6 +162,14 @@ function makeSummary( }; } +const missingSummary = Effect.fail( + new PullRequestOperationError({ + operation: "summary", + detail: "Pull request not found", + reason: "not-found", + }), +); + interface HarnessOptions { readonly invalidate?: PullRequestService.PullRequestService["Service"]["invalidate"]; readonly snapshot: TestShellSnapshot; @@ -892,7 +900,11 @@ describe("PullRequestSyncReactor", () => { makeThread("settled", { settledOverride: "settled", settledAt: "2026-08-21T00:00:00.000Z", - pullRequests: [makeLink(5, { state: "open" }), makeLink(6, { state: "closed" })], + pullRequests: [ + makeLink(5, { state: "open" }), + makeLink(6, { state: "closed" }), + makeLink(7), + ], }), ]), }); @@ -913,7 +925,7 @@ describe("PullRequestSyncReactor", () => { yield* sweepAgain(fixture, reactor); assert.deepStrictEqual( (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number).toSorted(), - [5, 6], + [5, 6, 7], ); }).pipe(Effect.provide(fixture.layer)); }), @@ -1011,6 +1023,151 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect( + "retries missing links every fifteen minutes and resumes normal reads after recovery", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const found = yield* Ref.make(false); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("missing", { pullRequests: [makeLink(9), makeLink(10, {})] }), + ]), + summary: (input) => + Ref.get(found).pipe( + Effect.flatMap((available) => + available ? Effect.succeed(makeSummary(input)) : missingSummary, + ), + ), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 4); + assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []); + + yield* Ref.set(found, true); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 4); + yield* sweepAgain(fixture, reactor); + const commands = yield* Ref.get(fixture.syncCommands); + assert.deepStrictEqual( + commands.map((command) => command.number), + [9], + ); + yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands)); + yield* sweepAgain(fixture, reactor); + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)) + .map((call) => call.number) + .toSorted((left, right) => left - right), + [9, 9, 9, 9, 10, 10, 10, 10], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect( + "holds missing links through sibling and link updates but honors a reader refresh", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("missing", { pullRequests: [makeLink(9)] })]), + summary: (input) => + input.number === 9 ? missingSummary : Effect.succeed(makeSummary(input)), + }); + const missingReads = Ref.get(fixture.summaryCalls).pipe( + Effect.map((calls) => calls.filter((call) => call.number === 9).length), + ); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + const linked = makeThread("missing", { pullRequests: [makeLink(9), makeLink(8)] }); + yield* Ref.set(fixture.snapshots, makeSnapshot([linked])); + yield* Queue.offer(fixture.domainEvents, { + type: "thread.pull-request-synced", + id: EventId.make("sibling-synced"), + threadId: linked.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: { ...v2PullRequestThread(linked), lastVisitedAt: null }, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + const afterSiblingSync = yield* missingReads; + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => command.number), + [8], + ); + + yield* Queue.offer(fixture.domainEvents, { + type: "thread.pull-request-link-updated", + id: EventId.make("missing-link-updated"), + threadId: linked.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: makeLink(9), + }); + // Let the event consumer finish even when it correctly queues no sweep. + yield* TestClock.adjust(0); + yield* reactor.drain; + const afterLinkUpdate = yield* missingReads; + + yield* Queue.offer(fixture.stateChanges, makeLink(9)); + yield* TestClock.adjust(0); + yield* reactor.drain; + const afterReaderRefresh = yield* missingReads; + yield* sweepAgain(fixture, reactor); + assert.deepStrictEqual( + { + afterSiblingSync, + afterLinkUpdate, + afterReaderRefresh, + afterSweep: yield* missingReads, + }, + { afterSiblingSync: 1, afterLinkUpdate: 1, afterReaderRefresh: 2, afterSweep: 2 }, + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("a pending stack retry respects the missing summary backoff", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const found = yield* Ref.make(true); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("missing", { pullRequests: [makeLink(4)] })]), + summary: (input) => + Ref.get(found).pipe( + Effect.flatMap((available) => + available ? Effect.succeed(makeSummary(input)) : missingSummary, + ), + ), + stack: () => + Effect.fail(new PullRequestOperationError({ operation: "stack", detail: "host down" })), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1); + yield* Ref.set(found, false); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 3); + assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("leaves a rate limited host unread until its pause ends", () => { const skips: Array> = []; const logger = Logger.make(({ logLevel, message }) => { From b8f7c4e2e931469d753f4abe281f46d81f463c43 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 06:27:00 +0000 Subject: [PATCH 4/5] fix(server): back off missing pull request sync reads --- .../PullRequestSyncReactor.ts | 34 +++++++++++++++---- 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 6bde4404fb7f1..3306ad363a25d 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -158,15 +158,20 @@ 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 missing = new Set(); // Rate limit pauses by project and host, since each project reads with its own credential. // A paused host refuses every read without asking it, so the sweep leaves its pull requests // due until the pause ends rather than failing each of them every minute. const pausedUntil = new Map(); const isDue = (key: string, entries: ReadonlyArray, nowMs: number): boolean => { - if (requested.has(key) || retryStacks.has(key)) return true; - if (entries.some((entry) => entry.link.snapshot === null)) return true; + if (requested.has(key)) return true; const active = entries.filter((entry) => isUnsettled(entry.thread)); + if (active.length === 0) return false; + const last = lastSyncedAt.get(key); + const slowDue = last === undefined || nowMs - last >= SLOW_SYNC_INTERVAL_MS; + if (missing.has(key)) return slowDue; + if (retryStacks.has(key) || active.some((entry) => entry.link.snapshot === null)) return true; // A merged snapshot without its merge time cannot settle the thread, so it keeps the slow // cadence until the host supplies one (Bitbucket snapshots saved before it was read). if ( @@ -177,8 +182,7 @@ export const make = Effect.gen(function* () { return false; } if (active.some((entry) => entry.link.snapshot?.state === "open")) return true; - const last = lastSyncedAt.get(key); - return last === undefined || nowMs - last >= SLOW_SYNC_INTERVAL_MS; + return slowDue; }; const logSkipped = @@ -204,6 +208,7 @@ export const make = Effect.gen(function* () { 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 missing) if (!groups.has(key)) missing.delete(key); for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key); const linkedThisSweep = new Set(); @@ -287,7 +292,19 @@ export const make = Effect.gen(function* () { if (generation !== undefined || missingMergedAt) { yield* pullRequests.invalidate({ reference: ref }); } - const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }); + const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }).pipe( + Effect.tapError((error) => + error.reason === "not-found" + ? Effect.sync(() => { + missing.add(key); + // Not-found is a host answer: consume only this refresh and start its backoff. + lastSyncedAt.set(key, nowMs); + if (requested.get(key) === generation) requested.delete(key); + }) + : Effect.void, + ), + ); + missing.delete(key); const fields = snapshotFieldsOf(summary); const needsStack = generation !== undefined || @@ -432,11 +449,14 @@ export const make = Effect.gen(function* () { Stream.runForEach(events, (event) => { switch (event.type) { case "thread.pull-request-link-updated": - return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void; + return event.payload.snapshot === null && + !missing.has(threadPullRequestKeyOf(event.payload)) + ? requestSync(event.payload) + : Effect.void; case "thread.pull-request-synced": return Effect.forEach( visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( - (link) => link.snapshot === null, + (link) => link.snapshot === null && !missing.has(threadPullRequestKeyOf(link)), ), requestSync, { discard: true }, From b41599d4928fff7cf4852c5ca68f54f49d97a968 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 06:55:51 +0000 Subject: [PATCH 5/5] fix(server): clear missing PR backoff when reading project changes --- .../PullRequestSyncReactor.test.ts | 37 +++++++++++++++ .../PullRequestSyncReactor.ts | 47 ++++++++++++++----- 2 files changed, 71 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 12a66a6864584..eb4eb1aec5336 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1072,6 +1072,43 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("retries a missing link when the remaining project uses different credentials", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const projectA = ProjectId.make("project-a"); + const projectB = ProjectId.make("project-b"); + const first = makeThread("first", { projectId: projectA, pullRequests: [makeLink(9)] }); + const second = makeThread("second", { projectId: projectB, pullRequests: [makeLink(9)] }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([first, second]), + summary: (input) => + input.projectId === projectA ? missingSummary : Effect.succeed(makeSummary(input)), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)).map((call) => call.projectId), + [projectA], + ); + assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []); + + yield* Ref.set(fixture.snapshots, makeSnapshot([{ ...first, pullRequests: [] }, second])); + yield* sweepAgain(fixture, reactor); + + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)).map((call) => call.projectId), + [projectA, projectB], + ); + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => command.threadId), + [second.id], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect( "holds missing links through sibling and link updates but honors a reader refresh", () => diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 3306ad363a25d..5578064e2cc11 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -1,6 +1,7 @@ import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl"; import { CommandId, + type ProjectId, type PullRequestSummary, type ThreadId, type ThreadPullRequestKey, @@ -158,7 +159,7 @@ 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 missing = new Set(); + const missing = new Map(); // Rate limit pauses by project and host, since each project reads with its own credential. // A paused host refuses every read without asking it, so the sweep leaves its pull requests // due until the pause ends rather than failing each of them every minute. @@ -208,7 +209,12 @@ export const make = Effect.gen(function* () { 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 missing) if (!groups.has(key)) missing.delete(key); + for (const [key, projectId] of missing) { + if (groups.get(key)?.[0]?.thread.projectId !== projectId) { + missing.delete(key); + lastSyncedAt.delete(key); + } + } for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key); const linkedThisSweep = new Set(); @@ -296,7 +302,7 @@ export const make = Effect.gen(function* () { Effect.tapError((error) => error.reason === "not-found" ? Effect.sync(() => { - missing.add(key); + missing.set(key, ref.projectId); // Not-found is a host answer: consume only this refresh and start its backoff. lastSyncedAt.set(key, nowMs); if (requested.get(key) === generation) requested.delete(key); @@ -432,6 +438,28 @@ export const make = Effect.gen(function* () { Effect.catchCause(logSkipped("pull request refresh after run skipped", { threadId })), ); + const requestUnsyncedLinks = Effect.fnUntraced( + function* (links: ReadonlyArray) { + const unsynced = links.filter((link) => link.snapshot === null); + const threads = unsynced.some((link) => missing.has(threadPullRequestKeyOf(link))) + ? yield* projections.getThreadsWithPullRequests() + : []; + for (const link of unsynced) { + const projectId = missing.get(threadPullRequestKeyOf(link)); + if (projectId !== undefined) { + const first = threads.find((thread) => + visibleThreadPullRequests(thread.pullRequests ?? []).some((candidate) => + threadPullRequestKeysEqual(candidate, link), + ), + ); + if (first === undefined || first.projectId === projectId) continue; + } + yield* requestSync(link); + } + }, + Effect.catchCause(logSkipped("pull request refresh after link update skipped", {})), + ); + const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn( "PullRequestSyncReactor.start", )(function* () { @@ -449,17 +477,10 @@ export const make = Effect.gen(function* () { Stream.runForEach(events, (event) => { switch (event.type) { case "thread.pull-request-link-updated": - return event.payload.snapshot === null && - !missing.has(threadPullRequestKeyOf(event.payload)) - ? requestSync(event.payload) - : Effect.void; + return requestUnsyncedLinks([event.payload]); case "thread.pull-request-synced": - return Effect.forEach( - visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( - (link) => link.snapshot === null && !missing.has(threadPullRequestKeyOf(link)), - ), - requestSync, - { discard: true }, + return requestUnsyncedLinks( + visibleThreadPullRequests(event.payload.pullRequests ?? []), ); // An agent can merge or close its pull request from a shell (`gh pr merge`), which // sends no merge notification. When a run that ran such a command ends, read the