| author | |
| committer | |
| log | 4df84a7d03ff084ff4c6dd7aef54623984690fb7 |
| tree | 14287fcd41a89100453354de2d80b54579c675fb |
| parent | 758dc290e900e35958d37fec41fed4a357ad60fa |
| signature | Signed by PGP key B5690EEEBB952194 |
Takes over #13841. A PR query refreshes on the server's refresh signal only while something reads it, and the server shares detail, activity, and preview for 60 seconds, or 10 minutes once merged.
Co-authored-by: Wout Stiens <71498452+StiensWout@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>4 files changed, 228 insertions(+), 13 deletions(-)
apps/server/src/pullRequest/PullRequestService.test.ts+68-5| ... | ... | @@ -236,7 +236,7 @@ it.effect("reuses only unexpired detail for previews", () => |
| 236 | 236 | yield* service.detail(ref); |
| 237 | 237 | assert.strictEqual((yield* service.preview(ref)).title, "Change request 1"); |
| 238 | 238 | assert.strictEqual(previewReads, 0); |
| 239 | yield* TestClock.adjust("16 seconds"); | |
| 239 | yield* TestClock.adjust("61 seconds"); | |
| 240 | 240 | const preview = yield* service.preview(ref); |
| 241 | 241 | assert.strictEqual(preview.title, "Updated title"); |
| 242 | 242 | assert.strictEqual(preview.state, "closed"); |
| ... | ... | @@ -244,6 +244,69 @@ it.effect("reuses only unexpired detail for previews", () => |
| 244 | 244 | }), |
| 245 | 245 | ); |
| 246 | 246 | |
| 247 | it.effect("shares detail and activity for a minute, and for ten once merged", () => | |
| 248 | Effect.gen(function* () { | |
| 249 | const reads: Array<string> = []; | |
| 250 | const service = yield* makeService({ | |
| 251 | projects: [project({ id: "p1", title: "web", workspaceRoot: "/w", repository: "acme/web" })], | |
| 252 | providers: [ | |
| 253 | fakeProvider("github", { | |
| 254 | getChangeRequest: ({ number }) => | |
| 255 | Effect.sync(() => { | |
| 256 | reads.push(`detail #${number}`); | |
| 257 | return { | |
| 258 | ...hostedChangeRequest("Description"), | |
| 259 | number, | |
| 260 | state: | |
| 261 | number === 1 | |
| 262 | ? ("open" as const) | |
| 263 | : number === 2 | |
| 264 | ? ("merged" as const) | |
| 265 | : ("closed" as const), | |
| 266 | }; | |
| 267 | }), | |
| 268 | getChangeRequestActivity: ({ number }) => | |
| 269 | Effect.sync(() => { | |
| 270 | reads.push(`activity #${number}`); | |
| 271 | return { | |
| 272 | comments: [], | |
| 273 | commentCount: 0, | |
| 274 | commentsTruncated: false, | |
| 275 | reviewThreads: [], | |
| 276 | commits: [], | |
| 277 | }; | |
| 278 | }), | |
| 279 | }), | |
| 280 | ], | |
| 281 | }); | |
| 282 | // What every client does on focus or a refresh signal for each panel it shows. | |
| 283 | // #1 is open, #2 merged, and #3 closed, which can reopen and so is held no longer than #1. | |
| 284 | const reread = Effect.forEach([1, 2, 3], (number) => { | |
| 285 | const ref = { projectId: "p1" as ProjectId, repository: "acme/web", number }; | |
| 286 | return Effect.andThen(service.detail(ref), service.activity(ref)); | |
| 287 | }).pipe( | |
| 288 | Effect.andThen(Effect.yieldNow), | |
| 289 | Effect.andThen(Effect.sync(() => reads.splice(0).toSorted())), | |
| 290 | ); | |
| 291 | ||
| 292 | const everything = [ | |
| 293 | "activity #1", | |
| 294 | "activity #2", | |
| 295 | "activity #3", | |
| 296 | "detail #1", | |
| 297 | "detail #2", | |
| 298 | "detail #3", | |
| 299 | ]; | |
| 300 | assert.deepStrictEqual(yield* reread, everything); | |
| 301 | yield* TestClock.adjust("59 seconds"); | |
| 302 | assert.deepStrictEqual(yield* reread, []); | |
| 303 | yield* TestClock.adjust("2 seconds"); | |
| 304 | assert.deepStrictEqual(yield* reread, ["activity #1", "activity #3", "detail #1", "detail #3"]); | |
| 305 | yield* TestClock.adjust("9 minutes"); | |
| 306 | assert.deepStrictEqual(yield* reread, everything); | |
| 307 | }), | |
| 308 | ); | |
| 309 | ||
| 247 | 310 | it.effect("does not wait for an in-flight detail read to display a preview", () => |
| 248 | 311 | Effect.gen(function* () { |
| 249 | 312 | const detailStarted = yield* Deferred.make<void>(); |
| ... | ... | @@ -1865,7 +1928,7 @@ for (const [provider, host] of [ |
| 1865 | 1928 | service[read](reference); |
| 1866 | 1929 | yield* Effect.all([request, request], { concurrency: 2 }); |
| 1867 | 1930 | assert.strictEqual(calls, 1); |
| 1868 | yield* TestClock.adjust("45 seconds"); | |
| 1931 | yield* TestClock.adjust("61 seconds"); | |
| 1869 | 1932 | limited = true; |
| 1870 | 1933 | assert.strictEqual((yield* Effect.exit(request))._tag, "Failure"); |
| 1871 | 1934 | assert.strictEqual(calls, 2); |
| ... | ... | @@ -4314,7 +4377,7 @@ it.effect("reads the fresh diff when detail or summary discovers a changed revis |
| 4314 | 4377 | assert.strictEqual((yield* service.diff(reference)).patch, "old patch"); |
| 4315 | 4378 | revision = "2026-07-02T00:01:00Z"; |
| 4316 | 4379 | patch = "new patch"; |
| 4317 | yield* TestClock.adjust("16 seconds"); | |
| 4380 | yield* TestClock.adjust("61 seconds"); | |
| 4318 | 4381 | yield* service.detail(reference); |
| 4319 | 4382 | yield* Effect.yieldNow; |
| 4320 | 4383 | assert.strictEqual((yield* service.detail(reference)).updatedAt, revision); |
| ... | ... | @@ -4698,7 +4761,7 @@ it.effect("answers a known pull request immediately while the host refreshes", ( |
| 4698 | 4761 | assert.strictEqual(first.body, "cached body"); |
| 4699 | 4762 | assert.strictEqual(first.additions, 4); |
| 4700 | 4763 | |
| 4701 | yield* TestClock.adjust("16 seconds"); | |
| 4764 | yield* TestClock.adjust("61 seconds"); | |
| 4702 | 4765 | const second = yield* service.detail(reference); |
| 4703 | 4766 | assert.strictEqual(second.body, "cached body"); |
| 4704 | 4767 | assert.strictEqual(second.additions, 4); |
| ... | ... | @@ -4962,7 +5025,7 @@ it.effect("keeps recent detail on a transient refresh failure but not after inva |
| 4962 | 5025 | }); |
| 4963 | 5026 | |
| 4964 | 5027 | yield* service.detail(reference); |
| 4965 | yield* TestClock.adjust("16 seconds"); | |
| 5028 | yield* TestClock.adjust("61 seconds"); | |
| 4966 | 5029 | failing = true; |
| 4967 | 5030 | const strict = yield* Effect.flip(service.detail({ ...reference, allowStale: false })); |
| 4968 | 5031 | assert.strictEqual(strict._tag, "PullRequestOperationError"); |
apps/server/src/pullRequest/PullRequestService.ts+25-6| ... | ... | @@ -61,6 +61,7 @@ import { |
| 61 | 61 | type PullRequestLabelCandidateList, |
| 62 | 62 | type PullRequestLabelChangeInput, |
| 63 | 63 | type PullRequestSetFilesViewedInput, |
| 64 | type PullRequestState, | |
| 64 | 65 | type PullRequestSubmitReviewInput, |
| 65 | 66 | PullRequestStack, |
| 66 | 67 | PullRequestSummary, |
| ... | ... | @@ -134,7 +135,21 @@ const REPOSITORY_SEARCH_CHUNK = 100; |
| 134 | 135 | * `invalidate` rather than a flag on the read, so an ordinary read can never opt out. |
| 135 | 136 | */ |
| 136 | 137 | const LIST_CACHE_TTL = Duration.seconds(30); |
| 137 | const DETAIL_CACHE_TTL = Duration.seconds(15); | |
| 138 | /** | |
| 139 | * Each connected client re-reads the pull request it shows on focus, on a timer, and after every | |
| 140 | * turn, so detail, activity, and preview answers are shared for the clients' own minute of | |
| 141 | * staleness. A merged change request cannot change again and is held for ten. A closed one is | |
| 142 | * not: it can reopen, and a watch started on it reads it to learn whether it did. | |
| 143 | */ | |
| 144 | const DETAIL_CACHE_TTL = Duration.seconds(60); | |
| 145 | const MERGED_DETAIL_CACHE_TTL = Duration.minutes(10); | |
| 146 | const detailTimeToLive = (state: PullRequestState | undefined) => | |
| 147 | state === "merged" ? MERGED_DETAIL_CACHE_TTL : DETAIL_CACHE_TTL; | |
| 148 | /** | |
| 149 | * Checks and the watch fingerprint stay short: they are how a running check's result reaches the | |
| 150 | * page and a watched pull request's change reaches its watch. | |
| 151 | */ | |
| 152 | const CHECKS_CACHE_TTL = Duration.seconds(15); | |
| 138 | 153 | const DIFF_CACHE_TTL = Duration.seconds(60); |
| 139 | 154 | /** A commit is content-addressed, so its own diff cannot change under its key. */ |
| 140 | 155 | const COMMIT_DIFF_CACHE_TTL = Duration.minutes(10); |
| ... | ... | @@ -2961,7 +2976,7 @@ export const make = Effect.gen(function* () { |
| 2961 | 2976 | }, |
| 2962 | 2977 | { |
| 2963 | 2978 | capacity: DETAIL_CACHE_CAPACITY, |
| 2964 | timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), | |
| 2979 | timeToLive: (exit) => (Exit.isSuccess(exit) ? CHECKS_CACHE_TTL : Duration.zero), | |
| 2965 | 2980 | }, |
| 2966 | 2981 | ); |
| 2967 | 2982 | |
| ... | ... | @@ -2985,7 +3000,7 @@ export const make = Effect.gen(function* () { |
| 2985 | 3000 | }, |
| 2986 | 3001 | { |
| 2987 | 3002 | capacity: DETAIL_CACHE_CAPACITY, |
| 2988 | timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), | |
| 3003 | timeToLive: (exit) => (Exit.isSuccess(exit) ? CHECKS_CACHE_TTL : Duration.zero), | |
| 2989 | 3004 | }, |
| 2990 | 3005 | ); |
| 2991 | 3006 | |
| ... | ... | @@ -3012,7 +3027,8 @@ export const make = Effect.gen(function* () { |
| 3012 | 3027 | }, |
| 3013 | 3028 | { |
| 3014 | 3029 | capacity: DETAIL_CACHE_CAPACITY, |
| 3015 | timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), | |
| 3030 | timeToLive: (exit) => | |
| 3031 | Exit.isSuccess(exit) ? detailTimeToLive(exit.value.state) : Duration.zero, | |
| 3016 | 3032 | }, |
| 3017 | 3033 | ); |
| 3018 | 3034 | const summaryFromDetail = ( |
| ... | ... | @@ -3073,7 +3089,9 @@ export const make = Effect.gen(function* () { |
| 3073 | 3089 | }, |
| 3074 | 3090 | { |
| 3075 | 3091 | capacity: DETAIL_CACHE_CAPACITY, |
| 3076 | timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), | |
| 3092 | // Activity carries no state of its own; the detail or summary read under the same key does. | |
| 3093 | timeToLive: (exit, key) => | |
| 3094 | Exit.isSuccess(exit) ? detailTimeToLive(lastGoodSummary.peek(key)?.state) : Duration.zero, | |
| 3077 | 3095 | }, |
| 3078 | 3096 | ); |
| 3079 | 3097 | |
| ... | ... | @@ -3083,7 +3101,8 @@ export const make = Effect.gen(function* () { |
| 3083 | 3101 | }, |
| 3084 | 3102 | { |
| 3085 | 3103 | capacity: DETAIL_CACHE_CAPACITY, |
| 3086 | timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), | |
| 3104 | timeToLive: (exit) => | |
| 3105 | Exit.isSuccess(exit) ? detailTimeToLive(exit.value.state) : Duration.zero, | |
| 3087 | 3106 | }, |
| 3088 | 3107 | ); |
| 3089 | 3108 | const preview: PullRequestService["Service"]["preview"] = (input) => { |
packages/client-runtime/src/state/runtime.test.ts+105-1| ... | ... | @@ -1,4 +1,4 @@ |
| 1 | import { describe, expect, it } from "@effect/vitest"; | |
| 1 | import { describe, expect, it, vi } from "@effect/vitest"; | |
| 2 | 2 | import { EnvironmentId } from "@t3tools/contracts"; |
| 3 | 3 | import * as Cause from "effect/Cause"; |
| 4 | 4 | import * as Deferred from "effect/Deferred"; |
| ... | ... | @@ -81,6 +81,10 @@ function queryConnectionState( |
| 81 | 81 | |
| 82 | 82 | const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness")(function* <A, E>( |
| 83 | 83 | execute: Effect.Effect<A, E>, |
| 84 | { | |
| 85 | refreshTrigger, | |
| 86 | idleTtlMs, | |
| 87 | }: { readonly refreshTrigger?: Atom.Atom<unknown>; readonly idleTtlMs?: number } = {}, | |
| 84 | 88 | ) { |
| 85 | 89 | const supervisorState = yield* SubscriptionRef.make(queryConnectionState()); |
| 86 | 90 | const supervisorSession = yield* SubscriptionRef.make(Option.some(QUERY_RPC_SESSION)); |
| ... | ... | @@ -111,6 +115,8 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" |
| 111 | 115 | label: "test.environment-query", |
| 112 | 116 | staleTimeMs: 60_000, |
| 113 | 117 | execute: () => execute, |
| 118 | ...(refreshTrigger === undefined ? {} : { refreshTrigger: () => refreshTrigger }), | |
| 119 | ...(idleTtlMs === undefined ? {} : { idleTtlMs }), | |
| 114 | 120 | }); |
| 115 | 121 | |
| 116 | 122 | return { |
| ... | ... | @@ -120,6 +126,18 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" |
| 120 | 126 | }; |
| 121 | 127 | }); |
| 122 | 128 | |
| 129 | /** | |
| 130 | * Reads a query's settled value without leaving it mounted. `getResult` resumes from inside the | |
| 131 | * query's listener, so it only unsubscribes once the reader yields. | |
| 132 | */ | |
| 133 | const readEnvironmentQuery = <A, E>( | |
| 134 | registry: AtomRegistry.AtomRegistry, | |
| 135 | atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>, | |
| 136 | ) => | |
| 137 | AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }).pipe( | |
| 138 | Effect.tap(() => Effect.yieldNow), | |
| 139 | ); | |
| 140 | ||
| 123 | 141 | const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* <A, E>( |
| 124 | 142 | atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>, |
| 125 | 143 | ) { |
| ... | ... | @@ -548,6 +566,92 @@ describe("environment query lifecycle", () => { |
| 548 | 566 | }), |
| 549 | 567 | ), |
| 550 | 568 | ); |
| 569 | ||
| 570 | it.effect("refreshes on its trigger while read, and once on its next read when idle", () => | |
| 571 | Effect.scoped( | |
| 572 | Effect.gen(function* () { | |
| 573 | let executions = 0; | |
| 574 | const trigger = Atom.make(0).pipe(Atom.keepAlive); | |
| 575 | const harness = yield* makeEnvironmentQueryHarness( | |
| 576 | Effect.sync(() => (executions += 1)), | |
| 577 | { refreshTrigger: trigger }, | |
| 578 | ); | |
| 579 | const registry = AtomRegistry.make(); | |
| 580 | yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); | |
| 581 | const read = readEnvironmentQuery(registry, harness.atom); | |
| 582 | ||
| 583 | const unmount = registry.mount(harness.atom); | |
| 584 | expect(yield* read).toBe(1); | |
| 585 | registry.set(trigger, 1); | |
| 586 | expect(yield* read).toBe(2); | |
| 587 | ||
| 588 | // Unmounted, the query stays alive on its idle TTL, but nothing reads it. | |
| 589 | unmount(); | |
| 590 | registry.set(trigger, 2); | |
| 591 | registry.set(trigger, 3); | |
| 592 | expect(executions).toBe(2); | |
| 593 | ||
| 594 | expect(yield* read).toBe(3); | |
| 595 | }), | |
| 596 | ), | |
| 597 | ); | |
| 598 | ||
| 599 | it.effect("keeps an idle query when its trigger recomputes to the same value", () => | |
| 600 | Effect.scoped( | |
| 601 | Effect.gen(function* () { | |
| 602 | let executions = 0; | |
| 603 | const settings = Atom.make({ theme: "dark" }).pipe(Atom.keepAlive); | |
| 604 | const trigger = Atom.make((get) => JSON.stringify([get(settings) !== null])); | |
| 605 | const harness = yield* makeEnvironmentQueryHarness( | |
| 606 | Effect.sync(() => (executions += 1)), | |
| 607 | { refreshTrigger: trigger }, | |
| 608 | ); | |
| 609 | const registry = AtomRegistry.make(); | |
| 610 | yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); | |
| 611 | const read = readEnvironmentQuery(registry, harness.atom); | |
| 612 | ||
| 613 | const unmount = registry.mount(harness.atom); | |
| 614 | expect(yield* read).toBe(1); | |
| 615 | unmount(); | |
| 616 | registry.set(settings, { theme: "light" }); | |
| 617 | expect(yield* read).toBe(1); | |
| 618 | expect(executions).toBe(1); | |
| 619 | }), | |
| 620 | ), | |
| 621 | ); | |
| 622 | ||
| 623 | it.effect("refreshes an idle query the registry swept and rebuilt after its trigger moved", () => | |
| 624 | Effect.scoped( | |
| 625 | Effect.gen(function* () { | |
| 626 | // Only the registry's idle-TTL sweeps run on these; effects keep their own scheduler. | |
| 627 | vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); | |
| 628 | yield* Effect.addFinalizer(() => Effect.sync(() => vi.useRealTimers())); | |
| 629 | let executions = 0; | |
| 630 | const trigger = Atom.make(0).pipe(Atom.keepAlive); | |
| 631 | const harness = yield* makeEnvironmentQueryHarness( | |
| 632 | Effect.sync(() => (executions += 1)), | |
| 633 | { refreshTrigger: trigger, idleTtlMs: 1_000 }, | |
| 634 | ); | |
| 635 | const registry = AtomRegistry.make({ timeoutResolution: 100 }); | |
| 636 | yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); | |
| 637 | const read = readEnvironmentQuery(registry, harness.atom); | |
| 638 | ||
| 639 | const unmount = registry.mount(harness.atom); | |
| 640 | expect(yield* read).toBe(1); | |
| 641 | unmount(); | |
| 642 | // Lets the registry start the query node's idle TTL. | |
| 643 | yield* Effect.yieldNow; | |
| 644 | vi.advanceTimersByTime(500); | |
| 645 | registry.set(trigger, 1); | |
| 646 | yield* Effect.yieldNow; | |
| 647 | // Sweeps the query node but not the data it read, whose idle TTL started with the trigger. | |
| 648 | vi.advanceTimersByTime(700); | |
| 649 | expect(executions).toBe(1); | |
| 650 | ||
| 651 | expect(yield* read).toBe(2); | |
| 652 | }), | |
| 653 | ), | |
| 654 | ); | |
| 551 | 655 | }); |
| 552 | 656 | |
| 553 | 657 | describe("Atom.fn mutation semantics", () => { |
packages/client-runtime/src/state/runtime.ts+30-1| ... | ... | @@ -489,6 +489,35 @@ export function followStreamInEnvironment<A, E, R>( |
| 489 | 489 | ); |
| 490 | 490 | } |
| 491 | 491 | |
| 492 | /** | |
| 493 | * Refreshes a query when `signal` changes, but only while something reads it. | |
| 494 | * `Atom.makeRefreshOnSignal` subscribes to the signal, so a query that outlives its view on an | |
| 495 | * idle TTL still re-runs on every change. Reading the signal as a dependency instead lets the | |
| 496 | * registry re-run a mounted query right away and only mark an idle one stale, which then | |
| 497 | * refreshes on its next read. | |
| 498 | */ | |
| 499 | const refreshOnSignalWhileRead = | |
| 500 | (signal: Atom.Atom<unknown>) => | |
| 501 | <A>(self: Atom.Atom<A>): Atom.Atom<A> => { | |
| 502 | // The signal value each registry's data was read under. Held outside the node because the | |
| 503 | // registry can sweep this node while `self` keeps the data, and the rebuilt node must still | |
| 504 | // see a change it missed. | |
| 505 | const readUnder = new WeakMap<AtomRegistry.AtomRegistry, unknown>(); | |
| 506 | return Atom.transform( | |
| 507 | self, | |
| 508 | (get) => { | |
| 509 | const current = get(signal); | |
| 510 | get.subscribe(self, (value) => get.setSelf(value)); | |
| 511 | const changed = | |
| 512 | readUnder.has(get.registry) && !Object.is(readUnder.get(get.registry), current); | |
| 513 | readUnder.set(get.registry, current); | |
| 514 | if (changed) get.refresh(self); | |
| 515 | return get.once(self); | |
| 516 | }, | |
| 517 | { initialValueTarget: self }, | |
| 518 | ); | |
| 519 | }; | |
| 520 | ||
| 492 | 521 | export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( |
| 493 | 522 | runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>, |
| 494 | 523 | options: EnvironmentQueryAtomOptions< |
| ... | ... | @@ -578,7 +607,7 @@ export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>( |
| 578 | 607 | return ( |
| 579 | 608 | refreshTrigger === undefined |
| 580 | 609 | ? intervalQuery |
| 581 | : intervalQuery.pipe(Atom.makeRefreshOnSignal(refreshTrigger)) | |
| 610 | : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger)) | |
| 582 | 611 | ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); |
| 583 | 612 | }); |
| 584 | 613 | return (target) => family(environmentRpcKey(target)); |