feat(vue): support query-scoped soft invalidation

.changeset/soft-work-invalidations.md

@@ -0,0 +1,5 @@
+---
+"@effect-app/vue": patch
+---
+
+Allow query consumers to opt into non-blocking invalidation refetches.

packages/vue/src/atomQuery.ts

@@ -32,7 +32,7 @@ import { isHttpClientError } from "effect/unstable/http/HttpClientError"
 import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"
 import * as Atom from "effect/unstable/reactivity/Atom"
 import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry"
-import { clearQueryReadDependencies, getQueryReadDependencies, setQueryReadDependencies } from "./dependencyMetadata.ts"
+import { clearQueryReadDependencies, getQueryReadDependencies, type QueryInvalidationMode, registerQueryInvalidationMode, setQueryReadDependencies } from "./dependencyMetadata.ts"
 import { reportRuntimeError } from "./lib.ts"
 import { beginLiveQueryFetch, endLiveQueryFetch, type LiveQueryOptions, registerLiveQuery } from "./liveQueryInvalidation.ts"
 
@@ -115,6 +115,17 @@ const atomsForKeys = (keys: ReadonlyArray<unknown>): ReadonlyArray<Atom.Atom<Asy
   return [...atoms]
 }
 
+/** Refresh registered 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)
+    yield* Effect.forEach(atoms, captureAtomQueryParentSpan, { discard: true, concurrency: "inherit" })
+    if (atoms.length === 0) return
+    yield* Effect.forEach(atoms, (atom) => Effect.sync(() => defaultRegistry.refresh(atom)), {
+      discard: true
+    })
+  })
+
 /**
  * Invalidate the given keys and AWAIT the result. `keyAtoms` resolves all matching hierarchical
  * keys to a deduplicated set of query atoms. Refresh that set directly: sending the whole key set
@@ -213,6 +224,7 @@ export interface AtomQueryOptions {
   readonly staleTime?: Duration.Input
   /** dispose-when-idle (TanStack gcTime; default 5min). "infinity" => keepAlive */
   readonly gcTime?: Duration.Input | "infinity"
