authorgravatar for 97427849+flamboh@users.noreply.github.comoliver <97427849+flamboh@users.noreply.github.com> 2026-10-10 00:24:11-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 02:24:11-05:00
log6d5ea190a4a54b01dfe6500e032526839e3258d1
treed783be7084e05fe7b75a101364e1ed25eafc2873
parent42f71dfe0768ee768283d50140e7d360a635ee28
signature Signed by PGP key B5690EEEBB952194

fix(server): thread PR badges catch up when another environment reads the PR (#17729)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

9 files changed, 139 insertions(+), 0 deletions(-)

apps/server/src/auth/RpcAuthorization.ts+1
...@@ -135,6 +135,7 @@ export const RPC_REQUIRED_SCOPES = {...@@ -135,6 +135,7 @@ export const RPC_REQUIRED_SCOPES = {
135 // Read scope like the reads it un-caches: refreshing is part of reading, and a read-only135 // Read scope like the reads it un-caches: refreshing is part of reading, and a read-only
136 // client pressing refresh must not be told it may not look again.136 // client pressing refresh must not be told it may not look again.
137 [WS_METHODS.pullRequestsInvalidate]: AuthOrchestrationReadScope,137 [WS_METHODS.pullRequestsInvalidate]: AuthOrchestrationReadScope,
138 [WS_METHODS.pullRequestsReportState]: AuthOrchestrationReadScope,
138 [WS_METHODS.pullRequestsSubscribeRefreshes]: AuthOrchestrationReadScope,139 [WS_METHODS.pullRequestsSubscribeRefreshes]: AuthOrchestrationReadScope,
139 // The candidate list is a read like the detail beside it; asking somebody for a review is a140 // The candidate list is a read like the detail beside it; asking somebody for a review is a
140 // write like every other one.141 // write like every other one.
apps/server/src/observability/RpcInstrumentation.ts+1
...@@ -118,6 +118,7 @@ const RPC_AGGREGATES = {...@@ -118,6 +118,7 @@ const RPC_AGGREGATES = {
118 [WS_METHODS.pullRequestsSetReaction]: "pull-requests",118 [WS_METHODS.pullRequestsSetReaction]: "pull-requests",
119 [WS_METHODS.pullRequestsSetFilesViewed]: "pull-requests",119 [WS_METHODS.pullRequestsSetFilesViewed]: "pull-requests",
120 [WS_METHODS.pullRequestsInvalidate]: "pull-requests",120 [WS_METHODS.pullRequestsInvalidate]: "pull-requests",
121 [WS_METHODS.pullRequestsReportState]: "pull-requests",
121 [WS_METHODS.pullRequestsSubscribeRefreshes]: "pull-requests",122 [WS_METHODS.pullRequestsSubscribeRefreshes]: "pull-requests",
122 [WS_METHODS.pullRequestsReviewerCandidates]: "pull-requests",123 [WS_METHODS.pullRequestsReviewerCandidates]: "pull-requests",
123 [WS_METHODS.pullRequestsRequestReviewers]: "pull-requests",124 [WS_METHODS.pullRequestsRequestReviewers]: "pull-requests",
apps/server/src/pullRequest/PullRequestService.test.ts+42
...@@ -4972,6 +4972,48 @@ it.effect("announces state a detail read sees first or newly", () =>...@@ -4972,6 +4972,48 @@ it.effect("announces state a detail read sees first or newly", () =>
4972 }),4972 }),
4973);4973);
49744974
4975it.effect("announces a reported state without trusting it", () =>
4976 Effect.gen(function* () {
4977 let detail = { state: "open" as "open" | "closed", updatedAt: "2026-07-02T00:00:00Z" };
4978 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
4979 const service = yield* makeService({
4980 projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],
4981 providers: [
4982 fakeProvider("github", {
4983 getChangeRequest: () => Effect.succeed({ ...hostedChangeRequest("body"), ...detail }),
4984 }),
4985 ],
4986 });
4987 const announced: Array<string> = [];
4988 yield* Stream.runForEach(yield* service.subscribeStateChanges, (key) =>
4989 Effect.sync(() => announced.push(`${key.host}/${key.repository}#${key.number}`)),
4990 ).pipe(Effect.forkChild({ startImmediately: true }));
4991 const note = (state: "open" | "closed" | "merged") =>
4992 service.reportState({ reference, state }).pipe(Effect.andThen(Effect.yieldNow));
4993 const readDetail = Effect.gen(function* () {
4994 yield* service.invalidate({ reference });
4995 yield* service.detail(reference);
4996 yield* Effect.yieldNow;
4997 });
4998
4999 yield* readDetail;
5000 yield* note("closed");
5001 yield* note("closed");
5002 assert.strictEqual(announced.length, 2);
5003 yield* TestClock.adjust("5 minutes");
5004 yield* note("closed");
5005 assert.strictEqual(announced.length, 3);
5006
5007 yield* note("open");
5008 assert.strictEqual(announced.length, 4);
5009
5010 yield* note("merged");
5011 detail = { state: "closed", updatedAt: "2026-07-03T00:00:00Z" };
5012 yield* readDetail;
5013 assert.strictEqual(announced.length, 6);
5014 }),
5015);
5016
4975it.effect("does not let a stale detail reopen overwrite a fresher linked summary", () =>5017it.effect("does not let a stale detail reopen overwrite a fresher linked summary", () =>
4976 Effect.gen(function* () {5018 Effect.gen(function* () {
4977 const gate = yield* Deferred.make<void>();5019 const gate = yield* Deferred.make<void>();
apps/server/src/pullRequest/PullRequestService.ts+41
...@@ -42,6 +42,7 @@ import {...@@ -42,6 +42,7 @@ import {
42 type PullRequestFilesViewedResult,42 type PullRequestFilesViewedResult,
43 type PullRequestDiffResult,43 type PullRequestDiffResult,
44 type PullRequestInvalidateInput,44 type PullRequestInvalidateInput,
45 type PullRequestReportStateInput,
45 type PullRequestListEntry,46 type PullRequestListEntry,
46 type PullRequestListFilters,47 type PullRequestListFilters,
47 type PullRequestListInput,48 type PullRequestListInput,
...@@ -160,6 +161,8 @@ const detailTimeToLive = (state: PullRequestState | undefined) =>...@@ -160,6 +161,8 @@ const detailTimeToLive = (state: PullRequestState | undefined) =>
160 * page and a watched pull request's change reaches its watch.161 * page and a watched pull request's change reaches its watch.
161 */162 */
162const CHECKS_CACHE_TTL = Duration.seconds(15);163const CHECKS_CACHE_TTL = Duration.seconds(15);
164/** How long a repeated reported state is skipped, so a wrong one cannot hide a real change for long. */
165const REPORTED_STATE_DEDUPE_MS = 5 * 60 * 1_000;
163const DIFF_CACHE_TTL = Duration.seconds(60);166const DIFF_CACHE_TTL = Duration.seconds(60);
164/** A commit is content-addressed, so its own diff cannot change under its key. */167/** A commit is content-addressed, so its own diff cannot change under its key. */
165const COMMIT_DIFF_CACHE_TTL = Duration.minutes(10);168const COMMIT_DIFF_CACHE_TTL = Duration.minutes(10);
...@@ -244,6 +247,13 @@ export class PullRequestService extends Context.Service<...@@ -244,6 +247,13 @@ export class PullRequestService extends Context.Service<
244 never,247 never,
245 Scope.Scope248 Scope.Scope
246 >;249 >;
250 /**
251 * The state a read routed to another environment saw. A new one reaches
252 * `subscribeStateChanges`, so the host is asked again; the report itself is never trusted.
253 */
254 readonly reportState: (
255 input: PullRequestReportStateInput,
256 ) => Effect.Effect<void, PullRequestError>;
247 readonly subscribeRefreshes: Stream.Stream<number>;257 readonly subscribeRefreshes: Stream.Stream<number>;
248 readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect<void>;258 readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect<void>;
249 readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;259 readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;
...@@ -3115,6 +3125,36 @@ export const make = Effect.gen(function* () {...@@ -3115,6 +3125,36 @@ export const make = Effect.gen(function* () {
3115 })3125 })
3116 : Effect.void;3126 : Effect.void;
3117 });3127 });
3128 // The last state each routed read reported, kept apart from `detailStates`: an unverified report
3129 // must never outrank a reading this environment made itself.
3130 const reportedStates = new Map<
3131 string,
3132 { readonly state: PullRequestState; readonly atMs: number }
3133 >();
3134 const reportState: PullRequestService["Service"]["reportState"] = ({ reference, state }) =>
3135 Effect.all([canonicalRef(reference), Clock.currentTimeMillis]).pipe(
3136 Effect.flatMap(([ref, nowMs]) =>
3137 Effect.suspend(() => {
3138 const scope = refScope(ref);
3139 const reported = reportedStates.get(scope);
3140 if (reported?.state === state && nowMs - reported.atMs < REPORTED_STATE_DEDUPE_MS)
3141 return Effect.void;
3142 reportedStates.delete(scope);
3143 if (reportedStates.size >= REF_EPOCH_CAPACITY) {
3144 const oldest = reportedStates.keys().next().value;
3145 if (oldest !== undefined) reportedStates.delete(oldest);
3146 }
3147 reportedStates.set(scope, { state, atMs: nowMs });
3148 return ref.host === undefined
3149 ? Effect.void
3150 : PubSub.publish(stateChanges, {
3151 host: ref.host,
3152 repository: ref.repository,
3153 number: ref.number,
3154 });
3155 }),
3156 ),
3157 );
3118 const detail: PullRequestService["Service"]["detail"] = (input) => {3158 const detail: PullRequestService["Service"]["detail"] = (input) => {
3119 const key = refCacheKey(input);3159 const key = refCacheKey(input);
3120 // Record the summary from a host or cache read, not the stale value3160 // Record the summary from a host or cache read, not the stale value
...@@ -3432,6 +3472,7 @@ export const make = Effect.gen(function* () {...@@ -3432,6 +3472,7 @@ export const make = Effect.gen(function* () {
3432 subscribeStateChanges: PubSub.subscribe(stateChanges).pipe(3472 subscribeStateChanges: PubSub.subscribe(stateChanges).pipe(
3433 Effect.map((subscription) => Stream.fromSubscription(subscription)),3473 Effect.map((subscription) => Stream.fromSubscription(subscription)),
3434 ),3474 ),
3475 reportState,
3435 subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe(3476 subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe(
3436 Stream.filter((revision) => revision > 0),3477 Stream.filter((revision) => revision > 0),
3437 ),3478 ),
apps/server/src/ws.ts+1
...@@ -2518,6 +2518,7 @@ const layerWsRpc = (...@@ -2518,6 +2518,7 @@ const layerWsRpc = (
2518 withPullRequestViewer(input, pullRequests.setThreadResolution(input)),2518 withPullRequestViewer(input, pullRequests.setThreadResolution(input)),
2519 [WS_METHODS.pullRequestsSetReaction]: (input) =>2519 [WS_METHODS.pullRequestsSetReaction]: (input) =>
2520 withPullRequestViewer(input, pullRequests.setReaction(input)),2520 withPullRequestViewer(input, pullRequests.setReaction(input)),
2521 [WS_METHODS.pullRequestsReportState]: (input) => pullRequests.reportState(input),
2521 [WS_METHODS.pullRequestsInvalidate]: (input) =>2522 [WS_METHODS.pullRequestsInvalidate]: (input) =>
2522 pullRequests.invalidate(input, { notifyReaders: true }).pipe(2523 pullRequests.invalidate(input, { notifyReaders: true }).pipe(
2523 // A reader asking for fresh host state also wants the thread badges it feeds to2524 // A reader asking for fresh host state also wants the thread badges it feeds to
packages/client-runtime/src/state/pullRequestRouting.ts+22
...@@ -2,6 +2,7 @@ import {...@@ -2,6 +2,7 @@ import {
2 WS_METHODS,2 WS_METHODS,
3 PullRequestRef,3 PullRequestRef,
4 PullRequestInvalidateInput,4 PullRequestInvalidateInput,
5 PullRequestState,
5 type EnvironmentId,6 type EnvironmentId,
6 type PullRequestRoutingResult,7 type PullRequestRoutingResult,
7 type PullRequestRoutingIdentityResult,8 type PullRequestRoutingIdentityResult,
...@@ -57,6 +58,7 @@ const writes = new Set<string>([...@@ -57,6 +58,7 @@ const writes = new Set<string>([
57 WS_METHODS.pullRequestsSetLabels,58 WS_METHODS.pullRequestsSetLabels,
58]);59]);
59const isRef = Schema.is(PullRequestRef);60const isRef = Schema.is(PullRequestRef);
61const hasState = Schema.is(Schema.Struct({ state: PullRequestState }));
60const isInvalidation = Schema.is(PullRequestInvalidateInput);62const isInvalidation = Schema.is(PullRequestInvalidateInput);
61const readTimeout = (environmentId: EnvironmentId) =>63const readTimeout = (environmentId: EnvironmentId) =>
62 Effect.timeoutOrElse({64 Effect.timeoutOrElse({
...@@ -208,6 +210,21 @@ const invalidateTarget = Effect.fn("PullRequestRouting.invalidateTarget")(functi...@@ -208,6 +210,21 @@ const invalidateTarget = Effect.fn("PullRequestRouting.invalidateTarget")(functi
208 );210 );
209});211});
210212
213/**
214 * The origin owns the threads linked to this pull request, but a routed read never reaches it.
215 * Tell it which state the read saw so it checks again. Best effort, and never holds up the read.
216 */
217const reportState = (
218 { allowStale: _allowStale, expectedAccountId: _expectedAccountId, ...reference }: PullRequestRef,
219 state: PullRequestState,
220) =>
221 request(WS_METHODS.pullRequestsReportState, { reference, state }).pipe(
222 Effect.timeoutOption("5 seconds"),
223 Effect.ignore,
224 Effect.forkDetach,
225 Effect.asVoid,
226 );
227
211/** Credentials stay on their environments. Only a verified host and account cross the wire. */228/** Credentials stay on their environments. Only a verified host and account cross the wire. */
212export function createPullRequestRouter() {229export function createPullRequestRouter() {
213 const routedRequest = Effect.fn("PullRequestRouting.request")(function* <230 const routedRequest = Effect.fn("PullRequestRouting.request")(function* <
...@@ -356,6 +373,11 @@ export function createPullRequestRouter() {...@@ -356,6 +373,11 @@ export function createPullRequestRouter() {
356 ? guardedSource373 ? guardedSource
357 : registry.run(id, request(tag, routedInput))374 : registry.run(id, request(tag, routedInput))
358 ).pipe(375 ).pipe(
376 Effect.tap((result) =>
377 id !== origin.target.environmentId && hasState(result)
378 ? reportState(ref, result.state)
379 : Effect.void,
380 ),
359 Effect.tap(() =>381 Effect.tap(() =>
360 Effect.sync(() => {382 Effect.sync(() => {
361 const entry = used.get(refKey) ?? {383 const entry = used.get(refKey) ?? {
packages/client-runtime/src/state/pullRequests.test.ts+11
...@@ -995,6 +995,7 @@ it.effect.each(...@@ -995,6 +995,7 @@ it.effect.each(
995it.live("keeps source workspace metadata when an alternate answers a detail read", () =>995it.live("keeps source workspace metadata when an alternate answers a detail read", () =>
996 Effect.scoped(996 Effect.scoped(
997 Effect.gen(function* () {997 Effect.gen(function* () {
998 const noted = yield* Deferred.make<unknown>();
998 const clientFor = (local: boolean) =>999 const clientFor = (local: boolean) =>
999 ({1000 ({
1000 [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>1001 [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
...@@ -1013,8 +1014,12 @@ it.live("keeps source workspace metadata when an alternate answers a detail read...@@ -1013,8 +1014,12 @@ it.live("keeps source workspace metadata when an alternate answers a detail read
1013 projectTitle: "local project",1014 projectTitle: "local project",
1014 workspaceRoot: "/Users/local/repo",1015 workspaceRoot: "/Users/local/repo",
1015 title: "github title",1016 title: "github title",
1017 state: "closed",
1018 updatedAt: "2026-10-10T01:52:36Z",
1016 })1019 })
1017 : Effect.never,1020 : Effect.never,
1021 [WS_METHODS.pullRequestsReportState]: (input: unknown) =>
1022 Deferred.succeed(noted, input).pipe(Effect.asVoid),
1018 }) as unknown as WsRpcProtocolClient;1023 }) as unknown as WsRpcProtocolClient;
1019 const { environmentRegistry, supervisor } = yield* makeTestRuntime(1024 const { environmentRegistry, supervisor } = yield* makeTestRuntime(
1020 clientFor(false),1025 clientFor(false),
...@@ -1035,6 +1040,12 @@ it.live("keeps source workspace metadata when an alternate answers a detail read...@@ -1035,6 +1040,12 @@ it.live("keeps source workspace metadata when an alternate answers a detail read
1035 projectTitle: "source project",1040 projectTitle: "source project",
1036 workspaceRoot: "/srv/source/repo",1041 workspaceRoot: "/srv/source/repo",
1037 title: "github title",1042 title: "github title",
1043 state: "closed",
1044 updatedAt: "2026-10-10T01:52:36Z",
1045 });
1046 expect(yield* Deferred.await(noted)).toEqual({
1047 reference: { projectId: "project-1", repository: "acme/web", number: 7 },
1048 state: "closed",
1038 });1049 });
1039 }),1050 }),
1040 ),1051 ),
packages/contracts/src/pullRequest.ts+11
...@@ -849,6 +849,17 @@ export const PullRequestInvalidateInput = Schema.Struct({...@@ -849,6 +849,17 @@ export const PullRequestInvalidateInput = Schema.Struct({
849});849});
850export type PullRequestInvalidateInput = typeof PullRequestInvalidateInput.Type;850export type PullRequestInvalidateInput = typeof PullRequestInvalidateInput.Type;
851851
852/**
853 * The state a read routed to another environment saw, reported to the environment the read was
854 * for so its thread links can catch up. A hint only: that environment confirms with the host
855 * before writing anything.
856 */
857export const PullRequestReportStateInput = Schema.Struct({
858 reference: PullRequestRef,
859 state: PullRequestState,
860});
861export type PullRequestReportStateInput = typeof PullRequestReportStateInput.Type;
862
852export const PullRequestDetail = Schema.Struct({863export const PullRequestDetail = Schema.Struct({
853 provider: SourceControlProviderKind,864 provider: SourceControlProviderKind,
854 capabilities: PullRequestCapabilities,865 capabilities: PullRequestCapabilities,
packages/contracts/src/rpc.ts+9
...@@ -171,6 +171,7 @@ import {...@@ -171,6 +171,7 @@ import {
171 PullRequestDiffFileContentsResult,171 PullRequestDiffFileContentsResult,
172 PullRequestFilesViewedResult,172 PullRequestFilesViewedResult,
173 PullRequestInvalidateInput,173 PullRequestInvalidateInput,
174 PullRequestReportStateInput,
174 PullRequestListInput,175 PullRequestListInput,
175 PullRequestListResult,176 PullRequestListResult,
176 PullRequestListStatsInput,177 PullRequestListStatsInput,
...@@ -543,6 +544,7 @@ export const WS_METHODS = {...@@ -543,6 +544,7 @@ export const WS_METHODS = {
543 pullRequestsSetThreadResolution: "pullRequests.setThreadResolution",544 pullRequestsSetThreadResolution: "pullRequests.setThreadResolution",
544 pullRequestsSetReaction: "pullRequests.setReaction",545 pullRequestsSetReaction: "pullRequests.setReaction",
545 pullRequestsInvalidate: "pullRequests.invalidate",546 pullRequestsInvalidate: "pullRequests.invalidate",
547 pullRequestsReportState: "pullRequests.reportState",
546 pullRequestsSubscribeRefreshes: "pullRequests.subscribeRefreshes",548 pullRequestsSubscribeRefreshes: "pullRequests.subscribeRefreshes",
547 pullRequestsReviewerCandidates: "pullRequests.reviewerCandidates",549 pullRequestsReviewerCandidates: "pullRequests.reviewerCandidates",
548 pullRequestsRequestReviewers: "pullRequests.requestReviewers",550 pullRequestsRequestReviewers: "pullRequests.requestReviewers",
...@@ -1087,6 +1089,12 @@ const WsPullRequestsInvalidateRpc = Rpc.make(WS_METHODS.pullRequestsInvalidate,...@@ -1087,6 +1089,12 @@ const WsPullRequestsInvalidateRpc = Rpc.make(WS_METHODS.pullRequestsInvalidate,
1087 error: PullRequestRpcError,1089 error: PullRequestRpcError,
1088});1090});
10891091
1092const WsPullRequestsReportStateRpc = Rpc.make(WS_METHODS.pullRequestsReportState, {
1093 payload: PullRequestReportStateInput,
1094 success: Schema.Void,
1095 error: PullRequestRpcError,
1096});
1097
1090const WsPullRequestsSubscribeRefreshesRpc = Rpc.make(WS_METHODS.pullRequestsSubscribeRefreshes, {1098const WsPullRequestsSubscribeRefreshesRpc = Rpc.make(WS_METHODS.pullRequestsSubscribeRefreshes, {
1091 payload: Schema.Struct({}),1099 payload: Schema.Struct({}),
1092 success: NonNegativeInt,1100 success: NonNegativeInt,
...@@ -1916,6 +1924,7 @@ export const WsRpcGroup = RpcGroup.make(...@@ -1916,6 +1924,7 @@ export const WsRpcGroup = RpcGroup.make(
1916 WsPullRequestsSetThreadResolutionRpc,1924 WsPullRequestsSetThreadResolutionRpc,
1917 WsPullRequestsSetReactionRpc,1925 WsPullRequestsSetReactionRpc,
1918 WsPullRequestsInvalidateRpc,1926 WsPullRequestsInvalidateRpc,
1927 WsPullRequestsReportStateRpc,
1919 WsPullRequestsSubscribeRefreshesRpc,1928 WsPullRequestsSubscribeRefreshesRpc,
1920 WsPullRequestsReviewerCandidatesRpc,1929 WsPullRequestsReviewerCandidatesRpc,
1921 WsPullRequestsRequestReviewersRpc,1930 WsPullRequestsRequestReviewersRpc,