authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-03 21:11:57-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-03 21:11:57-04:00
log2b96220f00de09b61327c8e881b7f509f8cc5a79
tree6bc2ecf88dc91b07a35fc5c53cb3c3fd27d2b1a0
parent343db2c328af56fe0fa7672f3af055550832689b
signature Signed by PGP key B5690EEEBB952194

fix(server): settle branch threads immediately on pull request merge (#9528)

Co-authored-by: maria-rcks <254055478+maria-rcks@users.noreply.github.com>

3 files changed, 150 insertions(+), 13 deletions(-)

apps/server/src/git/GitManager.ts+7-1
...@@ -120,7 +120,13 @@ const SHORT_SHA_LENGTH = 7;...@@ -120,7 +120,13 @@ const SHORT_SHA_LENGTH = 7;
120const TOAST_DESCRIPTION_MAX = 72;120const TOAST_DESCRIPTION_MAX = 72;
121const STATUS_RESULT_CACHE_TTL = Duration.seconds(1);121const STATUS_RESULT_CACHE_TTL = Duration.seconds(1);
122const STATUS_RESULT_CACHE_CAPACITY = 2_048;122const STATUS_RESULT_CACHE_CAPACITY = 2_048;
123const PR_LOOKUP_CACHE_TTL = Duration.minutes(2);123// Matches the automatic settlement sweep cadence so every background sweep
124// reads fresh branch state: an external merge settles within about a minute
125// instead of waiting out a longer cache. Unpublished branches never reach the
126// host (a local probe answers first), and failed lookups still back off
127// exponentially via prLookupFailureTtl, so throttling pressure still drops
128// under 429s instead of amplifying it.
129const PR_LOOKUP_CACHE_TTL = Duration.seconds(60);
124const PR_LOOKUP_FAILURE_BASE_TTL = Duration.seconds(20);130const PR_LOOKUP_FAILURE_BASE_TTL = Duration.seconds(20);
125const PR_LOOKUP_FAILURE_MAX_TTL = Duration.minutes(15);131const PR_LOOKUP_FAILURE_MAX_TTL = Duration.minutes(15);
126const PR_LOOKUP_CACHE_CAPACITY = 2_048;132const PR_LOOKUP_CACHE_CAPACITY = 2_048;
apps/server/src/orchestration/ThreadSettlementReactor.test.ts+113-1
...@@ -161,6 +161,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -161,6 +161,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
161 }>161 }>
162 >([]);162 >([]);
163 const summaryRecovery = yield* Ref.make<ReadonlyArray<boolean | undefined>>([]);163 const summaryRecovery = yield* Ref.make<ReadonlyArray<boolean | undefined>>([]);
164 const invalidatedCwds = yield* Ref.make<ReadonlyArray<string>>([]);
164165
165 const updateSettings = (patch: ServerSettingsPatch) =>166 const updateSettings = (patch: ServerSettingsPatch) =>
166 Effect.gen(function* () {167 Effect.gen(function* () {
...@@ -221,7 +222,10 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -221,7 +222,10 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
221 Effect.andThen(Ref.get(snapshots)),222 Effect.andThen(Ref.get(snapshots)),
222 ),223 ),
223 }),224 }),
224 Layer.mock(GitManager)({ branchPullRequest }),225 Layer.mock(GitManager)({
226 branchPullRequest,
227 invalidateStatus: (cwd) => Ref.update(invalidatedCwds, (cwds) => [...cwds, cwd]),
228 }),
225 Layer.mock(PullRequestService)({229 Layer.mock(PullRequestService)({
226 summary: pullRequestSummary,230 summary: pullRequestSummary,
227 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(231 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(
...@@ -251,6 +255,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -251,6 +255,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
251 branchCalls,255 branchCalls,
252 summaryCalls,256 summaryCalls,
253 summaryRecovery,257 summaryRecovery,
258 invalidatedCwds,
254 updateSettings,259 updateSettings,
255 publishMerge: PubSub.publish(mergedPullRequests, {260 publishMerge: PubSub.publish(mergedPullRequests, {
256 projectId: PROJECT_ID,261 projectId: PROJECT_ID,
...@@ -446,6 +451,113 @@ describe("ThreadSettlementReactor", () => {...@@ -446,6 +451,113 @@ describe("ThreadSettlementReactor", () => {
446 ),451 ),
447 );452 );
448453
454 it.effect(
455 "settles branch threads on a pull request merge without waiting for the next sweep",
456 () =>
457 Effect.scoped(
458 Effect.gen(function* () {
459 yield* TestClock.setTime(Date.parse(NOW));
460 const state = yield* Ref.make<"open" | "merged">("open");
461 const mergedThreadSettled = yield* Deferred.make<void>();
462 const fixture = yield* makeHarness({
463 snapshot: makeSnapshot([makeThread("branch-thread", { branch: "saved-feature" })]),
464 branchPullRequest: () =>
465 Ref.get(state).pipe(
466 Effect.map((pullRequestState) => ({ state: pullRequestState, updatedAt: NOW })),
467 ),
468 onDispatch: () => Deferred.succeed(mergedThreadSettled, undefined),
469 });
470
471 yield* Effect.gen(function* () {
472 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
473 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
474 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
475 assert.deepStrictEqual(yield* Ref.get(fixture.invalidatedCwds), []);
476
477 yield* Ref.set(state, "merged");
478 yield* fixture.publishMerge;
479 yield* Deferred.await(mergedThreadSettled);
480
481 assert.deepStrictEqual(
482 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
483 [ThreadId.make("branch-thread")],
484 );
485 assert.deepStrictEqual(yield* Ref.get(fixture.invalidatedCwds), ["/workspace/project"]);
486 yield* reactor.drain;
487 }).pipe(Effect.provide(fixture.layer));
488 }),
489 ),
490 );
491
492 it.effect("a merge does not settle threads linked to an unrelated pull request", () =>
493 Effect.scoped(
494 Effect.gen(function* () {
495 yield* TestClock.setTime(Date.parse(NOW));
496 const mergedThreadSettled = yield* Deferred.make<void>();
497 const mergeLookupStarted = yield* Deferred.make<void>();
498 const releaseMergeLookup = yield* Deferred.make<void>();
499 const lookupCount = yield* Ref.make(0);
500 const fixture = yield* makeHarness({
501 snapshot: makeSnapshot([
502 makeThread("merged-in-app", {
503 linkedPullRequest: {
504 projectId: PROJECT_ID,
505 repository: "owner/repository",
506 number: 42,
507 url: "https://example.test/owner/repository/pull/42",
508 },
509 }),
510 makeThread("unrelated-linked", {
511 linkedPullRequest: {
512 projectId: PROJECT_ID,
513 repository: "owner/repository",
514 number: 99,
515 url: "https://example.test/owner/repository/pull/99",
516 },
517 }),
518 ]),
519 pullRequestSummary: (input) =>
520 Ref.updateAndGet(lookupCount, (count) => count + 1).pipe(
521 // The initial sweep looks up both linked threads; the merge
522 // sweep only looks up the unrelated one, since the merged
523 // thread settles from the event itself.
524 Effect.tap((count) =>
525 count === 3 ? Deferred.succeed(mergeLookupStarted, undefined) : Effect.void,
526 ),
527 Effect.tap((count) =>
528 count === 3 ? Deferred.await(releaseMergeLookup) : Effect.void,
529 ),
530 Effect.map(() => makePullRequestSummary({ ...input, state: "open" })),
531 ),
532 onDispatch: () => Deferred.succeed(mergedThreadSettled, undefined),
533 });
534
535 yield* Effect.gen(function* () {
536 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
537 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
538 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
539
540 yield* fixture.publishMerge;
541 yield* Deferred.await(mergeLookupStarted);
542 yield* Deferred.await(mergedThreadSettled);
543 yield* Deferred.succeed(releaseMergeLookup, undefined);
544 yield* reactor.drain;
545
546 assert.deepStrictEqual(
547 (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
548 [ThreadId.make("merged-in-app")],
549 );
550 assert.deepStrictEqual(
551 (yield* Ref.get(fixture.summaryCalls))
552 .map((call) => call.number)
553 .toSorted((left, right) => left - right),
554 [42, 99, 99],
555 );
556 }).pipe(Effect.provide(fixture.layer));
557 }),
558 ),
559 );
560
449 it.effect("uses fresh settlement settings after lookup and ignores unrelated changes", () =>561 it.effect("uses fresh settlement settings after lookup and ignores unrelated changes", () =>
450 Effect.scoped(562 Effect.scoped(
451 Effect.gen(function* () {563 Effect.gen(function* () {
apps/server/src/orchestration/ThreadSettlementReactor.ts+30-11
...@@ -46,16 +46,11 @@ export const make = Effect.gen(function* () {...@@ -46,16 +46,11 @@ export const make = Effect.gen(function* () {
46 const snapshot = yield* snapshots.getShellSnapshot();46 const snapshot = yield* snapshots.getShellSnapshot();
47 const now = DateTime.formatIso(yield* DateTime.now);47 const now = DateTime.formatIso(yield* DateTime.now);
48 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));48 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
49 const candidates = snapshot.threads.filter(49 // A merge event re-sweeps every candidate, not just the threads linked to
50 (thread) =>50 // the merged pull request: most threads carry no link and settle from
51 isAutoSettlementCandidate(thread, now) &&51 // their branch lookup, which would otherwise wait for the next minute's
52 (mergedPullRequest === null ||52 // sweep on a possibly stale cached answer.
53 (thread.linkedPullRequest != null &&53 const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));
54 thread.linkedPullRequest.projectId === mergedPullRequest.projectId &&
55 thread.linkedPullRequest.repository.toLowerCase() ===
56 mergedPullRequest.repository.toLowerCase() &&
57 thread.linkedPullRequest.number === mergedPullRequest.number)),
58 );
59 // Use the same cwd as the sidebar so both paths share GitManager's PR cache.54 // Use the same cwd as the sidebar so both paths share GitManager's PR cache.
60 const lookupCwdByThreadId = new Map<string, string>();55 const lookupCwdByThreadId = new Map<string, string>();
61 yield* Effect.forEach(56 yield* Effect.forEach(
...@@ -76,6 +71,19 @@ export const make = Effect.gen(function* () {...@@ -76,6 +71,19 @@ export const make = Effect.gen(function* () {
76 }),71 }),
77 { concurrency: 8, discard: true },72 { concurrency: 8, discard: true },
78 );73 );
74 if (mergedPullRequest !== null) {
75 // The merge just confirmed a terminal state the lookup caches can still
76 // call open (branch answers live two minutes, the sweep runs every
77 // minute). Drop the swept checkouts' cached answers so the merge settles
78 // its branch threads now instead of on a later sweep. Threads linked to
79 // the merged pull request settle from the event itself below and need no
80 // lookup, so they are absent from this map by construction.
81 const cwds = [...new Set(lookupCwdByThreadId.values())];
82 yield* Effect.forEach(cwds, (cwd) => git.invalidateStatus(cwd), {
83 concurrency: 8,
84 discard: true,
85 });
86 }
79 const lookupKey = (thread: (typeof candidates)[number]) => {87 const lookupKey = (thread: (typeof candidates)[number]) => {
80 if (thread.linkedPullRequest != null) {88 if (thread.linkedPullRequest != null) {
81 return JSON.stringify([89 return JSON.stringify([
...@@ -97,7 +105,18 @@ export const make = Effect.gen(function* () {...@@ -97,7 +105,18 @@ export const make = Effect.gen(function* () {
97 thread: (typeof candidates)[number],105 thread: (typeof candidates)[number],
98 ) {106 ) {
99 if (thread.linkedPullRequest != null) {107 if (thread.linkedPullRequest != null) {
100 if (mergedPullRequest !== null) {108 // The event carries the merged state, so only the threads linked to
109 // that exact pull request settle from it. Every other linked thread
110 // falls through to a fresh summary lookup below: the merge sweep
111 // covers all candidates, and an unrelated merge must never settle
112 // them.
113 if (
114 mergedPullRequest !== null &&
115 thread.linkedPullRequest.projectId === mergedPullRequest.projectId &&
116 thread.linkedPullRequest.repository.toLowerCase() ===
117 mergedPullRequest.repository.toLowerCase() &&
118 thread.linkedPullRequest.number === mergedPullRequest.number
119 ) {
101 return {120 return {
102 state: "merged",121 state: "merged",
103 updatedAt: mergedPullRequest.mergedAt,122 updatedAt: mergedPullRequest.mergedAt,