diff --git a/.changeset/live-query-observer.md b/.changeset/live-query-observer.md new file mode 100644 index 0000000000..95565b091d --- /dev/null +++ b/.changeset/live-query-observer.md @@ -0,0 +1,16 @@ +--- +'@tanstack/db': minor +'@tanstack/react-db': patch +'@tanstack/vue-db': patch +'@tanstack/svelte-db': patch +'@tanstack/solid-db': patch +'@tanstack/angular-db': patch +--- + +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 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/angular-db/src/index.ts b/packages/angular-db/src/index.ts index f8cac4086b..392bfc6d7c 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, @@ -249,28 +249,25 @@ 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. + // 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`, + }) - // 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 a8cc7ec803..6a5b5ba66a 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/angular-db/tests/inject-live-query.test.ts b/packages/angular-db/tests/inject-live-query.test.ts index 81fbb12a92..a2dceb0591 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), @@ -104,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), } @@ -118,26 +138,33 @@ 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) } }, __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) diff --git a/packages/db/src/collection/changes.ts b/packages/db/src/collection/changes.ts index dc07cd3f18..e5bdfb8450 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 137fd5f595..13887a43d9 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/errors.ts b/packages/db/src/errors.ts index 0bfd2f9969..710025b654 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/index.ts b/packages/db/src/index.ts index 71e264d712..bf4e16a817 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 0000000000..83899bdfe6 --- /dev/null +++ b/packages/db/src/live-query-observer.ts @@ -0,0 +1,340 @@ +import { LiveQueryObserverDisposedError } from './errors.js' +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 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, + 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 +} + +/** + * 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, + 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 wholesale: boolean + private cachedRevision = -1 + private cachedStatus: CollectionStatus | undefined + private cachedSnapshot: LiveQuerySnapshot = DISABLED_SNAPSHOT + 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 + 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 + } + + getSnapshot(): LiveQuerySnapshot { + const collection = this.collection + if (!collection) return DISABLED_SNAPSHOT + + // 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.cachedRevision !== collection._stateRevision || + this.cachedStatus !== collection.status + ) { + this.cachedRevision = collection._stateRevision + this.cachedStatus = collection.status + 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() { + return (stateCache ??= new Map(readEntries())) + }, + get data() { + dataCache ??= readEntries().map(([, value]) => value) + return singleResult ? dataCache[0] : dataCache + }, + collection, + status: collection.status, + ...getLiveQueryStatusFlags(collection.status), + isEnabled: true, + } + } + return this.cachedSnapshot + } + + 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() + } else { + // 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 () => { + if (!record.active) return + record.active = false + this.subscriptions.delete(record) + if (this.subscriptions.size === 0) this.detach() + } + } + + /** 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 + + record.listener(seedChanges) + } + + private attach(): void { + 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 + // 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. + 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 + 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 = () => { + statusUnsub() + subscription?.unsubscribe() + } + this.collectionUnsub = release + subscription = collection.subscribeChanges( + (changes) => notify(changes as Array>), + { includeInitialState: !this.wholesale }, + ) + if (this.collectionUnsub !== release) { + subscription.unsubscribe() + return + } + } + + private detach(): void { + this.collectionUnsub?.() + this.collectionUnsub = null + } + + private emit(changes: Array> | undefined): void { + 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()! + // 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 { + await this.collection?.preload() + } + + dispose(): void { + if (this.disposed) return + this.disposed = true + this.detach() + for (const subRecord of this.subscriptions) subRecord.active = false + this.subscriptions.clear() + this.publicationQueue.length = 0 + } +} + +export interface CreateLiveQueryObserverOptions { + /** + * 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. Nothing is + * delivered synchronously during `subscribe`, which keeps + * `useSyncExternalStore`-style consumers safe by construction. + */ + mode?: `granular` | `wholesale` +} + +/** + * 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, + TKey extends string | number, +>( + collection: Collection | null | undefined, + options: CreateLiveQueryObserverOptions = {}, +): LiveQueryObserver { + return new LiveQueryObserverImpl( + collection ?? null, + options.mode === `wholesale`, + ) +} diff --git a/packages/db/tests/conformance/contract.ts b/packages/db/tests/conformance/contract.ts index 486fb4920d..11534231b0 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 8e02911c69..1a2d21c70b 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/db/tests/live-query-observer.test.ts b/packages/db/tests/live-query-observer.test.ts new file mode 100644 index 0000000000..eea7593815 --- /dev/null +++ b/packages/db/tests/live-query-observer.test.ts @@ -0,0 +1,534 @@ +import { describe, expect, it, vi } from 'vitest' +import { createCollection } from '../src/collection/index.js' +import { createLiveQueryObserver } from '../src/live-query-observer.js' +import { + mockSyncCollectionOptions, + mockSyncCollectionOptionsNoInitialState, +} 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, + }), + ) +} + +/** 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 +} + +/** 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) + + 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(`delivers nothing synchronously during a wholesale subscribe`, () => { + const observer = createLiveQueryObserver(makeSource() as any, { + mode: `wholesale`, + }) + let notified = false + observer.subscribe(() => { + notified = true + }) + // No bootstrap replay in wholesale mode: useSyncExternalStore-style + // consumers are never notified inside their own subscribe call. + expect(notified).toBe(false) + 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() + }) + + 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() + }) + + it(`a resubscribe before a microtask cannot leak a stale bootstrap`, async () => { + const observer = createLiveQueryObserver(makeSource() as any) + + // 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() + expect(notifications).toBe(1) // and nothing else afterwards + 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 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((lateChanges) => { + if (lateChanges?.some((c) => c.key === `4`)) { + lateListenerRow4Deliveries++ + } + }) + } + }) + + source.utils.begin() + source.utils.write({ type: `insert`, value: { id: `4`, name: `D` } }) + source.utils.commit() + + // 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() + }) + + it(`still delivers the in-flight publication to a listener removed during dispatch`, () => { + const source = makeSource() + const observer = createLiveQueryObserver(source as any) + + let row5Deliveries = 0 + let unsubB: (() => void) | null = null + observer.subscribe(() => { + unsubB?.() + unsubB = null + }) + unsubB = observer.subscribe((changes) => { + if (changes?.some((c) => c.key === `5`)) row5Deliveries++ + }) + + 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(row5Deliveries).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(`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(`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(`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(`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 (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.length).toBe(notifiesBefore + 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(`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) + + 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). + 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 8dc3d0a31b..2ada390770 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,38 @@ export function useLiveQuery( } } - // Reset refs when collection changes + // 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) { - versionRef.current = 0 - snapshotRef.current = null + // 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, { + mode: `wholesale`, + }) } + 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/react-db/tests/conformance.test.tsx b/packages/react-db/tests/conformance.test.tsx index 303bbd89e6..3916d67b79 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/react-db/tests/useLiveQuery.eager-onstorechange.test.tsx b/packages/react-db/tests/useLiveQuery.eager-onstorechange.test.tsx index 207f93156f..29f801e837 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() }) 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 0000000000..874d020625 --- /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)) + }) +}) diff --git a/packages/solid-db/src/useLiveQuery.ts b/packages/solid-db/src/useLiveQuery.ts index 7759915f32..9b53e1d887 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' @@ -350,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 [] } @@ -360,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() @@ -389,36 +400,40 @@ 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) + // 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(() => { - 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/solid-db/tests/conformance.test.tsx b/packages/solid-db/tests/conformance.test.tsx index 2dcea6ab03..ddb1cfbd08 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 378c2a0fc7..e1df013e41 100644 --- a/packages/solid-db/tests/useLiveQuery.test.tsx +++ b/packages/solid-db/tests/useLiveQuery.test.tsx @@ -523,6 +523,125 @@ 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(`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({ diff --git a/packages/svelte-db/src/useLiveQuery.svelte.ts b/packages/svelte-db/src/useLiveQuery.svelte.ts index 0d9dd9d7fe..71dc7a12a7 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, @@ -382,13 +384,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 @@ -396,81 +411,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/svelte-db/tests/conformance.svelte.test.ts b/packages/svelte-db/tests/conformance.svelte.test.ts index 5b502b02c7..b68d412d16 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/src/useLiveQuery.ts b/packages/vue-db/src/useLiveQuery.ts index 762cbda0a6..c12fdb8eea 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 { diff --git a/packages/vue-db/tests/conformance.test.ts b/packages/vue-db/tests/conformance.test.ts index 432876f4f5..eef6e0a2f6 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),