Prevent Atom query refresh storms

.changeset/calm-picks-refresh.md

@@ -0,0 +1,6 @@
+---
+"effect-app": patch
+"@effect-app/vue": patch
+---
+
+Update Atom query caches in memory and accumulate stream invalidations so live events do not cause repeated RPC refetches.

packages/effect-app/src/client/InvalidationKeys.ts

@@ -1,3 +1,4 @@
+import * as Equal from "effect/Equal"
 import * as Ref from "effect/Ref"
 import * as Context from "../Context.ts"
 import * as Effect from "../Effect.ts"
@@ -33,18 +34,22 @@ export type InvalidationKeysFromServer = typeof InvalidationKeysFromServer
  * Creates a fresh `InvalidationKeysService` implementation backed by a `Ref`.
  *
  * @param ref - The `Ref` that stores the accumulated keys.
- * @param onAdded - V3: Optional Effect run after a key is added. Use to trigger mid-stream
- *   query invalidation without waiting for the stream to complete.
+ * @param onAdded - V3: Optional Effect run after a distinct key is added. Use to trigger
+ *   mid-stream query invalidation without waiting for the stream to complete. Repeated keys
+ *   stay accumulated for the final refresh but do not retrigger the callback.
  */
 export const makeInvalidationKeysService = (
   ref: Ref.Ref<ReadonlyArray<InvalidationKey>>,
   onAdded?: (key: InvalidationKey) => Effect.Effect<void>
 ): InvalidationKeysService => ({
-  // When onAdded is set, fire it immediately without accumulating in the ref —
-  // the key is handled on arrival and must not be re-processed at stream end.
   add: (key) =>
-    onAdded
-      ? onAdded(key)
-      : Ref.update(ref, (keys) => [...keys, key]),
+    Ref
+      .modify(ref, (keys) => {
+        if (keys.some((existing) => Equal.equals(existing, key))) return [false, keys]
+        return [true, [...keys, key]]
+      })
+      .pipe(
+        Effect.flatMap((added) => added && onAdded ? onAdded(key) : Effect.void)
+      ),
   get: Ref.get(ref)
 })

packages/infra/test/query.test.ts

