diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 4ba438a97da9..f32d90757df2 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -229,7 +229,7 @@ it.effect("reuses only unexpired detail for previews", () => yield* service.detail(ref); assert.strictEqual((yield* service.preview(ref)).title, "Change request 1"); assert.strictEqual(previewReads, 0); - yield* TestClock.adjust("16 seconds"); + yield* TestClock.adjust("61 seconds"); const preview = yield* service.preview(ref); assert.strictEqual(preview.title, "Updated title"); assert.strictEqual(preview.state, "closed"); @@ -237,6 +237,55 @@ it.effect("reuses only unexpired detail for previews", () => }), ); +it.effect("shares detail and activity for a minute, and for ten once merged", () => + Effect.gen(function* () { + const reads: Array = []; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/w", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + getChangeRequest: ({ number }) => + Effect.sync(() => { + reads.push(`detail #${number}`); + return { + ...hostedChangeRequest("Description"), + number, + state: number === 1 ? ("open" as const) : ("merged" as const), + }; + }), + getChangeRequestActivity: ({ number }) => + Effect.sync(() => { + reads.push(`activity #${number}`); + return { + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }; + }), + }), + ], + }); + // What every client does on focus or a refresh signal for each panel it shows. + const reread = Effect.forEach([1, 2], (number) => { + const ref = { projectId: "p1" as ProjectId, repository: "acme/web", number }; + return Effect.andThen(service.detail(ref), service.activity(ref)); + }).pipe( + Effect.andThen(Effect.yieldNow), + Effect.andThen(Effect.sync(() => reads.splice(0).toSorted())), + ); + + assert.deepStrictEqual(yield* reread, ["activity #1", "activity #2", "detail #1", "detail #2"]); + yield* TestClock.adjust("59 seconds"); + assert.deepStrictEqual(yield* reread, []); + yield* TestClock.adjust("2 seconds"); + assert.deepStrictEqual(yield* reread, ["activity #1", "detail #1"]); + yield* TestClock.adjust("9 minutes"); + assert.deepStrictEqual(yield* reread, ["activity #1", "activity #2", "detail #1", "detail #2"]); + }), +); + it.effect("does not wait for an in-flight detail read to display a preview", () => Effect.gen(function* () { const detailStarted = yield* Deferred.make(); @@ -4121,7 +4170,7 @@ it.effect("reads the fresh diff when detail or summary discovers a changed revis assert.strictEqual((yield* service.diff(reference)).patch, "old patch"); revision = "2026-07-02T00:01:00Z"; patch = "new patch"; - yield* TestClock.adjust("16 seconds"); + yield* TestClock.adjust("61 seconds"); yield* service.detail(reference); yield* Effect.yieldNow; assert.strictEqual((yield* service.detail(reference)).updatedAt, revision); @@ -4491,7 +4540,7 @@ it.effect("answers a known pull request immediately while the host refreshes", ( assert.strictEqual(first.body, "cached body"); assert.strictEqual(first.additions, 4); - yield* TestClock.adjust("16 seconds"); + yield* TestClock.adjust("61 seconds"); const second = yield* service.detail(reference); assert.strictEqual(second.body, "cached body"); assert.strictEqual(second.additions, 4); @@ -4728,7 +4777,7 @@ it.effect("keeps recent detail on a transient refresh failure but not after inva }); yield* service.detail(reference); - yield* TestClock.adjust("16 seconds"); + yield* TestClock.adjust("61 seconds"); failing = true; const strict = yield* Effect.flip(service.detail({ ...reference, allowStale: false })); assert.strictEqual(strict._tag, "PullRequestOperationError"); diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 866a45e49b09..54281c5e1542 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -59,6 +59,7 @@ import { type PullRequestLabelCandidateList, type PullRequestLabelChangeInput, type PullRequestSetFilesViewedInput, + type PullRequestState, type PullRequestSubmitReviewInput, PullRequestStack, PullRequestSummary, @@ -128,7 +129,15 @@ const REPOSITORY_SEARCH_CHUNK = 100; * `invalidate` rather than a flag on the read, so an ordinary read can never opt out. */ const LIST_CACHE_TTL = Duration.seconds(30); -const DETAIL_CACHE_TTL = Duration.seconds(15); +/** + * Each connected client re-reads the pull request it shows on focus, on a timer, and after every + * turn, so detail, activity, and preview answers are shared for the clients' own minute of + * staleness. A merged or closed change request stops moving on its own and is held for ten. + */ +const DETAIL_CACHE_TTL = Duration.seconds(60); +const SETTLED_DETAIL_CACHE_TTL = Duration.minutes(10); +const detailTimeToLive = (state: PullRequestState | undefined) => + state === "merged" || state === "closed" ? SETTLED_DETAIL_CACHE_TTL : DETAIL_CACHE_TTL; const DIFF_CACHE_TTL = Duration.seconds(60); /** A commit is content-addressed, so its own diff cannot change under its key. */ const COMMIT_DIFF_CACHE_TTL = Duration.minutes(10); @@ -2880,7 +2889,8 @@ export const make = Effect.gen(function* () { }, { capacity: DETAIL_CACHE_CAPACITY, - timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), + timeToLive: (exit) => + Exit.isSuccess(exit) ? detailTimeToLive(exit.value.state) : Duration.zero, }, ); const summaryFromDetail = ( @@ -2941,7 +2951,9 @@ export const make = Effect.gen(function* () { }, { capacity: DETAIL_CACHE_CAPACITY, - timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), + // Activity carries no state of its own; the detail or summary read under the same key does. + timeToLive: (exit, key) => + Exit.isSuccess(exit) ? detailTimeToLive(lastGoodSummary.peek(key)?.state) : Duration.zero, }, ); @@ -2951,7 +2963,8 @@ export const make = Effect.gen(function* () { }, { capacity: DETAIL_CACHE_CAPACITY, - timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), + timeToLive: (exit) => + Exit.isSuccess(exit) ? detailTimeToLive(exit.value.state) : Duration.zero, }, ); const preview: PullRequestService["Service"]["preview"] = (input) => { diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index d04f313d26e5..37b0cdfabb0c 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "@effect/vitest"; +import { describe, expect, it, vi } from "@effect/vitest"; import { EnvironmentId } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; @@ -81,6 +81,10 @@ function queryConnectionState( const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness")(function* ( execute: Effect.Effect, + { + refreshTrigger, + idleTtlMs, + }: { readonly refreshTrigger?: Atom.Atom; readonly idleTtlMs?: number } = {}, ) { const supervisorState = yield* SubscriptionRef.make(queryConnectionState()); const supervisorSession = yield* SubscriptionRef.make(Option.some(QUERY_RPC_SESSION)); @@ -111,6 +115,8 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" label: "test.environment-query", staleTimeMs: 60_000, execute: () => execute, + ...(refreshTrigger === undefined ? {} : { refreshTrigger: () => refreshTrigger }), + ...(idleTtlMs === undefined ? {} : { idleTtlMs }), }); return { @@ -120,6 +126,18 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" }; }); +/** + * Reads a query's settled value without leaving it mounted. `getResult` resumes from inside the + * query's listener, so it only unsubscribes once the reader yields. + */ +const readEnvironmentQuery = ( + registry: AtomRegistry.AtomRegistry, + atom: Atom.Atom>, +) => + AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }).pipe( + Effect.tap(() => Effect.yieldNow), + ); + const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* ( atom: Atom.Atom>, ) { @@ -548,6 +566,92 @@ describe("environment query lifecycle", () => { }), ), ); + + it.effect("refreshes on its trigger while read, and once on its next read when idle", () => + Effect.scoped( + Effect.gen(function* () { + let executions = 0; + const trigger = Atom.make(0).pipe(Atom.keepAlive); + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => (executions += 1)), + { refreshTrigger: trigger }, + ); + const registry = AtomRegistry.make(); + yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); + const read = readEnvironmentQuery(registry, harness.atom); + + const unmount = registry.mount(harness.atom); + expect(yield* read).toBe(1); + registry.set(trigger, 1); + expect(yield* read).toBe(2); + + // Unmounted, the query stays alive on its idle TTL, but nothing reads it. + unmount(); + registry.set(trigger, 2); + registry.set(trigger, 3); + expect(executions).toBe(2); + + expect(yield* read).toBe(3); + }), + ), + ); + + it.effect("keeps an idle query when its trigger recomputes to the same value", () => + Effect.scoped( + Effect.gen(function* () { + let executions = 0; + const settings = Atom.make({ theme: "dark" }).pipe(Atom.keepAlive); + const trigger = Atom.make((get) => JSON.stringify([get(settings) !== null])); + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => (executions += 1)), + { refreshTrigger: trigger }, + ); + const registry = AtomRegistry.make(); + yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); + const read = readEnvironmentQuery(registry, harness.atom); + + const unmount = registry.mount(harness.atom); + expect(yield* read).toBe(1); + unmount(); + registry.set(settings, { theme: "light" }); + expect(yield* read).toBe(1); + expect(executions).toBe(1); + }), + ), + ); + + it.effect("refreshes an idle query the registry swept and rebuilt after its trigger moved", () => + Effect.scoped( + Effect.gen(function* () { + // Only the registry's idle-TTL sweeps run on these; effects keep their own scheduler. + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + yield* Effect.addFinalizer(() => Effect.sync(() => vi.useRealTimers())); + let executions = 0; + const trigger = Atom.make(0).pipe(Atom.keepAlive); + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => (executions += 1)), + { refreshTrigger: trigger, idleTtlMs: 1_000 }, + ); + const registry = AtomRegistry.make({ timeoutResolution: 100 }); + yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); + const read = readEnvironmentQuery(registry, harness.atom); + + const unmount = registry.mount(harness.atom); + expect(yield* read).toBe(1); + unmount(); + // Lets the registry start the query node's idle TTL. + yield* Effect.yieldNow; + vi.advanceTimersByTime(500); + registry.set(trigger, 1); + yield* Effect.yieldNow; + // Sweeps the query node but not the data it read, whose idle TTL started with the trigger. + vi.advanceTimersByTime(700); + expect(executions).toBe(1); + + expect(yield* read).toBe(2); + }), + ), + ); }); describe("Atom.fn mutation semantics", () => { diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index 0b9b0006e9fe..b0eefec106fe 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -483,6 +483,35 @@ export function followStreamInEnvironment( ); } +/** + * Refreshes a query when `signal` changes, but only while something reads it. + * `Atom.makeRefreshOnSignal` subscribes to the signal, so a query that outlives its view on an + * idle TTL still re-runs on every change. Reading the signal as a dependency instead lets the + * registry re-run a mounted query right away and only mark an idle one stale, which then + * refreshes on its next read. + */ +const refreshOnSignalWhileRead = + (signal: Atom.Atom) => + (self: Atom.Atom): Atom.Atom => { + // The signal value each registry's data was read under. Held outside the node because the + // registry can sweep this node while `self` keeps the data, and the rebuilt node must still + // see a change it missed. + const readUnder = new WeakMap(); + return Atom.transform( + self, + (get) => { + const current = get(signal); + get.subscribe(self, (value) => get.setSelf(value)); + const changed = + readUnder.has(get.registry) && !Object.is(readUnder.get(get.registry), current); + readUnder.set(get.registry, current); + if (changed) get.refresh(self); + return get.once(self); + }, + { initialValueTarget: self }, + ); + }; + export function createEnvironmentQueryAtomFamily( runtime: Atom.AtomRuntime, options: EnvironmentQueryAtomOptions< @@ -566,7 +595,7 @@ export function createEnvironmentQueryAtomFamily( return ( refreshTrigger === undefined ? intervalQuery - : intervalQuery.pipe(Atom.makeRefreshOnSignal(refreshTrigger)) + : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger)) ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); }); return (target) => family(environmentRpcKey(target));