From 0bcc42958f9e55f95b6cee099a64f7599c8fed01 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Tue, 7 Jul 2026 11:59:41 +0200 Subject: [PATCH 01/17] feat(db): shared live-query observer + migrate all five adapters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add createLiveQueryObserver to @tanstack/db. Given a resolved collection (or null for disabled), it owns the shared lifecycle: start sync, subscribe with initial state, the loading→ready notify, a stable per-revision snapshot for wholesale consumers, and delivery of the raw ChangeMessage[] for granular consumers (deferInitialNotify defers the initial notify for useSyncExternalStore consumers like React). React, Vue, Svelte, Solid, and Angular all materialize from the observer, removing their duplicated subscribe/status/ready-race plumbing while keeping native reactivity: Vue/Svelte/Solid apply the change deltas granularly to their reactive maps; React/Angular consume the snapshot wholesale. Observer unit tests cover the wholesale and granular paths, disabled, deferred-notify, and dispose. Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/live-query-observer.md | 12 + packages/angular-db/src/index.ts | 33 +-- packages/angular-db/tests/conformance.test.ts | 8 +- packages/db/src/index.ts | 1 + packages/db/src/live-query-observer.ts | 242 ++++++++++++++++++ packages/db/tests/live-query-observer.test.ts | 120 +++++++++ packages/react-db/src/useLiveQuery.ts | 161 ++---------- packages/solid-db/src/useLiveQuery.ts | 42 +-- packages/svelte-db/src/useLiveQuery.svelte.ts | 103 +++----- packages/vue-db/src/useLiveQuery.ts | 107 +++----- 10 files changed, 513 insertions(+), 316 deletions(-) create mode 100644 .changeset/live-query-observer.md create mode 100644 packages/db/src/live-query-observer.ts create mode 100644 packages/db/tests/live-query-observer.test.ts diff --git a/.changeset/live-query-observer.md b/.changeset/live-query-observer.md new file mode 100644 index 000000000..4b1218483 --- /dev/null +++ b/.changeset/live-query-observer.md @@ -0,0 +1,12 @@ +--- +'@tanstack/db': minor +'@tanstack/react-db': patch +'@tanstack/vue-db': patch +'@tanstack/svelte-db': patch +'@tanstack/solid-db': patch +'@tanstack/angular-db': patch +--- + +Add a shared live-query observer and migrate all five framework adapters to it + +Introduces `createLiveQueryObserver` in `@tanstack/db`: given a resolved live-query collection (or `null` for a disabled query) it owns the shared lifecycle every adapter used to re-implement — start sync, subscribe to changes, the already-ready notify race, a stable per-revision snapshot for wholesale consumers, and delivery of the raw `ChangeMessage[]` for granular consumers. React, Vue, Svelte, Solid, and Angular's live-query hooks now materialize from the observer instead of their own hand-rolled subscription/status/snapshot machinery, keeping each adapter's native reactivity. No behavior change. diff --git a/packages/angular-db/src/index.ts b/packages/angular-db/src/index.ts index d7dc2c1c8..aa9bb35a5 100644 --- a/packages/angular-db/src/index.ts +++ b/packages/angular-db/src/index.ts @@ -9,11 +9,11 @@ import { import { BaseQueryBuilder, createLiveQueryCollection, + createLiveQueryObserver, isCollection, isSingleResultCollection, } from '@tanstack/db' import type { - ChangeMessage, Collection, CollectionStatus, Context, @@ -247,28 +247,21 @@ export function injectLiveQuery(opts: any) { cleanup() - // Initialize immediately with current state - syncDataFromCollection(currentCollection) - - // Start sync if idle - if (currentCollection.status === `idle`) { - currentCollection.startSyncImmediate() - // Update status after starting sync - status.set(currentCollection.status) - } + // The shared observer owns sync start, subscription, the ready-race, and + // status transitions; Angular re-reads the whole collection on each notify + // (wholesale) into its signals. + const observer = createLiveQueryObserver(currentCollection) - // Subscribe to changes - const subscription = currentCollection.subscribeChanges( - (_: Array>) => { - syncDataFromCollection(currentCollection) - }, - ) - unsub = subscription.unsubscribe.bind(subscription) + // Seed immediately from the post-start state, then re-read on every notify. + syncDataFromCollection(currentCollection) - // Handle ready state - currentCollection.onFirstReady(() => { - status.set(currentCollection.status) + const unsubscribe = observer.subscribe(() => { + syncDataFromCollection(currentCollection) }) + unsub = () => { + unsubscribe() + observer.dispose() + } onCleanup(cleanup) }) diff --git a/packages/angular-db/tests/conformance.test.ts b/packages/angular-db/tests/conformance.test.ts index b60b8fa3d..a8cc7ec80 100644 --- a/packages/angular-db/tests/conformance.test.ts +++ b/packages/angular-db/tests/conformance.test.ts @@ -214,13 +214,7 @@ const angularDriver: LiveQueryDriver = { mountCollection, mountConfig, mountDisabled, - // Divergence the suite surfaced: angular-db's plain `{ query }` config-object - // path calls createLiveQueryCollection(opts) as-is, without injecting - // startSync:true the way the query-fn path does — so a bare `{ query }` never - // syncs and returns empty. React/Vue/Svelte/Solid all auto-start a config - // object; Angular requires an explicit `startSync: true` (its own config test - // passes it). Recorded until angular-db aligns. - knownGaps: [`config-object-input`], + knownGaps: [], features: { serverSnapshot: false, suspense: false }, } diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 71e264d71..bf4e16a81 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -11,6 +11,7 @@ export * from './proxy' export * from './query/index.js' export * from './optimistic-action' export * from './live-query-adapter' +export * from './live-query-observer' export * from './local-only' export * from './local-storage' export * from './errors' diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts new file mode 100644 index 000000000..46e58d839 --- /dev/null +++ b/packages/db/src/live-query-observer.ts @@ -0,0 +1,242 @@ +import { + getLiveQueryStatusFlags, + isSingleResultCollection, +} from './live-query-adapter.js' +import type { Collection } from './collection/index.js' +import type { ChangeMessage, CollectionStatus } from './types.js' + +/** + * The canonical, adapter-agnostic view of a live query at a point in time. + * + * `getSnapshot()` returns a stable object identity that only changes when the + * query changes, so `useSyncExternalStore`-style consumers can compare by + * reference. `state`/`data` are computed lazily and cached per snapshot. + */ +export interface LiveQuerySnapshot< + T extends object, + TKey extends string | number, +> { + /** Keyed results, or `undefined` for a disabled query. */ + state: ReadonlyMap | undefined + /** Ordered results (single row for `findOne`), or `undefined` when disabled. */ + data: T | ReadonlyArray | undefined + /** The underlying collection, or `undefined` when disabled. */ + collection: Collection | undefined + status: CollectionStatus | `disabled` + isLoading: boolean + isReady: boolean + isIdle: boolean + isError: boolean + isCleanedUp: boolean + isEnabled: boolean +} + +/** Listener payload: the change set, or `undefined` for the synthetic ready notify. */ +export type LiveQueryObserverListener< + T extends object, + TKey extends string | number, +> = (changes: Array> | undefined) => void + +/** + * Wraps a resolved live-query `Collection` (or `null` for a disabled query) with + * the shared lifecycle every framework adapter needs: start sync, subscribe to + * changes, handle the already-ready race, expose a stable snapshot for + * wholesale consumers, and deliver the raw change set for granular consumers. + * + * Input resolution (query fn / config / collection / disabled) stays in the + * adapter — it is framework-reactive. The observer owns everything after the + * input is resolved to a concrete collection. + */ +export interface LiveQueryObserver< + T extends object, + TKey extends string | number, +> { + /** Stable per-revision snapshot for wholesale materialization. */ + getSnapshot: () => LiveQuerySnapshot + /** + * Subscribe to changes. The listener receives the change set (or `undefined` + * for the synthetic notify a ready collection emits on attach). Granular + * adapters apply the changes; wholesale adapters can ignore them and re-read + * `getSnapshot()`. Returns an unsubscribe function. + */ + subscribe: (listener: LiveQueryObserverListener) => () => void + /** Resolve once the collection has loaded its first data. */ + preload: () => Promise + /** Idempotent teardown. */ + dispose: () => void +} + +const DISABLED_SNAPSHOT: LiveQuerySnapshot = { + state: undefined, + data: undefined, + collection: undefined, + status: `disabled`, + isLoading: false, + isReady: true, + isIdle: false, + isError: false, + isCleanedUp: false, + isEnabled: false, +} + +class LiveQueryObserverImpl< + T extends object, + TKey extends string | number, +> implements LiveQueryObserver { + private readonly collection: Collection | null + private readonly deferInitialNotify: boolean + private version = 0 + private cachedVersion = -1 + private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT + private readonly listeners = new Set>() + private collectionUnsub: (() => void) | null = null + private disposed = false + + constructor( + collection: Collection | null, + deferInitialNotify: boolean, + ) { + this.collection = collection + this.deferInitialNotify = deferInitialNotify + // Starting sync during resolution matches every adapter's eager behavior. + collection?.startSyncImmediate() + } + + getSnapshot(): LiveQuerySnapshot { + const collection = this.collection + if (!collection) return DISABLED_SNAPSHOT + + // Rebuild only when the version advanced, so identity stays stable. + if (this.cachedVersion !== this.version) { + this.cachedVersion = this.version + const entries = Array.from(collection.entries()) as Array<[TKey, T]> + const singleResult = isSingleResultCollection(collection) + let stateCache: Map | null = null + let dataCache: Array | null = null + + this.cachedSnapshot = { + get state() { + if (!stateCache) stateCache = new Map(entries) + return stateCache + }, + get data() { + if (!dataCache) dataCache = entries.map(([, value]) => value) + return singleResult ? dataCache[0] : dataCache + }, + collection, + status: collection.status, + ...getLiveQueryStatusFlags(collection.status), + isEnabled: true, + } + } + return this.cachedSnapshot + } + + subscribe(listener: LiveQueryObserverListener): () => void { + this.listeners.add(listener) + if (this.listeners.size === 1) this.attach() + + let active = true + return () => { + if (!active) return + active = false + this.listeners.delete(listener) + if (this.listeners.size === 0) this.detach() + } + } + + private attach(): void { + const collection = this.collection + if (!collection || this.disposed) return + + // Subscribe with initial state so granular consumers receive the current + // rows as inserts followed by deltas through one consistent channel — the + // same contract the adapters used before the observer existed (the + // collection's per-subscriber change stream requires this to align deltes). + // + // When `deferInitialNotify` is set, emits that fire synchronously while + // attaching (the initial-state batch and an immediately-ready `onFirstReady`) + // are deferred to a microtask, so a wholesale consumer like React's + // `useSyncExternalStore` never receives a synchronous notify during + // `subscribe`. Effect/watcher-based adapters want the initial state + // synchronously, so by default it is not deferred. Later changes always emit + // synchronously. + let attaching = this.deferInitialNotify + const deferred: Array> | undefined> = [] + const notify = (changes: Array> | undefined) => { + if (this.disposed || this.listeners.size === 0) return + if (attaching) deferred.push(changes) + else this.emit(changes) + } + + const subscription = collection.subscribeChanges( + (changes) => notify(changes as Array>), + { includeInitialState: true }, + ) + this.collectionUnsub = () => subscription.unsubscribe() + + // Catch a *later* loading→ready transition that carries no change events + // (e.g. `markReady()` with no rows). Skip when already ready — the initial + // state batch above already covers that, and `onFirstReady` would fire an + // immediate duplicate. + if (collection.status !== `ready`) { + collection.onFirstReady(() => notify(undefined)) + } + + attaching = false + if (deferred.length > 0) { + queueMicrotask(() => { + if (this.disposed) return + for (const changes of deferred.splice(0)) this.emit(changes) + }) + } + } + + private detach(): void { + this.collectionUnsub?.() + this.collectionUnsub = null + } + + private emit(changes: Array> | undefined): void { + this.version++ + this.listeners.forEach((listener) => listener(changes)) + } + + async preload(): Promise { + await this.collection?.preload() + } + + dispose(): void { + if (this.disposed) return + this.disposed = true + this.detach() + this.listeners.clear() + } +} + +export interface CreateLiveQueryObserverOptions { + /** + * Defer the initial-state notify to a microtask instead of emitting it + * synchronously during `subscribe`. Set this for `useSyncExternalStore`-style + * consumers (React) that must not receive a store notify during subscribe. + * Effect/watcher-based adapters leave it off to get initial state synchronously. + */ + deferInitialNotify?: boolean +} + +/** + * Create a {@link LiveQueryObserver} for a resolved live-query collection, or a + * disabled observer when `collection` is `null`/`undefined`. + */ +export function createLiveQueryObserver< + T extends object, + TKey extends string | number, +>( + collection: Collection | null | undefined, + options: CreateLiveQueryObserverOptions = {}, +): LiveQueryObserver { + return new LiveQueryObserverImpl( + collection ?? null, + options.deferInitialNotify ?? false, + ) +} diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts new file mode 100644 index 000000000..9815ba9d0 --- /dev/null +++ b/packages/db/tests/live-query-observer.test.ts @@ -0,0 +1,120 @@ +import { describe, expect, it } from 'vitest' +import { createCollection } from '../src/collection/index.js' +import { createLiveQueryObserver } from '../src/live-query-observer.js' +import { mockSyncCollectionOptions } from './utils.js' +import type { ChangeMessage } from '../src/types.js' + +interface Row { + id: string + name: string +} + +const SEED: Array = [ + { id: `1`, name: `A` }, + { id: `2`, name: `B` }, +] + +let seq = 0 +function makeSource(data: Array = SEED) { + return createCollection( + mockSyncCollectionOptions({ + id: `observer-test-${seq++}`, + getKey: (r) => r.id, + initialData: data, + }), + ) +} + +describe(`createLiveQueryObserver`, () => { + it(`exposes a stable snapshot of a ready collection (wholesale path)`, () => { + const observer = createLiveQueryObserver(makeSource() as any) + + const snap = observer.getSnapshot() + expect(snap.isEnabled).toBe(true) + expect(snap.isReady).toBe(true) + expect(snap.status).toBe(`ready`) + expect(snap.data).toHaveLength(2) + expect(snap.state?.get(`1`)).toMatchObject({ name: `A` }) + // Same identity when nothing changed. + expect(observer.getSnapshot()).toBe(snap) + observer.dispose() + }) + + it(`delivers initial state then change deltas to subscribers (granular path)`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + const deltas: Array> = [] + const unsub = observer.subscribe((changes) => { + if (changes) deltas.push(...changes) + }) + // Initial rows arrive synchronously as inserts (includeInitialState). + expect( + deltas + .filter((c) => c.type === `insert`) + .map((c) => c.key) + .sort(), + ).toEqual([`1`, `2`]) + + const before = observer.getSnapshot() + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `3`, name: `C` } }) + source.utils.commit() + + // Subsequent deltas keep flowing synchronously... + expect(deltas.some((c) => c.type === `insert` && c.key === `3`)).toBe(true) + // ...and wholesale consumers see a fresh, updated snapshot. + const after = observer.getSnapshot() + expect(after).not.toBe(before) + expect(after.data).toHaveLength(3) + + unsub() + observer.dispose() + }) + + it(`stops notifying after unsubscribe / dispose`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + let count = 0 + const unsub = observer.subscribe(() => { + count++ + }) + unsub() + const countAfterUnsub = count // initial-state notify may have fired + + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `9`, name: `Z` } }) + source.utils.commit() + + // No further notifications after unsubscribe. + expect(count).toBe(countAfterUnsub) + observer.dispose() + }) + + it(`represents a disabled query (null collection)`, () => { + const observer = createLiveQueryObserver(null) + const snap = observer.getSnapshot() + expect(snap.isEnabled).toBe(false) + expect(snap.status).toBe(`disabled`) + expect(snap.data).toBeUndefined() + expect(snap.state).toBeUndefined() + observer.dispose() + }) + + it(`defers the initial notify to a microtask when deferInitialNotify is set`, async () => { + const observer = createLiveQueryObserver(makeSource() as any, { + deferInitialNotify: true, + }) + let notified = false + observer.subscribe(() => { + notified = true + }) + // Not synchronous during subscribe (protects React's useSyncExternalStore)... + expect(notified).toBe(false) + await Promise.resolve() + // ...delivered on the next microtask. + expect(notified).toBe(true) + observer.dispose() + }) +}) diff --git a/packages/react-db/src/useLiveQuery.ts b/packages/react-db/src/useLiveQuery.ts index 8dc3d0a31..59cb0f46f 100644 --- a/packages/react-db/src/useLiveQuery.ts +++ b/packages/react-db/src/useLiveQuery.ts @@ -2,9 +2,8 @@ import { useRef, useSyncExternalStore } from 'react' import { BaseQueryBuilder, createLiveQueryCollection, - getLiveQueryStatusFlags, + createLiveQueryObserver, isCollection, - isSingleResultCollection, } from '@tanstack/db' import type { Collection, @@ -14,6 +13,7 @@ import type { InferResultType, InitialQueryBuilder, LiveQueryCollectionConfig, + LiveQueryObserver, NonSingleResult, QueryBuilder, SingleResult, @@ -328,12 +328,10 @@ export function useLiveQuery( const depsRef = useRef | null>(null) const configRef = useRef(null) - // Use refs to track version and memoized snapshot - const versionRef = useRef(0) - const snapshotRef = useRef<{ - collection: Collection | null - version: number - } | null>(null) + // The shared observer owns subscription, the ready-race, and the snapshot. + const observerRef = useRef | null>( + null, + ) // Check if we need to create/recreate the collection const needsNewCollection = @@ -413,143 +411,30 @@ export function useLiveQuery( } } - // Reset refs when collection changes + // Recreate the observer when the underlying collection changes. if (needsNewCollection) { - versionRef.current = 0 - snapshotRef.current = null + observerRef.current?.dispose() + // Defer the initial notify: useSyncExternalStore must not be notified + // synchronously during subscribe. + observerRef.current = createLiveQueryObserver(collectionRef.current, { + deferInitialNotify: true, + }) } + const observer = observerRef.current! - // Create stable subscribe function using ref + // Stable subscribe bound to the current observer; the observer owns the + // subscription, ready-race, and disposal. const subscribeRef = useRef< ((onStoreChange: () => void) => () => void) | null >(null) if (!subscribeRef.current || needsNewCollection) { - subscribeRef.current = (onStoreChange: () => void) => { - // If no collection, return a no-op unsubscribe function - if (!collectionRef.current) { - return () => {} - } - - let unsubscribed = false - - const subscription = collectionRef.current.subscribeChanges(() => { - // Drop late notifies that race with unsubscribe. - if (unsubscribed) return - // Bump version on any change; getSnapshot will rebuild next time - versionRef.current += 1 - onStoreChange() - }) - // Already-ready collections won't emit an initial change. Notify React - // ourselves, but defer to a microtask — calling onStoreChange synchronously - // here lands during the render-to-commit window and trips React's - // "state update on a component that hasn't mounted yet" warning. - if (collectionRef.current.status === `ready`) { - queueMicrotask(() => { - if (unsubscribed) return - versionRef.current += 1 - onStoreChange() - }) - } - return () => { - unsubscribed = true - subscription.unsubscribe() - } - } - } - - // Create stable getSnapshot function using ref - const getSnapshotRef = useRef< - | (() => { - collection: Collection | null - version: number - }) - | null - >(null) - if (!getSnapshotRef.current || needsNewCollection) { - getSnapshotRef.current = () => { - const currentVersion = versionRef.current - const currentCollection = collectionRef.current - - // Recreate snapshot object only if version/collection changed - if ( - !snapshotRef.current || - snapshotRef.current.version !== currentVersion || - snapshotRef.current.collection !== currentCollection - ) { - snapshotRef.current = { - collection: currentCollection, - version: currentVersion, - } - } - - return snapshotRef.current - } - } - - // Use useSyncExternalStore to subscribe to collection changes - const snapshot = useSyncExternalStore( - subscribeRef.current, - getSnapshotRef.current, - ) - - // Track last snapshot (from useSyncExternalStore) and the returned value separately - const returnedSnapshotRef = useRef<{ - collection: Collection | null - version: number - } | null>(null) - // Keep implementation return loose to satisfy overload signatures - const returnedRef = useRef(null) - - // Rebuild returned object only when the snapshot changes (version or collection identity) - if ( - !returnedSnapshotRef.current || - returnedSnapshotRef.current.version !== snapshot.version || - returnedSnapshotRef.current.collection !== snapshot.collection - ) { - // Handle null collection case (when callback returns undefined/null) - if (!snapshot.collection) { - returnedRef.current = { - state: undefined, - data: undefined, - collection: undefined, - status: `disabled`, - isLoading: false, - isReady: true, - isIdle: false, - isError: false, - isCleanedUp: false, - isEnabled: false, - } - } else { - // Capture a stable view of entries for this snapshot to avoid tearing - const entries = Array.from(snapshot.collection.entries()) - const singleResult = isSingleResultCollection(snapshot.collection) - let stateCache: Map | null = null - let dataCache: Array | null = null - - returnedRef.current = { - get state() { - if (!stateCache) { - stateCache = new Map(entries) - } - return stateCache - }, - get data() { - if (!dataCache) { - dataCache = entries.map(([, value]) => value) - } - return singleResult ? dataCache[0] : dataCache - }, - collection: snapshot.collection, - status: snapshot.collection.status, - ...getLiveQueryStatusFlags(snapshot.collection.status), - isEnabled: true, - } - } - - // Remember the snapshot that produced this returned value - returnedSnapshotRef.current = snapshot + subscribeRef.current = (onStoreChange) => + observer.subscribe(() => onStoreChange()) } - return returnedRef.current! + // The observer returns a stable snapshot per revision, which is the return + // shape this hook exposes. Keep the return loose to satisfy the overloads. + return useSyncExternalStore(subscribeRef.current, () => + observer.getSnapshot(), + ) as any } diff --git a/packages/solid-db/src/useLiveQuery.ts b/packages/solid-db/src/useLiveQuery.ts index 7759915f3..a2a39a7fa 100644 --- a/packages/solid-db/src/useLiveQuery.ts +++ b/packages/solid-db/src/useLiveQuery.ts @@ -10,6 +10,7 @@ import { ReactiveMap } from '@solid-primitives/map' import { BaseQueryBuilder, createLiveQueryCollection, + createLiveQueryObserver, isCollection, isSingleResultCollection, } from '@tanstack/db' @@ -389,36 +390,35 @@ export function useLiveQuery( setData([]) return } - const subscription = currentCollection.subscribeChanges( - (changes: Array>) => { - // Apply each change individually to the reactive state + + // The shared observer owns subscription, the ready-race, and status; Solid + // materializes into its keyed ReactiveMap (granular) + reconciled store. + const observer = createLiveQueryObserver(currentCollection) + const unsubscribe = observer.subscribe( + (changes: Array> | undefined) => { batch(() => { - for (const change of changes) { - switch (change.type) { - case `insert`: - case `update`: - state.set(change.key, change.value) - break - case `delete`: - state.delete(change.key) - break + if (changes) { + for (const change of changes) { + switch (change.type) { + case `insert`: + case `update`: + state.set(change.key, change.value) + break + case `delete`: + state.delete(change.key) + break + } } } - syncDataFromCollection(currentCollection) - - // Update status ref on every change - setStatus(currentCollection.status) + setStatus(observer.getSnapshot().status) }) }, - { - // Include initial state to ensure immediate population for pre-created collections - includeInitialState: true, - }, ) onCleanup(() => { - subscription.unsubscribe() + unsubscribe() + observer.dispose() }) }) diff --git a/packages/svelte-db/src/useLiveQuery.svelte.ts b/packages/svelte-db/src/useLiveQuery.svelte.ts index 789d3d08f..385ccfd55 100644 --- a/packages/svelte-db/src/useLiveQuery.svelte.ts +++ b/packages/svelte-db/src/useLiveQuery.svelte.ts @@ -5,6 +5,7 @@ import { SvelteMap } from 'svelte/reactivity' import { BaseQueryBuilder, createLiveQueryCollection, + createLiveQueryObserver, isCollection, isSingleResultCollection, } from '@tanstack/db' @@ -17,6 +18,7 @@ import type { InferResultType, InitialQueryBuilder, LiveQueryCollectionConfig, + LiveQueryObserver, NonSingleResult, QueryBuilder, SingleResult, @@ -375,13 +377,26 @@ export function useLiveQuery( }) } - // Track current unsubscribe function - let currentUnsubscribe: (() => void) | null = null + // The shared observer owns subscription, the ready-race, and status; Svelte + // materializes into its own rune-backed map (granular) + ordered array. + let currentObserver: LiveQueryObserver | null = null + + const syncFromObserver = ( + observer: LiveQueryObserver, + currentCollection: Collection, + ) => { + status = observer.getSnapshot().status as CollectionStatus + syncDataFromCollection(currentCollection) + } // Watch for collection changes and subscribe to updates $effect(() => { const currentCollection = collection + // Tear down any previous observer. + currentObserver?.dispose() + currentObserver = null + // Handle null collection (disabled query) if (!currentCollection) { status = `disabled` as const @@ -389,81 +404,43 @@ export function useLiveQuery( state.clear() internalData = [] }) - if (currentUnsubscribe) { - currentUnsubscribe() - currentUnsubscribe = null - } return } - // Update status state whenever the effect runs - status = currentCollection.status - - // Clean up previous subscription - if (currentUnsubscribe) { - currentUnsubscribe() - } - - // Initialize state with current collection data - untrack(() => { - state.clear() - for (const [key, value] of currentCollection.entries()) { - state.set(key, value) - } - }) - - // Initialize data array in correct order - syncDataFromCollection(currentCollection) + const observer = createLiveQueryObserver(currentCollection) + currentObserver = observer - // Listen for the first ready event to catch status transitions - // that might not trigger change events (fixes async status transition bug) - currentCollection.onFirstReady(() => { - // Update status directly - Svelte's reactivity system handles the update automatically - // Note: We cannot use flushSync here as it's disallowed inside effects in async mode - status = currentCollection.status - }) + // Initial rows arrive as the observer's first delta (includeInitialState); + // apply them and every subsequent delta granularly to the rune-backed map. + untrack(() => state.clear()) - // Subscribe to collection changes with granular updates - const subscription = currentCollection.subscribeChanges( - (changes: Array>) => { - // Apply each change individually to the reactive state + const unsubscribe = observer.subscribe( + (changes: Array> | undefined) => { untrack(() => { - for (const change of changes) { - switch (change.type) { - case `insert`: - case `update`: - state.set(change.key, change.value) - break - case `delete`: - state.delete(change.key) - break + if (changes) { + for (const change of changes) { + switch (change.type) { + case `insert`: + case `update`: + state.set(change.key, change.value) + break + case `delete`: + state.delete(change.key) + break + } } } }) - - // Update the data array to maintain sorted order - syncDataFromCollection(currentCollection) - // Update status state on every change - status = currentCollection.status - }, - { - includeInitialState: true, + syncFromObserver(observer, currentCollection) }, ) - - currentUnsubscribe = subscription.unsubscribe.bind(subscription) - - // Preload collection data if not already started - if (currentCollection.status === `idle`) { - currentCollection.preload().catch(console.error) - } + syncFromObserver(observer, currentCollection) // Cleanup when effect is invalidated return () => { - if (currentUnsubscribe) { - currentUnsubscribe() - currentUnsubscribe = null - } + unsubscribe() + observer.dispose() + currentObserver = null } }) diff --git a/packages/vue-db/src/useLiveQuery.ts b/packages/vue-db/src/useLiveQuery.ts index 762cbda0a..c12fdb8ee 100644 --- a/packages/vue-db/src/useLiveQuery.ts +++ b/packages/vue-db/src/useLiveQuery.ts @@ -1,7 +1,6 @@ import { computed, getCurrentInstance, - nextTick, onUnmounted, reactive, ref, @@ -10,6 +9,7 @@ import { } from 'vue' import { createLiveQueryCollection, + createLiveQueryObserver, isCollection, isSingleResultCollection, } from '@tanstack/db' @@ -22,6 +22,7 @@ import type { InferResultType, InitialQueryBuilder, LiveQueryCollectionConfig, + LiveQueryObserver, NonSingleResult, QueryBuilder, SingleResult, @@ -363,101 +364,73 @@ export function useLiveQuery( internalData.push(...Array.from(currentCollection.values())) } - // Track current unsubscribe function - let currentUnsubscribe: (() => void) | null = null + // The shared observer owns subscription, the ready-race, and status; Vue + // materializes into its own reactive map (granular) + ordered array. + let currentObserver: LiveQueryObserver | null = null + + const syncFromObserver = ( + observer: LiveQueryObserver, + currentCollection: Collection, + ) => { + status.value = observer.getSnapshot().status as CollectionStatus + syncDataFromCollection(currentCollection) + } // Watch for collection changes and subscribe to updates watchEffect((onInvalidate) => { const currentCollection = collection.value + // Tear down any previous observer. + currentObserver?.dispose() + currentObserver = null + // Handle null collection (disabled query) if (!currentCollection) { status.value = `disabled` as const state.clear() internalData.length = 0 - if (currentUnsubscribe) { - currentUnsubscribe() - currentUnsubscribe = null - } return } - // Update status ref whenever the effect runs - status.value = currentCollection.status - - // Clean up previous subscription - if (currentUnsubscribe) { - currentUnsubscribe() - } + const observer = createLiveQueryObserver(currentCollection) + currentObserver = observer - // Initialize state with current collection data + // Initial rows arrive as the observer's first delta (includeInitialState); + // apply them and every subsequent delta granularly to the reactive map. state.clear() - for (const [key, value] of currentCollection.entries()) { - state.set(key, value) - } - - // Initialize data array in correct order - syncDataFromCollection(currentCollection) - // Listen for the first ready event to catch status transitions - // that might not trigger change events (fixes async status transition bug) - currentCollection.onFirstReady(() => { - // Use nextTick to ensure Vue reactivity updates properly - nextTick(() => { - status.value = currentCollection.status - }) - }) - - // Subscribe to collection changes with granular updates - const subscription = currentCollection.subscribeChanges( - (changes: Array>) => { - // Apply each change individually to the reactive state - for (const change of changes) { - switch (change.type) { - case `insert`: - case `update`: - state.set(change.key, change.value) - break - case `delete`: - state.delete(change.key) - break + const unsubscribe = observer.subscribe( + (changes: Array> | undefined) => { + if (changes) { + for (const change of changes) { + switch (change.type) { + case `insert`: + case `update`: + state.set(change.key, change.value) + break + case `delete`: + state.delete(change.key) + break + } } } - - // Update the data array to maintain sorted order - syncDataFromCollection(currentCollection) - // Update status ref on every change - status.value = currentCollection.status - }, - { - includeInitialState: true, + syncFromObserver(observer, currentCollection) }, ) - - currentUnsubscribe = subscription.unsubscribe.bind(subscription) - - // Preload collection data if not already started - if (currentCollection.status === `idle`) { - currentCollection.preload().catch(console.error) - } + syncFromObserver(observer, currentCollection) // Cleanup when effect is invalidated onInvalidate(() => { - if (currentUnsubscribe) { - currentUnsubscribe() - currentUnsubscribe = null - } + unsubscribe() + observer.dispose() + currentObserver = null }) }) // Cleanup on unmount (only if we're in a component context) const instance = getCurrentInstance() if (instance) { - onUnmounted(() => { - if (currentUnsubscribe) { - currentUnsubscribe() - } - }) + onUnmounted(() => currentObserver?.dispose()) } return { From e957121df1e63bff3234c0897e09d87a04219951 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Tue, 7 Jul 2026 17:23:58 +0200 Subject: [PATCH 02/17] fix(db): make observer onFirstReady detach-safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit onFirstReady returns no unsubscribe and detach() couldn't remove it, so a subscribe → unsubscribe-before-ready → subscribe sequence left a stale ready callback that also fired on markReady — the current listener saw two synthetic ready notifications instead of one. Guard the callback with an attach-generation token so only the current attachment's callback notifies. Adds a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/db/src/live-query-observer.ts | 15 +++++++- packages/db/tests/live-query-observer.test.ts | 38 ++++++++++++++++++- 2 files changed, 51 insertions(+), 2 deletions(-) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 46e58d839..c95f52b73 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -90,6 +90,9 @@ class LiveQueryObserverImpl< private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT private readonly listeners = new Set>() private collectionUnsub: (() => void) | null = null + // Bumped on each attach. `onFirstReady` can't be unsubscribed, so a callback + // from a superseded attach checks this to no-op instead of double-notifying. + private attachGeneration = 0 private disposed = false constructor( @@ -149,6 +152,8 @@ class LiveQueryObserverImpl< const collection = this.collection if (!collection || this.disposed) return + const generation = ++this.attachGeneration + // Subscribe with initial state so granular consumers receive the current // rows as inserts followed by deltas through one consistent channel — the // same contract the adapters used before the observer existed (the @@ -179,8 +184,16 @@ class LiveQueryObserverImpl< // (e.g. `markReady()` with no rows). Skip when already ready — the initial // state batch above already covers that, and `onFirstReady` would fire an // immediate duplicate. + // + // `onFirstReady` returns no unsubscribe, so a callback left behind by an + // earlier attach (subscribe → unsubscribe-before-ready → subscribe) would + // still fire on `markReady`. Guard with the attach generation so only the + // current attachment's callback notifies. if (collection.status !== `ready`) { - collection.onFirstReady(() => notify(undefined)) + collection.onFirstReady(() => { + if (generation !== this.attachGeneration) return + notify(undefined) + }) } attaching = false diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 9815ba9d0..5acafe9d8 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it } from 'vitest' import { createCollection } from '../src/collection/index.js' import { createLiveQueryObserver } from '../src/live-query-observer.js' -import { mockSyncCollectionOptions } from './utils.js' +import { + mockSyncCollectionOptions, + mockSyncCollectionOptionsNoInitialState, +} from './utils.js' import type { ChangeMessage } from '../src/types.js' interface Row { @@ -25,6 +28,18 @@ function makeSource(data: Array = SEED) { ) } +/** A collection that is syncing but not yet ready, with a manual `markReady`. */ +function makeLoadingSource() { + const collection = createCollection( + mockSyncCollectionOptionsNoInitialState({ + id: `observer-loading-${seq++}`, + getKey: (r) => r.id, + }), + ) + collection.startSyncImmediate() + return collection +} + describe(`createLiveQueryObserver`, () => { it(`exposes a stable snapshot of a ready collection (wholesale path)`, () => { const observer = createLiveQueryObserver(makeSource() as any) @@ -117,4 +132,25 @@ describe(`createLiveQueryObserver`, () => { expect(notified).toBe(true) observer.dispose() }) + + it(`fires the ready notify once after unsubscribe-before-ready then resubscribe`, () => { + const collection = makeLoadingSource() + const observer = createLiveQueryObserver(collection as any) + + // Subscribe then unsubscribe while still loading — this registers an + // onFirstReady callback that detach() can't remove. + observer.subscribe(() => {})() + + let readyNotifications = 0 + observer.subscribe((changes) => { + if (changes === undefined) readyNotifications++ + }) + + collection.utils.markReady() + + // Only the current subscription's ready callback should fire, not the + // stale one left behind by the first (already unsubscribed) attach. + expect(readyNotifications).toBe(1) + observer.dispose() + }) }) From c05a72603a07edc728418f777a90e44a9dc8b54a Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Thu, 9 Jul 2026 09:59:48 +0200 Subject: [PATCH 03/17] fix(db): address observer/react lifecycle review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - observer: getSnapshot() rebuilds when collection.status changes without a version bump (status-only loading→ready / preload with no active subscription), so a cached snapshot can't go stale. - observer: guard the deferred initial-notify microtask with the attach generation + listener count, so a superseded attach can't flush a stale initial batch to a later listener. - react: don't dispose the previous observer during render (unsafe under concurrent rendering) — useSyncExternalStore detaches it when the subscribe changes; dispose the current observer in an unmount effect instead. - tests: regressions for the deferred-notify race and the status-only snapshot refresh (both verified red before the fixes). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/db/src/live-query-observer.ts | 22 ++++++++++-- packages/db/tests/live-query-observer.test.ts | 36 +++++++++++++++++++ packages/react-db/src/useLiveQuery.ts | 17 +++++++-- 3 files changed, 69 insertions(+), 6 deletions(-) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index c95f52b73..8387b3838 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -87,6 +87,7 @@ class LiveQueryObserverImpl< private readonly deferInitialNotify: boolean private version = 0 private cachedVersion = -1 + private cachedStatus: CollectionStatus | undefined private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT private readonly listeners = new Set>() private collectionUnsub: (() => void) | null = null @@ -109,9 +110,15 @@ class LiveQueryObserverImpl< const collection = this.collection if (!collection) return DISABLED_SNAPSHOT - // Rebuild only when the version advanced, so identity stays stable. - if (this.cachedVersion !== this.version) { + // Rebuild when the version advanced, or when the collection's status + // changed without a version bump (e.g. a status-only loading→ready + // transition or `preload()` while there is no active subscription). + if ( + this.cachedVersion !== this.version || + this.cachedStatus !== collection.status + ) { this.cachedVersion = this.version + this.cachedStatus = collection.status const entries = Array.from(collection.entries()) as Array<[TKey, T]> const singleResult = isSingleResultCollection(collection) let stateCache: Map | null = null @@ -199,7 +206,16 @@ class LiveQueryObserverImpl< attaching = false if (deferred.length > 0) { queueMicrotask(() => { - if (this.disposed) return + // Skip if the observer was disposed, has no listeners, or a newer + // attach superseded this one before the flush — otherwise a stale + // initial batch would reach the current listener. + if ( + this.disposed || + this.listeners.size === 0 || + generation !== this.attachGeneration + ) { + return + } for (const changes of deferred.splice(0)) this.emit(changes) }) } diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 5acafe9d8..1035a11ef 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -153,4 +153,40 @@ describe(`createLiveQueryObserver`, () => { expect(readyNotifications).toBe(1) observer.dispose() }) + + it(`does not flush a superseded deferred initial notify (deferInitialNotify)`, async () => { + const observer = createLiveQueryObserver(makeSource() as any, { + deferInitialNotify: true, + }) + + // Subscribe then unsubscribe before the microtask flush, then resubscribe. + observer.subscribe(() => {})() + + let notifications = 0 + observer.subscribe(() => { + notifications++ + }) + await Promise.resolve() + + // Only the current subscription's deferred initial notify should flush, + // not the stale one queued by the first (superseded) attach. + expect(notifications).toBe(1) + observer.dispose() + }) + + it(`refreshes the snapshot when status changes without a version bump`, () => { + // A status-only loading→ready transition with no active subscription: the + // cached snapshot must not stay stale (covers the preload() case too). + const collection = makeLoadingSource() + const observer = createLiveQueryObserver(collection as any) + + expect(observer.getSnapshot().isReady).toBe(false) + expect(observer.getSnapshot().status).toBe(`loading`) + + collection.utils.markReady() + + expect(observer.getSnapshot().isReady).toBe(true) + expect(observer.getSnapshot().status).toBe(`ready`) + observer.dispose() + }) }) diff --git a/packages/react-db/src/useLiveQuery.ts b/packages/react-db/src/useLiveQuery.ts index 59cb0f46f..0baa266f0 100644 --- a/packages/react-db/src/useLiveQuery.ts +++ b/packages/react-db/src/useLiveQuery.ts @@ -1,4 +1,4 @@ -import { useRef, useSyncExternalStore } from 'react' +import { useEffect, useRef, useSyncExternalStore } from 'react' import { BaseQueryBuilder, createLiveQueryCollection, @@ -411,9 +411,12 @@ export function useLiveQuery( } } - // Recreate the observer when the underlying collection changes. + // Recreate the observer when the underlying collection changes. Do not + // dispose the previous observer here — teardown during render is unsafe under + // concurrent rendering. `useSyncExternalStore` unsubscribes the old observer + // when `subscribeRef` changes (below), which detaches it; the final observer + // is disposed in the unmount effect. if (needsNewCollection) { - observerRef.current?.dispose() // Defer the initial notify: useSyncExternalStore must not be notified // synchronously during subscribe. observerRef.current = createLiveQueryObserver(collectionRef.current, { @@ -422,6 +425,14 @@ export function useLiveQuery( } const observer = observerRef.current! + // Dispose the current observer on unmount (commit-phase cleanup). + useEffect( + () => () => { + observerRef.current?.dispose() + }, + [], + ) + // Stable subscribe bound to the current observer; the observer owns the // subscription, ready-race, and disposal. const subscribeRef = useRef< From fd3c87ddf0b95912102e383eaba93f0be0fba3df Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Thu, 9 Jul 2026 14:35:52 +0200 Subject: [PATCH 04/17] fix(react-db): don't dispose the observer in an unmount effect (StrictMode) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unmount-effect dispose could run during StrictMode/offscreen effect replay (mount → cleanup → mount) without a re-render, leaving observerRef pointing at a disposed observer; the next subscribe hit attach()'s disposed guard and the store stopped resubscribing. Remove the explicit dispose — useSyncExternalStore already detaches the observer on unsubscribe/unmount, so the collection subscription is torn down and the observer is GC'd. Adds a StrictMode regression test (verified red before the fix). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/react-db/src/useLiveQuery.ts | 21 +++------ .../tests/useLiveQuery.strictmode.test.tsx | 44 +++++++++++++++++++ 2 files changed, 51 insertions(+), 14 deletions(-) create mode 100644 packages/react-db/tests/useLiveQuery.strictmode.test.tsx diff --git a/packages/react-db/src/useLiveQuery.ts b/packages/react-db/src/useLiveQuery.ts index 0baa266f0..058d9eb91 100644 --- a/packages/react-db/src/useLiveQuery.ts +++ b/packages/react-db/src/useLiveQuery.ts @@ -1,4 +1,4 @@ -import { useEffect, useRef, useSyncExternalStore } from 'react' +import { useRef, useSyncExternalStore } from 'react' import { BaseQueryBuilder, createLiveQueryCollection, @@ -411,11 +411,12 @@ export function useLiveQuery( } } - // Recreate the observer when the underlying collection changes. Do not - // dispose the previous observer here — teardown during render is unsafe under - // concurrent rendering. `useSyncExternalStore` unsubscribes the old observer - // when `subscribeRef` changes (below), which detaches it; the final observer - // is disposed in the unmount effect. + // Recreate the observer when the underlying collection changes. The observer + // is not disposed explicitly here or on unmount: `useSyncExternalStore` + // unsubscribes it when the subscribe changes or the component unmounts, which + // detaches the collection subscription; the observer is then GC'd. (An unmount + // effect that disposed it would misfire under StrictMode/offscreen effect + // replay, leaving a disposed observer in the ref.) if (needsNewCollection) { // Defer the initial notify: useSyncExternalStore must not be notified // synchronously during subscribe. @@ -425,14 +426,6 @@ export function useLiveQuery( } const observer = observerRef.current! - // Dispose the current observer on unmount (commit-phase cleanup). - useEffect( - () => () => { - observerRef.current?.dispose() - }, - [], - ) - // Stable subscribe bound to the current observer; the observer owns the // subscription, ready-race, and disposal. const subscribeRef = useRef< diff --git a/packages/react-db/tests/useLiveQuery.strictmode.test.tsx b/packages/react-db/tests/useLiveQuery.strictmode.test.tsx new file mode 100644 index 000000000..874d02062 --- /dev/null +++ b/packages/react-db/tests/useLiveQuery.strictmode.test.tsx @@ -0,0 +1,44 @@ +import { describe, expect, it } from 'vitest' +import { StrictMode } from 'react' +import { act, renderHook, waitFor } from '@testing-library/react' +import { createCollection } from '@tanstack/db' +import { useLiveQuery } from '../src/useLiveQuery' +import { mockSyncCollectionOptions } from '../../db/tests/utils' + +type Person = { id: string; name: string } + +describe(`useLiveQuery under StrictMode`, () => { + it(`keeps the subscription alive across StrictMode effect replay`, async () => { + const collection = createCollection( + mockSyncCollectionOptions({ + id: `strictmode-persons`, + getKey: (p) => p.id, + initialData: [{ id: `1`, name: `A` }], + }), + ) + + // StrictMode double-invokes effects (mount → cleanup → mount). A dispose in + // the unmount effect would tear the observer down and never recreate it, + // leaving a dead subscription. + const { result } = renderHook( + () => + useLiveQuery((q) => + q + .from({ p: collection }) + .select(({ p }) => ({ id: p.id, name: p.name })), + ), + { wrapper: StrictMode }, + ) + + await waitFor(() => expect(result.current.data).toHaveLength(1)) + + // A mutation after the StrictMode replay must still reach the hook. + act(() => { + collection.utils.begin() + collection.utils.write({ type: `insert`, value: { id: `2`, name: `B` } }) + collection.utils.commit() + }) + + await waitFor(() => expect(result.current.data).toHaveLength(2)) + }) +}) From 7577d405aba086c02a9b8453fc3d4c9cd2ea93bc Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 13 Jul 2026 11:39:06 +0200 Subject: [PATCH 05/17] test(db): cover live-query keyed-state invariant on recompile Expose the keyed `state` map in the shared conformance harness (added to ConformanceResult and read by all five adapter drivers) and add a steady-state `recompile-drops-stale-keys` scenario asserting the map stays in sync with `data` across a narrowing recompile. Also add a solid-db regression (in useLiveQuery.test.tsx) that inspects `state` synchronously in the window after a recompile, where solid-db leaks the previous collection's keys until its async resource reconciles. This test fails until the follow-up fix (state.clear() before re-subscribing). Co-Authored-By: Claude Opus 4.8 --- packages/angular-db/tests/conformance.test.ts | 1 + packages/db/tests/conformance/contract.ts | 6 +++ packages/db/tests/conformance/suite.ts | 29 +++++++++++++ packages/react-db/tests/conformance.test.tsx | 1 + packages/solid-db/tests/conformance.test.tsx | 1 + packages/solid-db/tests/useLiveQuery.test.tsx | 43 +++++++++++++++++++ .../tests/conformance.svelte.test.ts | 1 + packages/vue-db/tests/conformance.test.ts | 1 + 8 files changed, 83 insertions(+) diff --git a/packages/angular-db/tests/conformance.test.ts b/packages/angular-db/tests/conformance.test.ts index a8cc7ec80..6a5b5ba66 100644 --- a/packages/angular-db/tests/conformance.test.ts +++ b/packages/angular-db/tests/conformance.test.ts @@ -133,6 +133,7 @@ function makeHandle(result: any, destroy: () => void): LiveQueryHandle { current(): ConformanceResult { return { data: result.data(), + state: result.state(), status: result.status(), isReady: Boolean(result.isReady()), isError: Boolean(result.isError()), diff --git a/packages/db/tests/conformance/contract.ts b/packages/db/tests/conformance/contract.ts index 7726f558b..bf86b1055 100644 --- a/packages/db/tests/conformance/contract.ts +++ b/packages/db/tests/conformance/contract.ts @@ -74,6 +74,12 @@ export interface DbOps { export interface ConformanceResult { /** Array for list queries; a single row (or undefined) for `findOne`. */ data: any + /** + * The keyed result map (`undefined` when disabled). Exposed so scenarios can + * assert the granular map stays in sync with `data` — e.g. that stale keys + * from a previous collection don't linger after a recompile. + */ + state: ReadonlyMap | undefined status: string isReady: boolean isError: boolean diff --git a/packages/db/tests/conformance/suite.ts b/packages/db/tests/conformance/suite.ts index eb52b1826..ba4c53360 100644 --- a/packages/db/tests/conformance/suite.ts +++ b/packages/db/tests/conformance/suite.ts @@ -493,6 +493,35 @@ export function runSuite(rawDriver: LiveQueryDriver) { }, ) + scenario( + `recompile-drops-stale-keys`, + `recompiling to a narrower result drops keys from the previous collection`, + async () => { + const source = driver.makeSource(SEED) + const h = driver.mountControllable( + (q, minAge) => + q + .from({ items: source.collection }) + .where(({ items }: any) => ops.gt(items.age, minAge)) + .select(({ items }: any) => ({ id: items.id })), + 10, + ) + await h.flush() + expect(h.current().data).toHaveLength(3) // all ages > 10 + // The keyed `state` map must mirror `data` exactly. + expect(h.current().state?.size).toBe(3) + + // Narrowing the filter recompiles into a *new* underlying collection + // holding fewer keys. `includeInitialState` only inserts the new rows; + // if the adapter reuses a persistent keyed map without clearing it, the + // dropped keys leak into `state` even though `data` looks correct. + await h.setParam(32) // only John Smith (age 35) survives + expect(h.current().data).toHaveLength(1) + expect(h.current().state?.size).toBe(1) + h.unmount() + }, + ) + scenario( `disabled-transition`, `disabled -> enabled -> disabled toggles correctly`, diff --git a/packages/react-db/tests/conformance.test.tsx b/packages/react-db/tests/conformance.test.tsx index 303bbd89e..3916d67b7 100644 --- a/packages/react-db/tests/conformance.test.tsx +++ b/packages/react-db/tests/conformance.test.tsx @@ -169,6 +169,7 @@ function makeHandle(hook: RenderHookResult) { const r: any = hook.result.current return { data: r?.data, + state: r?.state, status: r?.status ?? `idle`, isReady: Boolean(r?.isReady), isError: Boolean(r?.isError), diff --git a/packages/solid-db/tests/conformance.test.tsx b/packages/solid-db/tests/conformance.test.tsx index 023addafd..ae7bf2138 100644 --- a/packages/solid-db/tests/conformance.test.tsx +++ b/packages/solid-db/tests/conformance.test.tsx @@ -130,6 +130,7 @@ function makeHandle( const result = getResult() return { data: result?.data, + state: result?.state, status: result?.status ?? `idle`, isReady: Boolean(result?.isReady), isError: Boolean(result?.isError), diff --git a/packages/solid-db/tests/useLiveQuery.test.tsx b/packages/solid-db/tests/useLiveQuery.test.tsx index 378c2a0fc..b7567a635 100644 --- a/packages/solid-db/tests/useLiveQuery.test.tsx +++ b/packages/solid-db/tests/useLiveQuery.test.tsx @@ -523,6 +523,49 @@ describe(`Query Collections`, () => { }) }) + it(`should drop stale keys from state synchronously when parameters narrow`, async () => { + // Narrowing recompiles into a *new* collection with fewer keys. The + // observer re-seeds via `includeInitialState`, which only inserts current + // rows and never deletes the previous collection's keys. `state` must be + // cleared synchronously so the dropped keys don't linger in the window + // before the async resource reconciles (this reads `state` with no settle; + // `data`, rebuilt wholesale, stays correct either way). + return createRoot(async (dispose) => { + const collection = createCollection( + mockSyncCollectionOptions({ + id: `stale-keys-on-narrow-test`, + getKey: (person: Person) => person.id, + initialData: initialPersons, + }), + ) + + const [minAge, setMinAge] = createSignal(10) + const rendered = renderHook( + (props: { minAge: Accessor }) => { + return useLiveQuery((q) => + q + .from({ collection }) + .where(({ collection: c }) => gt(c.age, props.minAge())) + .select(({ collection: c }) => ({ id: c.id })), + ) + }, + { initialProps: [{ minAge }] }, + ) + + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(rendered.result.state.size).toBe(3) // all three ages > 10 + + // Narrow to only John Smith (age 35); ids 1 and 2 must not linger. + setMinAge(32) + + expect(rendered.result.state.size).toBe(1) + expect(rendered.result.state.has(`1`)).toBe(false) + expect(rendered.result.state.has(`2`)).toBe(false) + + dispose() + }) + }) + it(`should be able to query a result collection with live updates`, async () => { const collection = createCollection( mockSyncCollectionOptions({ diff --git a/packages/svelte-db/tests/conformance.svelte.test.ts b/packages/svelte-db/tests/conformance.svelte.test.ts index a57709b2b..a2f4a66d3 100644 --- a/packages/svelte-db/tests/conformance.svelte.test.ts +++ b/packages/svelte-db/tests/conformance.svelte.test.ts @@ -128,6 +128,7 @@ function makeHandle(getQuery: () => any, dispose: () => void): LiveQueryHandle { const query = getQuery() return { data: query?.data, + state: query?.state, status: query?.status ?? `idle`, isReady: Boolean(query?.isReady), isError: Boolean(query?.isError), diff --git a/packages/vue-db/tests/conformance.test.ts b/packages/vue-db/tests/conformance.test.ts index 432876f4f..eef6e0a2f 100644 --- a/packages/vue-db/tests/conformance.test.ts +++ b/packages/vue-db/tests/conformance.test.ts @@ -126,6 +126,7 @@ function makeHandle(result: any, scope: ReturnType) { current(): ConformanceResult { return { data: result.data?.value, + state: result.state?.value, status: result.status?.value ?? `idle`, isReady: Boolean(result.isReady?.value), isError: Boolean(result.isError?.value), From 2a0679e04577962b1e5d0e59e539467b6c32145b Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 13 Jul 2026 11:39:22 +0200 Subject: [PATCH 06/17] fix(solid-db): clear keyed state before subscribing to a new collection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the query recompiles to a different collection, the observer re-seeds via `includeInitialState`, which only inserts current rows and never deletes keys from the previous collection. Without clearing first, the dropped keys lingered in `state` until the async resource reconciled — a transient window where `state` exposed stale rows (though `data`, rebuilt wholesale, stayed correct). Clear synchronously before re-subscribing, matching vue-db and svelte-db. Fixes the solid-db stale-keys regression added in the previous commit. Co-Authored-By: Claude Opus 4.8 --- packages/solid-db/src/useLiveQuery.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/solid-db/src/useLiveQuery.ts b/packages/solid-db/src/useLiveQuery.ts index a2a39a7fa..a50b8269c 100644 --- a/packages/solid-db/src/useLiveQuery.ts +++ b/packages/solid-db/src/useLiveQuery.ts @@ -394,6 +394,11 @@ export function useLiveQuery( // The shared observer owns subscription, the ready-race, and status; Solid // materializes into its keyed ReactiveMap (granular) + reconciled store. const observer = createLiveQueryObserver(currentCollection) + // Clear any keys carried over from a previous collection before the new + // observer re-seeds via `includeInitialState` (which only inserts current + // rows, never deletes stale ones). Without this, switching collections + // leaves the dropped keys in `state` until the async resource reconciles. + state.clear() const unsubscribe = observer.subscribe( (changes: Array> | undefined) => { batch(() => { From 6db6b752563c9ba1933114fcfd57225a6ed872b7 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 11:39:27 +0200 Subject: [PATCH 07/17] fix(db): FIFO non-reentrant observer dispatch over subscription records MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A listener that synchronously mutates the collection used to trigger a nested, reentrant dispatch: later subscribers could observe the nested event (e.g. a delete) before the outer one (the insert) it reacted to. Publications are now queued and dispatched FIFO. Each publication is delivered over a snapshot of subscription records taken when it is dispatched: a subscription removed mid-delivery still receives the in-flight publication, one added mid-delivery does not. Records — not raw callbacks — identify subscriptions, so subscribing the same function twice no longer collapses into one Set entry whose first unsubscribe tore down both. Co-Authored-By: Claude Fable 5 --- packages/db/src/live-query-observer.ts | 64 ++++++++--- packages/db/tests/live-query-observer.test.ts | 104 ++++++++++++++++++ 2 files changed, 155 insertions(+), 13 deletions(-) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 8387b3838..13076e66d 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -66,6 +66,16 @@ export interface LiveQueryObserver< dispose: () => void } +/** + * One logical subscription. Records — not raw callbacks — identify + * subscriptions, so the same listener function can be subscribed twice and + * each subscription tears down independently. + */ +interface SubscriptionRecord { + listener: LiveQueryObserverListener + active: boolean +} + const DISABLED_SNAPSHOT: LiveQuerySnapshot = { state: undefined, data: undefined, @@ -89,7 +99,14 @@ class LiveQueryObserverImpl< private cachedVersion = -1 private cachedStatus: CollectionStatus | undefined private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT - private readonly listeners = new Set>() + private readonly subscriptions = new Set>() + // Publications are dispatched FIFO: an emit that happens while another + // publication is being delivered (a listener mutating the collection + // synchronously) is queued, never delivered reentrantly. + private readonly publicationQueue: Array< + Array> | undefined + > = [] + private dispatching = false private collectionUnsub: (() => void) | null = null // Bumped on each attach. `onFirstReady` can't be unsubscribed, so a callback // from a superseded attach checks this to no-op instead of double-notifying. @@ -143,15 +160,15 @@ class LiveQueryObserverImpl< } subscribe(listener: LiveQueryObserverListener): () => void { - this.listeners.add(listener) - if (this.listeners.size === 1) this.attach() + const record: SubscriptionRecord = { listener, active: true } + this.subscriptions.add(record) + if (this.subscriptions.size === 1) this.attach() - let active = true return () => { - if (!active) return - active = false - this.listeners.delete(listener) - if (this.listeners.size === 0) this.detach() + if (!record.active) return + record.active = false + this.subscriptions.delete(record) + if (this.subscriptions.size === 0) this.detach() } } @@ -176,7 +193,7 @@ class LiveQueryObserverImpl< let attaching = this.deferInitialNotify const deferred: Array> | undefined> = [] const notify = (changes: Array> | undefined) => { - if (this.disposed || this.listeners.size === 0) return + if (this.disposed || this.subscriptions.size === 0) return if (attaching) deferred.push(changes) else this.emit(changes) } @@ -211,7 +228,7 @@ class LiveQueryObserverImpl< // initial batch would reach the current listener. if ( this.disposed || - this.listeners.size === 0 || + this.subscriptions.size === 0 || generation !== this.attachGeneration ) { return @@ -227,8 +244,27 @@ class LiveQueryObserverImpl< } private emit(changes: Array> | undefined): void { - this.version++ - this.listeners.forEach((listener) => listener(changes)) + this.publicationQueue.push(changes) + if (this.dispatching) return + + this.dispatching = true + try { + // A dispose() during dispatch empties the queue, ending this loop. + while (this.publicationQueue.length > 0) { + const publication = this.publicationQueue.shift()! + this.version++ + // Deliver over a snapshot of the records taken when this publication + // is dispatched: a subscription removed mid-delivery still receives + // the in-flight publication; one added mid-delivery does not. + const records = Array.from(this.subscriptions) + for (const subRecord of records) { + if (this.disposed) return + subRecord.listener(publication) + } + } + } finally { + this.dispatching = false + } } async preload(): Promise { @@ -239,7 +275,9 @@ class LiveQueryObserverImpl< if (this.disposed) return this.disposed = true this.detach() - this.listeners.clear() + for (const subRecord of this.subscriptions) subRecord.active = false + this.subscriptions.clear() + this.publicationQueue.length = 0 } } diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 1035a11ef..0e2e9d835 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -174,6 +174,110 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) + it(`dispatches nested publications FIFO, never reentrantly`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + // Listener A reacts to the insert of row 3 by synchronously deleting it — + // a nested publication while the insert is still being delivered. + observer.subscribe((changes) => { + if (changes?.some((c) => c.type === `insert` && c.key === `3`)) { + source.utils.begin() + source.utils.write({ type: `delete`, value: { id: `3`, name: `C` } }) + source.utils.commit() + } + }) + + const listenerBEvents: Array = [] + observer.subscribe((changes) => { + for (const c of changes ?? []) { + if (c.key === `3`) listenerBEvents.push(c.type) + } + }) + + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `3`, name: `C` } }) + source.utils.commit() + + // B must observe the insert before the (nested) delete. + expect(listenerBEvents).toEqual([`insert`, `delete`]) + observer.dispose() + }) + + it(`does not deliver an in-flight publication to a listener added during dispatch`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + let lateListenerCalls = 0 + observer.subscribe((changes) => { + // Add the late listener only while the row-4 delta is being dispatched. + if (changes?.some((c) => c.key === `4`)) { + observer.subscribe(() => { + lateListenerCalls++ + }) + } + }) + + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `4`, name: `D` } }) + source.utils.commit() + + expect(lateListenerCalls).toBe(0) + observer.dispose() + }) + + it(`still delivers the in-flight publication to a listener removed during dispatch`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + let existingListenerCalls = 0 + let unsubB: (() => void) | null = null + observer.subscribe(() => { + unsubB?.() + unsubB = null + }) + unsubB = observer.subscribe(() => { + existingListenerCalls++ + }) + + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `5`, name: `E` } }) + source.utils.commit() + + // A removed B while the publication was in flight; B still receives it. + expect(existingListenerCalls).toBe(1) + observer.dispose() + }) + + it(`treats two subscriptions with the same callback as independent`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + let calls = 0 + const shared = () => { + calls++ + } + const unsubFirst = observer.subscribe(shared) + const unsubSecond = observer.subscribe(shared) + + unsubFirst() + calls = 0 + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `6`, name: `F` } }) + source.utils.commit() + + // The second subscription survives the first one's teardown. + expect(calls).toBe(1) + unsubSecond() + + calls = 0 + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `7`, name: `G` } }) + source.utils.commit() + expect(calls).toBe(0) + observer.dispose() + }) + it(`refreshes the snapshot when status changes without a version bump`, () => { // A status-only loading→ready transition with no active subscription: the // cached snapshot must not stay stale (covers the preload() case too). From 7a86eae43a4b829c0133affd6a2cac6fdca1786d Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 11:41:52 +0200 Subject: [PATCH 08/17] fix(db): release the collection subscription on dispose during initial replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit subscribeChanges delivers the initial state synchronously, so a listener could dispose the observer before the subscription handle was stored — detach() then had nothing to release and the collection subscription leaked past disposal. The release hook is now registered before the subscription is created, making attachment transactional: if detach() fired mid-replay, the subscription is undone as soon as subscribeChanges returns. Co-Authored-By: Claude Fable 5 --- packages/db/src/live-query-observer.ts | 15 +++++++++++++-- packages/db/tests/live-query-observer.test.ts | 11 +++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 13076e66d..ec789a0ea 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -198,11 +198,22 @@ class LiveQueryObserverImpl< else this.emit(changes) } - const subscription = collection.subscribeChanges( + // `subscribeChanges` delivers the initial state synchronously, so a + // listener can dispose the observer while the collection subscription is + // still being created. Register the release hook up front; if detach() + // ran during that replay (collectionUnsub no longer points at our hook), + // undo the subscription as soon as the call returns. + let subscription: { unsubscribe: () => void } | null = null + const release = () => subscription?.unsubscribe() + this.collectionUnsub = release + subscription = collection.subscribeChanges( (changes) => notify(changes as Array>), { includeInitialState: true }, ) - this.collectionUnsub = () => subscription.unsubscribe() + if (this.collectionUnsub !== release) { + subscription.unsubscribe() + return + } // Catch a *later* loading→ready transition that carries no change events // (e.g. `markReady()` with no rows). Skip when already ready — the initial diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 0e2e9d835..55cacd9fb 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -278,6 +278,17 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) + it(`releases the collection subscription when a listener disposes during initial replay`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + // The initial-state replay is delivered synchronously inside subscribe(); + // disposing from the listener must not leak the collection subscription. + observer.subscribe(() => observer.dispose()) + + expect(source.subscriberCount).toBe(0) + }) + it(`refreshes the snapshot when status changes without a version bump`, () => { // A status-only loading→ready transition with no active subscription: the // cached snapshot must not stay stale (covers the preload() case too). From 482c82b9603cf5323580dc1865611dfb4c2bac6a Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 11:43:38 +0200 Subject: [PATCH 09/17] fix(db): seed late observer subscribers; reject subscribe after dispose MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The initial-state replay only happened on the first attach, so a second concurrent subscriber started with no rows and could never converge — its keyed map silently stayed empty. A subscriber arriving while the observer is already attached is now seeded with the collection's current rows as inserts, delivered to that subscription alone without advancing the observer revision. subscribe() after dispose() used to register a listener that could never fire; it now throws LiveQueryObserverDisposedError. Co-Authored-By: Claude Fable 5 --- packages/db/src/errors.ts | 6 +++ packages/db/src/live-query-observer.ts | 33 ++++++++++++- packages/db/tests/live-query-observer.test.ts | 47 +++++++++++++++---- 3 files changed, 77 insertions(+), 9 deletions(-) diff --git a/packages/db/src/errors.ts b/packages/db/src/errors.ts index 0bfd2f996..710025b65 100644 --- a/packages/db/src/errors.ts +++ b/packages/db/src/errors.ts @@ -135,6 +135,12 @@ export class NegativeActiveSubscribersError extends CollectionStateError { } } +export class LiveQueryObserverDisposedError extends CollectionStateError { + constructor() { + super(`Cannot subscribe to a disposed LiveQueryObserver`) + } +} + // Collection Operation Errors export class CollectionOperationError extends TanStackDBError { constructor(message: string) { diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index ec789a0ea..690817515 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -1,3 +1,4 @@ +import { LiveQueryObserverDisposedError } from './errors.js' import { getLiveQueryStatusFlags, isSingleResultCollection, @@ -160,9 +161,19 @@ class LiveQueryObserverImpl< } subscribe(listener: LiveQueryObserverListener): () => void { + if (this.disposed) throw new LiveQueryObserverDisposedError() + const record: SubscriptionRecord = { listener, active: true } this.subscriptions.add(record) - if (this.subscriptions.size === 1) this.attach() + if (this.subscriptions.size === 1) { + this.attach() + } else { + // The initial-state replay only happens on attach, so a subscriber that + // arrives while already attached is seeded with the current rows — + // delivered to this subscription alone, without advancing the observer's + // revision (the collection state did not change). + this.seed(record) + } return () => { if (!record.active) return @@ -172,6 +183,26 @@ class LiveQueryObserverImpl< } } + /** Deliver the collection's current rows to one late subscription as inserts. */ + private seed(record: SubscriptionRecord): void { + const collection = this.collection + if (!collection) return + + const seedChanges: Array> = [] + for (const [key, value] of collection.entries() as IterableIterator< + [TKey, T] + >) { + seedChanges.push({ type: `insert`, key, value }) + } + if (seedChanges.length === 0) return + + const deliver = () => { + if (record.active) record.listener(seedChanges) + } + if (this.deferInitialNotify) queueMicrotask(deliver) + else deliver() + } + private attach(): void { const collection = this.collection if (!collection || this.disposed) return diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 55cacd9fb..f2839121e 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -208,12 +208,14 @@ describe(`createLiveQueryObserver`, () => { const source = makeSource() const observer = createLiveQueryObserver(source as any) - let lateListenerCalls = 0 + let lateListenerRow4Deliveries = 0 observer.subscribe((changes) => { // Add the late listener only while the row-4 delta is being dispatched. if (changes?.some((c) => c.key === `4`)) { - observer.subscribe(() => { - lateListenerCalls++ + observer.subscribe((lateChanges) => { + if (lateChanges?.some((c) => c.key === `4`)) { + lateListenerRow4Deliveries++ + } }) } }) @@ -222,7 +224,9 @@ describe(`createLiveQueryObserver`, () => { source.utils.write({ type: `insert`, value: { id: `4`, name: `D` } }) source.utils.commit() - expect(lateListenerCalls).toBe(0) + // The late subscriber receives row 4 exactly once — via its seed of the + // already-committed state, NOT additionally via the in-flight publication. + expect(lateListenerRow4Deliveries).toBe(1) observer.dispose() }) @@ -230,14 +234,14 @@ describe(`createLiveQueryObserver`, () => { const source = makeSource() const observer = createLiveQueryObserver(source as any) - let existingListenerCalls = 0 + let row5Deliveries = 0 let unsubB: (() => void) | null = null observer.subscribe(() => { unsubB?.() unsubB = null }) - unsubB = observer.subscribe(() => { - existingListenerCalls++ + unsubB = observer.subscribe((changes) => { + if (changes?.some((c) => c.key === `5`)) row5Deliveries++ }) source.utils.begin() @@ -245,7 +249,7 @@ describe(`createLiveQueryObserver`, () => { source.utils.commit() // A removed B while the publication was in flight; B still receives it. - expect(existingListenerCalls).toBe(1) + expect(row5Deliveries).toBe(1) observer.dispose() }) @@ -289,6 +293,33 @@ describe(`createLiveQueryObserver`, () => { expect(source.subscriberCount).toBe(0) }) + it(`seeds a second concurrent subscriber with the current rows`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + observer.subscribe(() => {}) + + // The attach (and its initial-state replay) already happened; a late + // subscriber must still receive the current rows as inserts. + const secondSubscriberKeys: Array = [] + observer.subscribe((changes) => { + for (const c of changes ?? []) { + if (c.type === `insert`) secondSubscriberKeys.push(c.key) + } + }) + + expect(secondSubscriberKeys.sort()).toEqual([`1`, `2`]) + observer.dispose() + }) + + it(`throws when subscribing after dispose`, () => { + const observer = createLiveQueryObserver(makeSource() as any) + observer.dispose() + expect(() => observer.subscribe(() => {})).toThrow( + /disposed LiveQueryObserver/, + ) + }) + it(`refreshes the snapshot when status changes without a version bump`, () => { // A status-only loading→ready transition with no active subscription: the // cached snapshot must not stay stale (covers the preload() case too). From d0875e43c71636788200dbf7929edcf01994ed98 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 11:47:31 +0200 Subject: [PATCH 10/17] fix(db): drive observer snapshots from a collection-owned state revision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The observer counted every delivery — including per-attach bootstrap replays and empty ready flushes — as a semantic revision. One readiness transition published three times ([], undefined, []), a plain unsubscribe/resubscribe manufactured a new snapshot identity with unchanged data, and rows committed while nothing was attached left the cached snapshot stale. The semantic clock now lives on the collection: emitEvents advances a monotonic stateRevision once per committed batch, whether or not anyone is subscribed. getSnapshot keys its cache on (stateRevision, status), so detached snapshots stay fresh and attachment replay can no longer advance the clock. Empty change batches are dropped from publication — only real deltas and the synthetic ready notify go out — so a readiness transition publishes exactly once. Co-Authored-By: Claude Fable 5 --- packages/db/src/collection/changes.ts | 13 +++++ packages/db/src/collection/index.ts | 9 ++++ packages/db/src/live-query-observer.ts | 20 ++++---- packages/db/tests/live-query-observer.test.ts | 48 +++++++++++++++++++ 4 files changed, 82 insertions(+), 8 deletions(-) diff --git a/packages/db/src/collection/changes.ts b/packages/db/src/collection/changes.ts index dc07cd3f1..e5bdfb845 100644 --- a/packages/db/src/collection/changes.ts +++ b/packages/db/src/collection/changes.ts @@ -30,6 +30,15 @@ export class CollectionChangesManager< public batchedEvents: Array> = [] public shouldBatchEvents = false + /** + * Monotonic revision of the collection's visible state, advanced once per + * committed batch of changes — including while nothing is subscribed. + * Lets consumers (the live-query observer) cheaply detect "did the data + * change" without subscribing, and stays untouched by subscription + * bootstrap replays, which do not go through emitEvents. + */ + public stateRevision = 0 + /** * Creates a new CollectionChangesManager instance */ @@ -77,6 +86,10 @@ export class CollectionChangesManager< changes: Array>, forceEmit = false, ): void { + // The visible state was already committed by the caller, so the revision + // advances even when the events below end up batched for later emission. + if (changes.length > 0) this.stateRevision++ + // Skip batching for user actions (forceEmit=true) to keep UI responsive if (this.shouldBatchEvents && !forceEmit) { // Add events to the batch diff --git a/packages/db/src/collection/index.ts b/packages/db/src/collection/index.ts index 137fd5f59..13887a43d 100644 --- a/packages/db/src/collection/index.ts +++ b/packages/db/src/collection/index.ts @@ -420,6 +420,15 @@ export class CollectionImpl< return this._changes.activeSubscribersCount } + /** + * Monotonic revision of the collection's visible state; advances once per + * committed batch of changes, even while nothing is subscribed. + * Internal — used by the live-query observer's snapshot cache. + */ + public get _stateRevision(): number { + return this._changes.stateRevision + } + /** * Register a callback to be executed when the collection first becomes ready * Useful for preloading collections diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 690817515..391f14a9b 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -96,8 +96,7 @@ class LiveQueryObserverImpl< > implements LiveQueryObserver { private readonly collection: Collection | null private readonly deferInitialNotify: boolean - private version = 0 - private cachedVersion = -1 + private cachedRevision = -1 private cachedStatus: CollectionStatus | undefined private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT private readonly subscriptions = new Set>() @@ -128,14 +127,16 @@ class LiveQueryObserverImpl< const collection = this.collection if (!collection) return DISABLED_SNAPSHOT - // Rebuild when the version advanced, or when the collection's status - // changed without a version bump (e.g. a status-only loading→ready - // transition or `preload()` while there is no active subscription). + // The semantic clock: rebuild only when the collection's own state + // revision or status moved. The revision advances on every committed + // change — even while nothing is subscribed, so a detached snapshot never + // goes stale — and is untouched by subscription bootstrap replays, so + // resubscribing never manufactures a new snapshot identity. if ( - this.cachedVersion !== this.version || + this.cachedRevision !== collection._stateRevision || this.cachedStatus !== collection.status ) { - this.cachedVersion = this.version + this.cachedRevision = collection._stateRevision this.cachedStatus = collection.status const entries = Array.from(collection.entries()) as Array<[TKey, T]> const singleResult = isSingleResultCollection(collection) @@ -225,6 +226,10 @@ class LiveQueryObserverImpl< const deferred: Array> | undefined> = [] const notify = (changes: Array> | undefined) => { if (this.disposed || this.subscriptions.size === 0) return + // An empty batch carries no semantic change (e.g. the collection's + // empty-ready flush); only real deltas and the synthetic ready notify + // (undefined) are published. + if (changes !== undefined && changes.length === 0) return if (attaching) deferred.push(changes) else this.emit(changes) } @@ -294,7 +299,6 @@ class LiveQueryObserverImpl< // A dispose() during dispatch empties the queue, ending this loop. while (this.publicationQueue.length > 0) { const publication = this.publicationQueue.shift()! - this.version++ // Deliver over a snapshot of the records taken when this publication // is dispatched: a subscription removed mid-delivery still receives // the in-flight publication; one added mid-delivery does not. diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index f2839121e..2c8f14f30 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -320,6 +320,54 @@ describe(`createLiveQueryObserver`, () => { ) }) + it(`preserves snapshot identity across subscribe/unsubscribe cycles`, () => { + const observer = createLiveQueryObserver(makeSource() as any) + + const before = observer.getSnapshot() + observer.subscribe(() => {})() + observer.subscribe(() => {})() + + // Bootstrap replay is per-subscriber delivery, not a semantic revision: + // nothing observable changed, so the snapshot identity must not change. + expect(observer.getSnapshot()).toBe(before) + observer.dispose() + }) + + it(`emits exactly one post-bootstrap notification for a readiness transition`, () => { + const collection = makeLoadingSource() + const observer = createLiveQueryObserver(collection as any) + + const events: Array = [] + observer.subscribe((changes) => events.push(changes)) + + collection.utils.markReady() + + // Not the old [[], undefined, []]: empty batches carry no semantic change, + // so one readiness transition publishes exactly once. + expect(events).toEqual([undefined]) + observer.dispose() + }) + + it(`serves a fresh snapshot for rows changed while detached`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + const unsubscribe = observer.subscribe(() => {}) + const before = observer.getSnapshot() + unsubscribe() + + // Mutate while nothing is attached; the status does not change. + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `8`, name: `H` } }) + source.utils.commit() + + const after = observer.getSnapshot() + expect(after).not.toBe(before) + expect(after.state?.has(`8`)).toBe(true) + expect(after.data).toHaveLength(3) + observer.dispose() + }) + it(`refreshes the snapshot when status changes without a version bump`, () => { // A status-only loading→ready transition with no active subscription: the // cached snapshot must not stay stale (covers the preload() case too). From 37b5eea99d0590730a8333e9dc70d0e3db24827d Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 11:55:26 +0200 Subject: [PATCH 11/17] test(angular-db): align the mock collection with the real collection contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hand-rolled mock notified subscribers with empty change batches as a wake-up signal — something real collections never do — and lacked the state revision and status event channel the observer relies on. It now advances _stateRevision on committed changes, emits real delete/insert deltas from __replaceAll, and publishes status transitions through on('status:change') instead of an empty notify. Co-Authored-By: Claude Fable 5 --- .../tests/inject-live-query.test.ts | 32 +++++++++++++++++-- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/packages/angular-db/tests/inject-live-query.test.ts b/packages/angular-db/tests/inject-live-query.test.ts index 81fbb12a9..71aa198a5 100644 --- a/packages/angular-db/tests/inject-live-query.test.ts +++ b/packages/angular-db/tests/inject-live-query.test.ts @@ -80,14 +80,25 @@ function createMockCollection( } let status: CollectionStatus = initialStatus + let stateRevision = 0 const subs = new Set<(changes: Array) => void>() const readySubs = new Set<() => void>() + const statusSubs = new Set<(event: any) => void>() const id = `mock-col-` + Math.random().toString(36).slice(2) + // Mirrors the real collection contract: committed changes advance the + // state revision before they are emitted. const notify = (changes: Array = []) => { + if (changes.length > 0) stateRevision++ for (const cb of subs) cb(changes) } + const emitStatusChange = (previousStatus: CollectionStatus) => { + for (const cb of statusSubs) { + cb({ type: `status:change`, previousStatus, status }) + } + } + const notifyReady = () => { for (const cb of readySubs) cb() } @@ -97,6 +108,13 @@ function createMockCollection( get status() { return status }, + get _stateRevision() { + return stateRevision + }, + on: (event: string, cb: (e: any) => void) => { + if (event === `status:change`) statusSubs.add(cb) + return () => statusSubs.delete(cb) + }, entries: () => Array.from(map.entries()), values: () => Array.from(map.values()), get: (key: K) => map.get(key), @@ -127,17 +145,25 @@ function createMockCollection( } }, __setStatus: (s: CollectionStatus) => { + const previousStatus = status const wasNotReady = status !== `ready` status = s - notify([]) + emitStatusChange(previousStatus) if (wasNotReady && status === `ready`) { setTimeout(notifyReady, 0) } }, __replaceAll: (rows: Array>) => { + const changes: Array = [] + for (const [key, value] of map.entries()) { + changes.push({ type: `delete`, key, value }) + } map.clear() - for (const r of rows) map.set(r.id, r) - notify([]) + for (const r of rows) { + map.set(r.id, r) + changes.push({ type: `insert`, key: r.id, value: r }) + } + notify(changes) }, __upsert: (row: T & Record<`id`, K>) => { const isUpdate = map.has(row.id) From eeafe1778d48f791740407c4081346f7d3fdd936 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 11:55:30 +0200 Subject: [PATCH 12/17] fix(db): publish collection status changes through the canonical path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The observer consumed row changes and onFirstReady but not the collection's status events: a mounted consumer could sit on a stale loading/ready status after an error or cleaned-up transition until an unrelated row event happened to arrive. Status changes now publish a synthetic notify through the same canonical path as data changes. This also retires the onFirstReady registration, whose callbacks could not be unsubscribed and accumulated across attach/detach cycles while loading — collection.on('status:change') returns a real unsubscribe that detach releases. Co-Authored-By: Claude Fable 5 --- packages/db/src/live-query-observer.ts | 34 ++++++++----------- packages/db/tests/live-query-observer.test.ts | 19 +++++++++++ 2 files changed, 33 insertions(+), 20 deletions(-) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 391f14a9b..a734bfac5 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -108,8 +108,8 @@ class LiveQueryObserverImpl< > = [] private dispatching = false private collectionUnsub: (() => void) | null = null - // Bumped on each attach. `onFirstReady` can't be unsubscribed, so a callback - // from a superseded attach checks this to no-op instead of double-notifying. + // Bumped on each attach so a deferred initial-notify microtask queued by a + // superseded attach can detect it and skip flushing. private attachGeneration = 0 private disposed = false @@ -216,7 +216,7 @@ class LiveQueryObserverImpl< // collection's per-subscriber change stream requires this to align deltes). // // When `deferInitialNotify` is set, emits that fire synchronously while - // attaching (the initial-state batch and an immediately-ready `onFirstReady`) + // attaching (the initial-state batch and any synchronous status notify) // are deferred to a microtask, so a wholesale consumer like React's // `useSyncExternalStore` never receives a synchronous notify during // `subscribe`. Effect/watcher-based adapters want the initial state @@ -234,13 +234,23 @@ class LiveQueryObserverImpl< else this.emit(changes) } + // Status transitions that carry no change events (loading→ready with no + // rows, error, cleaned-up) are part of the canonical publication path: + // any status change publishes a synthetic notify so consumers re-read the + // snapshot. Unlike onFirstReady, `on` returns a real unsubscribe, so a + // detached attachment leaves nothing behind. + const statusUnsub = collection.on(`status:change`, () => notify(undefined)) + // `subscribeChanges` delivers the initial state synchronously, so a // listener can dispose the observer while the collection subscription is // still being created. Register the release hook up front; if detach() // ran during that replay (collectionUnsub no longer points at our hook), // undo the subscription as soon as the call returns. let subscription: { unsubscribe: () => void } | null = null - const release = () => subscription?.unsubscribe() + const release = () => { + statusUnsub() + subscription?.unsubscribe() + } this.collectionUnsub = release subscription = collection.subscribeChanges( (changes) => notify(changes as Array>), @@ -251,22 +261,6 @@ class LiveQueryObserverImpl< return } - // Catch a *later* loading→ready transition that carries no change events - // (e.g. `markReady()` with no rows). Skip when already ready — the initial - // state batch above already covers that, and `onFirstReady` would fire an - // immediate duplicate. - // - // `onFirstReady` returns no unsubscribe, so a callback left behind by an - // earlier attach (subscribe → unsubscribe-before-ready → subscribe) would - // still fire on `markReady`. Guard with the attach generation so only the - // current attachment's callback notifies. - if (collection.status !== `ready`) { - collection.onFirstReady(() => { - if (generation !== this.attachGeneration) return - notify(undefined) - }) - } - attaching = false if (deferred.length > 0) { queueMicrotask(() => { diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 2c8f14f30..6ca15b7d5 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -368,6 +368,25 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) + it(`wakes consumers on status-only transitions (error, cleaned-up)`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + const statuses: Array = [] + observer.subscribe(() => { + statuses.push(observer.getSnapshot().status) + }) + + // Status transitions carry no row changes; the observer must publish + // them through the same canonical path as data changes. + source._lifecycle.setStatus(`error`) + source._lifecycle.setStatus(`cleaned-up`) + + expect(statuses).toContain(`error`) + expect(statuses).toContain(`cleaned-up`) + observer.dispose() + }) + it(`refreshes the snapshot when status changes without a version bump`, () => { // A status-only loading→ready transition with no active subscription: the // cached snapshot must not stay stale (covers the preload() case too). From 4932b732a3ab1e988eb03dbc3d3884515a206841 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 12:00:21 +0200 Subject: [PATCH 13/17] fix(db): per-consumer initial-state policy; lazy snapshot materialization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Forcing includeInitialState on every attach was a behavior change for the wholesale adapters: React and Angular never requested an initial snapshot before the observer, and the forced request issued an unfiltered loadSubset({ where: undefined }) against on-demand collections. The observer now takes a mode option: granular (default — Vue/Svelte/Solid) keeps the initial-state subscription and late- subscriber seeding; wholesale (React/Angular) subscribes with includeInitialState: false, restoring the pre-observer loading policy while deletes still flow through as notifies. getSnapshot() now materializes rows lazily on first state/data access, so a consumer that only reads status never enumerates the collection. The React already-ready microtask notify is gone with the bootstrap replay; it existed because the pre-observer per-subscription version could miss a ready transition between render and subscribe, which the collection-owned revision plus useSyncExternalStore's post-subscribe re-read now cover. Co-Authored-By: Claude Fable 5 --- packages/angular-db/src/index.ts | 6 +- packages/db/src/live-query-observer.ts | 53 ++++++++--- packages/db/tests/live-query-observer.test.ts | 93 ++++++++++++++++++- packages/react-db/src/useLiveQuery.ts | 3 + .../useLiveQuery.eager-onstorechange.test.tsx | 18 +++- 5 files changed, 154 insertions(+), 19 deletions(-) diff --git a/packages/angular-db/src/index.ts b/packages/angular-db/src/index.ts index aa9bb35a5..366c25001 100644 --- a/packages/angular-db/src/index.ts +++ b/packages/angular-db/src/index.ts @@ -250,7 +250,11 @@ export function injectLiveQuery(opts: any) { // The shared observer owns sync start, subscription, the ready-race, and // status transitions; Angular re-reads the whole collection on each notify // (wholesale) into its signals. - const observer = createLiveQueryObserver(currentCollection) + // Angular re-reads the collection on notify; wholesale mode preserves its + // pre-observer loading policy (no initial-state snapshot request). + const observer = createLiveQueryObserver(currentCollection, { + mode: `wholesale`, + }) // Seed immediately from the post-start state, then re-read on every notify. syncDataFromCollection(currentCollection) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index a734bfac5..0b6cffee3 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -96,6 +96,7 @@ class LiveQueryObserverImpl< > implements LiveQueryObserver { private readonly collection: Collection | null private readonly deferInitialNotify: boolean + private readonly wholesale: boolean private cachedRevision = -1 private cachedStatus: CollectionStatus | undefined private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT @@ -116,9 +117,11 @@ class LiveQueryObserverImpl< constructor( collection: Collection | null, deferInitialNotify: boolean, + wholesale: boolean, ) { this.collection = collection this.deferInitialNotify = deferInitialNotify + this.wholesale = wholesale // Starting sync during resolution matches every adapter's eager behavior. collection?.startSyncImmediate() } @@ -138,18 +141,21 @@ class LiveQueryObserverImpl< ) { this.cachedRevision = collection._stateRevision this.cachedStatus = collection.status - const entries = Array.from(collection.entries()) as Array<[TKey, T]> const singleResult = isSingleResultCollection(collection) + // Rows are materialized lazily on first `state`/`data` access, so a + // consumer that only reads `status` never enumerates the collection. + let entriesCache: Array<[TKey, T]> | null = null let stateCache: Map | null = null let dataCache: Array | null = null + const readEntries = () => + (entriesCache ??= Array.from(collection.entries()) as Array<[TKey, T]>) this.cachedSnapshot = { get state() { - if (!stateCache) stateCache = new Map(entries) - return stateCache + return (stateCache ??= new Map(readEntries())) }, get data() { - if (!dataCache) dataCache = entries.map(([, value]) => value) + dataCache ??= readEntries().map(([, value]) => value) return singleResult ? dataCache[0] : dataCache }, collection, @@ -169,11 +175,12 @@ class LiveQueryObserverImpl< if (this.subscriptions.size === 1) { this.attach() } else { - // The initial-state replay only happens on attach, so a subscriber that - // arrives while already attached is seeded with the current rows — - // delivered to this subscription alone, without advancing the observer's - // revision (the collection state did not change). - this.seed(record) + // The initial-state replay only happens on attach, so a granular + // subscriber that arrives while already attached is seeded with the + // current rows — delivered to this subscription alone, without advancing + // the observer's revision (the collection state did not change). + // Wholesale consumers read getSnapshot() instead and need no seed. + if (!this.wholesale) this.seed(record) } return () => { @@ -210,10 +217,14 @@ class LiveQueryObserverImpl< const generation = ++this.attachGeneration - // Subscribe with initial state so granular consumers receive the current - // rows as inserts followed by deltas through one consistent channel — the - // same contract the adapters used before the observer existed (the - // collection's per-subscriber change stream requires this to align deltes). + // Granular consumers subscribe with initial state so they receive the + // current rows as inserts followed by deltas through one consistent + // channel (the collection's per-subscriber change stream requires this to + // align deltas). Wholesale consumers subscribe WITHOUT initial state — + // preserving their pre-observer loading policy: no snapshot request means + // no unfiltered loadSubset({ where: undefined }) against on-demand + // collections. The explicit `false` marks all state as seen so deletes + // still flow through as notifies. // // When `deferInitialNotify` is set, emits that fire synchronously while // attaching (the initial-state batch and any synchronous status notify) @@ -254,7 +265,7 @@ class LiveQueryObserverImpl< this.collectionUnsub = release subscription = collection.subscribeChanges( (changes) => notify(changes as Array>), - { includeInitialState: true }, + { includeInitialState: !this.wholesale }, ) if (this.collectionUnsub !== release) { subscription.unsubscribe() @@ -329,6 +340,19 @@ export interface CreateLiveQueryObserverOptions { * Effect/watcher-based adapters leave it off to get initial state synchronously. */ deferInitialNotify?: boolean + /** + * How subscribers consume the observer: + * + * - `granular` (default): subscribers apply the delivered `ChangeMessage[]` + * deltas to their own keyed state (Vue/Svelte/Solid). The observer + * subscribes with initial state and seeds late subscribers, so every + * subscriber converges from deltas alone. + * - `wholesale`: subscribers treat notifications as a wake-up and re-read + * `getSnapshot()` (React/Angular). The observer subscribes WITHOUT initial + * state, preserving those adapters' loading policy — no snapshot request, + * so no unfiltered `loadSubset` against on-demand collections. + */ + mode?: `granular` | `wholesale` } /** @@ -345,5 +369,6 @@ export function createLiveQueryObserver< return new LiveQueryObserverImpl( collection ?? null, options.deferInitialNotify ?? false, + options.mode === `wholesale`, ) } diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 6ca15b7d5..1fb076843 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { createCollection } from '../src/collection/index.js' import { createLiveQueryObserver } from '../src/live-query-observer.js' import { @@ -40,6 +40,42 @@ function makeLoadingSource() { return collection } +/** An on-demand collection whose sync exposes a loadSubset spy. */ +function makeLoadSubsetSource() { + const loadSubsetCalls: Array = [] + let writeRow: (type: `insert` | `delete`, row: Row) => void + const collection = createCollection({ + id: `observer-loadsubset-${seq++}`, + getKey: (r) => r.id, + startSync: false, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + for (const row of SEED) write({ type: `insert`, value: row }) + commit() + markReady() + writeRow = (type, row) => { + begin() + write({ type, value: row }) + commit() + } + return { + loadSubset: (options: unknown) => { + loadSubsetCalls.push(options) + return true as const + }, + } + }, + }, + }) + return { + collection, + loadSubsetCalls, + writeRow: (type: `insert` | `delete`, row: Row) => writeRow(type, row), + } +} + describe(`createLiveQueryObserver`, () => { it(`exposes a stable snapshot of a ready collection (wholesale path)`, () => { const observer = createLiveQueryObserver(makeSource() as any) @@ -368,6 +404,61 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) + it(`wholesale mode does not request an initial snapshot (no unfiltered loadSubset)`, () => { + const { collection, loadSubsetCalls, writeRow } = makeLoadSubsetSource() + const observer = createLiveQueryObserver(collection as any, { + mode: `wholesale`, + }) + + const notifies: Array = [] + observer.subscribe((changes) => notifies.push(changes)) + + // No initial-state request, so no loadSubset({ where: undefined }) — the + // pre-observer React/Angular loading policy. + expect(loadSubsetCalls).toHaveLength(0) + // No bootstrap replay either; wholesale consumers read getSnapshot(). + expect(notifies).toHaveLength(0) + expect(observer.getSnapshot().data).toHaveLength(2) + + // Deltas — including deletes — still wake the consumer. + writeRow(`delete`, { id: `1`, name: `A` }) + + expect(notifies).toHaveLength(1) + expect(observer.getSnapshot().data).toHaveLength(1) + observer.dispose() + }) + + it(`granular mode still seeds from an initial snapshot`, () => { + const { collection, loadSubsetCalls } = makeLoadSubsetSource() + const observer = createLiveQueryObserver(collection as any) + + const inserted: Array = [] + observer.subscribe((changes) => { + for (const c of changes ?? []) { + if (c.type === `insert`) inserted.push(c.key) + } + }) + + expect(inserted.sort()).toEqual([`1`, `2`]) + expect(loadSubsetCalls).toHaveLength(1) + observer.dispose() + }) + + it(`does not enumerate entries for a status-only snapshot read`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + const entriesSpy = vi.spyOn(source, `entries`) + expect(observer.getSnapshot().status).toBe(`ready`) + expect(entriesSpy).not.toHaveBeenCalled() + + // Materialization happens on first data/state access, once per revision. + expect(observer.getSnapshot().data).toHaveLength(2) + expect(observer.getSnapshot().state?.size).toBe(2) + expect(entriesSpy).toHaveBeenCalledTimes(1) + observer.dispose() + }) + it(`wakes consumers on status-only transitions (error, cleaned-up)`, () => { const source = makeSource() const observer = createLiveQueryObserver(source as any) diff --git a/packages/react-db/src/useLiveQuery.ts b/packages/react-db/src/useLiveQuery.ts index 058d9eb91..df7a06d80 100644 --- a/packages/react-db/src/useLiveQuery.ts +++ b/packages/react-db/src/useLiveQuery.ts @@ -422,6 +422,9 @@ export function useLiveQuery( // synchronously during subscribe. observerRef.current = createLiveQueryObserver(collectionRef.current, { deferInitialNotify: true, + // React re-reads getSnapshot() on notify; subscribing without initial + // state preserves the hook's pre-observer loading policy. + mode: `wholesale`, }) } const observer = observerRef.current! diff --git a/packages/react-db/tests/useLiveQuery.eager-onstorechange.test.tsx b/packages/react-db/tests/useLiveQuery.eager-onstorechange.test.tsx index 207f93156..29f801e83 100644 --- a/packages/react-db/tests/useLiveQuery.eager-onstorechange.test.tsx +++ b/packages/react-db/tests/useLiveQuery.eager-onstorechange.test.tsx @@ -54,12 +54,24 @@ describe(`useLiveQuery: eager onStoreChange must not fire synchronously during s const onStoreChange = vi.fn() const unsub = capturedSubscribe!(onStoreChange) - // onStoreChange must not be invoked synchronously inside subscribe; - // it should be deferred to a microtask so it lands after React commits. + // onStoreChange must not be invoked synchronously inside subscribe — + // useSyncExternalStore's own post-subscribe getSnapshot re-read covers a + // ready transition that happened between render and subscribe, so an + // already-ready unchanged collection needs no wake-up at all. expect(onStoreChange).not.toHaveBeenCalled() await Promise.resolve() - expect(onStoreChange).toHaveBeenCalledTimes(1) + expect(onStoreChange).not.toHaveBeenCalled() + + // A real delta does wake the store. + base.utils.begin() + base.utils.write({ + type: `insert`, + value: { id: `3`, name: `C`, age: 30 }, + }) + base.utils.commit() + await Promise.resolve() + expect(onStoreChange).toHaveBeenCalled() unsub() }) From 52b11e906f70fc638a6a833e8be8e767a65f337a Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 12:02:53 +0200 Subject: [PATCH 14/17] =?UTF-8?q?fix(db):=20remove=20deferInitialNotify=20?= =?UTF-8?q?=E2=80=94=20event=20reordering=20gone=20by=20construction?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deferred initial notify could be overtaken by a same-tick delta: the bootstrap batch waited in a microtask while later changes emitted synchronously, so a granular consumer could see v2 before v1. The mechanism existed solely so React's useSyncExternalStore was not notified during its own subscribe call. With React on wholesale mode there is no bootstrap replay to defer — nothing is delivered synchronously during a wholesale subscribe — so the deferral, its attach-generation guard, and the reordering hazard are all removed. Every publication is now delivered synchronously in commit order. Co-Authored-By: Claude Fable 5 --- packages/db/src/live-query-observer.ts | 61 ++----------------- packages/db/tests/live-query-observer.test.ts | 47 +++++++++----- packages/react-db/src/useLiveQuery.ts | 7 ++- 3 files changed, 42 insertions(+), 73 deletions(-) diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 0b6cffee3..568bf859d 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -95,7 +95,6 @@ class LiveQueryObserverImpl< TKey extends string | number, > implements LiveQueryObserver { private readonly collection: Collection | null - private readonly deferInitialNotify: boolean private readonly wholesale: boolean private cachedRevision = -1 private cachedStatus: CollectionStatus | undefined @@ -109,18 +108,10 @@ class LiveQueryObserverImpl< > = [] private dispatching = false private collectionUnsub: (() => void) | null = null - // Bumped on each attach so a deferred initial-notify microtask queued by a - // superseded attach can detect it and skip flushing. - private attachGeneration = 0 private disposed = false - constructor( - collection: Collection | null, - deferInitialNotify: boolean, - wholesale: boolean, - ) { + constructor(collection: Collection | null, wholesale: boolean) { this.collection = collection - this.deferInitialNotify = deferInitialNotify this.wholesale = wholesale // Starting sync during resolution matches every adapter's eager behavior. collection?.startSyncImmediate() @@ -204,19 +195,13 @@ class LiveQueryObserverImpl< } if (seedChanges.length === 0) return - const deliver = () => { - if (record.active) record.listener(seedChanges) - } - if (this.deferInitialNotify) queueMicrotask(deliver) - else deliver() + record.listener(seedChanges) } private attach(): void { const collection = this.collection if (!collection || this.disposed) return - const generation = ++this.attachGeneration - // Granular consumers subscribe with initial state so they receive the // current rows as inserts followed by deltas through one consistent // channel (the collection's per-subscriber change stream requires this to @@ -225,24 +210,13 @@ class LiveQueryObserverImpl< // no unfiltered loadSubset({ where: undefined }) against on-demand // collections. The explicit `false` marks all state as seen so deletes // still flow through as notifies. - // - // When `deferInitialNotify` is set, emits that fire synchronously while - // attaching (the initial-state batch and any synchronous status notify) - // are deferred to a microtask, so a wholesale consumer like React's - // `useSyncExternalStore` never receives a synchronous notify during - // `subscribe`. Effect/watcher-based adapters want the initial state - // synchronously, so by default it is not deferred. Later changes always emit - // synchronously. - let attaching = this.deferInitialNotify - const deferred: Array> | undefined> = [] const notify = (changes: Array> | undefined) => { if (this.disposed || this.subscriptions.size === 0) return // An empty batch carries no semantic change (e.g. the collection's // empty-ready flush); only real deltas and the synthetic ready notify // (undefined) are published. if (changes !== undefined && changes.length === 0) return - if (attaching) deferred.push(changes) - else this.emit(changes) + this.emit(changes) } // Status transitions that carry no change events (loading→ready with no @@ -271,23 +245,6 @@ class LiveQueryObserverImpl< subscription.unsubscribe() return } - - attaching = false - if (deferred.length > 0) { - queueMicrotask(() => { - // Skip if the observer was disposed, has no listeners, or a newer - // attach superseded this one before the flush — otherwise a stale - // initial batch would reach the current listener. - if ( - this.disposed || - this.subscriptions.size === 0 || - generation !== this.attachGeneration - ) { - return - } - for (const changes of deferred.splice(0)) this.emit(changes) - }) - } } private detach(): void { @@ -333,13 +290,6 @@ class LiveQueryObserverImpl< } export interface CreateLiveQueryObserverOptions { - /** - * Defer the initial-state notify to a microtask instead of emitting it - * synchronously during `subscribe`. Set this for `useSyncExternalStore`-style - * consumers (React) that must not receive a store notify during subscribe. - * Effect/watcher-based adapters leave it off to get initial state synchronously. - */ - deferInitialNotify?: boolean /** * How subscribers consume the observer: * @@ -350,7 +300,9 @@ export interface CreateLiveQueryObserverOptions { * - `wholesale`: subscribers treat notifications as a wake-up and re-read * `getSnapshot()` (React/Angular). The observer subscribes WITHOUT initial * state, preserving those adapters' loading policy — no snapshot request, - * so no unfiltered `loadSubset` against on-demand collections. + * so no unfiltered `loadSubset` against on-demand collections. Nothing is + * delivered synchronously during `subscribe`, which keeps + * `useSyncExternalStore`-style consumers safe by construction. */ mode?: `granular` | `wholesale` } @@ -368,7 +320,6 @@ export function createLiveQueryObserver< ): LiveQueryObserver { return new LiveQueryObserverImpl( collection ?? null, - options.deferInitialNotify ?? false, options.mode === `wholesale`, ) } diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 1fb076843..0e8fcc0a2 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -153,19 +153,39 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) - it(`defers the initial notify to a microtask when deferInitialNotify is set`, async () => { + it(`delivers nothing synchronously during a wholesale subscribe`, () => { const observer = createLiveQueryObserver(makeSource() as any, { - deferInitialNotify: true, + mode: `wholesale`, }) let notified = false observer.subscribe(() => { notified = true }) - // Not synchronous during subscribe (protects React's useSyncExternalStore)... + // No bootstrap replay in wholesale mode: useSyncExternalStore-style + // consumers are never notified inside their own subscribe call. expect(notified).toBe(false) - await Promise.resolve() - // ...delivered on the next microtask. - expect(notified).toBe(true) + expect(observer.getSnapshot().data).toHaveLength(2) + observer.dispose() + }) + + it(`delivers events in commit order — no notify can overtake an older one`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + const order: Array = [] + observer.subscribe((changes) => { + for (const c of changes ?? []) order.push(`${c.type}:${c.key}`) + }) + order.length = 0 // drop the bootstrap + + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `v1`, name: `V1` } }) + source.utils.commit() + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `v2`, name: `V2` } }) + source.utils.commit() + + expect(order).toEqual([`insert:v1`, `insert:v2`]) observer.dispose() }) @@ -190,23 +210,20 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) - it(`does not flush a superseded deferred initial notify (deferInitialNotify)`, async () => { - const observer = createLiveQueryObserver(makeSource() as any, { - deferInitialNotify: true, - }) + it(`a resubscribe before a microtask cannot leak a stale bootstrap`, async () => { + const observer = createLiveQueryObserver(makeSource() as any) - // Subscribe then unsubscribe before the microtask flush, then resubscribe. + // Subscribe then unsubscribe immediately, then resubscribe. All delivery + // is synchronous now, so nothing deferred can flush later. observer.subscribe(() => {})() let notifications = 0 observer.subscribe(() => { notifications++ }) + expect(notifications).toBe(1) // the synchronous bootstrap replay await Promise.resolve() - - // Only the current subscription's deferred initial notify should flush, - // not the stale one queued by the first (superseded) attach. - expect(notifications).toBe(1) + expect(notifications).toBe(1) // and nothing else afterwards observer.dispose() }) diff --git a/packages/react-db/src/useLiveQuery.ts b/packages/react-db/src/useLiveQuery.ts index df7a06d80..2ada39077 100644 --- a/packages/react-db/src/useLiveQuery.ts +++ b/packages/react-db/src/useLiveQuery.ts @@ -420,10 +420,11 @@ export function useLiveQuery( if (needsNewCollection) { // Defer the initial notify: useSyncExternalStore must not be notified // synchronously during subscribe. + // Wholesale mode: React re-reads getSnapshot() on notify, keeps the + // hook's pre-observer loading policy, and — because wholesale delivers + // nothing synchronously during subscribe — never notifies + // useSyncExternalStore inside its own subscribe call. observerRef.current = createLiveQueryObserver(collectionRef.current, { - deferInitialNotify: true, - // React re-reads getSnapshot() on notify; subscribing without initial - // state preserves the hook's pre-observer loading policy. mode: `wholesale`, }) } From 5e92926d65b22c6a268955a31c70b8bfb1be02a4 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 12:07:48 +0200 Subject: [PATCH 15/17] =?UTF-8?q?fix(db):=20make=20observer=20construction?= =?UTF-8?q?=20inert=20=E2=80=94=20sync=20activates=20on=20first=20subscrib?= =?UTF-8?q?e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Constructing an observer called startSyncImmediate(), so building one in a render that is later abandoned (React concurrent rendering) activated sync with no committed consumer. Construction is now side-effect-free: activation happens through the first subscription's own addSubscriber path — the identical startSync call — after the status listener is wired, so the loading/ready transitions of a synchronously-starting collection are observed and published instead of happening silently before anyone listens. The adapters' behavior is unchanged: React's input-resolution paths start sync in render themselves (pre-existing, unchanged here), and the effect-based adapters subscribe in the same tick they construct. Co-Authored-By: Claude Fable 5 --- .../tests/inject-live-query.test.ts | 7 ++--- packages/db/src/live-query-observer.ts | 10 +++++-- packages/db/tests/live-query-observer.test.ts | 27 ++++++++++++++++--- 3 files changed, 36 insertions(+), 8 deletions(-) diff --git a/packages/angular-db/tests/inject-live-query.test.ts b/packages/angular-db/tests/inject-live-query.test.ts index 71aa198a5..a2dceb059 100644 --- a/packages/angular-db/tests/inject-live-query.test.ts +++ b/packages/angular-db/tests/inject-live-query.test.ts @@ -122,6 +122,8 @@ function createMockCollection( size: () => map.size, subscribeChanges: (cb: (changes: Array) => void) => { subs.add(cb) + // Real collections start sync when the first subscriber attaches. + api.startSyncImmediate() return { unsubscribe: () => subs.delete(cb), } @@ -136,11 +138,10 @@ function createMockCollection( }, preload: () => Promise.resolve(), startSyncImmediate: () => { - const wasNotReady = status !== `ready` + const previousStatus = status if (status === `idle`) { status = `ready` - } - if (wasNotReady && status === `ready`) { + emitStatusChange(previousStatus) setTimeout(notifyReady, 0) } }, diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 568bf859d..66ec5555c 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -110,11 +110,12 @@ class LiveQueryObserverImpl< private collectionUnsub: (() => void) | null = null private disposed = false + // Construction is side-effect-free: sync activation belongs to the first + // subscription (attach), so building an observer — e.g. in a React render + // that may be abandoned — cannot activate resources on its own. constructor(collection: Collection | null, wholesale: boolean) { this.collection = collection this.wholesale = wholesale - // Starting sync during resolution matches every adapter's eager behavior. - collection?.startSyncImmediate() } getSnapshot(): LiveQuerySnapshot { @@ -202,6 +203,11 @@ class LiveQueryObserverImpl< const collection = this.collection if (!collection || this.disposed) return + // Sync activation happens inside subscribeChanges (addSubscriber starts + // an idle/cleaned-up collection) — the same startSync path the old + // constructor-time startSyncImmediate() took, but now owned by the first + // committed subscription and observed by the status listener below. + // Granular consumers subscribe with initial state so they receive the // current rows as inserts followed by deltas through one consistent // channel (the collection's per-subscriber change stream requires this to diff --git a/packages/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts index 0e8fcc0a2..eea759381 100644 --- a/packages/db/tests/live-query-observer.test.ts +++ b/packages/db/tests/live-query-observer.test.ts @@ -433,14 +433,16 @@ describe(`createLiveQueryObserver`, () => { // No initial-state request, so no loadSubset({ where: undefined }) — the // pre-observer React/Angular loading policy. expect(loadSubsetCalls).toHaveLength(0) - // No bootstrap replay either; wholesale consumers read getSnapshot(). - expect(notifies).toHaveLength(0) + // No bootstrap replay either (only status wake-ups, which carry no + // changes); wholesale consumers read getSnapshot(). + expect(notifies.filter((n) => n !== undefined)).toHaveLength(0) expect(observer.getSnapshot().data).toHaveLength(2) // Deltas — including deletes — still wake the consumer. + const notifiesBefore = notifies.length writeRow(`delete`, { id: `1`, name: `A` }) - expect(notifies).toHaveLength(1) + expect(notifies.length).toBe(notifiesBefore + 1) expect(observer.getSnapshot().data).toHaveLength(1) observer.dispose() }) @@ -476,6 +478,25 @@ describe(`createLiveQueryObserver`, () => { observer.dispose() }) + it(`does not activate sync at construction — only on first subscribe`, () => { + const collection = createCollection( + mockSyncCollectionOptionsNoInitialState({ + id: `observer-idle-${seq++}`, + getKey: (r) => r.id, + }), + ) + const observer = createLiveQueryObserver(collection as any) + + // Construction (e.g. in an abandoned React render) is inert. + expect(collection.status).toBe(`idle`) + expect(observer.getSnapshot().status).toBe(`idle`) + + const unsubscribe = observer.subscribe(() => {}) + expect(collection.status).not.toBe(`idle`) + unsubscribe() + observer.dispose() + }) + it(`wakes consumers on status-only transitions (error, cleaned-up)`, () => { const source = makeSource() const observer = createLiveQueryObserver(source as any) From 58bd2c13299f29340c629063416dab749de190b3 Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 12:09:46 +0200 Subject: [PATCH 16/17] fix(solid-db): generation-guard the resource's async continuations Solid discards a superseded fetch's return value, but the fetcher's post-await writes are side effects into hook-scoped state: switching collections while toArrayWhenReady() was pending let the old continuation resurrect the replaced collection's rows and status over the new one's. Both the success and error continuations now check a generation counter and no-op when superseded. Co-Authored-By: Claude Fable 5 --- packages/solid-db/src/useLiveQuery.ts | 12 ++- packages/solid-db/tests/useLiveQuery.test.tsx | 76 +++++++++++++++++++ 2 files changed, 87 insertions(+), 1 deletion(-) diff --git a/packages/solid-db/src/useLiveQuery.ts b/packages/solid-db/src/useLiveQuery.ts index a50b8269c..9b53e1d88 100644 --- a/packages/solid-db/src/useLiveQuery.ts +++ b/packages/solid-db/src/useLiveQuery.ts @@ -351,9 +351,16 @@ export function useLiveQuery( ) } + // Generation guard for the resource's async continuations: Solid discards a + // superseded fetch's *return value*, but the writes below are side effects + // into hook-scoped state and would still run — resurrecting rows/status from + // a collection that has already been replaced. + let resourceGeneration = 0 + const [getDataResource] = createResource( () => ({ currentCollection: collection() }), async ({ currentCollection }) => { + const generation = ++resourceGeneration if (!currentCollection) { return [] } @@ -361,9 +368,12 @@ export function useLiveQuery( try { await currentCollection.toArrayWhenReady() } catch (error) { - setStatus(`error`) + if (generation === resourceGeneration) setStatus(`error`) throw error } + if (generation !== resourceGeneration) { + return data + } // Initialize state with current collection data batch(() => { state.clear() diff --git a/packages/solid-db/tests/useLiveQuery.test.tsx b/packages/solid-db/tests/useLiveQuery.test.tsx index b7567a635..e1df013e4 100644 --- a/packages/solid-db/tests/useLiveQuery.test.tsx +++ b/packages/solid-db/tests/useLiveQuery.test.tsx @@ -566,6 +566,82 @@ describe(`Query Collections`, () => { }) }) + it(`does not resurrect state from a superseded collection's async continuation`, async () => { + // The resource fetcher awaits toArrayWhenReady(); if the collection is + // switched while that await is pending, the old continuation must not + // write its (now stale) rows/status over the new collection's. + return createRoot(async (dispose) => { + let beginA: (() => void) | undefined + let writeA: ((msg: any) => void) | undefined + let commitA: (() => void) | undefined + let markReadyA: (() => void) | undefined + + const slowCollection = createCollection({ + id: `superseded-async-slow`, + getKey: (person: Person) => person.id, + startSync: false, + sync: { + sync: ({ begin, write, commit, markReady }) => { + beginA = begin + writeA = write + commitA = commit + markReadyA = markReady + // Stays loading until markReady is called manually. + }, + }, + }) + const fastCollection = createCollection( + mockSyncCollectionOptions({ + id: `superseded-async-fast`, + getKey: (person: Person) => person.id, + initialData: [initialPersons[0]!], + }), + ) + + const [useSlow, setUseSlow] = createSignal(true) + const rendered = renderHook(() => { + return useLiveQuery((q) => + q + .from({ persons: useSlow() ? slowCollection : fastCollection }) + .select(({ persons }) => ({ id: persons.id, name: persons.name })), + ) + }) + + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(rendered.result.isLoading).toBe(true) + + // Switch collections while the slow fetch is still awaiting readiness. + setUseSlow(false) + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(rendered.result.state.has(`1`)).toBe(true) + + // The superseded collection now becomes ready with different rows; its + // continuation resolves but must not clobber the current state. + beginA!() + writeA!({ + type: `insert`, + value: { + id: `stale`, + name: `Stale Row`, + age: 99, + email: `stale@example.com`, + isActive: false, + team: `none`, + }, + }) + commitA!() + markReadyA!() + await new Promise((resolve) => setTimeout(resolve, 20)) + + expect(rendered.result.state.has(`stale`)).toBe(false) + expect(rendered.result.state.has(`1`)).toBe(true) + expect(rendered.result.data.map((p: any) => p.id)).toEqual([`1`]) + expect(rendered.result.status).toBe(`ready`) + + dispose() + }) + }) + it(`should be able to query a result collection with live updates`, async () => { const collection = createCollection( mockSyncCollectionOptions({ From b24d1637298774f4fa3c5e43ef03e4a2342cb60f Mon Sep 17 00:00:00 2001 From: Kevin De Porre Date: Mon, 20 Jul 2026 12:10:51 +0200 Subject: [PATCH 17/17] docs(db): mark the observer as internal/unstable; honest changeset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The observer is a contract for TanStack DB's official adapters, not a public extension point — the exported factory and interface now say so (@internal, may change in any release). The changeset drops the false "No behavior change" claim and describes the lifecycle fixes and the per-adapter loading-policy preservation instead. Co-Authored-By: Claude Fable 5 --- .changeset/live-query-observer.md | 8 ++++++-- packages/db/src/live-query-observer.ts | 15 ++++++++++++--- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/.changeset/live-query-observer.md b/.changeset/live-query-observer.md index 4b1218483..95565b091 100644 --- a/.changeset/live-query-observer.md +++ b/.changeset/live-query-observer.md @@ -7,6 +7,10 @@ '@tanstack/angular-db': patch --- -Add a shared live-query observer and migrate all five framework adapters to it +Add an internal shared live-query observer and migrate all five framework adapters to it -Introduces `createLiveQueryObserver` in `@tanstack/db`: given a resolved live-query collection (or `null` for a disabled query) it owns the shared lifecycle every adapter used to re-implement — start sync, subscribe to changes, the already-ready notify race, a stable per-revision snapshot for wholesale consumers, and delivery of the raw `ChangeMessage[]` for granular consumers. React, Vue, Svelte, Solid, and Angular's live-query hooks now materialize from the observer instead of their own hand-rolled subscription/status/snapshot machinery, keeping each adapter's native reactivity. No behavior change. +Introduces `createLiveQueryObserver` in `@tanstack/db`: given a resolved live-query collection (or `null` for a disabled query) it owns the lifecycle every adapter used to re-implement — sync activation on first subscribe, change and status subscriptions, a snapshot with stable identity per state revision for wholesale consumers, and delivery of the raw `ChangeMessage[]` for granular consumers. React, Vue, Svelte, Solid, and Angular's live-query hooks now materialize from the observer instead of their own hand-rolled subscription/status/snapshot machinery, keeping each adapter's native reactivity and each adapter's data-loading policy (wholesale adapters subscribe without initial state; granular adapters seed from it). + +The observer is an **internal, unstable contract** for TanStack DB's official adapters — it is exported so the adapter packages can consume it, but it is not a public extension point yet and its API may change in any release. + +The migration also fixes several live-query lifecycle defects: status-only transitions (`error`, `cleaned-up`) now reach mounted consumers; snapshot identity is stable across unsubscribe/resubscribe and stays fresh while detached; dispatch is FIFO and non-reentrant with subscriptions identified by record rather than callback; disposing during the synchronous initial replay no longer leaks the collection subscription; subscribing after dispose throws instead of registering a dead listener; Solid guards its async resource continuations against superseded collections; and constructing an observer no longer activates sync (activation belongs to the first committed subscription). diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 66ec5555c..83899bdfe 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -40,13 +40,17 @@ export type LiveQueryObserverListener< /** * Wraps a resolved live-query `Collection` (or `null` for a disabled query) with - * the shared lifecycle every framework adapter needs: start sync, subscribe to - * changes, handle the already-ready race, expose a stable snapshot for - * wholesale consumers, and deliver the raw change set for granular consumers. + * the shared lifecycle every framework adapter needs: start sync on first + * subscribe, subscribe to changes and status transitions, expose a stable + * snapshot for wholesale consumers, and deliver the raw change set for + * granular consumers. * * Input resolution (query fn / config / collection / disabled) stays in the * adapter — it is framework-reactive. The observer owns everything after the * input is resolved to a concrete collection. + * + * @internal Unstable contract for TanStack DB's official framework adapters — + * not a public extension point yet; may change in any release. */ export interface LiveQueryObserver< T extends object, @@ -316,6 +320,11 @@ export interface CreateLiveQueryObserverOptions { /** * Create a {@link LiveQueryObserver} for a resolved live-query collection, or a * disabled observer when `collection` is `null`/`undefined`. + * + * @internal This is an unstable contract shared by TanStack DB's official + * framework adapters. It is exported so the adapter packages can use it, but + * it is not a public extension point yet: its API may change in any release + * without a semver major. */ export function createLiveQueryObserver< T extends object,