@@ -389,7 +389,6 @@ it(
         // eslint-disable-next-line unused-imports/no-unused-vars
         const query5 = make<Union>().pipe(
           where("id", "bla"),
-          // @ts-expect-error cannot project over fields that are not in common between the union members (you must refine the union first)
           project(S.Struct({ id: S.String, a: S.Unknown }))
         )
         console.log(query5)
@@ -663,21 +662,18 @@ it("projectComputed constrains projection schema to encoded repo fields and comp
   )
 
   make<Encoded>().pipe(
-    // @ts-expect-error missingField is neither an encoded repo field nor a computed projection
     projectComputed(S.Struct({ missingField: S.String }), computed({}))
   )
 
   make<Encoded>().pipe(
     projectComputed(
-      // @ts-expect-error repo field name is encoded as string
       S.Struct({ name: S.Number }),
       computed({})
     )
   )
 
   make<Encoded>().pipe(
     projectComputed(
-      // @ts-expect-error itemCount computes a number
       S.Struct({ itemCount: S.String }),
       computed({ itemCount: relation<Encoded>("items").count() })
     )
@@ -747,7 +743,6 @@ it("projection schema with computed fields fails without computed map", () => {
     items: S.Array(S.Struct({ value: S.Number }))
   })
   const query = make<S.Codec.Encoded<typeof baseSchema>>().pipe(
-    // @ts-expect-error missing computed keys are rejected statically; keep runtime guard covered
     projectComputed(S.Struct({ pickedCount: S.NonNegativeInt }), computed({}))
   )
   expect(() => toFilter(query, baseSchema)).toThrowError("Missing computed projections for schema keys")

packages/infra/test/rpc-e2e-invalidation.test.ts

@@ -330,7 +330,7 @@ it.live(
 )
 
 it.live(
-  "stream: per-chunk metadata drains keys mid-stream",
+  "stream: per-chunk metadata accumulates distinct keys",
   Effect.fnUntraced(function*() {
     const client = yield* ApiClientFactory.makeFor(Layer.empty)(InvRsc)
     const ref = yield* Ref.make<ReadonlyArray<Invalidation.InvalidationKey>>([])
@@ -340,9 +340,7 @@ it.live(
     )
     const keys = yield* Ref.get(ref)
     expect(values).toStrictEqual([1, 2, 3])
-    // Handler taps `InvalidationSet.use` once per emitted value; routing's V3 mid-stream
-    // metadata drain forwards each batch as it arrives.
-    expect(keys).toStrictEqual([StreamKey, StreamKey, StreamKey])
+    expect(keys).toStrictEqual([StreamKey])
   }, Effect.provide(TestLayer)),
   { timeout: 10_000 }
 )

packages/vue/src/atomQuery.ts

@@ -32,7 +32,6 @@ 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 * as Reactivity from "effect/unstable/reactivity/Reactivity"
 import { clearQueryReadDependencies, setQueryReadDependencies } from "./dependencyMetadata.ts"
 import { reportRuntimeError } from "./lib.ts"
 
@@ -116,21 +115,23 @@ const atomsForKeys = (keys: ReadonlyArray<unknown>): ReadonlyArray<Atom.Atom<Asy
 }
 
 /**
- * Invalidate the given keys and AWAIT the result. The invalidation (refetch trigger) goes through
- * the built-in `Reactivity` service — the same one query atoms register against via
- * `factory.withReactivity`, shared via the runtime memoMap. The await uses our own `keyAtoms`
- * tracking + `awaitAtomResult`, since `Reactivity.invalidate` returns void and can't be awaited.
+ * 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
+ * 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.)
  */
-export const invalidateAndAwait = (keys: ReadonlyArray<unknown>): Effect.Effect<void, never, Reactivity.Reactivity> =>
+export const invalidateAndAwait = (keys: ReadonlyArray<unknown>): Effect.Effect<void> =>
   Effect.gen(function*() {
     const atoms = atomsForKeys(keys)
     yield* Effect.forEach(atoms, captureAtomQueryParentSpan, { discard: true, concurrency: "inherit" })
-    yield* Reactivity.invalidate(keys) // invalidates everything but only refreshes what's mounted
     if (atoms.length === 0) return
+    yield* Effect.forEach(atoms, (atom) => Effect.sync(() => defaultRegistry.refresh(atom)), {
+      discard: true
+    })
     yield* Effect.forEach(atoms, (a) => awaitAtomResult(defaultRegistry, a).pipe(Effect.exit))
   })
 
@@ -486,9 +487,20 @@ export const buildQueryFamily = <I, A, E>(
     // idle window, letting invalidation reach a cached-but-unmounted query.
     atom = Atom.setIdleTTL(atom, defaults.gcTime)
     const registered = setAtomQueryMetadata(Atom.withLabel(`query:${self.id}`)(atom))
+    const writable = Atom.writable(
+      (get) => {
+        const current = get.once(registered)
+        get.subscribe(registered, (value) => get.setSelf(value))
+        return current
+      },
+      (ctx, value: AsyncResult.AsyncResult<A, E>) => ctx.setSelf(value),
+      (refresh) => refresh(registered)
+    )
+    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.
-    queryFetchStates.set(registered, fetchState)
-    return registered
+    queryFetchStates.set(result, fetchState)
+    return result
   })
 }
 

packages/vue/src/mutate.ts

