authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-10-05 09:12:55-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 09:12:55-03:00
log1a3f7ad5085606463a602bd756ca9c04d106fffe
treedc9c270c551639f95fd6ad7fcc9412c130e11ce0
parent7ffa2184a7cfc78d870bf6cbfa4064c51868aef2
signature Signed by PGP key B5690EEEBB952194

perf(prs): share concurrent github routing metadata probes (#15853)

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

2 files changed, 327 insertions(+), 3 deletions(-)

packages/client-runtime/src/state/pullRequestRouting.ts+65-2
......@@ -3,9 +3,13 @@ import {
33 PullRequestRef,
44 PullRequestInvalidateInput,
55 type EnvironmentId,
6 type PullRequestRoutingResult,
7 type PullRequestRoutingIdentityResult,
68} from "@t3tools/contracts";
79import * as Effect from "effect/Effect";
810import * as Cause from "effect/Cause";
11import * as Deferred from "effect/Deferred";
12import * as Exit from "effect/Exit";
913import * as Option from "effect/Option";
1014import * as Schema from "effect/Schema";
1115import * as SubscriptionRef from "effect/SubscriptionRef";
......@@ -18,6 +22,7 @@ import {
1822} from "../connection/githubRoutingPermissions.ts";
1923import type { ConnectionCatalogEntry } from "../connection/catalog.ts";
2024import * as ConnectionProfileStore from "../connection/profileStore.ts";
25import type { RpcSession } from "../rpc/session.ts";
2126import {
2227 request,
2328 EnvironmentRpcUnavailableError,
......@@ -77,6 +82,44 @@ const isUnregistered = Schema.is(EnvironmentRegistry.EnvironmentNotRegisteredErr
7782const encodeKey = Schema.encodeSync(
7883 Schema.fromJsonString(Schema.Array(Schema.NullOr(Schema.String))),
7984);
85type PendingMetadata<A> = WeakMap<RpcSession, Map<string, Deferred.Deferred<Exit.Exit<A>>>>;
86const pendingRouting: PendingMetadata<PullRequestRoutingResult | null> = new WeakMap();
87const pendingIdentities: PendingMetadata<PullRequestRoutingIdentityResult | null> = new WeakMap();
88
89/** Share only active read probes. Completed probes and mutations always read the current account. */
90function shareMetadataRead<A, R>(
91 pending: PendingMetadata<A>,
92 key: string,
93 probe: Effect.Effect<A, never, R>,
94): Effect.Effect<A, never, R | EnvironmentSupervisor.EnvironmentSupervisor> {
95 return Effect.uninterruptibleMask((restore) =>
96 Effect.gen(function* () {
97 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
98 const session = yield* SubscriptionRef.get(supervisor.session);
99 if (Option.isNone(session)) return yield* restore(probe);
100 const entries =
101 pending.get(session.value) ?? new Map<string, Deferred.Deferred<Exit.Exit<A>>>();
102 pending.set(session.value, entries);
103 const existing = entries.get(key);
104 if (existing !== undefined) {
105 const result = yield* restore(Deferred.await(existing));
106 // A cancelled first reader must not cancel the other readers waiting on its probe.
107 return yield* restore(
108 Exit.hasInterrupts(result) ? shareMetadataRead(pending, key, probe) : result,
109 );
110 }
111 const result = Deferred.makeUnsafe<Exit.Exit<A>>();
112 entries.set(key, result);
113 return yield* restore(probe).pipe(
114 Effect.onExit((exit) =>
115 Effect.sync(() => entries.delete(key)).pipe(
116 Effect.andThen(Deferred.succeed(result, exit)),
117 ),
118 ),
119 );
120 }),
121 );
122}
80123
81124function isLocal(entry: ConnectionCatalogEntry): boolean {
82125 const url =
......@@ -276,12 +319,19 @@ export function createPullRequestRouter() {
276319 }
277320 if (alternatives.length === 0) return yield* finish(source);
278321
279 const identity = yield* request(WS_METHODS.pullRequestsRouting, ref).pipe(
322 const identityProbe = request(WS_METHODS.pullRequestsRouting, ref).pipe(
280323 Effect.timeout("2 seconds"),
281324 Effect.catchCause((cause) =>
282325 Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(null),
283326 ),
284327 );
328 const identity = yield* reads.has(tag)
329 ? shareMetadataRead(
330 pendingRouting,
331 encodeKey([ref.projectId, ref.host?.toLowerCase() ?? null, ref.repository.toLowerCase()]),
332 identityProbe,
333 )
334 : identityProbe;
285335 // Old servers and unknown accounts retain the existing path.
286336 if (identity === null || identity.provider !== "github") return yield* finish(source);
287337
......@@ -338,8 +388,21 @@ export function createPullRequestRouter() {
338388 if (!(yield* routingAllowed(registry, origin.target.environmentId, id, writes.has(tag))))
339389 return yield* visit(index + 1);
340390 // An older server would discard expectedAccountId. Verify it implements the guard first.
391 const alternateProbe = request(WS_METHODS.pullRequestsRoutingIdentity, {
392 host: identity.host,
393 }).pipe(
394 Effect.timeout("2 seconds"),
395 Effect.catchCause((cause) =>
396 Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(null),
397 ),
398 );
341399 const alternate = yield* registry
342 .run(id, request(WS_METHODS.pullRequestsRoutingIdentity, { host: identity.host }))
400 .run(
401 id,
402 reads.has(tag)
403 ? shareMetadataRead(pendingIdentities, identity.host.toLowerCase(), alternateProbe)
404 : alternateProbe,
405 )
343406 .pipe(
344407 Effect.timeout("2 seconds"),
345408 Effect.catchCause((cause) =>
packages/client-runtime/src/state/pullRequests.test.ts+262-1
......@@ -236,6 +236,267 @@ for (const scenario of [
236236 );
237237}
238238
239it.effect("shares active read identity probes while keeping later reads and mutations fresh", () =>
240 Effect.scoped(
241 Effect.gen(function* () {
242 const probes: string[] = [];
243 const reads: number[] = [];
244 const mutations: string[] = [];
245 let accountId = "123";
246 const clientFor = (local: boolean) =>
247 ({
248 [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
249 Effect.yieldNow.pipe(
250 Effect.andThen(
251 Effect.sync(() => {
252 probes.push(local ? "local" : "origin");
253 return { host: "github.com", provider: "github", viewer: "maria", accountId };
254 }),
255 ),
256 ),
257 [WS_METHODS.pullRequestsSummary]: (input: {
258 number: number;
259 expectedAccountId: string;
260 }) =>
261 Effect.sync(() => {
262 expect(local).toBe(true);
263 expect(input.expectedAccountId).toBe(accountId);
264 reads.push(input.number);
265 return null;
266 }),
267 [WS_METHODS.pullRequestsRunAction]: (input: { expectedAccountId: string }) =>
268 Effect.sync(() => mutations.push(input.expectedAccountId)),
269 [WS_METHODS.pullRequestsInvalidate]: () => Effect.void,
270 }) as unknown as WsRpcProtocolClient;
271 const { environmentRegistry, supervisor } = yield* makeTestRuntime(
272 clientFor(false),
273 clientFor(true),
274 );
275 const reference = {
276 projectId: ProjectId.make("project-1"),
277 repository: "acme/web",
278 number: 7,
279 };
280 yield* Effect.gen(function* () {
281 yield* Effect.all(
282 Array.from({ length: 4 }, (_, index) =>
283 createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
284 ...reference,
285 number: index + 1,
286 }),
287 ),
288 { concurrency: "unbounded" },
289 );
290 expect(probes).toEqual(["origin", "local"]);
291 expect(reads.toSorted()).toEqual([1, 2, 3, 4]);
292 accountId = "456";
293 yield* createPullRequestRouter()(WS_METHODS.pullRequestsSummary, reference);
294 yield* createPullRequestRouter()(WS_METHODS.pullRequestsRunAction, {
295 ...reference,
296 action: "close",
297 });
298 expect(probes).toEqual(["origin", "local", "origin", "local", "origin", "local"]);
299 expect(mutations).toEqual(["456"]);
300 }).pipe(
301 Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
302 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
303 Effect.provideService(GitHubRoutingPermissions, trustedRouting),
304 );
305 }),
306 ),
307);
308
309it.effect.each([
310 { probe: "origin", cancelled: "first" },
311 { probe: "origin", cancelled: "waiter" },
312 { probe: "alternate", cancelled: "first" },
313 { probe: "alternate", cancelled: "waiter" },
314] as const)(
315 "preserves a shared $probe probe's other reader when the $cancelled cancels",
316 ({ probe, cancelled }) =>
317 Effect.scoped(
318 Effect.gen(function* () {
319 const started = yield* Deferred.make<void>();
320 const release = yield* Deferred.make<void>();
321 let probes = 0;
322 const clientFor = (local: boolean) =>
323 ({
324 [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
325 Effect.gen(function* () {
326 if (local === (probe === "alternate")) {
327 probes++;
328 yield* Deferred.succeed(started, undefined);
329 yield* Deferred.await(release);
330 }
331 return {
332 host: "github.com",
333 provider: "github",
334 viewer: "maria",
335 accountId: "123",
336 };
337 }),
338 [WS_METHODS.pullRequestsSummary]: () => Effect.succeed(null),
339 }) as unknown as WsRpcProtocolClient;
340 const { environmentRegistry, supervisor } = yield* makeTestRuntime(
341 clientFor(false),
342 clientFor(true),
343 );
344 const read = () =>
345 createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
346 projectId: ProjectId.make("project-1"),
347 repository: "acme/web",
348 number: 7,
349 }).pipe(
350 Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
351 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
352 Effect.provideService(GitHubRoutingPermissions, trustedRouting),
353 );
354 const first = yield* read().pipe(Effect.forkChild);
355 yield* Deferred.await(started);
356 const waiter = yield* read().pipe(Effect.forkChild);
357 yield* Effect.yieldNow;
358 yield* Fiber.interrupt(cancelled === "first" ? first : waiter);
359 yield* Deferred.succeed(release, undefined);
360 expect(yield* Fiber.join(cancelled === "first" ? waiter : first)).toBeNull();
361 expect(probes).toBe(cancelled === "first" ? 2 : 1);
362 yield* read();
363 expect(probes).toBe(cancelled === "first" ? 3 : 2);
364 }),
365 ),
366);
367
368it.effect.each(["revoked", "disabled"] as const)(
369 "rechecks routing after an alternate with a shared probe is %s",
370 (change) =>
371 Effect.scoped(
372 Effect.gen(function* () {
373 const started = yield* Deferred.make<void>();
374 const release = yield* Deferred.make<void>();
375 const reads: string[] = [];
376 let trusted = true;
377 let probes = 0;
378 const clientFor = (local: boolean) =>
379 ({
380 [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
381 Effect.gen(function* () {
382 if (local) {
383 probes++;
384 yield* Deferred.succeed(started, undefined);
385 yield* Deferred.await(release);
386 }
387 return {
388 host: "github.com",
389 provider: "github",
390 viewer: "maria",
391 accountId: "123",
392 };
393 }),
394 [WS_METHODS.pullRequestsSummary]: () =>
395 Effect.sync(() => {
396 reads.push(local ? "local" : "origin");
397 return null;
398 }),
399 }) as unknown as WsRpcProtocolClient;
400 const { environmentRegistry, supervisor } = yield* makeTestRuntime(
401 clientFor(false),
402 clientFor(true),
403 );
404 const readers = yield* Effect.all(
405 Array.from({ length: 2 }, () =>
406 createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
407 projectId: ProjectId.make("project-1"),
408 repository: "acme/web",
409 number: 7,
410 }),
411 ),
412 { concurrency: "unbounded" },
413 ).pipe(
414 Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
415 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
416 Effect.provideService(GitHubRoutingPermissions, {
417 ...trustedRouting,
418 get: () => Effect.sync(() => (trusted ? "read-write" : "off")),
419 }),
420 Effect.forkChild,
421 );
422 yield* Deferred.await(started);
423 yield* Effect.yieldNow;
424 if (change === "revoked") trusted = false;
425 else
426 yield* SubscriptionRef.update(
427 environmentRegistry.entries,
428 (entries) =>
429 new Map(
430 [...entries].map(([id, entry]) => [
431 id,
432 id === TARGET.environmentId ? entry : { ...entry, enabled: false },
433 ]),
434 ),
435 );
436 yield* Deferred.succeed(release, undefined);
437 expect(yield* Fiber.join(readers)).toEqual([null, null]);
438 expect(probes).toBe(1);
439 expect(reads).toEqual(["origin", "origin"]);
440 }),
441 ),
442);
443
444it.effect.each(["origin", "alternate"] as const)(
445 "keeps pending metadata from an old %s connection separate from its replacement",
446 (reconnected) =>
447 Effect.scoped(
448 Effect.gen(function* () {
449 const started = yield* Deferred.make<void>();
450 const release = yield* Deferred.make<void>();
451 const probes: string[] = [];
452 const clientFor = (local: boolean, replacement = false) =>
453 ({
454 [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
455 Effect.gen(function* () {
456 if (local === (reconnected === "alternate")) {
457 probes.push(replacement ? "new" : "old");
458 if (!replacement) {
459 yield* Deferred.succeed(started, undefined);
460 yield* Deferred.await(release);
461 }
462 }
463 return {
464 host: "github.com",
465 provider: "github",
466 viewer: "maria",
467 accountId: "123",
468 };
469 }),
470 [WS_METHODS.pullRequestsSummary]: () => Effect.succeed(null),
471 }) as unknown as WsRpcProtocolClient;
472 const { environmentRegistry, supervisor, localSupervisor } = yield* makeTestRuntime(
473 clientFor(false),
474 clientFor(true),
475 );
476 const read = () =>
477 createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
478 projectId: ProjectId.make("project-1"),
479 repository: "acme/web",
480 number: 7,
481 }).pipe(
482 Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
483 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
484 Effect.provideService(GitHubRoutingPermissions, trustedRouting),
485 );
486 const first = yield* read().pipe(Effect.forkChild);
487 yield* Deferred.await(started);
488 yield* SubscriptionRef.set(
489 (reconnected === "origin" ? supervisor : localSupervisor).session,
490 Option.some(session(clientFor(reconnected === "alternate", true))),
491 );
492 expect(yield* read()).toBeNull();
493 expect(probes).toEqual(["old", "new"]);
494 yield* Deferred.succeed(release, undefined);
495 expect(yield* Fiber.join(first)).toBeNull();
496 }),
497 ),
498);
499
239500it.effect.each(["github", "gitlab", "bitbucket", "azure-devops"] as const)(
240501 "routes %s viewed marks to their storage environment",
241502 (provider) =>
......@@ -417,7 +678,7 @@ const makeTestRuntime = Effect.fn("makeTestRuntime")(function* (
417678 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
418679 Effect.sync(() => registry.dispose()),
419680 );
420 return { runtime, atoms, registry, environmentRegistry, supervisor };
681 return { runtime, atoms, registry, environmentRegistry, supervisor, localSupervisor };
421682});
422683
423684it.effect.each(["default", "origin-off", "destination-off", "read-only"] as const)(