+  readonly invalidation?: QueryInvalidationMode
   /**
    * Revalidate a stale query on window focus AND on network reconnect (default on, matching
    * tanstack refetchOnWindowFocus + refetchOnReconnect).
@@ -365,6 +377,13 @@ export const withQueryOptions = <A, E>(
   setAtomQueryMetadata(self, opts)
   const staleTime: Duration.Input = opts.staleTime ?? defaults.staleTime
   let atom = self
+  if (liveKey !== undefined) {
+    atom = Atom.transform(atom, (get) => {
+      const unregister = registerQueryInvalidationMode(liveKey, opts.invalidation ?? "await")
+      get.addFinalizer(unregister)
+      return get(self)
+    }, { initialValueTarget: self })
+  }
   if (opts.live && liveKey !== undefined) {
     const liveOptions = opts.live === true ? {} : opts.live
     atom = Atom.transform(atom, (get) => {

packages/vue/src/dependencyMetadata.ts

@@ -7,6 +7,33 @@ import * as Hash from "effect/Hash"
 // cached-within-ttl) — the atom equivalent of the former tanstack query cache.
 type Entry = { readonly key: ReadonlyArray<unknown>; readonly reads: DataDependencies.DataDependencies }
 const readDependencies = new Map<number, Entry>()
+export type QueryInvalidationMode = "await" | "soft"
+type InvalidationModeEntry = { awaitSubscribers: number; softSubscribers: number }
+const invalidationModes = new Map<number, InvalidationModeEntry>()
+
+export const registerQueryInvalidationMode = (
+  key: ReadonlyArray<unknown>,
+  mode: QueryInvalidationMode
+): () => void => {
+  const hash = Hash.hash(key)
+  const entry = invalidationModes.get(hash) ?? { awaitSubscribers: 0, softSubscribers: 0 }
+  if (mode === "soft") entry.softSubscribers++
+  else entry.awaitSubscribers++
+  invalidationModes.set(hash, entry)
+  let active = true
+  return () => {
+    if (!active) return
+    active = false
+    if (mode === "soft") entry.softSubscribers--
+    else entry.awaitSubscribers--
+    if (entry.awaitSubscribers === 0 && entry.softSubscribers === 0) invalidationModes.delete(hash)
+  }
+}
+
+export const getQueryInvalidationMode = (key: ReadonlyArray<unknown>): QueryInvalidationMode => {
+  const entry = invalidationModes.get(Hash.hash(key))
+  return entry !== undefined && entry.awaitSubscribers === 0 && entry.softSubscribers > 0 ? "soft" : "await"
+}
 
 export const setQueryReadDependencies = (
   key: ReadonlyArray<unknown>,
@@ -40,3 +67,18 @@ export const getDerivedInvalidationKeys = (
   }
   return keys
 }
+
+export const partitionInvalidationKeys = (
+  keys: ReadonlyArray<ReadonlyArray<unknown>>
+): {
+  readonly awaitKeys: ReadonlyArray<ReadonlyArray<unknown>>
+  readonly softKeys: ReadonlyArray<ReadonlyArray<unknown>>
+} => {
+  const awaitKeys: Array<ReadonlyArray<unknown>> = []
+  const softKeys: Array<ReadonlyArray<unknown>> = []
+  for (const key of keys) {
+    if (getQueryInvalidationMode(key) === "soft") softKeys.push(key)
+    else awaitKeys.push(key)
+  }
+  return { awaitKeys, softKeys }
+}

packages/vue/src/internal/tanstackQuery.ts

@@ -15,7 +15,7 @@ import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"
 import * as Atom from "effect/unstable/reactivity/Atom"
 import { computed, type MaybeRefOrGetter, shallowRef, toValue, watch, type WatchSource } from "vue"
 import { replaceEqualDeep } from "../atomQuery.ts"
-import { clearQueryReadDependencies, setQueryReadDependencies } from "../dependencyMetadata.ts"
+import { clearQueryReadDependencies, registerQueryInvalidationMode, setQueryReadDependencies } from "../dependencyMetadata.ts"
 import { reportRuntimeError } from "../lib.ts"
 import type { QueryInvalidator } from "../mutate.ts"
 import type { CustomDefinedInitialQueryOptions, CustomDefinedPlaceholderQueryOptions, CustomUndefinedInitialQueryOptions, CustomUseQueryOptions, MakeQuery2, QueryCacheUpdater, QueryHandle, QueryObserverResult, RefetchOptions } from "../query.ts"
@@ -111,6 +111,15 @@ export const makeTanstackQueryInvalidator = (queryClient: QueryClient): QueryInv
         },
         { discard: true, concurrency: "inherit" }
       )
+    }),
+  invalidateSoft: (keys) =>
+    Effect.gen(function*() {
+      const span = yield* Effect.currentParentSpan.pipe(Effect.orElseSucceed(() => undefined))
+      yield* Effect.forEach(
+        keys,
+        (queryKey) => Effect.promise(() => queryClient.invalidateQueries({ queryKey }, { updateMeta: { span } })),
+        { discard: true, concurrency: "inherit" }
+      )
     })
 })
 
@@ -165,15 +174,21 @@ export const makeTanstackQuery = <R>(
       ...(options?.refetchInterval !== undefined ? { refetchInterval: options.refetchInterval } : {}),
       ...(options?.select !== undefined ? { select: options.select } : {})
     }
+    const resolvedQueryKey = computed(() => {
+      const input = resolveInput(arg, options?.mode)
+      return fullQueryKey(q, queryKey, input)
+    })
+    watch(
+      resolvedQueryKey,
+      (key, _, onCleanup) => onCleanup(registerQueryInvalidationMode(key, options?.invalidation ?? "await")),
+      { immediate: true }
+    )
     const tanstack = useTanstackQuery<A, CauseException<E>, TData>({
       ...tanstackOptions,
       enabled,
       throwOnError: false,
       retry: (retryCount: number, error: unknown) => isRetryable(error) && retryCount < 5,
-      queryKey: computed(() => {
-        const input = resolveInput(arg, options?.mode)
-        return fullQueryKey(q, queryKey, input)
-      }),
+      queryKey: resolvedQueryKey,
       queryFn: (
         { meta, signal }: {
           readonly meta?: { readonly span?: Tracer.AnySpan | undefined } | undefined

packages/vue/src/makeClient.ts

@@ -19,7 +19,7 @@ import * as Struct from "effect/Struct"
 import type * as AsyncResult from "effect/unstable/reactivity/AsyncResult"
 import * as Reactivity from "effect/unstable/reactivity/Reactivity"
 import { type ComputedRef, effectScope, onBeforeUnmount, onScopeDispose, ref, type WatchSource } from "vue"
-import { type AtomClientRuntime, invalidateAndAwait, makeAtomClientRuntime } from "./atomQuery.ts"
+import { type AtomClientRuntime, invalidateAndAwait, invalidateSoft, makeAtomClientRuntime } from "./atomQuery.ts"
 import { type Commander, CommanderStatic, type Progress } from "./commander.ts"
 import { makeTanstackQuery, makeTanstackQueryCacheUpdater, makeTanstackQueryClient, makeTanstackQueryInvalidator } from "./internal/tanstackQuery.ts"
 import { type I18n } from "./intl.ts"
@@ -720,6 +720,10 @@ const makeResolvedAtomQueryInvalidator = <R>(getContext: () => Context.Context<R
     invalidateAndAwait: (keys) =>
       invalidateAndAwait(keys).pipe(
         Effect.provideService(Reactivity.Reactivity, getReactivity())
+      ),
+    invalidateSoft: (keys) =>
+      invalidateSoft(keys).pipe(
+        Effect.provideService(Reactivity.Reactivity, getReactivity())
       )
   }
 }

packages/vue/src/mutate.ts

@@ -13,8 +13,9 @@ import * as Stream from "effect/Stream"
 import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"
 import type * as Reactivity from "effect/unstable/reactivity/Reactivity"
 import { computed, type ComputedRef, shallowRef } from "vue"
-import { invalidateAndAwait } from "./atomQuery.ts"
-import { getDerivedInvalidationKeys } from "./dependencyMetadata.ts"
+import { invalidateAndAwait, invalidateSoft } from "./atomQuery.ts"
+import { getDerivedInvalidationKeys, partitionInvalidationKeys } from "./dependencyMetadata.ts"
+import { reportRuntimeError } from "./lib.ts"
 
 export type GetQueryKey = (h: { id: string; options?: ClientForOptions }) => string[]
 
@@ -119,10 +120,12 @@ export type QueryInvalidationEffect<R = never> = (
 ) => Effect.Effect<void, never, R>
 export interface QueryInvalidator<R = never> {
   readonly invalidateAndAwait: QueryInvalidationEffect<R>
+  readonly invalidateSoft?: QueryInvalidationEffect<R>
 }
 
 export const atomQueryInvalidator: QueryInvalidator<Reactivity.Reactivity> = {
-  invalidateAndAwait
+  invalidateAndAwait,
+  invalidateSoft
 }
 
 export const combineQueryInvalidators = <R>(
@@ -133,6 +136,12 @@ export const combineQueryInvalidators = <R>(
       invalidators,
       (invalidator) => invalidator.invalidateAndAwait(keys),
       { discard: true, concurrency: "inherit" }
+    ),
+  invalidateSoft: (keys) =>
+    Effect.forEach(
+      invalidators,
+      (invalidator) => invalidator.invalidateSoft?.(keys) ?? Effect.void,
+      { discard: true, concurrency: "inherit" }
     )
 })
 
@@ -322,18 +331,34 @@ const buildInvalidateCache = <RInvalidator>(
 
       if (!isReadonlyArrayNonEmpty(keys)) return Effect.void
 
+      const { awaitKeys, softKeys } = partitionInvalidationKeys(keys)
+
       return Effect
         .andThen(
           Effect.annotateCurrentSpan({
             keys,
+            awaitKeys,
+            softKeys,
             clientKeys,
             serverKeys,
             derivedKeys,
             writeDependencies: [...writeDependencies]
           }),
           // refetch + AWAIT every live query registered under these keys, so by the time the
           // mutation resolves the affected queries are fresh.
-          queryInvalidator.invalidateAndAwait(keys)
+          Effect.gen(function*() {
+            if (softKeys.length > 0) {
+              yield* (queryInvalidator.invalidateSoft?.(softKeys) ?? Effect.void).pipe(
+                Effect.catchCause((cause) =>
+                  reportRuntimeError(cause, { invalidation: "soft", keys: softKeys }).pipe(Effect.asVoid)
+                ),
+                Effect.forkDetach({ startImmediately: true })
+              )
+            }
+            if (awaitKeys.length > 0) {
+              yield* queryInvalidator.invalidateAndAwait(awaitKeys)
+            }
+          })
         )
         .pipe(
           Effect.tap(Effect.sleep(0.1)), // allow for refs to update etc

packages/vue/src/query.ts

@@ -18,6 +18,7 @@ import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"
 import * as Atom from "effect/unstable/reactivity/Atom"
 import { computed, type ComputedRef, effectScope, type MaybeRefOrGetter, onBeforeUnmount, onMounted, onScopeDispose, ref, toValue, type WatchSource } from "vue"
 import { type AtomClientRuntime, type AtomQueryOptions, awaitAtomResult, buildQueryFamily, buildStreamQueryFamily, disabledQueryAtom, isStaleResult, queryKeyForAtom, refreshAtomWithCurrentSpan, staleTimeMsOf, withQueryOptions } from "./atomQuery.ts"
+import type { QueryInvalidationMode } from "./dependencyMetadata.ts"
 import type { LiveQueryOptions } from "./liveQueryInvalidation.ts"
 import { latestDefined } from "./suspense.ts"
 
@@ -203,6 +204,7 @@ export interface CustomUseQueryOptions<
   /** poll: re-fetch every N ms (tanstack refetchInterval) */
   readonly refetchInterval?: number
   readonly live?: boolean | LiveQueryOptions
