From bf852808da9b9c636ad9cd57926ae873219b7c42 Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:57:57 +1100 Subject: [PATCH 1/2] fix(client-runtime): stop query polling after the last reader leaves --- .../client-runtime/src/state/runtime.test.ts | 379 +++++++++++++++++- packages/client-runtime/src/state/runtime.ts | 83 ++-- 2 files changed, 438 insertions(+), 24 deletions(-) diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index 5b348d012027c..344b3c9d47c13 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; import { EnvironmentId, type PreviewEvent, @@ -88,6 +89,8 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" execute: Effect.Effect, { refreshTrigger, + refreshIntervalMs, + staleTimeMs, idleTtlMs, persistence, initialState, @@ -96,6 +99,8 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" onAnswer, }: { readonly refreshTrigger?: Atom.Atom; + readonly refreshIntervalMs?: number; + readonly staleTimeMs?: number; readonly onAnswer?: QueryAnswerHandler; readonly isFinal?: (value: A) => boolean; readonly idleTtlMs?: number; @@ -139,10 +144,11 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" ); const family = createEnvironmentQueryAtomFamily(runtime, { label: "test.environment-query", - staleTimeMs: 60_000, + staleTimeMs: staleTimeMs ?? 60_000, ...(persistence === undefined ? {} : { persistence }), execute: () => execute, ...(refreshTrigger === undefined ? {} : { refreshTrigger: () => refreshTrigger }), + ...(refreshIntervalMs === undefined ? {} : { refreshIntervalMs }), ...(idleTtlMs === undefined ? {} : { idleTtlMs }), ...(isFinal === undefined ? {} : { isFinal }), ...(onAnswer === undefined ? {} : { onAnswer }), @@ -183,6 +189,21 @@ const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* return registry; }); +// Drain registry cleanup in microtasks so fake query intervals cannot overtake +// native setImmediate disposal when the clock advances by several intervals. +const makeTimerQueryRegistry = () => + AtomRegistry.make({ + scheduleTask: (run) => { + let cancelled = false; + queueMicrotask(() => { + if (!cancelled) run(); + }); + return () => { + cancelled = true; + }; + }, + }); + describe("preview event subscriptions", () => { it.effect("delivers every open and close when RPC events arrive in batches", () => Effect.scoped( @@ -840,6 +861,362 @@ describe("environment query lifecycle", () => { ), ); + it.effect.each(["interval", "interval and signal"] as const)( + "polls only for real readers and retains cached reopen with %s", + (mode) => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const answers = yield* Queue.make(); + const trigger = Atom.make(0).pipe(Atom.keepAlive); + const harness = yield* makeEnvironmentQueryHarness( + Effect.promise(() => Promise.resolve(++executions)).pipe( + Effect.tap((value) => Queue.offer(answers, value)), + ), + { + staleTimeMs: 0, + refreshIntervalMs: 10_000, + ...(mode === "interval" ? {} : { refreshTrigger: trigger }), + }, + ); + const read = readEnvironmentQuery(registry, harness.atom); + const first = registry.mount(harness.atom); + const second = registry.mount(harness.atom); + expect(yield* Queue.take(answers)).toBe(1); + expect(yield* read).toBe(1); + first(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + expect(yield* Queue.take(answers)).toBe(2); + expect(yield* read).toBe(2); + + second(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + expect(executions).toBe(2); + + const remount = registry.mount(harness.atom); + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(2); + expect(yield* Queue.take(answers)).toBe(3); + expect(yield* read).toBe(3); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + expect(yield* Queue.take(answers)).toBe(4); + expect(yield* read).toBe(4); + if (mode === "interval and signal") { + registry.set(trigger, 1); + expect(yield* Queue.take(answers)).toBe(5); + expect(yield* read).toBe(5); + } + const beforeIdle = executions; + remount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + if (mode === "interval and signal") { + registry.set(trigger, 2); + registry.set(trigger, 3); + } + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + expect(executions).toBe(beforeIdle); + const finalMount = registry.mount(harness.atom); + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(beforeIdle); + expect(yield* Queue.take(answers)).toBe(beforeIdle + 1); + expect(yield* read).toBe(beforeIdle + 1); + expect(executions).toBe(beforeIdle + 1); + finalMount(); + }), + ), + ); + + it.effect.each( + (["sync", "async", "failure", "previous success"] as const).flatMap((completion) => + ["stale", "signal", "signal and session"].map((invalidation) => ({ + completion, + invalidation, + })), + ), + )( + "revalidates once after idle $invalidation with $completion completion", + ({ completion, invalidation }) => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const trigger = Atom.make(0).pipe(Atom.keepAlive); + const harness = yield* makeEnvironmentQueryHarness( + Effect.suspend(() => { + executions++; + if ( + completion === "failure" || + (completion === "previous success" && executions > 1) + ) { + return Effect.fail(new TestQueryError({ message: "query failed" })); + } + return completion === "async" + ? Effect.promise(() => Promise.resolve(executions)) + : Effect.succeed(executions); + }), + { refreshTrigger: trigger, refreshIntervalMs: 10_000, staleTimeMs: 0 }, + ); + const read = readEnvironmentQuery(registry, harness.atom).pipe(Effect.exit); + const unmount = registry.mount(harness.atom); + yield* read; + expect(executions).toBe(1); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + if (invalidation !== "stale") registry.set(trigger, 1); + if (invalidation === "signal and session") { + yield* SubscriptionRef.set( + harness.supervisorSession, + Option.some({ ...QUERY_RPC_SESSION }), + ); + } + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect(executions).toBe(1); + const remount = registry.mount(harness.atom); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + const result = yield* read; + if (completion === "failure" || completion === "previous success") { + expect(Exit.isFailure(result)).toBe(true); + if (Exit.isFailure(result)) { + expect(Cause.squash(result.cause)).toBeInstanceOf(TestQueryError); + } + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe( + completion === "previous success" ? 1 : null, + ); + } else { + expect(result).toEqual(Exit.succeed(2)); + } + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect(executions).toBe(2); + remount(); + }), + ), + ); + + it.effect.each([undefined, 10_000])( + "retains a disposable idle signal alongside cached data (interval %s)", + (refreshIntervalMs) => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const trigger = Atom.make(0); + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => ++executions), + { + refreshTrigger: trigger, + ...(refreshIntervalMs === undefined ? {} : { refreshIntervalMs }), + staleTimeMs: 60_000, + }, + ); + const read = readEnvironmentQuery(registry, harness.atom); + const unmount = registry.mount(harness.atom); + expect(yield* read).toBe(1); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + registry.set(trigger, 1); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + registry.set(trigger, 2); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + expect(executions).toBe(1); + const remount = registry.mount(harness.atom); + expect(yield* read).toBe(2); + expect(executions).toBe(2); + remount(); + }), + ), + ); + + it.effect.each(["interval", "interval and signal"] as const)( + "discards data and invalidation state together after cache expiry with %s", + (mode) => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const trigger = Atom.make(0); + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => ++executions), + { + refreshIntervalMs: 1_000, + idleTtlMs: 5_000, + staleTimeMs: 60_000, + ...(mode === "interval" ? {} : { refreshTrigger: trigger }), + }, + ); + const read = readEnvironmentQuery(registry, harness.atom); + const unmount = registry.mount(harness.atom); + expect(yield* read).toBe(1); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + expect(executions).toBe(1); + registry.set(trigger, 1); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + const remount = registry.mount(harness.atom); + expect(yield* read).toBe(2); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect(executions).toBe(2); + remount(); + }), + ), + ); + + it.effect.each([ + { refreshIntervalMs: 60_000, idleTtlMs: 5_000 }, + { refreshIntervalMs: 30 * 60_000, idleTtlMs: 60 * 60_000 }, + ])("stops polling independently of cache TTL ($idleTtlMs)", (options) => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => ++executions), + { ...options, refreshTrigger: Atom.make(0).pipe(Atom.keepAlive) }, + ); + const unmount = registry.mount(harness.atom); + expect(yield* readEnvironmentQuery(registry, harness.atom)).toBe(1); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(options.refreshIntervalMs)); + expect(yield* readEnvironmentQuery(registry, harness.atom)).toBe(2); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(3 * options.refreshIntervalMs)); + expect(executions).toBe(2); + }), + ), + ); + + it.effect("stops interval and signal refreshes for a final query result", () => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const trigger = Atom.make(0).pipe(Atom.keepAlive); + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => ++executions), + { + refreshIntervalMs: 10_000, + refreshTrigger: trigger, + isFinal: (value) => value >= 2, + staleTimeMs: 0, + }, + ); + const unmount = registry.mount(harness.atom); + const read = readEnvironmentQuery(registry, harness.atom); + expect(yield* read).toBe(1); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + expect(yield* read).toBe(2); + registry.set(trigger, 1); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + expect(executions).toBe(2); + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(2); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + const remount = registry.mount(harness.atom); + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(2); + expect(yield* read).toBe(2); + expect(executions).toBe(2); + registry.refresh(harness.atom); + expect(yield* read).toBe(3); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + expect(executions).toBe(3); + remount(); + }), + ), + ); + + it.effect("cancels an obsolete periodic query when its session is replaced", () => + Effect.scoped( + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const registry = makeTimerQueryRegistry(); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + let executions = 0; + const pendingStarted = Latch.makeUnsafe(); + const pendingStopped = Latch.makeUnsafe(); + const finishObsolete = Latch.makeUnsafe(); + const harness = yield* makeEnvironmentQueryHarness( + Effect.suspend(() => { + executions += 1; + if (executions !== 2) return Effect.succeed(executions); + pendingStarted.openUnsafe(); + return finishObsolete.await.pipe( + Effect.as(-1), + Effect.ensuring(Effect.sync(() => pendingStopped.openUnsafe())), + ); + }), + { refreshIntervalMs: 10_000, refreshTrigger: Atom.make(0).pipe(Atom.keepAlive) }, + ); + const observed: Array> = []; + const unmount = registry.subscribe(harness.atom, (result) => observed.push(result)); + const read = readEnvironmentQuery(registry, harness.atom); + expect(yield* read).toBe(1); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + yield* pendingStarted.await; + yield* SubscriptionRef.set( + harness.supervisorSession, + Option.some({ ...QUERY_RPC_SESSION }), + ); + expect(yield* read).toBe(3); + yield* pendingStopped.await; + finishObsolete.openUnsafe(); + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(3); + expect(observed.some(AsyncResult.isFailure)).toBe(false); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + expect(yield* read).toBe(4); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + expect(executions).toBe(4); + }), + ), + ); + it.effect("refreshes on its trigger while read, and once on its next read when idle", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index 79b1ec4ac5e81..6109db2f6d7bb 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -516,20 +516,18 @@ function runInEnvironment( const refreshOnSignalWhileRead = (signal: Atom.Atom, isFinal: (value: V) => boolean) => (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. + // Retain the revision used by the cached answer across idle invalidations. + // A newly created cache node has no answer to invalidate. 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); + const previous = get.self(); + const changed = Option.isSome(previous) && !Object.is(readUnder.get(get.registry), current); readUnder.set(get.registry, current); - if (changed && !isFinal(get.once(self))) get.refresh(self); - return get.once(self); + if (changed && !Option.exists(previous, isFinal)) get.refresh(self); + return get(self); }, { initialValueTarget: self }, ); @@ -595,6 +593,8 @@ export function createEnvironmentQueryAtomFamily( const family = Atom.family((key: string) => { const target = parseEnvironmentRpcKey(key); const idleTtlMs = options.idleTtlMs ?? 5 * 60_000; + const builds = new WeakMap(); + const refreshTrigger = options.refreshTrigger?.(target); const queryAtom = runtime .atom< A, @@ -603,6 +603,7 @@ export function createEnvironmentQueryAtomFamily( | EnvironmentRegistry.EnvironmentNotRegisteredError | EnvironmentRpcUnavailableError >((get) => { + builds.set(get.registry, (builds.get(get.registry) ?? 0) + 1); const connection = Option.getOrNull( AsyncResult.value(get(connectionAtom(target.environmentId))), ); @@ -695,25 +696,61 @@ export function createEnvironmentQueryAtomFamily( ); }); }) - .pipe( - Atom.swr({ - staleTime: options.staleTimeMs ?? 30_000, - revalidateOnMount: true, - }), - Atom.setIdleTTL(idleTtlMs), - ); + .pipe(Atom.setIdleTTL(0)); const isFinal = (result: AsyncResult.AsyncResult) => options.isFinal !== undefined && Option.exists(AsyncResult.value(result), options.isFinal); - const intervalQuery = - options.refreshIntervalMs === undefined + // Keep the signal dependency for exactly as long as the cached answer. Tracked + // dependencies retain disposable revisions without observing them while idle. + const cachedQuery = ( + refreshTrigger === undefined ? queryAtom - : queryAtom.pipe(refreshEveryUntilFinal(options.refreshIntervalMs, isFinal)); - const refreshTrigger = options.refreshTrigger?.(target); + : queryAtom.pipe(refreshOnSignalWhileRead(refreshTrigger, isFinal)) + ).pipe(Atom.setIdleTTL(idleTtlMs)); + const shouldRevalidate = (result: AsyncResult.AsyncResult) => { + if (result.waiting || isFinal(result) || AsyncResult.isInitial(result)) return false; + const success = AsyncResult.isSuccess(result) ? Option.some(result) : result.previousSuccess; + return ( + Option.isNone(success) || + // @effect-diagnostics-next-line globalDate:off - AsyncResult timestamps use wall time outside an Effect runtime. + Date.now() - success.value.timestamp >= (options.staleTimeMs ?? 30_000) + ); + }; + const revalidatingQuery = Atom.transform( + cachedQuery, + (get) => { + const before = builds.get(get.registry); + const current = get(cachedQuery); + // A stale signal or connection can rebuild the source during this read, + // even synchronously. It already owns this mount's revalidation. + if ( + Option.isNone(get.self()) && + before === builds.get(get.registry) && + shouldRevalidate(current) + ) { + let active = true; + get.addFinalizer(() => { + active = false; + }); + get.registry.schedulerAsync.makeDispatcher().scheduleTask(() => { + if ( + active && + before === builds.get(get.registry) && + shouldRevalidate(get.once(cachedQuery)) + ) { + get.refresh(cachedQuery); + } + }, 0); + } + return current; + }, + { initialValueTarget: cachedQuery }, + ); + // Only the live wrapper owns the timer; the cache never subscribes to it. return ( - refreshTrigger === undefined - ? intervalQuery - : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger, isFinal)) - ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); + options.refreshIntervalMs === undefined + ? revalidatingQuery + : revalidatingQuery.pipe(refreshEveryUntilFinal(options.refreshIntervalMs, isFinal)) + ).pipe(Atom.withLabel(`${options.label}:${key}`)); }); return (target) => family(environmentRpcKey(target)); } From 3431a69655f8274f77329223dfb7dfa7e75278e8 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:35:56 +1100 Subject: [PATCH 2/2] fix(client-runtime): a returning reader's catch-up poll no longer doubles a remount refresh Catch up only from a settled success, and queue the catch-up on the registry scheduler behind any revalidation the remount started, so a signal refresh or revalidateOnRemount check that already ran cancels it. Co-Authored-By: Claude Opus 5.5 --- .../client-runtime/src/state/runtime.test.ts | 61 +++++++++++++++++-- packages/client-runtime/src/state/runtime.ts | 32 ++++++---- 2 files changed, 74 insertions(+), 19 deletions(-) diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index bd87a7e9d1490..c86fe7980d281 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -104,6 +104,7 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" refreshTrigger, refreshIntervalMs, staleTimeMs, + revalidateOnRemount, idleTtlMs, persistence, initialState, @@ -114,6 +115,7 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" readonly refreshTrigger?: Atom.Atom; readonly refreshIntervalMs?: number; readonly staleTimeMs?: number; + readonly revalidateOnRemount?: boolean; readonly onAnswer?: QueryAnswerHandler; readonly isFinal?: (value: A) => boolean; readonly idleTtlMs?: number; @@ -162,6 +164,7 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" execute: () => execute, ...(refreshTrigger === undefined ? {} : { refreshTrigger: () => refreshTrigger }), ...(refreshIntervalMs === undefined ? {} : { refreshIntervalMs }), + ...(revalidateOnRemount === undefined ? {} : { revalidateOnRemount }), ...(idleTtlMs === undefined ? {} : { idleTtlMs }), ...(isFinal === undefined ? {} : { isFinal }), ...(onAnswer === undefined ? {} : { onAnswer }), @@ -1120,7 +1123,6 @@ describe("environment query lifecycle", () => { const remount = registry.mount(harness.atom); expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(2); - yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); expect(yield* Queue.take(answers)).toBe(3); expect(yield* read).toBe(3); yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); @@ -1142,7 +1144,6 @@ describe("environment query lifecycle", () => { expect(executions).toBe(beforeIdle); const finalMount = registry.mount(harness.atom); expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(beforeIdle); - yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); expect(yield* Queue.take(answers)).toBe(beforeIdle + 1); expect(yield* read).toBe(beforeIdle + 1); expect(executions).toBe(beforeIdle + 1); @@ -1151,9 +1152,13 @@ describe("environment query lifecycle", () => { ), ); - it.effect.each(["sync", "async", "failure", "previous success"] as const)( - "revalidates once after an idle signal with %s completion", - (completion) => + it.effect.each( + (["sync", "async", "failure", "previous success"] as const).flatMap((completion) => + [0, 30_000].map((idleMs) => ({ completion, idleMs })), + ), + )( + "revalidates once after an idle signal with $completion completion after $idleMs ms", + ({ completion, idleMs }) => Effect.scoped( Effect.gen(function* () { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); @@ -1188,7 +1193,7 @@ describe("environment query lifecycle", () => { unmount(); yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); registry.set(trigger, 1); - yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(idleMs)); expect(executions).toBe(1); const remount = registry.mount(harness.atom); yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); @@ -1281,6 +1286,50 @@ describe("environment query lifecycle", () => { ), ); + it.effect("revalidates a polled revalidateOnRemount query once when remounted", () => + Effect.scoped( + Effect.gen(function* () { + // Registry tasks run only on flush, so a zero-delay timer can overtake a queued revalidation. + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const queued: Array<() => void> = []; + const registry = AtomRegistry.make({ + scheduleTask: (run) => { + queued.push(run); + return () => { + const index = queued.indexOf(run); + if (index >= 0) queued.splice(index, 1); + }; + }, + }); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + registry.dispose(); + vi.useRealTimers(); + }), + ); + const flushTasks = () => { + while (queued.length > 0) queued.shift()?.(); + }; + let executions = 0; + const harness = yield* makeEnvironmentQueryHarness( + Effect.sync(() => ++executions), + { refreshIntervalMs: 10_000, staleTimeMs: 0, revalidateOnRemount: true }, + ); + const unmount = registry.mount(harness.atom); + flushTasks(); + expect(selectEnvironmentQueryData(registry.get(harness.atom))).toBe(1); + unmount(); + flushTasks(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(30_000)); + const remount = registry.mount(harness.atom); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + flushTasks(); + expect(executions).toBe(2); + remount(); + }), + ), + ); + it.effect("queries afresh once after a polled query's cache expires", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index 573f733ce7456..415960e5aa4da 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -554,7 +554,7 @@ const pollerByCache = new WeakMap, Atom.Atom>(); /** * Polls only while read: keep this outside every idle TTL so the timer ends with the last - * reader. A returning reader gets any poll it missed right away. + * reader. A returning reader gets a poll it missed right away. */ const refreshEveryUntilFinal = >( @@ -566,19 +566,25 @@ const refreshEveryUntilFinal = self, (get) => { const result = get(self); - if (!isFinal(result)) { - const lastSuccess = AsyncResult.isSuccess(result) - ? Option.some(result) - : AsyncResult.isFailure(result) - ? result.previousSuccess - : Option.none(); - const delay = - Option.isSome(get.self()) || result.waiting || Option.isNone(lastSuccess) - ? intervalMs - : // @effect-diagnostics-next-line globalDate:off - AsyncResult timestamps use wall time outside an Effect runtime. - Math.max(0, intervalMs - (Date.now() - lastSuccess.value.timestamp)); + if (isFinal(result)) return result; + const age = + Option.isNone(get.self()) && AsyncResult.isSuccess(result) && !result.waiting + ? // @effect-diagnostics-next-line globalDate:off - AsyncResult timestamps use wall time outside an Effect runtime. + Date.now() - result.timestamp + : 0; + if (age >= intervalMs) { + // Queued behind any revalidation this mount started; a refresh it makes rebuilds this + // node and cancels the catch-up. + let active = true; + get.addFinalizer(() => { + active = false; + }); + get.registry.schedulerAsync.makeDispatcher().scheduleTask(() => { + if (active) get.refresh(self); + }, 0); + } else { // @effect-diagnostics-next-line globalTimers:off - Mirrors `Atom.withRefresh`, outside an Effect runtime. - const handle = setTimeout(() => get.refresh(self), delay); + const handle = setTimeout(() => get.refresh(self), intervalMs - age); get.addFinalizer(() => clearTimeout(handle)); } return result;