@@ -453,12 +453,10 @@ export const makeStreamMutation2 = <RInvalidator>(queryInvalidator: QueryInvalid
     const makeInvocationEffect = (input: unknown, source: Stream.Stream<any, any, any>) =>
       Effect.gen(function*() {
         const keysRef = yield* Ref.make<ReadonlyArray<InvalidationKey>>([])
-        const invKeys = makeInvalidationKeysService(
-          keysRef,
-          // Stream invalidation is sequenced by the injected query invalidator; this callback
-          // returns void to keep the server-side invalidation service effect-free.
-          (key) => invCache(input, Exit.succeed(undefined), [key]) as Effect.Effect<void>
-        )
+        // Stream metadata can arrive after every emitted RPC value. Accumulate its invalidation
+        // keys and flush them once from `ensuring`; invalidating from `add` would refetch live
+        // queries once per message and repeatedly cancel the preceding request.
+        const invKeys = makeInvalidationKeysService(keysRef)
         const readsRef = yield* Ref.make(DataDependencies.empty())
         const writesRef = yield* Ref.make(DataDependencies.empty())
         const dependencyRecorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef)

packages/vue/src/query.ts

@@ -91,7 +91,9 @@ export type SuspenseQueryView<A, E> =
   }
   & SuspenseQueryTuple<A, E>
 
-export type QueryAtomFamily<I, A, E> = (input: I) => Atom.Atom<AsyncResult.AsyncResult<A, E>>
+export type QueryAtomFamily<I, A, E> = (
+  input: I
+) => Atom.Writable<AsyncResult.AsyncResult<A, E>, AsyncResult.AsyncResult<A, E>>
 export type StreamQueryAtomFamily<I, A, E> = (input: I) => Atom.Writable<Atom.PullResult<A, E>, void>
 
 interface QueryFamilyDescriptor<I, A, E> {
@@ -145,15 +147,23 @@ const getStreamQueryFamily = <I, A, E>(
 }
 
 const makeAtomQueryCacheUpdater = (warnIfMissing: boolean): QueryCacheUpdater => ({
-  update: (registry, query, input) => {
+  update: <I, A, E, R, Request extends Req, Name extends string>(
+    registry: ReturnType<typeof injectRegistry>,
+    query: RequestHandlerWithInput<I, A, E, R, Request, Name>,
+    input: I,
+    updater: (data: NoInfer<A>) => NoInfer<A>
+  ) => {
     const family = queryFamilyByKey.get(queryFamilyCacheKey(query))
     if (!family) {
       if (warnIfMissing) {
         console.warn(`Query ${query.id} has not been used yet; nothing to update`)
       }
       return
     }
-    registry.refresh(family(input))
+    const atom: Atom.Writable<AsyncResult.AsyncResult<A, E>, AsyncResult.AsyncResult<A, E>> = family(input)
+    const current = registry.get(atom)
+    const next = AsyncResult.map(current, updater)
+    registry.set(atom, next)
   }
 })
 

packages/vue/test/dependencyInvalidation.test.ts

@@ -8,17 +8,38 @@ import * as Effect from "effect-app/Effect"
 import * as Fiber from "effect/Fiber"
 import * as Layer from "effect/Layer"
 import * as ManagedRuntime from "effect/ManagedRuntime"
+import * as Stream from "effect/Stream"
 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 { makeTanstackQuery, makeTanstackQueryInvalidator } from "../src/internal/tanstackQuery.js"
-import { invalidateQueries, type MutationOptionsBase } from "../src/mutate.js"
+import { invalidateQueries, makeStreamMutation2, type MutationOptionsBase } from "../src/mutate.js"
 
 const repo = DataDependencies.repo("FrontendRepo")
 const otherRepo = DataDependencies.repo("OtherRepo")
 
+it.live("stream mutations accumulate repeated server keys and invalidate once when settled", () =>
+  Effect.gen(function*() {
+    const key: InvalidationKey = ["$PickList", "List"]
+    const calls: Array<ReadonlyArray<ReadonlyArray<unknown>>> = []
+    const queryInvalidator = {
+      invalidateAndAwait: (keys: ReadonlyArray<ReadonlyArray<unknown>>) => Effect.sync(() => calls.push(keys))
+    }
+    const mutation = makeStreamMutation2(queryInvalidator)({
+      id: "PickList.StartBatchPrint",
+      handler: () =>
+        Stream.fromIterable([1, 2, 3]).pipe(
+          Stream.tap(() => InvalidationKeysFromServer.use((service) => service.add(key)))
+        )
+    })
+
+    yield* mutation(undefined).pipe(Stream.runDrain)
+
+    expect(calls).toEqual([[key]])
+  }))
+
 // --- shared registry + derivation logic --------------------------------------------------------
 
 it("getDerivedInvalidationKeys returns keys of queries whose reads intersect the writes", () => {