fix(vue): invalidate only mounted queries

.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.

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
 

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<unknown>): ReadonlyArray<ReadonlyArray<unknown>> =>
@@ -53,9 +54,10 @@ const uniqueKeys = (keys: ReadonlyArray<ReadonlyArray<unknown>>): 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<number, Set<Atom.Atom<AsyncResult.AsyncResult<any, any>>>>()
 
 const trackByKeys =
@@ -115,11 +117,11 @@ const atomsForKeys = (keys: ReadonlyArray<unknown>): ReadonlyArray<Atom.Atom<Asy
   return [...atoms]
 }
 
-/** Refresh registered query atoms without making the triggering mutation await them. */
+/** Refresh observed query atoms without making the triggering mutation await them. */
 export const invalidateSoft = (keys: ReadonlyArray<unknown>): Effect.Effect<void> =>
   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<unknown>): Effect.Effect<void
  * through `Reactivity.invalidate` invokes one atom's registered callback once per matching key,
  * repeatedly superseding the same fetch when a mutation carries many row/prefix keys.
  *
- * Resolves once the affected queries have settled, so a mutation can `yield*` this and know the
- * affected queries are fresh. (The await reads via the module-global default registry — the one the
- * vue composables resolve via `injectRegistry`'s fallback.)
+ * Resolves once every **observed** matching query has settled, so a mutation can
+ * `yield*` this and know the on-screen data is fresh. Idle (observers === 0)
+ * matches are only marked stale — they do not block the mutation. Soft
+ * invalidation (`invalidateSoft`) still fires without awaiting.
  */
 export const invalidateAndAwait = (keys: ReadonlyArray<unknown>): Effect.Effect<void> =>
   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 = <I, A, E>(
     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 = <I, A, E>(
     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 = <I, A, E>(
     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)))
   })
 }
 

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<Atom.Atom<any>, QueryLifetime>()
+
+const rootOf = (atom: Atom.Atom<any>): Atom.Atom<any> => {
+  let target = atom
+  while (target.initialValueTarget !== undefined) {
+    target = target.initialValueTarget
+  }
+  return target
+}
+
+export const queryLifetimeOf = (atom: Atom.Atom<any>): QueryLifetime | undefined =>
+  lifetimes.get(rootOf(atom)) ?? lifetimes.get(atom)
+
+export const attachQueryLifetime = <A, E>(
+  atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>
+): 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 = <A, E>(
+  family: Atom.Atom<AsyncResult.AsyncResult<A, E>>
+): Atom.Atom<AsyncResult.AsyncResult<A, E>> => {
+  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 = <A, E>(
+  atoms: ReadonlyArray<Atom.Atom<AsyncResult.AsyncResult<A, E>>>
+): ReadonlyArray<Atom.Atom<AsyncResult.AsyncResult<A, E>>> => {
+  const active: Array<Atom.Atom<AsyncResult.AsyncResult<A, E>>> = []
+  for (const atom of atoms) {
+    const life = queryLifetimeOf(atom)
+    if (life === undefined || life.observers > 0) active.push(atom)
+    else life.invalidated = true
+  }
+  return active
+}

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<Reactivity.Reactivity>())
+  const reactivity = Context.get(baseContext, Reactivity.Reactivity)
+  const rt = makeAtomClientRuntime(
+    () => Layer.succeedContext(baseContext) as Layer.Layer<any, never, never>,
+    mrt.memoMap
+  )
+  return {
+    rt,
+    invalidate: (keys: ReadonlyArray<ReadonlyArray<unknown>>) =>
+      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<number, never> => 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<any> | 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<any> | 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<Latch.Latch> = []
+    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()
+  })
+})