From 916195bd8b9580252568efd568bc048b33032618 Mon Sep 17 00:00:00 2001 From: jfet97 Date: Mon, 26 Oct 2026 00:16:10 +0100 Subject: [PATCH 1/2] fix(vue): keep live queries registered across refreshes --- .changeset/steady-live-subscriptions.md | 5 ++ packages/vue/src/liveQueryInvalidation.ts | 7 +- .../vue/test/liveQueryInvalidation.test.ts | 70 ++++++++++++++++++- 3 files changed, 79 insertions(+), 3 deletions(-) create mode 100644 .changeset/steady-live-subscriptions.md diff --git a/.changeset/steady-live-subscriptions.md b/.changeset/steady-live-subscriptions.md new file mode 100644 index 000000000..e8ff1ed01 --- /dev/null +++ b/.changeset/steady-live-subscriptions.md @@ -0,0 +1,5 @@ +--- +"@effect-app/vue": patch +--- + +Keep live queries subscribed across successive repository writes by dispatching each write to a stable snapshot of the registered queries. diff --git a/packages/vue/src/liveQueryInvalidation.ts b/packages/vue/src/liveQueryInvalidation.ts index 234fcfcc9..171103072 100644 --- a/packages/vue/src/liveQueryInvalidation.ts +++ b/packages/vue/src/liveQueryInvalidation.ts @@ -44,7 +44,9 @@ const schedule = (entry: LiveQueryEntry) => { const dispatch = (writes: DataDependencies.DataDependencies) => { if (!DataDependencies.isNonEmpty(writes)) return - for (const entry of entries.values()) { + // a synchronous refresh can replace registrations while this write is being dispatched + const snapshot = [...entries.values()] + for (const entry of snapshot) { if (DataDependencies.intersects(entry.reads(), writes)) schedule(entry) } } @@ -59,7 +61,8 @@ const receive = (writes: DataDependencies.DataDependencies) => { const reset = () => { bufferedWrites = DataDependencies.empty() - for (const entry of entries.values()) { + const snapshot = [...entries.values()] + for (const entry of snapshot) { if (DataDependencies.isNonEmpty(entry.reads())) schedule(entry) } } diff --git a/packages/vue/test/liveQueryInvalidation.test.ts b/packages/vue/test/liveQueryInvalidation.test.ts index 9c2df6e9b..0a20c3121 100644 --- a/packages/vue/test/liveQueryInvalidation.test.ts +++ b/packages/vue/test/liveQueryInvalidation.test.ts @@ -1,5 +1,10 @@ -import { DataDependencies } from "effect-app/client" +import { defaultRegistry } from "@effect/atom-vue" +import { DataDependencies, makeQueryKey } from "effect-app/client" +import * as Effect from "effect-app/Effect" +import * as Layer from "effect-app/Layer" +import * as ManagedRuntime from "effect/ManagedRuntime" import { afterEach, describe, expect, it, vi } from "vitest" +import { awaitAtomResult, buildQueryFamily, invalidateAndAwait, makeAtomClientRuntime, withQueryOptions } from "../src/atomQuery.js" import { beginLiveQueryFetch, configureLiveQueryInvalidation, endLiveQueryFetch, registerLiveQuery } from "../src/liveQueryInvalidation.js" const repo = DataDependencies.repo("Inventory", ["item-1"]) @@ -13,6 +18,69 @@ describe("live query invalidation", () => { vi.useRealTimers() }) + it.each(["await", "soft"] as const)( + "keeps an observed %s query subscribed after successive repository-write refreshes", + async (invalidation) => { + const dependency = DataDependencies.repo("LiveOrders") + const mrt = ManagedRuntime.make(Layer.empty) + const rt = makeAtomClientRuntime(() => Layer.empty, mrt.memoMap) + let calls = 0 + let receive = (_writes: DataDependencies.DataDependencies) => {} + let reset = () => {} + let settled = Promise.resolve() + let refreshes = 0 + configureLiveQueryInvalidation({ + ready: () => Promise.resolve(), + subscribe: (onWrites, onReset) => { + receive = onWrites + reset = onReset + return () => { + receive = () => {} + reset = () => {} + } + }, + invalidate: (keys) => { + refreshes++ + settled = Effect.runPromise(invalidateAndAwait(keys)) + } + }) + const query = { + id: "LiveOrders.List", + handler: () => DataDependencies.read(dependency).pipe(Effect.andThen(Effect.sync(() => ++calls))) + } + const base = buildQueryFamily(rt, query)(undefined) + const atom = withQueryOptions(base, { live: true, invalidation, revalidateOnFocus: false }, [ + ...makeQueryKey(query), + undefined + ]) + const unmount = defaultRegistry.mount(atom) + try { + await Effect.runPromise(awaitAtomResult(defaultRegistry, atom)) + expect(calls).toBe(1) + receive(new Set([dependency])) + await settled + expect(calls).toBe(2) + expect(refreshes).toBe(1) + receive(new Set([dependency])) + await settled + expect(calls).toBe(3) + expect(refreshes).toBe(2) + reset() + await settled + expect(calls).toBe(4) + expect(refreshes).toBe(3) + receive(new Set([dependency])) + await settled + expect(calls).toBe(5) + expect(refreshes).toBe(4) + } finally { + unmount() + defaultRegistry.reset() + await mrt.dispose() + } + } + ) + it("invalidates only live queries whose recorded reads intersect", () => { const invalidations: Array>> = [] let receive = (_writes: DataDependencies.DataDependencies) => {} From fcc1faaa42389f0deadc9fd766f1da989b7b98b8 Mon Sep 17 00:00:00 2001 From: "omegent-app[bot]" <306514130+omegent-app[bot]@users.noreply.github.com> Date: Mon, 26 Oct 2026 08:43:47 +0000 Subject: [PATCH 2/2] fix(vue): skip stale live-query registrations and honor soft mode for live queries Co-Authored-By: Claude Opus 5.5 --- .changeset/steady-live-subscriptions.md | 2 +- packages/vue/src/atomQuery.ts | 5 +- packages/vue/src/liveQueryInvalidation.ts | 16 +++--- .../vue/test/liveQueryInvalidation.test.ts | 50 ++++++++++++++++++- 4 files changed, 61 insertions(+), 12 deletions(-) diff --git a/.changeset/steady-live-subscriptions.md b/.changeset/steady-live-subscriptions.md index e8ff1ed01..fc394ff13 100644 --- a/.changeset/steady-live-subscriptions.md +++ b/.changeset/steady-live-subscriptions.md @@ -2,4 +2,4 @@ "@effect-app/vue": patch --- -Keep live queries subscribed across successive repository writes by dispatching each write to a stable snapshot of the registered queries. +Keep live queries subscribed across successive repository writes: dispatch each write/reset to a snapshot of current registrations, skipping ones removed or replaced mid-dispatch. Live queries now also honor `invalidation: "soft"`. diff --git a/packages/vue/src/atomQuery.ts b/packages/vue/src/atomQuery.ts index e31750fab..e7dc7b290 100644 --- a/packages/vue/src/atomQuery.ts +++ b/packages/vue/src/atomQuery.ts @@ -394,14 +394,15 @@ export const withQueryOptions = ( } if (opts.live && liveKey !== undefined) { const liveOptions = opts.live === true ? {} : opts.live - atom = Atom.transform(atom, (get) => { + const inner = atom + atom = Atom.transform(inner, (get) => { const unregister = registerLiveQuery( liveKey, () => getQueryReadDependencies(liveKey), liveOptions ) get.addFinalizer(unregister) - return get(self) + return get(inner) }, { initialValueTarget: self }) } const revalidateOnFocus = opts.revalidateOnFocus ?? true diff --git a/packages/vue/src/liveQueryInvalidation.ts b/packages/vue/src/liveQueryInvalidation.ts index 171103072..48fdefc29 100644 --- a/packages/vue/src/liveQueryInvalidation.ts +++ b/packages/vue/src/liveQueryInvalidation.ts @@ -17,6 +17,7 @@ export interface LiveQueryInvalidationSource { } interface LiveQueryEntry { + readonly hash: number readonly key: ReadonlyArray readonly reads: () => DataDependencies.DataDependencies observers: number @@ -42,12 +43,13 @@ const schedule = (entry: LiveQueryEntry) => { entry.timer = setTimeout(() => invalidateEntry(entry), maxDelayMs) } +// a synchronous refresh can remove or replace registrations while dispatching +const isCurrent = (entry: LiveQueryEntry) => entries.get(entry.hash) === entry + const dispatch = (writes: DataDependencies.DataDependencies) => { if (!DataDependencies.isNonEmpty(writes)) return - // a synchronous refresh can replace registrations while this write is being dispatched - const snapshot = [...entries.values()] - for (const entry of snapshot) { - if (DataDependencies.intersects(entry.reads(), writes)) schedule(entry) + for (const entry of Array.from(entries.values())) { + if (isCurrent(entry) && DataDependencies.intersects(entry.reads(), writes)) schedule(entry) } } @@ -61,9 +63,8 @@ const receive = (writes: DataDependencies.DataDependencies) => { const reset = () => { bufferedWrites = DataDependencies.empty() - const snapshot = [...entries.values()] - for (const entry of snapshot) { - if (DataDependencies.isNonEmpty(entry.reads())) schedule(entry) + for (const entry of Array.from(entries.values())) { + if (isCurrent(entry) && DataDependencies.isNonEmpty(entry.reads())) schedule(entry) } } @@ -93,6 +94,7 @@ export const registerLiveQuery = ( current.delays.set(delay, (current.delays.get(delay) ?? 0) + 1) } else { entries.set(hash, { + hash, key, reads, observers: 1, diff --git a/packages/vue/test/liveQueryInvalidation.test.ts b/packages/vue/test/liveQueryInvalidation.test.ts index 0a20c3121..d807cda3c 100644 --- a/packages/vue/test/liveQueryInvalidation.test.ts +++ b/packages/vue/test/liveQueryInvalidation.test.ts @@ -4,7 +4,8 @@ import * as Effect from "effect-app/Effect" import * as Layer from "effect-app/Layer" import * as ManagedRuntime from "effect/ManagedRuntime" import { afterEach, describe, expect, it, vi } from "vitest" -import { awaitAtomResult, buildQueryFamily, invalidateAndAwait, makeAtomClientRuntime, withQueryOptions } from "../src/atomQuery.js" +import { awaitAtomResult, buildQueryFamily, invalidateAndAwait, invalidateSoft, makeAtomClientRuntime, withQueryOptions } from "../src/atomQuery.js" +import { getQueryInvalidationMode } from "../src/dependencyMetadata.js" import { beginLiveQueryFetch, configureLiveQueryInvalidation, endLiveQueryFetch, registerLiveQuery } from "../src/liveQueryInvalidation.js" const repo = DataDependencies.repo("Inventory", ["item-1"]) @@ -41,7 +42,11 @@ describe("live query invalidation", () => { }, invalidate: (keys) => { refreshes++ - settled = Effect.runPromise(invalidateAndAwait(keys)) + settled = Effect.runPromise( + keys.every((key) => getQueryInvalidationMode(key) === "soft") + ? invalidateSoft(keys).pipe(Effect.andThen(awaitAtomResult(defaultRegistry, atom))) + : invalidateAndAwait(keys) + ) } }) const query = { @@ -57,6 +62,7 @@ describe("live query invalidation", () => { try { await Effect.runPromise(awaitAtomResult(defaultRegistry, atom)) expect(calls).toBe(1) + expect(getQueryInvalidationMode([...makeQueryKey(query), undefined])).toBe(invalidation) receive(new Set([dependency])) await settled expect(calls).toBe(2) @@ -81,6 +87,46 @@ describe("live query invalidation", () => { } ) + it.each(["write", "reset"] as const)( + "skips registrations removed or replaced by a synchronous refresh during %s dispatch", + (event) => { + vi.useFakeTimers() + const invalidations: Array>> = [] + let receive = (_writes: DataDependencies.DataDependencies) => {} + let reset = () => {} + const first = ["$Inventory", "List", undefined] + const removed = ["$Inventory", "Get", { id: "item-1" }] + const replaced = ["$Inventory", "Get", { id: "item-2" }] + let unregisterRemoved = () => {} + let unregisterReplaced = () => {} + configureLiveQueryInvalidation({ + ready: () => Promise.resolve(), + subscribe: (onWrites, onReset) => { + receive = onWrites + reset = onReset + return () => {} + }, + invalidate: (keys) => { + invalidations.push(keys) + if (keys[0] !== first) return + unregisterRemoved() + unregisterReplaced() + unregisterReplaced = registerLiveQuery(replaced, () => new Set([repo]), { maxDelayMs: 1_000 }) + } + }) + + cleanups.push(registerLiveQuery(first, () => new Set([repo]), {})) + unregisterRemoved = registerLiveQuery(removed, () => new Set([repo]), {}) + unregisterReplaced = registerLiveQuery(replaced, () => new Set([repo]), {}) + cleanups.push(() => unregisterReplaced()) + + if (event === "write") receive(new Set([repo])) + else reset() + vi.runAllTimers() + expect(invalidations).toEqual([[first]]) + } + ) + it("invalidates only live queries whose recorded reads intersect", () => { const invalidations: Array>> = [] let receive = (_writes: DataDependencies.DataDependencies) => {}