diff --git a/.changeset/active-query-invalidation.md b/.changeset/active-query-invalidation.md new file mode 100644 index 000000000..de7f548f0 --- /dev/null +++ b/.changeset/active-query-invalidation.md @@ -0,0 +1,5 @@ +--- +"@effect-app/vue": minor +--- + +Invalidate only mounted queries (TanStack `refetchType: "active"`). Idle `gcTime` entries are marked stale and refetch on remount. `invalidateAndAwait` still waits for observed refetches to finish. diff --git a/packages/vue/docs/query-key-invalidation.md b/packages/vue/docs/query-key-invalidation.md index 1396eb9f7..68173e4e0 100644 --- a/packages/vue/docs/query-key-invalidation.md +++ b/packages/vue/docs/query-key-invalidation.md @@ -95,9 +95,10 @@ The await-tracking map (`keyAtoms` in `atomQuery.ts`) keys on the same `Hash.has 3. **O(depth) registrations per live query** (depth+1 prefixes, in both the reactivity handler map and `keyAtoms`). Negligible at typical depths of 2-4. 4. **Reaches only "alive" atoms** (mounted, or cached within `idleTTL`). `setIdleTTL` is - applied last in the atom chain so cached-but-unmounted queries stay registered and are - still hit; a query GC'd past idle TTL is gone and refetches fresh on next mount anyway — - matching TanStack `gcTime` semantics. + applied last in the atom chain so cached-but-unmounted queries stay registered. + Invalidation **marks** those idle entries stale (TanStack `refetchType: "active"`) and + **awaits** the refetch of any atom that still has observers. A query GC'd past idle TTL + is gone and refetches fresh on next mount — matching TanStack `gcTime` semantics. ## Verdict diff --git a/packages/vue/src/atomQuery.ts b/packages/vue/src/atomQuery.ts index 5ffe4f8a7..e59460cc5 100644 --- a/packages/vue/src/atomQuery.ts +++ b/packages/vue/src/atomQuery.ts @@ -35,6 +35,7 @@ import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry" import { clearQueryReadDependencies, getQueryReadDependencies, type QueryInvalidationMode, registerQueryInvalidationMode, setQueryReadDependencies } from "./dependencyMetadata.ts" import { reportRuntimeError } from "./lib.ts" import { beginLiveQueryFetch, endLiveQueryFetch, type LiveQueryOptions, registerLiveQuery } from "./liveQueryInvalidation.ts" +import { atomsToRefetch, observeQueryAtom } from "./queryLifetime.ts" /** All non-empty prefixes of a key, longest last. `[a,b,c]` -> `[[a],[a,b],[a,b,c]]`. */ const prefixesOf = (key: ReadonlyArray): ReadonlyArray> => @@ -53,9 +54,10 @@ const uniqueKeys = (keys: ReadonlyArray>): ReadonlyArray< } // --- awaitable invalidation ------------------------------------------------------------------- -// keyHash -> live query atoms registered under that key. A query atom is tracked while it is alive -// in the registry (mounted OR cached within idle-ttl) and removed on GC, so invalidation reaches -// cached-but-unmounted queries too (e.g. a list you navigated away from). +// keyHash -> query atoms registered under that key. A query atom stays in the map while it is +// alive in the registry (mounted OR cached within idle-ttl). Invalidation *marks* every matching +// atom stale; it only *refetches* atoms that currently have observers (TanStack refetchType: +// "active"). Idle cache entries refetch on the next mount. const keyAtoms = new Map>>>() const trackByKeys = @@ -115,11 +117,11 @@ const atomsForKeys = (keys: ReadonlyArray): ReadonlyArray): Effect.Effect => Effect .gen(function*() { - const atoms = atomsForKeys(keys) + const atoms = atomsToRefetch(atomsForKeys(keys)) yield* Effect.forEach(atoms, captureAtomQueryParentSpan, { discard: true, concurrency: "unbounded" }) if (atoms.length === 0) return yield* Effect.forEach(atoms, (atom) => Effect.sync(() => defaultRegistry.refresh(atom)), { @@ -134,19 +136,21 @@ export const invalidateSoft = (keys: ReadonlyArray): Effect.Effect): Effect.Effect => Effect .gen(function*() { - const atoms = atomsForKeys(keys) + const atoms = atomsToRefetch(atomsForKeys(keys)) yield* Effect.forEach(atoms, captureAtomQueryParentSpan, { discard: true, concurrency: "unbounded" }) if (atoms.length === 0) return yield* Effect.forEach(atoms, (atom) => Effect.sync(() => defaultRegistry.refresh(atom)), { discard: true }) + // Contract: observers > 0 ⇒ wait for this refresh to finish. yield* Effect.forEach(atoms, (a) => awaitAtomResult(defaultRegistry, a).pipe(Effect.exit)) }) .pipe(Effect.orDie) @@ -535,7 +539,8 @@ export const buildQueryFamily = ( atom = trackByKeys(reactivityKeys)(atom) atom = trackReadDependencies(fullKey, () => lastReads)(atom) // gcTime LAST so the whole chain (incl. the registration + tracking) stays alive through the - // idle window, letting invalidation reach a cached-but-unmounted query. + // idle window. Invalidation still finds the cached atom and marks it stale; it does not + // refetch until an observer remounts (TanStack refetchType: "active"). atom = Atom.setIdleTTL(atom, defaults.gcTime) const registered = setAtomQueryMetadata(Atom.withLabel(`query:${self.id}`)(atom)) const writable = Atom.writable( @@ -550,9 +555,12 @@ export const buildQueryFamily = ( const writableWithTarget = Object.assign(writable, { initialValueTarget: registered }) const result = setAtomQueryMetadata(Atom.withLabel(`query-cache:${self.id}`)(writableWithTarget)) // Key the fetch state by the atom `withQueryOptions` receives, so its mount hook can find it. + const observed = observeQueryAtom(result) queryFetchStates.set(result, fetchState) + queryFetchStates.set(observed, fetchState) atomQueryKeys.set(result, fullKey) - return result + atomQueryKeys.set(observed, fullKey) + return observed }) } @@ -581,7 +589,7 @@ export const buildStreamQueryFamily = ( atom = rt.factory.withReactivity(reactivityKeys)(atom) atom = trackWritableByKeys(reactivityKeys)(atom) atom = Atom.setIdleTTL(atom, defaults.gcTime) - return setAtomQueryMetadata(Atom.withLabel(`stream-query:${self.id}`)(atom)) + return observeQueryAtom(setAtomQueryMetadata(Atom.withLabel(`stream-query:${self.id}`)(atom))) }) } diff --git a/packages/vue/src/queryLifetime.ts b/packages/vue/src/queryLifetime.ts new file mode 100644 index 000000000..d5d7dce8d --- /dev/null +++ b/packages/vue/src/queryLifetime.ts @@ -0,0 +1,74 @@ +import type * as AsyncResult from "effect/unstable/reactivity/AsyncResult" +import * as Atom from "effect/unstable/reactivity/Atom" + +/** + * Per-query observer count + invalidation flag. Lives on the family atom so + * invalidate can mark stale and refetch only when someone is looking + * (TanStack refetchType: "active"). + */ +export interface QueryLifetime { + observers: number + invalidated: boolean +} + +const lifetimes = new WeakMap, QueryLifetime>() + +const rootOf = (atom: Atom.Atom): Atom.Atom => { + let target = atom + while (target.initialValueTarget !== undefined) { + target = target.initialValueTarget + } + return target +} + +export const queryLifetimeOf = (atom: Atom.Atom): QueryLifetime | undefined => + lifetimes.get(rootOf(atom)) ?? lifetimes.get(atom) + +export const attachQueryLifetime = ( + atom: Atom.Atom> +): QueryLifetime => { + const existing = queryLifetimeOf(atom) + if (existing !== undefined) return existing + const created: QueryLifetime = { observers: 0, invalidated: false } + lifetimes.set(rootOf(atom), created) + lifetimes.set(atom, created) + return created +} + +/** + * Count this subscription as an observer of `family`. On remount, consume a + * pending invalidation and refresh the family atom. + */ +export const observeQueryAtom = ( + family: Atom.Atom> +): Atom.Atom> => { + const life = attachQueryLifetime(family) + return Atom.transform(family, (get) => { + life.observers++ + get.addFinalizer(() => { + life.observers = Math.max(0, life.observers - 1) + }) + if (life.invalidated) { + life.invalidated = false + get.refresh(family) + } + return get(family) + }, { initialValueTarget: family }) +} + +/** + * Idle queries with a lifetime are marked stale. Queries that still have + * observers (or have no lifetime, e.g. ad-hoc registry mounts of a raw + * effect atom) are refetched now. + */ +export const atomsToRefetch = ( + atoms: ReadonlyArray>> +): ReadonlyArray>> => { + const active: Array>> = [] + for (const atom of atoms) { + const life = queryLifetimeOf(atom) + if (life === undefined || life.observers > 0) active.push(atom) + else life.invalidated = true + } + return active +} diff --git a/packages/vue/test/inactiveInvalidation.test.ts b/packages/vue/test/inactiveInvalidation.test.ts new file mode 100644 index 000000000..d6aea3fe2 --- /dev/null +++ b/packages/vue/test/inactiveInvalidation.test.ts @@ -0,0 +1,172 @@ +import { defaultRegistry, registryKey } from "@effect/atom-vue" +import { DataDependencies, makeQueryKey } from "effect-app/client" +import * as Context from "effect-app/Context" +import * as Effect from "effect-app/Effect" +import * as Option from "effect-app/Option" +import * as Latch from "effect/Latch" +import * as Layer from "effect/Layer" +import * as ManagedRuntime from "effect/ManagedRuntime" +import * as AsyncResult from "effect/unstable/reactivity/AsyncResult" +import * as Reactivity from "effect/unstable/reactivity/Reactivity" +import { describe, expect, it } from "vitest" +import { createApp, nextTick } from "vue" +import { buildQueryFamily, invalidateAndAwait, makeAtomClientRuntime, withQueryOptions } from "../src/atomQuery.js" +import { useAtomSuspense } from "../src/query.js" + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)) +const ticks = async (n: number) => { + for (let i = 0; i < n; i++) { + await nextTick() + await tick() + } +} + +const makeEnv = () => { + const mrt = ManagedRuntime.make(Reactivity.layer) + const baseContext = mrt.runSync(Effect.context()) + const reactivity = Context.get(baseContext, Reactivity.Reactivity) + const rt = makeAtomClientRuntime( + () => Layer.succeedContext(baseContext) as Layer.Layer, + mrt.memoMap + ) + return { + rt, + invalidate: (keys: ReadonlyArray>) => + invalidateAndAwait(keys).pipe(Effect.provideService(Reactivity.Reactivity, reactivity)) + } +} + +const makeCounted = (id: string, repo: DataDependencies.DataDependency) => { + let starts = 0 + const self = { + id, + handler: () => + Effect.gen(function*() { + yield* DataDependencies.read(repo) + return ++starts + }) + } + return { self, starts: () => starts } +} + +const result = (atom: any): AsyncResult.AsyncResult => defaultRegistry.get(atom) +const served = (atom: any): number | undefined => Option.getOrUndefined(AsyncResult.value(result(atom))) + +const queryAtom = (rt: any, self: any) => + withQueryOptions(buildQueryFamily(rt, self)(undefined), { gcTime: "infinity", revalidateOnFocus: false }) + +const mountSuspense = (atom: any) => { + let promise: Promise | undefined + const host = document.createElement("div") + const app = createApp({ + setup() { + promise = useAtomSuspense(() => atom) + return () => null + } + }) + app.provide(registryKey, defaultRegistry) + app.mount(host) + return { app, getPromise: () => promise } +} + +const settle = (p: Promise | undefined) => + Promise.race([ + Promise.resolve(p).then(() => "resolved" as const, () => "rejected" as const), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 40)) + ]) + +const fullKeyOf = (self: { id: string }) => [...makeQueryKey(self), undefined] + +describe("inactive invalidation (refetchType: active)", () => { + it("does not refetch an unmounted query; remount does", async () => { + defaultRegistry.reset() + const { rt, invalidate } = makeEnv() + const repo = DataDependencies.repo("IdleRepo") + const g = makeCounted("Idle.List", repo) + const atom = queryAtom(rt, g.self) + + const m1 = mountSuspense(atom) + await settle(m1.getPromise()) + await ticks(2) + expect(served(atom)).toBe(1) + + m1.app.unmount() + await ticks(2) + + await Effect.runPromise(invalidate([fullKeyOf(g.self)])) + await ticks(2) + expect(g.starts()).toBe(1) + + const m2 = mountSuspense(atom) + await settle(m2.getPromise()) + await ticks(2) + expect(g.starts()).toBeGreaterThanOrEqual(2) + expect(served(atom)).not.toBe(1) + m2.app.unmount() + defaultRegistry.reset() + }) + + it("still refetches a mounted query immediately", async () => { + defaultRegistry.reset() + const { rt, invalidate } = makeEnv() + const repo = DataDependencies.repo("ActiveRepo") + const g = makeCounted("Active.List", repo) + const atom = queryAtom(rt, g.self) + + const mounted = mountSuspense(atom) + await settle(mounted.getPromise()) + await ticks(2) + expect(g.starts()).toBe(1) + + await Effect.runPromise(invalidate([fullKeyOf(g.self)])) + await ticks(2) + expect(g.starts()).toBeGreaterThanOrEqual(2) + mounted.app.unmount() + defaultRegistry.reset() + }) + + it("waits for a mounted refetch to finish before invalidateAndAwait resolves", async () => { + defaultRegistry.reset() + const { rt, invalidate } = makeEnv() + const repo = DataDependencies.repo("AwaitRepo") + let starts = 0 + let completes = 0 + const latches: Array = [] + const self = { + id: "Await.List", + handler: () => + Effect.gen(function*() { + yield* DataDependencies.read(repo) + const n = ++starts + if (n > 1) { + const latch = Latch.makeUnsafe(false) + latches.push(latch) + yield* latch.await + } + completes++ + return n + }) + } + const atom = queryAtom(rt, self) + + const mounted = mountSuspense(atom) + await settle(mounted.getPromise()) + await ticks(2) + expect(completes).toBe(1) + + let settled = false + const waiting = Effect.runPromise(invalidate([fullKeyOf(self)])).then(() => { + settled = true + }) + await ticks(4) + expect(starts).toBeGreaterThanOrEqual(2) + expect(settled, "must not resolve while the mounted refetch is in flight").toBe(false) + + latches.splice(0).forEach((latch) => latch.openUnsafe()) + await waiting + expect(settled).toBe(true) + expect(completes).toBeGreaterThanOrEqual(2) + mounted.app.unmount() + defaultRegistry.reset() + }) +}) diff --git a/packages/vue/test/interrupt-refetch-repro.test.ts b/packages/vue/test/interrupt-refetch-repro.test.ts index 2b110300d..9fb2a744a 100644 --- a/packages/vue/test/interrupt-refetch-repro.test.ts +++ b/packages/vue/test/interrupt-refetch-repro.test.ts @@ -225,12 +225,12 @@ it("B1 — mutation fiber interrupted while its onExit invalidation-refetch runs const atom = buildQueryFamily(rt as any, g.self as any)(undefined) const unmount = defaultRegistry.mount(atom) - await Effect.runPromise(awaitAtomResult(defaultRegistry, atom) as any) // fetch #1 + await Effect.runPromise(awaitAtomResult(defaultRegistry, atom)) // fetch #1 expect(g.completes()).toBe(1) g.setBlocking(true) const mutation = invalidateQueries({ id: "Repro.B1.Save" }, undefined, invalidator)(writeCommand(repo), { id: "x" }) - const fiber = Effect.runFork(mutation as any) + const fiber = Effect.runFork(mutation) await ticks(2) expect(g.starts(), "invalidation should have started the refetch").toBe(2) @@ -239,7 +239,7 @@ it("B1 — mutation fiber interrupted while its onExit invalidation-refetch runs const interrupting = drain(fiber) g.releaseAll() await interrupting - await Effect.runPromise(awaitAtomResult(defaultRegistry, atom) as any) + await Effect.runPromise(awaitAtomResult(defaultRegistry, atom)) expect(g.completes(), "onExit invalidation must complete the refetch despite fiber interrupt").toBe(2) expect(served(atom)).toBe(2) @@ -266,7 +266,7 @@ it("B2 — query unmounts DURING a mutation-triggered refetch, then remounts: re g.setBlocking(true) const mutation = invalidateQueries({ id: "Repro.B2.Save" }, undefined, invalidator)(writeCommand(repo), { id: "x" }) - const fiber = Effect.runFork(mutation as any) + const fiber = Effect.runFork(mutation) await ticks(2) expect(g.starts(), "mutation should have started the refetch").toBe(2) @@ -301,10 +301,12 @@ it("B3 — query UNMOUNTED when the mutation invalidation lands, then remounts: m1.app.unmount() // navigate away: query is unmounted (cached) await ticks(2) - // Command writes the repo; invalidation marks the (unmounted) query dirty. + // Command writes the repo; invalidation marks the (unmounted) query dirty + // without refetching it (TanStack refetchType: "active"). const mutation = invalidateQueries({ id: "Repro.B3.Save" }, undefined, invalidator)(writeCommand(repo), { id: "x" }) await Effect.runPromise(mutation as any).then(() => {}, () => {}) await ticks(2) + expect(g.starts(), "unmounted invalidation must not refetch").toBe(1) const m2 = mountSuspense(atom) // remount: an invalidated query refetches on mount await settle(m2.getPromise()) @@ -495,7 +497,7 @@ it("SUP — invalidation supersedes a stale in-flight fetch; remaining listeners // A mutation invalidates the query while fetch #2 is still in-flight (stale) -> must supersede it. const mutation = invalidateQueries({ id: "Repro.SUP.Save" }, undefined, invalidator)(writeCommand(repo), { id: "x" }) - const fiber = Effect.runFork(mutation as any) + const fiber = Effect.runFork(mutation) await ticks(2) const superseded = g.starts()