Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 53 additions & 4 deletions apps/server/src/pullRequest/PullRequestService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -229,14 +229,63 @@ 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");
assert.strictEqual(previewReads, 1);
}),
);

it.effect("shares detail and activity for a minute, and for ten once merged", () =>
Effect.gen(function* () {
const reads: Array<string> = [];
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<void>();
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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");
Expand Down
21 changes: 17 additions & 4 deletions apps/server/src/pullRequest/PullRequestService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ import {
type PullRequestLabelCandidateList,
type PullRequestLabelChangeInput,
type PullRequestSetFilesViewedInput,
type PullRequestState,
type PullRequestSubmitReviewInput,
PullRequestStack,
PullRequestSummary,
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 = (
Expand Down Expand Up @@ -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,
Comment thread
StiensWout marked this conversation as resolved.
Comment thread
coderabbitai[bot] marked this conversation as resolved.
},
);

Expand All @@ -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) => {
Expand Down
106 changes: 105 additions & 1 deletion packages/client-runtime/src/state/runtime.test.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -81,6 +81,10 @@ function queryConnectionState(

const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness")(function* <A, E>(
execute: Effect.Effect<A, E>,
{
refreshTrigger,
idleTtlMs,
}: { readonly refreshTrigger?: Atom.Atom<unknown>; readonly idleTtlMs?: number } = {},
) {
const supervisorState = yield* SubscriptionRef.make(queryConnectionState());
const supervisorSession = yield* SubscriptionRef.make(Option.some(QUERY_RPC_SESSION));
Expand Down Expand Up @@ -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 {
Expand All @@ -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 = <A, E>(
registry: AtomRegistry.AtomRegistry,
atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
) =>
AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }).pipe(
Effect.tap(() => Effect.yieldNow),
);

const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* <A, E>(
atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
) {
Expand Down Expand Up @@ -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", () => {
Expand Down
31 changes: 30 additions & 1 deletion packages/client-runtime/src/state/runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -483,6 +483,35 @@ export function followStreamInEnvironment<A, E, R>(
);
}

/**
* 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<unknown>) =>
<A>(self: Atom.Atom<A>): Atom.Atom<A> => {
// 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<AtomRegistry.AtomRegistry, unknown>();
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<R, ER, Input, A, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry | R, ER>,
options: EnvironmentQueryAtomOptions<
Expand Down Expand Up @@ -566,7 +595,7 @@ export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>(
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));
Expand Down
Loading