From 3cccfe3b17d09aca92f6d22397e7a6fde0b4a530 Mon Sep 17 00:00:00 2001 From: "omegent-app[bot]" <306514130+omegent-app[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 08:32:32 +0000 Subject: [PATCH] fix(vue): flush stream write-deps on first arrival MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Server invalidation keys stay settlement-only so list queries are not refetched per chunk. The first write-dep set now invalidates matching queries while the stream is still open (job create → GetActiveJob). Co-authored-by: Enrico Polanski <16064771+enricopolanski@users.noreply.github.com> Co-authored-by: Patrick Roza <42661+patroza@users.noreply.github.com> --- .changeset/stream-first-write-flush.md | 5 ++ packages/vue/src/mutate.ts | 16 +++++-- .../vue/test/dependencyInvalidation.test.ts | 47 +++++++++++++++++++ wiki/repository-derived-query-invalidation.md | 11 +++-- 4 files changed, 72 insertions(+), 7 deletions(-) create mode 100644 .changeset/stream-first-write-flush.md diff --git a/.changeset/stream-first-write-flush.md b/.changeset/stream-first-write-flush.md new file mode 100644 index 000000000..25591fe83 --- /dev/null +++ b/.changeset/stream-first-write-flush.md @@ -0,0 +1,5 @@ +--- +"@effect-app/vue": patch +--- + +Flush stream mutation write-deps once when the first write arrives, then again on settlement. Long-running streams can refresh queries like GetActiveJob without invalidating list queries on every subsequent item write. diff --git a/packages/vue/src/mutate.ts b/packages/vue/src/mutate.ts index edc09d219..4554a36b8 100644 --- a/packages/vue/src/mutate.ts +++ b/packages/vue/src/mutate.ts @@ -478,18 +478,28 @@ export const makeStreamMutation2 = (queryInvalidator: QueryInvalid const makeInvocationEffect = (input: unknown, source: Stream.Stream) => Effect.gen(function*() { const keysRef = yield* Ref.make>([]) - // 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. + // Server invalidation keys stay settlement-only: flushing them from `add` refetched + // live queries once per chunk (One-Pick List storms). Write-deps flush once when the + // first write arrives (job create → GetActiveJob) and again from `ensuring`. const invKeys = makeInvalidationKeysService(keysRef) const readsRef = yield* Ref.make(DataDependencies.empty()) const writesRef = yield* Ref.make(DataDependencies.empty()) const dependencyRecorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef) const lastRef = yield* Ref.make(undefined) + let flushedFirstWrites = false + const flushFirstWrites = Effect.gen(function*() { + if (flushedFirstWrites) return + const writeDependencies = yield* Ref.get(writesRef) + if (!DataDependencies.isNonEmpty(writeDependencies)) return + flushedFirstWrites = true + const lastValue = yield* Ref.get(lastRef) + yield* invCache(input, Exit.succeed(lastValue), [], writeDependencies) + }) return source.pipe( Stream.provideService(InvalidationKeysFromServer, invKeys), Stream.provideService(DataDependencies.DataDependencyRecorder, dependencyRecorder), Stream.tap((v) => Ref.set(lastRef, v)), + Stream.tap(() => flushFirstWrites), Stream.ensuring( Effect.gen(function*() { const lastValue = yield* Ref.get(lastRef) diff --git a/packages/vue/test/dependencyInvalidation.test.ts b/packages/vue/test/dependencyInvalidation.test.ts index 26c291697..f88acc12e 100644 --- a/packages/vue/test/dependencyInvalidation.test.ts +++ b/packages/vue/test/dependencyInvalidation.test.ts @@ -5,6 +5,7 @@ import { QueryClient, VueQueryPlugin } from "@tanstack/vue-query" import { DataDependencies, type InvalidationKey, InvalidationKeysFromServer, makeQueryKey } from "effect-app/client" import * as Context from "effect-app/Context" import * as Effect from "effect-app/Effect" +import * as Deferred from "effect/Deferred" import * as Fiber from "effect/Fiber" import * as Layer from "effect/Layer" import * as ManagedRuntime from "effect/ManagedRuntime" @@ -40,6 +41,52 @@ it.live("stream mutations accumulate repeated server keys and invalidate once wh expect(calls).toEqual([[key]]) })) +it.live("stream mutations flush write-deps once when they first arrive, then again on settle", () => + Effect.gen(function*() { + const jobKey: InvalidationKey = ["$PickList", "GetActiveJob"] + const listKey: InvalidationKey = ["$PickList", "List"] + const jobRepo = DataDependencies.repo("PickJob") + const itemRepo = DataDependencies.repo("PickItem") + setQueryReadDependencies(jobKey, new Set([jobRepo])) + setQueryReadDependencies(listKey, new Set([itemRepo])) + + const calls: Array>> = [] + const firstFlush = yield* Deferred.make() + const gate = yield* Deferred.make() + const queryInvalidator = { + invalidateAndAwait: (keys: ReadonlyArray>) => + Effect + .sync(() => { + calls.push(keys) + }) + .pipe( + Effect.flatMap(() => calls.length === 1 ? Deferred.succeed(firstFlush, undefined) : Effect.void) + ) + } + const mutation = makeStreamMutation2(queryInvalidator)({ + id: "PickList.StartBatchPrint", + handler: () => + Stream.make(1).pipe( + Stream.tap(() => DataDependencies.write(jobRepo)), + Stream.concat(Stream.fromEffect(Deferred.await(gate).pipe(Effect.as(2)))), + Stream.tap((n) => n === 2 ? DataDependencies.write(itemRepo) : Effect.void) + ) + }) + + try { + const fiber = yield* Effect.forkChild(Stream.runDrain(mutation(undefined))) + yield* Deferred.await(firstFlush) + expect(calls).toEqual([[jobKey]]) + + yield* Deferred.succeed(gate, undefined) + yield* Fiber.join(fiber) + expect(calls).toEqual([[jobKey], [jobKey, listKey]]) + } finally { + clearQueryReadDependencies(jobKey) + clearQueryReadDependencies(listKey) + } + })) + // --- shared registry + derivation logic -------------------------------------------------------- it("getDerivedInvalidationKeys returns keys of queries whose reads intersect the writes", () => { diff --git a/wiki/repository-derived-query-invalidation.md b/wiki/repository-derived-query-invalidation.md index 7627cb7ff..7b1c6721a 100644 --- a/wiki/repository-derived-query-invalidation.md +++ b/wiki/repository-derived-query-invalidation.md @@ -37,8 +37,8 @@ type DataDependency = Each request has a `DataDependencyRecorder` in context. Code can record: ```ts -yield* DataDependencies.read(DataDependencies.repo("PickList")) -yield* DataDependencies.write(DataDependencies.repo("PickList")) +yield * DataDependencies.read(DataDependencies.repo("PickList")) +yield * DataDependencies.write(DataDependencies.repo("PickList")) ``` Repository operations do this automatically, so most resource handlers do not @@ -136,8 +136,11 @@ the local `DataDependencyRecorder`. This makes dependency propagation work for both direct RPC clients and the Vue query/mutation helpers. Stream commands emit dependency metadata in the same metadata chunks already -used for server-driven invalidation keys. Writes can therefore invalidate -queries mid-stream or when the stream completes. +used for server-driven invalidation keys. `@effect-app/vue` `makeStreamMutation2` +flushes **write-deps once** when the first write arrives (so a created job +refreshes `GetActiveJob` while the stream is still open) and again when the +stream settles. Server invalidation keys stay settlement-only — flushing them +per chunk refetched live list queries on every item. ## Vue cache integration