authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-07 01:38:47-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 01:38:47-07:00
logcd41c4ada0c70cc2eec95ecd7266f3dab010c58c
tree7c796311a649e14e877f3665a7a869a56a72e8d5
parentc93080b0a3bf4faffb375169c5ec06adce05eb2f
signature Signed by PGP key B5690EEEBB952194

fix(server): threads settle as soon as a client sees their PR merge (#16761)

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

4 files changed, 174 insertions(+), 10 deletions(-)

apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts+47
......@@ -15,6 +15,7 @@ import {
1515 type PullRequestRef,
1616 type PullRequestStack,
1717 type PullRequestSummary,
18 type ThreadPullRequestKey,
1819 type ThreadPullRequestLink,
1920 type ThreadPullRequestSnapshot,
2021} from "@t3tools/contracts";
......@@ -181,6 +182,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
181182 const summaryCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
182183 const stackCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
183184 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
185 const stateChanges = yield* Queue.unbounded<ThreadPullRequestKey>();
184186
185187 const summary: PullRequestService.PullRequestService["Service"]["summary"] = (
186188 input,
......@@ -218,6 +220,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
218220 summary,
219221 stack,
220222 invalidate: options.invalidate ?? (() => Effect.void),
223 subscribeStateChanges: Effect.succeed(Stream.fromQueue(stateChanges)),
221224 }),
222225 Layer.mock(ProjectionStore.ProjectionStoreV2)({
223226 // Mirrors the store's filter: active threads that have at least one link.
......@@ -269,6 +272,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
269272 summaryCalls,
270273 stackCalls,
271274 domainEvents,
275 stateChanges,
272276 layer: PullRequestSyncReactor.layer.pipe(Layer.provide(layerDependencies)),
273277 };
274278});
......@@ -654,6 +658,49 @@ describe("PullRequestSyncReactor", () => {
654658 ),
655659 );
656660
661 it.effect("syncs a pull request a reader saw merge without waiting for the sweep", () =>
662 Effect.scoped(
663 Effect.gen(function* () {
664 yield* TestClock.setTime(Date.parse(NOW));
665 const state = yield* Ref.make<PullRequestSummary["state"]>("open");
666 const invalidated = yield* Ref.make<ReadonlyArray<number>>([]);
667 const fixture = yield* makeHarness({
668 snapshot: makeSnapshot([
669 makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }),
670 ]),
671 summary: (input) =>
672 Ref.get(state).pipe(
673 Effect.map((current) =>
674 makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}),
675 ),
676 ),
677 invalidate: ({ reference }) =>
678 Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]),
679 });
680
681 yield* Effect.gen(function* () {
682 const reactor = yield* startAndSweep(fixture);
683 yield* Ref.set(state, "merged");
684
685 // The clock stays put: the next sweep is still a minute away.
686 yield* Queue.offer(fixture.stateChanges, {
687 host: "github.com",
688 repository: "owner/repository",
689 number: 7,
690 });
691 yield* Queue.take(fixture.snapshotReads);
692 yield* reactor.drain;
693
694 assert.deepStrictEqual(yield* Ref.get(invalidated), [7]);
695 assert.deepStrictEqual(
696 (yield* Ref.get(fixture.syncCommands)).map((command) => command.snapshot.state),
697 ["merged"],
698 );
699 }).pipe(Effect.provide(fixture.layer));
700 }),
701 ),
702 );
703
657704 it.effect("discovers externally reopened pull requests after fifteen minutes", () =>
658705 Effect.scoped(
659706 Effect.gen(function* () {
apps/server/src/orchestration-v2/PullRequestSyncReactor.ts+7
......@@ -424,6 +424,13 @@ export const make = Effect.gen(function* () {
424424 "PullRequestSyncReactor.start",
425425 )(function* () {
426426 const events = engine.streamDomainEvents;
427 // A client reading a pull request can see it merge or close before the next sweep does.
428 const stateChanges = yield* pullRequests.subscribeStateChanges;
429 yield* forkParked(
430 Stream.runForEach(stateChanges, requestSync).pipe(
431 Effect.catchCause(logSkipped("pull request state change stream failed", {})),
432 ),
433 );
427434 yield* forkParked(
428435 Stream.runForEach(events, (event) => {
429436 switch (event.type) {
apps/server/src/pullRequest/PullRequestService.test.ts+56
......@@ -4864,6 +4864,62 @@ it.effect("reuses an observed merged state for strict settlement reads", () =>
48644864 }),
48654865);
48664866
4867it.effect("announces state a detail read sees first or newly", () =>
4868 Effect.gen(function* () {
4869 let detail = {
4870 state: "open" as "open" | "closed" | "merged",
4871 updatedAt: "2026-07-02T00:00:00Z",
4872 };
4873 let summaryState: "open" | "merged" = "open";
4874 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
4875 const service = yield* makeService({
4876 projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],
4877 providers: [
4878 fakeProvider("github", {
4879 getChangeRequest: () => Effect.succeed({ ...hostedChangeRequest("body"), ...detail }),
4880 getChangeRequestSummary: () =>
4881 Effect.succeed({
4882 ...changeRequest(1, "2026-07-04T00:00:00Z"),
4883 state: summaryState,
4884 }),
4885 }),
4886 ],
4887 });
4888 const announced: Array<string> = [];
4889 yield* Stream.runForEach(yield* service.subscribeStateChanges, (key) =>
4890 Effect.sync(() => announced.push(`${key.host}/${key.repository}#${key.number}`)),
4891 ).pipe(Effect.forkChild({ startImmediately: true }));
4892 const readDetail = Effect.gen(function* () {
4893 yield* service.invalidate({ reference });
4894 yield* service.detail(reference);
4895 yield* Effect.yieldNow;
4896 });
4897
4898 // First sight announces; the same state again, even from a fresh read, does not.
4899 yield* readDetail;
4900 yield* readDetail;
4901 assert.deepStrictEqual(announced, ["github.com/acme/web#1"]);
4902
4903 detail = { state: "closed", updatedAt: "2026-07-03T00:00:00Z" };
4904 yield* readDetail;
4905 assert.strictEqual(announced.length, 2);
4906
4907 // A summary seeing the merge first does not hide it from the detail read after.
4908 summaryState = "merged";
4909 yield* service.summary(reference, { recoverTransientFailure: false });
4910 yield* Effect.yieldNow;
4911 assert.strictEqual(announced.length, 2);
4912 detail = { state: "merged", updatedAt: "2026-07-04T00:00:00Z" };
4913 yield* readDetail;
4914 assert.strictEqual(announced.length, 3);
4915
4916 // A detail read older than the merge cannot announce a reopen.
4917 detail = { state: "open", updatedAt: "2026-07-03T00:00:00Z" };
4918 yield* readDetail;
4919 assert.strictEqual(announced.length, 3);
4920 }),
4921);
4922
48674923it.effect("does not let a stale detail reopen overwrite a fresher linked summary", () =>
48684924 Effect.gen(function* () {
48694925 const gate = yield* Deferred.make<void>();
apps/server/src/pullRequest/PullRequestService.ts+64-10
......@@ -72,6 +72,7 @@ import {
7272 type PullRequestUpdateInput,
7373 type SourceControlProviderInfo,
7474 type SourceControlProviderKind,
75 type ThreadPullRequestKey,
7576} from "@t3tools/contracts";
7677import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
7778import { detectSourceControlProviderFromRemoteUrl } from "@t3tools/shared/sourceControl";
......@@ -98,6 +99,16 @@ export interface PullRequestMergeEvent extends PullRequestRef {
9899 readonly mergedAt: string;
99100}
100101
102type PullRequestReading = Pick<PullRequestSummary, "state" | "updatedAt" | "observedAt">;
103
104/** Whether `next` is a newer reading of a pull request than `current`. Merged is final. */
105function supersedesReading(current: PullRequestReading | undefined, next: PullRequestReading) {
106 if (current === undefined) return true;
107 if (current.state === "merged" && next.state !== "merged") return false;
108 if (next.updatedAt !== current.updatedAt) return next.updatedAt > current.updatedAt;
109 return (next.observedAt ?? -Infinity) >= (current.observedAt ?? -Infinity);
110}
111
101112/**
102113 * Rows per repository when the client does not ask for a page size, and rows per slice when a
103114 * listing is carried on from a cursor.
......@@ -225,6 +236,15 @@ export class PullRequestService extends Context.Service<
225236 never,
226237 Scope.Scope
227238 >;
239 /**
240 * Pull requests a detail read saw for the first time or in a new state (opened, closed,
241 * merged), so thread links can catch up before the next sync sweep. Best effort.
242 */
243 readonly subscribeStateChanges: Effect.Effect<
244 Stream.Stream<ThreadPullRequestKey>,
245 never,
246 Scope.Scope
247 >;
228248 readonly subscribeRefreshes: Stream.Stream<number>;
229249 readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect<void>;
230250 readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;
......@@ -666,6 +686,7 @@ const observeRead = Effect.fnUntraced(function* <A, E, R>(read: Effect.Effect<A,
666686
667687export const make = Effect.gen(function* () {
668688 const mergedPullRequests = yield* PubSub.sliding<PullRequestMergeEvent>(64);
689 const stateChanges = yield* PubSub.sliding<ThreadPullRequestKey>(64);
669690 const pullRequestRefreshes = yield* SubscriptionRef.make(0);
670691 const registry = yield* PullRequestProviderRegistry.PullRequestProviderRegistry;
671692 const serverSettings = yield* ServerSettings.ServerSettingsService;
......@@ -3057,13 +3078,39 @@ export const make = Effect.gen(function* () {
30573078 updatedAt: detail.updatedAt,
30583079 observedAt: detail.observedAt,
30593080 });
3060 const shouldReplaceHeldSummary = (key: string, next: PullRequestSummary) => {
3061 const current = lastGoodSummary.peek(key);
3062 if (current === undefined) return true;
3063 if (current.state === "merged" && next.state !== "merged") return false;
3064 if (next.updatedAt !== current.updatedAt) return next.updatedAt > current.updatedAt;
3065 return (next.observedAt ?? -Infinity) >= (current.observedAt ?? -Infinity);
3066 };
3081 const shouldReplaceHeldSummary = (key: string, next: PullRequestSummary) =>
3082 supersedesReading(lastGoodSummary.peek(key), next);
3083 // The newest state a detail read found per pull request, whatever cache epoch or credential
3084 // it read under. Summary reads stay out: they would hide a change from the detail read after.
3085 const detailStates = new Map<string, PullRequestReading>();
3086 /**
3087 * Announces a pull request a detail read sees first or in a new state, so a client showing a
3088 * fresh merge brings the thread's link and its settlement along, rather than leaving them to
3089 * the next sync sweep.
3090 */
3091 const noteDetailReading = (ref: PullRequestRef, next: PullRequestSummary) =>
3092 Effect.suspend(() => {
3093 const scope = refScope(ref);
3094 const current = detailStates.get(scope);
3095 if (!supersedesReading(current, next)) return Effect.void;
3096 detailStates.delete(scope);
3097 if (detailStates.size >= REF_EPOCH_CAPACITY) {
3098 const oldest = detailStates.keys().next().value;
3099 if (oldest !== undefined) detailStates.delete(oldest);
3100 }
3101 detailStates.set(scope, {
3102 state: next.state,
3103 updatedAt: next.updatedAt,
3104 observedAt: next.observedAt,
3105 });
3106 return current?.state !== next.state && ref.host !== undefined
3107 ? PubSub.publish(stateChanges, {
3108 host: ref.host,
3109 repository: ref.repository,
3110 number: ref.number,
3111 })
3112 : Effect.void;
3113 });
30673114 const detail: PullRequestService["Service"]["detail"] = (input) => {
30683115 const key = refCacheKey(input);
30693116 // Record the summary from a host or cache read, not the stale value
......@@ -3073,9 +3120,13 @@ export const make = Effect.gen(function* () {
30733120 const read = Cache.get(detailCache, key).pipe(
30743121 Effect.tap((value) => {
30753122 const summary = summaryFromDetail(value, lastGoodSummary.peek(key));
3076 return shouldReplaceHeldSummary(key, summary)
3077 ? lastGoodSummary.record(key, summary)
3078 : Effect.void;
3123 return noteDetailReading(input, summary).pipe(
3124 Effect.andThen(
3125 shouldReplaceHeldSummary(key, summary)
3126 ? lastGoodSummary.record(key, summary)
3127 : Effect.void,
3128 ),
3129 );
30793130 }),
30803131 );
30813132 return input.allowStale === false
......@@ -3374,6 +3425,9 @@ export const make = Effect.gen(function* () {
33743425 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(
33753426 Effect.map((subscription) => Stream.fromSubscription(subscription)),
33763427 ),
3428 subscribeStateChanges: PubSub.subscribe(stateChanges).pipe(
3429 Effect.map((subscription) => Stream.fromSubscription(subscription)),
3430 ),
33773431 subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe(
33783432 Stream.filter((revision) => revision > 0),
33793433 ),