+  readonly invalidation?: QueryInvalidationMode
   readonly select?: (data: TQueryFnData) => TData
   /** accepted for source compatibility; not used by the atom engine */
   readonly retry?: boolean | number
@@ -256,6 +258,7 @@ export interface AtomQueryNewOptions<TQueryFnData = unknown, TData = TQueryFnDat
   readonly refreshEvery?: number
   readonly refetchInterval?: number
   readonly live?: boolean | LiveQueryOptions
+  readonly invalidation?: QueryInvalidationMode
   readonly select?: (data: TQueryFnData) => TData
 }
 
@@ -327,6 +330,7 @@ const normalizeQueryOptions = (options?: {
   readonly refetchInterval?: number
   readonly refreshEvery?: number
   readonly live?: boolean | LiveQueryOptions
+  readonly invalidation?: QueryInvalidationMode
 }): AtomQueryOptions => {
   const out: {
     staleTime?: number
@@ -335,6 +339,7 @@ const normalizeQueryOptions = (options?: {
     structuralSharing?: boolean
     refetchInterval?: number
     live?: boolean | LiveQueryOptions
+    invalidation?: QueryInvalidationMode
   } = {}
   if (options?.staleTime !== undefined) out.staleTime = options.staleTime
   const gcTime = options?.idleTTL ?? options?.gcTime
@@ -345,6 +350,7 @@ const normalizeQueryOptions = (options?: {
   const refetchInterval = options?.refreshEvery ?? options?.refetchInterval
   if (refetchInterval !== undefined) out.refetchInterval = refetchInterval
   if (options?.live !== undefined) out.live = options.live
+  if (options?.invalidation !== undefined) out.invalidation = options.invalidation
   return out
 }
 
@@ -521,6 +527,7 @@ const observedAtom = <A, E>(
     readonly refetchInterval?: number
     readonly refreshEvery?: number
     readonly live?: boolean | LiveQueryOptions
+    readonly invalidation?: QueryInvalidationMode
   }
 ): Atom.Atom<AsyncResult.AsyncResult<A, E>> =>
   withQueryOptions(atom, normalizeQueryOptions(options), queryKeyForAtom(atom))

packages/vue/test/dependencyInvalidation.test.ts

@@ -13,7 +13,7 @@ import { TestClock } from "effect/testing"
 import * as Reactivity from "effect/unstable/reactivity/Reactivity"
 import { createApp, effectScope, ref } from "vue"
 import { awaitAtomResult, buildQueryFamily, invalidateAndAwait, makeAtomClientRuntime } from "../src/atomQuery.js"
-import { clearQueryReadDependencies, getDerivedInvalidationKeys, setQueryReadDependencies } from "../src/dependencyMetadata.js"
+import { clearQueryReadDependencies, getDerivedInvalidationKeys, partitionInvalidationKeys, registerQueryInvalidationMode, setQueryReadDependencies } from "../src/dependencyMetadata.js"
 import { makeTanstackQuery, makeTanstackQueryInvalidator } from "../src/internal/tanstackQuery.js"
 import { invalidateQueries, makeStreamMutation2, type MutationOptionsBase } from "../src/mutate.js"
 
@@ -58,6 +58,30 @@ it("getDerivedInvalidationKeys returns keys of queries whose reads intersect the
   }
 })
 
+it("uses the strictest active query invalidation mode", () => {
+  const overviewKey = ["$Overview", "$List", undefined]
+  const pickListKey = ["$PickList", "$List", undefined]
+
+  const unregisterSoft = registerQueryInvalidationMode(overviewKey, "soft")
+  const unregisterAwait = registerQueryInvalidationMode(pickListKey, "await")
+  const unregisterStrictOverview = registerQueryInvalidationMode(overviewKey, "await")
+  try {
+    expect(partitionInvalidationKeys([overviewKey, pickListKey])).toEqual({
+      awaitKeys: [overviewKey, pickListKey],
+      softKeys: []
+    })
+    unregisterStrictOverview()
+    expect(partitionInvalidationKeys([overviewKey, pickListKey])).toEqual({
+      awaitKeys: [pickListKey],
+      softKeys: [overviewKey]
+    })
+  } finally {
+    unregisterStrictOverview()
+    unregisterSoft()
+    unregisterAwait()
+  }
+})
+
 it("clearing read dependencies drops the query from derivation", () => {
   const inventoryKey = ["$Inventory", "List", undefined]
   setQueryReadDependencies(inventoryKey, new Set([repo]))
@@ -135,7 +159,7 @@ it("atom engine: a query records its read deps so a command's writes derive it",
 
   const unmount = defaultRegistry.mount(atom)
   try {
-    await Effect.runPromise(awaitAtomResult(defaultRegistry, atom) as any)
+    await Effect.runPromise(awaitAtomResult(defaultRegistry, atom))
     expect(runs).toBe(1)
 
     const fullKey = [...makeQueryKey(self), undefined]
@@ -224,7 +248,7 @@ it("atom engine: disposing the query atom clears its recorded reads", async () =
   const fullKey = [...makeQueryKey(self), undefined]
 
   const unmount = defaultRegistry.mount(atom)
-  await Effect.runPromise(awaitAtomResult(defaultRegistry, atom) as any)
+  await Effect.runPromise(awaitAtomResult(defaultRegistry, atom))
   expect(getDerivedInvalidationKeys(new Set([atomRepo]))).toContainEqual(fullKey)
 
   // Disposing the registry runs the atom's finalizers, including `trackReadDependencies`.
@@ -291,12 +315,12 @@ const makeAtomHarness = (queryRepo: DataDependencies.DataDependency): EngineHarn
   return {
     queryFullKey: [...makeQueryKey(self), undefined],
     serverInvalidationKey: makeQueryKey(self),
-    fetchInitial: () => Effect.runPromise(awaitAtomResult(defaultRegistry, atom) as any),
+    fetchInitial: () => Effect.runPromise(awaitAtomResult(defaultRegistry, atom)),
     runs: () => runs,
     runCommand: (options, command) =>
       Effect.runPromise(
         invalidateQueries({ id: "MatrixAtom.Save" }, options, invalidator)(command, { id: "x" })
-          .pipe(Effect.andThen(awaitAtomResult(defaultRegistry, atom).pipe(Effect.exit))) as any
+          .pipe(Effect.andThen(awaitAtomResult(defaultRegistry, atom).pipe(Effect.exit)))
       ),
     dispose: () => {
       unmount()