diff --git a/.changeset/live-query-window-controller.md b/.changeset/live-query-window-controller.md new file mode 100644 index 000000000..0f258cd5f --- /dev/null +++ b/.changeset/live-query-window-controller.md @@ -0,0 +1,10 @@ +--- +'@tanstack/db': patch +'@tanstack/react-db': patch +--- + +Add the unstable, internal `createLiveQueryWindowController` primitive for +forward pagination. It coordinates collection-scoped window leases, commits +pages only after subset loads succeed, restores windows after failures and +cleanup, and lets React's `useLiveInfiniteQuery` become a thin binding without +changing its public API or resetting pages for structurally equal dependencies. diff --git a/packages/db/src/collection/changes.ts b/packages/db/src/collection/changes.ts index 87d092f9b..b5b9dfb83 100644 --- a/packages/db/src/collection/changes.ts +++ b/packages/db/src/collection/changes.ts @@ -29,6 +29,7 @@ export class CollectionChangesManager< public changeSubscriptions = new Set() public batchedEvents: Array> = [] public shouldBatchEvents = false + private layoutChangeListeners = new Set<() => void>() /** * Monotonic revision of the collection's visible state, advanced once per @@ -41,8 +42,7 @@ export class CollectionChangesManager< /** * Monotonic revision advanced only for explicit layout-only publications. - * This distinguishes them from the legacy empty ready event, since both use - * an empty change batch at the public subscription boundary. + * Observers use it to detect reordered rows whose values did not change. */ public layoutRevision = 0 @@ -124,6 +124,13 @@ export class CollectionChangesManager< return } + // Notify both internal layout consumers and the public subscription API. + // Public subscribers historically receive an empty batch for order-only + // moves because there is no row-value ChangeMessage to publish. + if (rawEvents.length === 0) { + for (const listener of this.layoutChangeListeners) listener() + } + // Enrich all change messages with virtual properties // This uses the "add-if-missing" pattern to preserve pass-through semantics const enrichedEvents: Array< @@ -132,10 +139,16 @@ export class CollectionChangesManager< // Emit to all listeners for (const subscription of this.changeSubscriptions) { - subscription.emitEvents(enrichedEvents, layoutChanged) + subscription.emitEvents(enrichedEvents) } } + /** Subscribe to layout-only publications. Internal observer channel. */ + public subscribeLayoutChanges(listener: () => void): () => void { + this.layoutChangeListeners.add(listener) + return () => this.layoutChangeListeners.delete(listener) + } + /** * Subscribe to changes in the collection */ diff --git a/packages/db/src/collection/index.ts b/packages/db/src/collection/index.ts index c775e7ada..f9bb465fe 100644 --- a/packages/db/src/collection/index.ts +++ b/packages/db/src/collection/index.ts @@ -437,6 +437,11 @@ export class CollectionImpl< return this._changes.layoutRevision } + /** Subscribe to layout-only publications. Internal observer channel. */ + public _subscribeLayoutChanges(listener: () => void): () => void { + return this._changes.subscribeLayoutChanges(listener) + } + /** Mark the active sync transaction as layout-changing. Internal. */ public _markLayoutChange(): void { this._sync.markLayoutChange() diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts index 0f603ee71..7757be14a 100644 --- a/packages/db/src/collection/subscription.ts +++ b/packages/db/src/collection/subscription.ts @@ -99,7 +99,6 @@ export class CollectionSubscription // This prevents a flash of missing content between deletes and new inserts private isBufferingForTruncate = false private truncateBuffer: Array>> = [] - private truncateBufferHasLayoutChange = false private pendingTruncateRefetches: Set> = new Set() public get status(): SubscriptionStatus { @@ -249,14 +248,9 @@ export class CollectionSubscription // Flatten all buffered changes into a single array for atomic emission // This ensures consumers see all truncate changes (deletes + inserts) in one callback const merged = this.truncateBuffer.flat() - const layoutChanged = this.truncateBufferHasLayoutChange - if (merged.length > 0 || layoutChanged) { - const delivered = this.filteredCallback(merged) - if (layoutChanged && !delivered) this.filteredCallback([]) - } + if (merged.length > 0) this.filteredCallback(merged) this.truncateBuffer = [] - this.truncateBufferHasLayoutChange = false } setOrderByIndex(index: IndexInterface) { @@ -308,12 +302,13 @@ export class CollectionSubscription this.pendingLoadSubsetPromises.add(syncResult) this.setStatus(`loadingSubset`) - syncResult.finally(() => { + const finish = () => { this.pendingLoadSubsetPromises.delete(syncResult) if (this.pendingLoadSubsetPromises.size === 0) { this.setStatus(`ready`) } - }) + } + void syncResult.then(finish, finish) } } @@ -325,10 +320,7 @@ export class CollectionSubscription return this.snapshotSent } - emitEvents( - changes: Array>, - layoutChanged = false, - ): boolean { + emitEvents(changes: Array>): boolean { const newChanges = this.filterAndFlipChanges(changes) if (this.isBufferingForTruncate) { @@ -337,12 +329,9 @@ export class CollectionSubscription if (newChanges.length > 0) { this.truncateBuffer.push(newChanges) } - if (layoutChanged) this.truncateBufferHasLayoutChange = true return false } else { - const delivered = this.filteredCallback(newChanges) - if (layoutChanged && !delivered) return this.filteredCallback([]) - return delivered + return this.filteredCallback(newChanges) } } @@ -742,7 +731,6 @@ export class CollectionSubscription // Clean up truncate buffer state this.isBufferingForTruncate = false this.truncateBuffer = [] - this.truncateBufferHasLayoutChange = false this.pendingTruncateRefetches.clear() // Unload all subsets that this subscription loaded diff --git a/packages/db/src/collection/sync.ts b/packages/db/src/collection/sync.ts index 8fa49b163..d71710611 100644 --- a/packages/db/src/collection/sync.ts +++ b/packages/db/src/collection/sync.ts @@ -444,6 +444,18 @@ export class CollectionSyncManager< return this.pendingLoadSubsetPromises.size > 0 } + /** Wait for the subset loads that are active during the current operation. */ + public waitForCurrentLoadSubset(): true | Promise { + if (this.pendingLoadSubsetPromises.size === 0) return true + return this.waitForPendingLoadSubset() + } + + private async waitForPendingLoadSubset(): Promise { + do { + await Promise.all([...this.pendingLoadSubsetPromises]) + } while (this.pendingLoadSubsetPromises.size > 0) + } + /** * Tracks a load promise for isLoadingSubset state. * @internal This is for internal coordination (e.g., live-query glue code), not for general use. @@ -462,7 +474,7 @@ export class CollectionSyncManager< }) } - promise.finally(() => { + const finish = () => { const loadingEnding = this.pendingLoadSubsetPromises.size === 1 && this.pendingLoadSubsetPromises.has(promise) @@ -477,7 +489,8 @@ export class CollectionSyncManager< loadingSubsetTransition: `end`, }) } - }) + } + void promise.then(finish, finish) } /** diff --git a/packages/db/src/errors.ts b/packages/db/src/errors.ts index 710025b65..0281b13af 100644 --- a/packages/db/src/errors.ts +++ b/packages/db/src/errors.ts @@ -141,6 +141,12 @@ export class LiveQueryObserverDisposedError extends CollectionStateError { } } +export class LiveQueryWindowControllerDisposedError extends CollectionStateError { + constructor() { + super(`Cannot subscribe to a disposed LiveQueryWindowController`) + } +} + // 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 bf4e16a81..80814640d 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -12,6 +12,8 @@ export * from './query/index.js' export * from './optimistic-action' export * from './live-query-adapter' export * from './live-query-observer' +/** @internal Unstable adapter primitive for RFC #1623. */ +export * from './live-query-window-controller' 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 index 3bb35c818..1cb0cba28 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -44,7 +44,10 @@ export interface LiveQuerySnapshot< isEnabled: boolean } -/** Listener payload: the change set, or `undefined` for the synthetic ready notify. */ +/** + * Listener payload: changes, `[]` for an internal layout-only publication, or + * `undefined` for a synthetic status/ready notification. + */ export type LiveQueryObserverListener< T extends object, TKey extends string | number, @@ -357,11 +360,16 @@ class LiveQueryObserverImpl< const notify = ( changes: Array> | undefined, status: CollectionStatus = collection.status, + explicitLayoutChange = false, ) => { if (this.disposed || this.subscriptions.size === 0) return const layoutRevision = this.getCollectionLayoutRevision(collection) - let layoutChanged = false - if (changes !== undefined && changes.length === 0) { + let layoutChanged = explicitLayoutChange + if ( + !explicitLayoutChange && + changes !== undefined && + changes.length === 0 + ) { // Empty ready events predate the explicit layout signal and share its // empty-array payload. Only forward an empty batch when the collection // confirms that a new layout-only publication occurred. @@ -401,6 +409,17 @@ class LiveQueryObserverImpl< const statusUnsub = collection.on(`status:change`, ({ status }) => notify(undefined, status), ) + const subscribeLayoutChanges = ( + collection as Collection & { + _subscribeLayoutChanges?: (listener: () => void) => () => void + } + )._subscribeLayoutChanges + const layoutUnsub = + typeof subscribeLayoutChanges === `function` + ? subscribeLayoutChanges.call(collection, () => + notify([], collection.status, true), + ) + : () => {} // `subscribeChanges` delivers the initial state synchronously, so a // listener can dispose the observer while the collection subscription is @@ -410,6 +429,7 @@ class LiveQueryObserverImpl< let subscription: { unsubscribe: () => void } | null = null const release = () => { statusUnsub() + layoutUnsub() subscription?.unsubscribe() } this.collectionUnsub = release diff --git a/packages/db/src/live-query-window-controller.ts b/packages/db/src/live-query-window-controller.ts new file mode 100644 index 000000000..e22ca6170 --- /dev/null +++ b/packages/db/src/live-query-window-controller.ts @@ -0,0 +1,734 @@ +import { + LiveQueryWindowControllerDisposedError, + SetWindowRequiresOrderByError, +} from './errors.js' +import { getLiveQueryStatusFlags } from './live-query-adapter.js' +import { createLiveQueryObserver } from './live-query-observer.js' +import type { + LiveQueryObserver, + LiveQuerySnapshot, +} from './live-query-observer.js' +import type { Collection } from './collection/index.js' +import type { CollectionStatus } from './types.js' + +const DEFAULT_PAGE_SIZE = 20 + +type WindowResult = true | Promise + +type WindowTarget = object & { + utils?: { + setWindow?: (options: { offset: number; limit: number }) => WindowResult + getWindow?: () => { offset: number; limit: number } | undefined + } +} + +type PendingWindow = { + generation: number + limit: number + promise: Promise +} + +class WindowCoordinator { + private readonly leases = new Map() + private readonly leaseVersions = new Map() + private readonly initialWindow: { offset: number; limit: number } | undefined + private appliedLimit: number | undefined + private pending: PendingWindow | undefined + private generation = 0 + private leaseVersion = 0 + + constructor(private readonly target: WindowTarget) { + this.initialWindow = target.utils?.getWindow?.() + } + + request(lease: symbol, limit: number): WindowResult { + const previousLimit = this.leases.get(lease) + const previousVersion = this.leaseVersions.get(lease) + const version = ++this.leaseVersion + this.leases.set(lease, limit) + this.leaseVersions.set(lease, version) + + let result: WindowResult + try { + result = this.applyDesiredWindow() + } catch (error) { + this.rollbackLease(lease, version, previousLimit, previousVersion) + this.appliedLimit = undefined + throw error + } + + if (result === true) return true + return result.catch(async (error: unknown) => { + if (this.rollbackLease(lease, version, previousLimit, previousVersion)) { + this.generation++ + this.pending = undefined + this.appliedLimit = undefined + try { + if (this.leases.size === 0) { + this.restoreInitialWindow() + } else { + const rollback = this.applyDesiredWindow() + if (rollback !== true) await rollback + } + } catch { + // Preserve the failure from the requested window. + } + } + throw error + }) + } + + isLeaseSatisfied(lease: symbol, minimumLimit: number): boolean { + const limit = this.leases.get(lease) + if (limit === undefined || limit < minimumLimit) return false + const desiredLimit = this.getDesiredLimit() + const currentWindow = this.target.utils?.getWindow?.() + return ( + currentWindow === undefined || + (currentWindow.offset === 0 && currentWindow.limit === desiredLimit) + ) + } + + release(lease: symbol): void { + if (!this.leases.delete(lease)) return + this.leaseVersions.delete(lease) + + // A pending request may still mutate the physical operator, but it no longer + // establishes the accepted window for the remaining lease set. + this.generation++ + this.pending = undefined + this.appliedLimit = undefined + + if (this.leases.size === 0) { + return + } + + try { + const result = this.applyDesiredWindow() + if (result !== true) { + void result.catch(() => { + // Unsubscribe has no async error channel. Leave the physical window + // unaccepted so the next request retries it. + this.appliedLimit = undefined + }) + } + } catch { + // The remaining controller will retry on its next request. + this.appliedLimit = undefined + } + } + + private getDesiredLimit(): number | undefined { + let desired: number | undefined + for (const limit of this.leases.values()) { + desired = desired === undefined ? limit : Math.max(desired, limit) + } + return desired + } + + private rollbackLease( + lease: symbol, + version: number, + previousLimit: number | undefined, + previousVersion: number | undefined, + ): boolean { + if (this.leaseVersions.get(lease) !== version) return false + if (previousLimit === undefined) { + this.leases.delete(lease) + this.leaseVersions.delete(lease) + } else { + this.leases.set(lease, previousLimit) + if (previousVersion === undefined) { + this.leaseVersions.delete(lease) + } else { + this.leaseVersions.set(lease, previousVersion) + } + } + return true + } + + private restoreInitialWindow(): void { + const setWindow = this.target.utils?.setWindow + if (!this.initialWindow || typeof setWindow !== `function`) return + try { + const result = setWindow.call(this.target.utils, this.initialWindow) + if (result !== true) void result.catch(() => {}) + } catch { + // Release has no error channel. A future lease will retry its own window. + } + } + + private applyDesiredWindow(): WindowResult { + const limit = this.getDesiredLimit() + if (limit === undefined) return true + if (this.pending?.limit === limit) return this.pending.promise + if (this.pending) { + // `setWindow` mutates the physical operator before its load promise + // settles. A different desired window must therefore be applied again, + // even when it matches the last settled limit. + this.generation++ + this.pending = undefined + this.appliedLimit = undefined + } + const currentWindow = this.target.utils?.getWindow?.() + if ( + limit === this.appliedLimit && + currentWindow?.offset === 0 && + currentWindow.limit === limit + ) { + return true + } + + const setWindow = this.target.utils?.setWindow + if (typeof setWindow !== `function`) { + throw new SetWindowRequiresOrderByError() + } + + const generation = ++this.generation + const result = setWindow.call(this.target.utils, { offset: 0, limit }) + if (result === true) { + if (generation === this.generation && this.getDesiredLimit() === limit) { + this.appliedLimit = limit + } + return true + } + + const promise = result.then( + () => { + if ( + generation === this.generation && + this.getDesiredLimit() === limit + ) { + this.appliedLimit = limit + } + if (this.pending?.generation === generation) { + this.pending = undefined + } + }, + (error: unknown) => { + if (this.pending?.generation === generation) { + this.pending = undefined + } + throw error + }, + ) + this.pending = { generation, limit, promise } + return promise + } +} + +const windowCoordinators = new WeakMap() + +function getWindowCoordinator(target: WindowTarget): WindowCoordinator { + let coordinator = windowCoordinators.get(target) + if (!coordinator) { + coordinator = new WindowCoordinator(target) + windowCoordinators.set(target, coordinator) + } + return coordinator +} + +/** + * A page-windowed view of a live query at a point in time. + * + * @internal This contract is unstable while RFC #1623 is being implemented. + */ +export interface LiveQueryWindowSnapshot< + T extends object, + TKey extends string | number, +> { + /** Rows across all committed pages, with the peek-ahead row removed. */ + data: ReadonlyArray + /** Rows grouped into committed pages of `pageSize`. */ + pages: ReadonlyArray> + /** `initialPageParam + i` for each committed page. */ + pageParams: ReadonlyArray + hasNextPage: boolean + isFetchingNextPage: boolean + /** The last pagination failure, cleared when a retry begins. */ + error: unknown + /** Keyed results for the physical window, or `undefined` when disabled. */ + state: ReadonlyMap | undefined + collection: Collection | undefined + status: CollectionStatus | `disabled` + isLoading: boolean + isReady: boolean + isIdle: boolean + isError: boolean + isCleanedUp: boolean + isEnabled: boolean +} + +/** @internal This contract is unstable while RFC #1623 is being implemented. */ +export interface CreateLiveQueryWindowControllerOptions { + /** Rows per page (default 20). Non-positive values use the default. */ + pageSize?: number + /** Value of the first page's `pageParam` (default 0). */ + initialPageParam?: number + /** Committed pages to preserve when a framework binding changes page shape. */ + initialPageCount?: number +} + +/** @internal This contract is unstable while RFC #1623 is being implemented. */ +export interface LiveQueryWindowController< + T extends object, + TKey extends string | number, +> { + getSnapshot: () => LiveQueryWindowSnapshot + subscribe: (listener: () => void) => () => void + /** Load one more page, resolving only after that page is committed. */ + fetchNextPage: () => Promise + /** Reset to the first page, resolving after the smaller window is accepted. */ + reset: () => Promise + preload: () => Promise + dispose: () => void +} + +interface CachedFrom { + observerSnapshot: unknown + committedPageCount: number + isFetchingNextPage: boolean + hasPaginationError: boolean + paginationError: unknown + failedHasNextPage: boolean +} + +interface SubscriptionRecord { + listener: () => void + active: boolean +} + +interface Publication { + targets: Array +} + +class LiveQueryWindowControllerImpl< + T extends object, + TKey extends string | number, +> implements LiveQueryWindowController { + private readonly observer: LiveQueryObserver + private readonly collection: Collection | null + private readonly coordinator: WindowCoordinator | null + private readonly lease = Symbol(`liveQueryWindowLease`) + private readonly pageSize: number + private readonly initialPageParam: number + + private committedPageCount: number + private isFetchingNextPage = false + private hasPaginationError = false + private paginationError: unknown + private failedHasNextPage = false + private windowGeneration = 0 + private pendingWindowGeneration: number | undefined + private leaseActive = false + private leaseGeneration = 0 + private inFlightLeaseHolders = 0 + + private readonly subscriptions = new Set() + private readonly publicationQueue: Array = [] + private dispatching = false + private blockDelivery = false + private transitionDepth = 0 + private transitionNeedsNotify = false + private observerUnsub: (() => void) | null = null + private cachedSnapshot: LiveQueryWindowSnapshot | null = null + private cachedFrom: CachedFrom | null = null + private disposed = false + + constructor( + collection: Collection | null, + options: CreateLiveQueryWindowControllerOptions, + ) { + this.collection = collection + this.coordinator = collection + ? getWindowCoordinator(collection as unknown as WindowTarget) + : null + this.pageSize = + options.pageSize !== undefined && options.pageSize > 0 + ? options.pageSize + : DEFAULT_PAGE_SIZE + this.initialPageParam = options.initialPageParam ?? 0 + const initialPageCount = Math.floor(options.initialPageCount ?? 1) + this.committedPageCount = Number.isFinite(initialPageCount) + ? Math.max(1, initialPageCount) + : 1 + // The controller listener carries no delta payload, so wholesale is the + // only coherent observer contract and guarantees non-reentrant subscribe. + this.observer = createLiveQueryObserver(collection, { + mode: `wholesale`, + }) + } + + getSnapshot(): LiveQueryWindowSnapshot { + const observerSnapshot = this.observer.getSnapshot() + const cached = this.cachedSnapshot + if ( + cached && + this.cachedFrom && + this.cachedFrom.observerSnapshot === observerSnapshot && + this.cachedFrom.committedPageCount === this.committedPageCount && + this.cachedFrom.isFetchingNextPage === this.isFetchingNextPage && + this.cachedFrom.hasPaginationError === this.hasPaginationError && + this.cachedFrom.paginationError === this.paginationError && + this.cachedFrom.failedHasNextPage === this.failedHasNextPage + ) { + return cached + } + + const enabled = observerSnapshot.isEnabled + const rows = + enabled && Array.isArray(observerSnapshot.data) + ? (observerSnapshot.data as ReadonlyArray) + : [] + const totalRequested = this.committedPageCount * this.pageSize + const computedHasNextPage = enabled && rows.length > totalRequested + const hasNextPage = this.hasPaginationError + ? this.failedHasNextPage + : computedHasNextPage + + const pageCount = enabled ? this.committedPageCount : 0 + const pages: Array> = [] + const pageParams: Array = [] + for (let i = 0; i < pageCount; i++) { + pages.push(rows.slice(i * this.pageSize, (i + 1) * this.pageSize)) + pageParams.push(this.initialPageParam + i) + } + + const status = this.hasPaginationError ? `error` : observerSnapshot.status + const statusFlags = this.hasPaginationError + ? getLiveQueryStatusFlags(`error`) + : observerSnapshot + this.cachedSnapshot = { + data: rows.slice(0, totalRequested), + pages, + pageParams, + hasNextPage, + isFetchingNextPage: this.isFetchingNextPage, + error: this.hasPaginationError ? this.paginationError : undefined, + state: observerSnapshot.state, + collection: observerSnapshot.collection, + status, + isLoading: statusFlags.isLoading, + isReady: statusFlags.isReady, + isIdle: statusFlags.isIdle, + isError: statusFlags.isError, + isCleanedUp: observerSnapshot.isCleanedUp, + isEnabled: observerSnapshot.isEnabled, + } + this.cachedFrom = { + observerSnapshot, + committedPageCount: this.committedPageCount, + isFetchingNextPage: this.isFetchingNextPage, + hasPaginationError: this.hasPaginationError, + paginationError: this.paginationError, + failedHasNextPage: this.failedHasNextPage, + } + return this.cachedSnapshot + } + + subscribe(listener: () => void): () => void { + if (this.disposed) throw new LiveQueryWindowControllerDisposedError() + + const record: SubscriptionRecord = { listener, active: true } + this.subscriptions.add(record) + if (this.subscriptions.size === 1) { + this.blockDelivery = true + let observerUnsub: (() => void) | null = null + try { + // Store the desired physical window before observer activation can + // compile or restart the live-query pipeline. + const windowResult = this.ensureLeaseActive(this.committedPageCount) + const leaseGeneration = this.leaseGeneration + observerUnsub = this.observer.subscribe(() => this.onObserverNotify()) + this.observerUnsub = observerUnsub + if (windowResult !== true) { + this.trackAttachmentFailure(windowResult, leaseGeneration) + } + } catch (error) { + observerUnsub?.() + this.observerUnsub = null + this.deactivateLease() + record.active = false + this.subscriptions.delete(record) + throw error + } finally { + this.blockDelivery = false + } + } + + return () => { + if (!record.active) return + record.active = false + this.subscriptions.delete(record) + if (this.subscriptions.size === 0) { + this.observerUnsub?.() + this.observerUnsub = null + if (this.inFlightLeaseHolders === 0) this.deactivateLease() + } + } + } + + fetchNextPage(): Promise { + if (this.disposed || this.isFetchingNextPage) return Promise.resolve() + if (!this.getSnapshot().hasNextPage) return Promise.resolve() + return this.requestPageCount(this.committedPageCount + 1, true) + } + + reset(): Promise { + if (this.disposed) return Promise.resolve() + if ( + this.committedPageCount === 1 && + !this.hasPaginationError && + !this.isFetchingNextPage && + this.pendingWindowGeneration === undefined + ) { + return Promise.resolve() + } + return this.requestPageCount(1, false) + } + + async preload(): Promise { + if (this.disposed) throw new LiveQueryWindowControllerDisposedError() + + const hadPaginationError = this.hasPaginationError + this.hasPaginationError = false + this.paginationError = undefined + this.acquireInFlightLease() + try { + const result = this.ensureLeaseActive(this.committedPageCount) + if (result !== true) await result + await this.observer.preload() + this.failedHasNextPage = false + if (hadPaginationError) this.notify() + } catch (error) { + this.hasPaginationError = true + this.paginationError = error + this.failedHasNextPage = this.getComputedHasNextPage() + this.notify() + throw error + } finally { + this.releaseInFlightLease() + } + } + + dispose(): void { + if (this.disposed) return + this.disposed = true + this.windowGeneration++ + this.pendingWindowGeneration = undefined + this.observerUnsub?.() + this.observerUnsub = null + this.deactivateLease() + this.observer.dispose() + for (const record of this.subscriptions) record.active = false + this.subscriptions.clear() + this.publicationQueue.length = 0 + } + + private requestPageCount( + requestedPageCount: number, + fetchingNextPage: boolean, + ): Promise { + const generation = ++this.windowGeneration + const previousHasNextPage = this.getSnapshot().hasNextPage + this.pendingWindowGeneration = undefined + this.acquireInFlightLease() + + this.beginTransition() + this.isFetchingNextPage = fetchingNextPage + this.hasPaginationError = false + this.paginationError = undefined + if (fetchingNextPage) this.notify() + + let result: WindowResult + try { + result = this.activateLease(requestedPageCount) + } catch (error) { + this.isFetchingNextPage = false + this.hasPaginationError = true + this.paginationError = error + this.failedHasNextPage = previousHasNextPage + this.notify() + this.endTransition() + this.releaseInFlightLease() + return Promise.reject(error) + } + + if (result === true) { + if (!this.disposed && generation === this.windowGeneration) { + this.committedPageCount = requestedPageCount + this.isFetchingNextPage = false + this.failedHasNextPage = false + this.notify() + } + this.endTransition() + this.releaseInFlightLease() + return Promise.resolve() + } + + this.pendingWindowGeneration = generation + this.endTransition() + + return result + .then( + () => { + if (this.disposed || generation !== this.windowGeneration) return + this.beginTransition() + this.pendingWindowGeneration = undefined + this.committedPageCount = requestedPageCount + this.isFetchingNextPage = false + this.failedHasNextPage = false + this.notify() + this.endTransition() + }, + (error: unknown) => { + if (!this.disposed && generation === this.windowGeneration) { + this.beginTransition() + this.pendingWindowGeneration = undefined + this.isFetchingNextPage = false + this.hasPaginationError = true + this.paginationError = error + this.failedHasNextPage = previousHasNextPage + this.notify() + this.endTransition() + } + throw error + }, + ) + .finally(() => { + this.releaseInFlightLease() + }) + } + + private acquireInFlightLease(): void { + this.inFlightLeaseHolders++ + } + + private releaseInFlightLease(): void { + this.inFlightLeaseHolders-- + if (this.inFlightLeaseHolders === 0 && this.subscriptions.size === 0) { + this.deactivateLease() + } + } + + private activateLease(pageCount: number): WindowResult { + this.leaseGeneration++ + if (!this.coordinator || !this.collection) return true + this.leaseActive = true + return this.coordinator.request(this.lease, pageCount * this.pageSize + 1) + } + + private ensureLeaseActive(pageCount: number): WindowResult { + const minimumLimit = pageCount * this.pageSize + 1 + if ( + this.leaseActive && + this.coordinator?.isLeaseSatisfied(this.lease, minimumLimit) + ) { + return true + } + return this.activateLease(pageCount) + } + + private deactivateLease(): void { + if (!this.leaseActive || !this.coordinator) return + this.leaseGeneration++ + this.leaseActive = false + this.coordinator.release(this.lease) + } + + private trackAttachmentFailure( + result: Promise, + leaseGeneration: number, + ): void { + void result.catch((error: unknown) => { + if ( + this.disposed || + !this.leaseActive || + leaseGeneration !== this.leaseGeneration + ) { + return + } + this.beginTransition() + this.hasPaginationError = true + this.paginationError = error + this.failedHasNextPage = this.getComputedHasNextPage() + this.notify() + this.endTransition() + }) + } + + private getComputedHasNextPage(): boolean { + const snapshot: LiveQuerySnapshot = this.observer.getSnapshot() + return ( + snapshot.isEnabled && + Array.isArray(snapshot.data) && + snapshot.data.length > this.committedPageCount * this.pageSize + ) + } + + private onObserverNotify(): void { + this.notify() + } + + private beginTransition(): void { + this.transitionDepth++ + } + + private endTransition(): void { + this.transitionDepth-- + if (this.transitionDepth === 0 && this.transitionNeedsNotify) { + this.transitionNeedsNotify = false + this.publish() + } + } + + private notify(): void { + if (this.transitionDepth > 0) { + this.transitionNeedsNotify = true + return + } + this.publish() + } + + private publish(): void { + if (this.disposed || this.blockDelivery || this.subscriptions.size === 0) { + return + } + + this.publicationQueue.push({ targets: [...this.subscriptions] }) + if (this.dispatching) return + + this.dispatching = true + try { + while (this.publicationQueue.length > 0) { + const publication = this.publicationQueue.shift()! + for (const record of publication.targets) { + if (this.hasBeenDisposed()) return + if (!record.active) continue + record.listener() + } + } + } finally { + this.dispatching = false + } + } + + private hasBeenDisposed(): boolean { + return this.disposed + } +} + +/** + * Create an internal forward-window controller for an ordered live query. + * + * @internal This factory is unstable while RFC #1623 is being implemented. + */ +export function createLiveQueryWindowController< + T extends object, + TKey extends string | number, +>( + collection: Collection | null | undefined, + options: CreateLiveQueryWindowControllerOptions = {}, +): LiveQueryWindowController { + return new LiveQueryWindowControllerImpl(collection ?? null, options) +} diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts index e324119c6..e497150e5 100644 --- a/packages/db/src/query/live/collection-config-builder.ts +++ b/packages/db/src/query/live/collection-config-builder.ts @@ -123,6 +123,7 @@ export class CollectionConfigBuilder< public liveQueryCollection?: Collection private windowFn: ((options: WindowOptions) => void) | undefined + private readonly initialWindow: WindowOptions | undefined private currentWindow: WindowOptions | undefined private maybeRunGraphFn: (() => void) | undefined @@ -176,6 +177,12 @@ export class CollectionConfigBuilder< query: config.query, requireObjectResult: true, }) + this.initialWindow = this.query.orderBy?.length + ? { + offset: this.query.offset ?? 0, + limit: this.query.limit ?? Infinity, + } + : undefined this.collections = extractCollectionsFromQuery(this.query) const collectionAliasesById = extractCollectionAliases(this.query) @@ -275,38 +282,36 @@ export class CollectionConfigBuilder< throw new SetWindowRequiresOrderByError() } - this.currentWindow = options - this.windowFn(options) - this.maybeRunGraphFn?.() - - // Check if loading a subset was triggered - if (this.liveQueryCollection?.isLoadingSubset) { - // Loading was triggered, return a promise that resolves when it completes - return new Promise((resolve) => { - const unsubscribe = this.liveQueryCollection!.on( - `loadingSubset:change`, - (event) => { - if (!event.isLoadingSubset) { - unsubscribe() - resolve() - } - }, - ) - }) + const previousWindow = this.currentWindow ?? this.initialWindow + try { + this.windowFn(options) + this.maybeRunGraphFn?.() + this.currentWindow = options + } catch (error) { + if (previousWindow) { + try { + this.windowFn(previousWindow) + this.maybeRunGraphFn?.() + } catch { + // Recovery is best-effort; preserve the error from the requested + // window rather than replacing it with a rollback failure. + } + } + throw error } - // No loading was triggered - return true + return this.liveQueryCollection?._sync.waitForCurrentLoadSubset() ?? true } getWindow(): { offset: number; limit: number } | undefined { // Only return window if this is a windowed query (has orderBy and windowFn) - if (!this.windowFn || !this.currentWindow) { + const window = this.currentWindow ?? this.initialWindow + if (!this.windowFn || !window) { return undefined } return { - offset: this.currentWindow.offset ?? 0, - limit: this.currentWindow.limit ?? 0, + offset: window.offset ?? 0, + limit: window.limit ?? 0, } } @@ -655,6 +660,8 @@ export class CollectionConfigBuilder< // Clear current sync session state this.currentSyncConfig = undefined this.currentSyncState = undefined + this.maybeRunGraphFn = undefined + this.currentWindow = undefined // Clear all pending graph runs to prevent memory leaks from in-flight transactions // that may flush after the sync session ends @@ -709,6 +716,12 @@ export class CollectionConfigBuilder< this.optimizableOrderByCollections, (windowFn: (options: WindowOptions) => void) => { this.windowFn = windowFn + // `setWindow` mutates the compiled top-K operator, which is replaced + // whenever a cleaned-up live query compiles a fresh pipeline. Keep the + // desired window on the builder and replay it into each new operator. + if (this.currentWindow) { + windowFn(this.currentWindow) + } }, ) diff --git a/packages/db/src/query/live/collection-subscriber.ts b/packages/db/src/query/live/collection-subscriber.ts index 69c8220ad..9cfbf5852 100644 --- a/packages/db/src/query/live/collection-subscriber.ts +++ b/packages/db/src/query/live/collection-subscriber.ts @@ -234,11 +234,12 @@ export class CollectionSubscriber< const handleLoadSubsetResult = (result: Promise | true) => { if (result instanceof Promise) { this.pendingOrderedLoadPromise = result - result.finally(() => { + const finish = () => { if (this.pendingOrderedLoadPromise === result) { this.pendingOrderedLoadPromise = undefined } - }) + } + void result.then(finish, finish) } onLoadSubsetResult(result) } diff --git a/packages/db/tests/live-query-order-only-move.test.ts b/packages/db/tests/live-query-order-only-move.test.ts index 2637bf8a3..5d6bf5181 100644 --- a/packages/db/tests/live-query-order-only-move.test.ts +++ b/packages/db/tests/live-query-order-only-move.test.ts @@ -240,6 +240,27 @@ describe(`order-only move (RFC #1623 phase 4)`, () => { observer.dispose() }) + it(`exposes layout-only signals as empty public change batches`, async () => { + const source = makeSource() + const lq = await makeOrderedByAge(source) + const publications: Array> = [] + const subscription = lq.subscribeChanges( + (changes) => publications.push(changes), + { includeInitialState: false }, + ) + + source.utils.begin() + source.utils.write({ + type: `update`, + value: { id: `2`, name: `Bob`, age: 99 }, + }) + source.utils.commit() + await flush() + + expect(publications).toEqual([[]]) + subscription.unsubscribe() + }) + it(`bumps the layout revision on membership changes too`, async () => { const source = makeSource() const lq = await makeOrderedByAge(source) @@ -338,7 +359,7 @@ describe(`order-only move (RFC #1623 phase 4)`, () => { }) source.utils.commit() - expect(publications).toEqual([[]]) + expect(publications).toEqual([]) subscription.unsubscribe() }) @@ -381,11 +402,10 @@ describe(`order-only move (RFC #1623 phase 4)`, () => { await lq.preload() const childCollection = (lq.get(`p1`) as any).children + const observer = createLiveQueryObserver(childCollection) let notifications = 0 - const subscription = childCollection.subscribeChanges( - () => notifications++, - { includeInitialState: false }, - ) + observer.subscribe(() => notifications++) + notifications = 0 children.utils.begin() children.utils.write({ @@ -399,7 +419,7 @@ describe(`order-only move (RFC #1623 phase 4)`, () => { `c1`, ]) expect(notifications).toBe(1) - subscription.unsubscribe() + observer.dispose() }) // The includes flush is recursive, so the order-only-move handling must hold @@ -461,11 +481,10 @@ describe(`order-only move (RFC #1623 phase 4)`, () => { const teamCollection = (lq.get(`o1`) as any).teams const memberCollection = teamCollection.get(`t1`).members + const observer = createLiveQueryObserver(memberCollection) let notifications = 0 - const subscription = memberCollection.subscribeChanges( - () => notifications++, - { includeInitialState: false }, - ) + observer.subscribe(() => notifications++) + notifications = 0 // Move m1 behind m2 (position 1 -> 3); projected { id, name } unchanged. members.utils.begin() @@ -480,6 +499,6 @@ describe(`order-only move (RFC #1623 phase 4)`, () => { `m1`, ]) expect(notifications).toBe(1) - subscription.unsubscribe() + observer.dispose() }) }) diff --git a/packages/db/tests/live-query-window-controller.test.ts b/packages/db/tests/live-query-window-controller.test.ts new file mode 100644 index 000000000..c67e5d462 --- /dev/null +++ b/packages/db/tests/live-query-window-controller.test.ts @@ -0,0 +1,898 @@ +import { describe, expect, it, vi } from 'vitest' +import { BTreeIndex } from '../src/index.js' +import { createCollection } from '../src/collection/index.js' +import { createLiveQueryCollection } from '../src/query/live-query-collection.js' +import { LIVE_QUERY_INTERNAL } from '../src/query/live/internal.js' +import { LiveQueryWindowControllerDisposedError } from '../src/errors.js' +import { createLiveQueryWindowController } from '../src/live-query-window-controller.js' +import { mockSyncCollectionOptions } from './utils.js' + +interface Row { + id: string + n: number +} + +const ROWS: Array = [1, 2, 3, 4, 5].map((n) => ({ id: String(n), n })) + +let seq = 0 +function makeSource(initialData: Array = ROWS) { + return createCollection( + mockSyncCollectionOptions({ + id: `window-ctrl-${seq++}`, + getKey: (r) => r.id, + initialData, + }), + ) +} + +/** Ordered live query with page 1's peek-ahead window baked in, as the React adapter builds it. */ +function makeOrderedLiveQuery( + source: ReturnType, + pageSize: number, +) { + return createLiveQueryCollection({ + query: (q) => + q + .from({ r: source }) + .orderBy(({ r }) => r.n, `asc`) + .limit(pageSize + 1) + .offset(0) + .select(({ r }) => ({ id: r.id, n: r.n })), + startSync: true, + gcTime: 1, + }) +} + +const flush = () => new Promise((r) => setTimeout(r, 0)) + +const ids = (snap: { data: ReadonlyArray }) => snap.data.map((r) => r.id) + +describe(`createLiveQueryWindowController`, () => { + it(`exposes the first page with a peek-ahead hasNextPage`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + await flush() + + const snap = controller.getSnapshot() + expect(ids(snap)).toEqual([`1`, `2`]) + expect(snap.pages.map((p) => p.map((r) => r.id))).toEqual([[`1`, `2`]]) + expect(snap.pageParams).toEqual([0]) + expect(snap.hasNextPage).toBe(true) + expect(snap.isFetchingNextPage).toBe(false) + controller.dispose() + }) + + it(`loads further pages via fetchNextPage until the source is exhausted`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + await flush() + + controller.fetchNextPage() + await flush() + let snap = controller.getSnapshot() + expect(ids(snap)).toEqual([`1`, `2`, `3`, `4`]) + expect(snap.pages.map((p) => p.map((r) => r.id))).toEqual([ + [`1`, `2`], + [`3`, `4`], + ]) + expect(snap.pageParams).toEqual([0, 1]) + expect(snap.hasNextPage).toBe(true) + + controller.fetchNextPage() + await flush() + snap = controller.getSnapshot() + // 5 rows total; the 3rd page is a partial page and there is no peek row. + expect(ids(snap)).toEqual([`1`, `2`, `3`, `4`, `5`]) + expect(snap.pages.map((p) => p.map((r) => r.id))).toEqual([ + [`1`, `2`], + [`3`, `4`], + [`5`], + ]) + expect(snap.hasNextPage).toBe(false) + controller.dispose() + }) + + it(`fetchNextPage is a no-op when there is no next page`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 10) // pageSize > row count + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 10, + }) + controller.subscribe(() => {}) + await lq.preload() + await flush() + expect(controller.getSnapshot().hasNextPage).toBe(false) + + controller.fetchNextPage() + await flush() + expect(ids(controller.getSnapshot())).toEqual([`1`, `2`, `3`, `4`, `5`]) + expect(controller.getSnapshot().pages).toHaveLength(1) + controller.dispose() + }) + + it(`represents an empty enabled query as one empty page`, async () => { + const lq = makeOrderedLiveQuery(makeSource([]), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + + const snapshot = controller.getSnapshot() + expect(snapshot.isEnabled).toBe(true) + expect(snapshot.data).toEqual([]) + expect(snapshot.pages).toEqual([[]]) + expect(snapshot.hasNextPage).toBe(false) + controller.dispose() + }) + + it(`uses the default page size when pageSize is zero`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 0, + }) + controller.subscribe(() => {}) + await lq.preload() + + const snapshot = controller.getSnapshot() + expect(ids(snapshot)).toEqual([`1`, `2`, `3`, `4`, `5`]) + expect(snapshot.pages).toHaveLength(1) + expect(snapshot.hasNextPage).toBe(false) + controller.dispose() + }) + + it(`reset returns to the first page`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + await flush() + + controller.fetchNextPage() + await flush() + expect(controller.getSnapshot().pages).toHaveLength(2) + + controller.reset() + await flush() + const snap = controller.getSnapshot() + expect(ids(snap)).toEqual([`1`, `2`]) + expect(snap.pages).toHaveLength(1) + controller.dispose() + }) + + it(`notifies subscribers on data changes and page changes`, async () => { + const source = makeSource() + const lq = makeOrderedLiveQuery(source, 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let notifications = 0 + controller.subscribe(() => notifications++) + await lq.preload() + await flush() + + notifications = 0 + controller.fetchNextPage() + await flush() + expect(notifications).toBeGreaterThan(0) + controller.dispose() + }) + + it(`retries a window that throws synchronously`, () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const setWindow = vi + .spyOn(lq.utils, `setWindow`) + .mockImplementationOnce(() => { + throw new Error(`window failed`) + }) + .mockReturnValue(true) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + + expect(() => controller.subscribe(() => {})).toThrow(`window failed`) + const unsubscribe = controller.subscribe(() => {}) + + expect(setWindow).toHaveBeenCalledTimes(2) + unsubscribe() + controller.dispose() + }) + + it(`restores the initial operator window when a graph run throws`, () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const builder = lq.utils[LIVE_QUERY_INTERNAL].getBuilder() + const originalWindowFn = Reflect.get(builder, `windowFn`) as (options: { + offset?: number + limit?: number + }) => void + const windowFn = vi.fn(originalWindowFn) + const requestedError = new Error(`requested window failed`) + const maybeRunGraph = vi + .fn() + .mockImplementationOnce(() => { + throw requestedError + }) + .mockImplementationOnce(() => { + throw new Error(`rollback failed`) + }) + Reflect.set(builder, `windowFn`, windowFn) + Reflect.set(builder, `maybeRunGraphFn`, maybeRunGraph) + + expect(() => lq.utils.setWindow({ offset: 0, limit: 5 })).toThrow( + requestedError, + ) + expect(windowFn).toHaveBeenNthCalledWith(1, { offset: 0, limit: 5 }) + expect(windowFn).toHaveBeenNthCalledWith(2, { offset: 0, limit: 3 }) + expect(maybeRunGraph).toHaveBeenCalledTimes(2) + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + }) + + it(`keeps the committed page retryable when a window load rejects`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + await flush() + expect(controller.getSnapshot().hasNextPage).toBe(true) + + const failure = new Error(`load failed`) + vi.spyOn(lq.utils, `setWindow`).mockRejectedValueOnce(failure) + + await expect(Promise.resolve(controller.fetchNextPage())).rejects.toThrow( + `load failed`, + ) + + expect(controller.getSnapshot().pages).toHaveLength(1) + expect(controller.getSnapshot().hasNextPage).toBe(true) + expect((controller.getSnapshot() as { error?: unknown }).error).toBe( + failure, + ) + + await controller.fetchNextPage() + expect(controller.getSnapshot().pages).toHaveLength(2) + controller.dispose() + }) + + it(`clears a preload error after a successful retry`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const failure = new Error(`preload failed`) + vi.spyOn(lq.utils, `setWindow`).mockRejectedValueOnce(failure) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + + await expect(controller.preload()).rejects.toBe(failure) + expect(controller.getSnapshot().error).toBe(failure) + + await controller.preload() + expect(controller.getSnapshot().isError).toBe(false) + expect(controller.getSnapshot().error).toBeUndefined() + controller.dispose() + }) + + it(`publishes one coherent loading snapshot and one settled snapshot`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + const snapshots: Array<{ pages: number; fetching: boolean }> = [] + controller.subscribe(() => { + const snapshot = controller.getSnapshot() + snapshots.push({ + pages: snapshot.pages.length, + fetching: snapshot.isFetchingNextPage, + }) + }) + await lq.preload() + await flush() + snapshots.length = 0 + + let resolveWindow!: () => void + vi.spyOn(lq.utils, `setWindow`).mockReturnValueOnce( + new Promise((resolve) => { + resolveWindow = resolve + }), + ) + + const fetch = Promise.resolve(controller.fetchNextPage()) + expect(snapshots).toEqual([{ pages: 1, fetching: true }]) + + resolveWindow() + await fetch + expect(snapshots).toEqual([ + { pages: 1, fetching: true }, + { pages: 2, fetching: false }, + ]) + controller.dispose() + }) + + it(`does not shrink the physical window when preload overlaps a page fetch`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + + const originalSetWindow = lq.utils.setWindow.bind(lq.utils) + let resolveExpansion!: () => void + vi.spyOn(lq.utils, `setWindow`).mockImplementation((options) => { + const result = originalSetWindow(options) + if (options.limit !== 5) return result + return new Promise((resolve) => { + resolveExpansion = resolve + }) + }) + + const expansion = controller.fetchNextPage() + const preload = controller.preload() + resolveExpansion() + await Promise.all([expansion, preload]) + + expect(controller.getSnapshot().pages).toHaveLength(2) + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 5 }) + controller.dispose() + }) + + it(`publishes source changes while a page fetch is pending`, async () => { + const source = makeSource() + const lq = makeOrderedLiveQuery(source, 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let notifications = 0 + controller.subscribe(() => notifications++) + await lq.preload() + notifications = 0 + + vi.spyOn(lq.utils, `setWindow`).mockReturnValueOnce( + new Promise(() => {}), + ) + void controller.fetchNextPage() + notifications = 0 + + source.utils.begin() + source.utils.write({ + type: `update`, + value: { id: `1`, n: 0 }, + }) + source.utils.commit() + await flush() + + expect(notifications).toBeGreaterThan(0) + expect(controller.getSnapshot().data[0]).toMatchObject({ id: `1`, n: 0 }) + controller.dispose() + }) + + it(`surfaces a real async subset-load failure from setWindow`, async () => { + const remoteRows = [...ROWS] + let rejectLoads = false + const failure = new Error(`remote page failed`) + const source = createCollection({ + id: `window-ctrl-rejecting-source-${seq++}`, + getKey: (row) => row.id, + syncMode: `on-demand`, + startSync: true, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: (options) => + new Promise((resolve, reject) => { + queueMicrotask(() => { + if (rejectLoads) { + reject(failure) + return + } + begin() + remoteRows.slice(0, options.limit).forEach((row) => { + write({ type: `insert`, value: row }) + }) + commit() + resolve() + }) + }), + } + }, + }, + }) + const lq = makeOrderedLiveQuery(source, 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await controller.preload() + expect(controller.getSnapshot().hasNextPage).toBe(true) + + rejectLoads = true + await expect(controller.fetchNextPage()).rejects.toBe(failure) + expect(controller.getSnapshot().pages).toHaveLength(1) + expect(controller.getSnapshot().error).toBe(failure) + controller.dispose() + }) + + it(`reset supersedes an in-flight page expansion`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + + let resolveExpansion!: () => void + const setWindow = vi + .spyOn(lq.utils, `setWindow`) + .mockReturnValueOnce( + new Promise((resolve) => { + resolveExpansion = resolve + }), + ) + .mockReturnValueOnce(true) + + const expansion = controller.fetchNextPage() + expect(controller.getSnapshot().isFetchingNextPage).toBe(true) + + await controller.reset() + expect(controller.getSnapshot().pages).toHaveLength(1) + expect(controller.getSnapshot().isFetchingNextPage).toBe(false) + expect(setWindow).toHaveBeenNthCalledWith(1, { offset: 0, limit: 5 }) + expect(setWindow).toHaveBeenNthCalledWith(2, { offset: 0, limit: 3 }) + + resolveExpansion() + await expansion + expect(controller.getSnapshot().pages).toHaveLength(1) + controller.dispose() + }) + + it(`retains an unsubscribed lease until overlapping requests settle`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + await lq.preload() + + const originalSetWindow = lq.utils.setWindow.bind(lq.utils) + const resolvers: Array<() => void> = [] + const setWindow = vi + .spyOn(lq.utils, `setWindow`) + .mockImplementation((options) => { + originalSetWindow(options) + if (resolvers.length >= 2) return true + return new Promise((resolve) => resolvers.push(resolve)) + }) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + + const expansion = controller.fetchNextPage() + const reset = controller.reset() + expect(setWindow).toHaveBeenCalledTimes(2) + + resolvers[0]!() + await expansion + + const competingController = createLiveQueryWindowController( + lq as any, + { pageSize: 1 }, + ) + competingController.subscribe(() => {}) + expect(setWindow).toHaveBeenCalledTimes(2) + + resolvers[1]!() + await reset + expect(controller.getSnapshot().pages).toHaveLength(1) + competingController.dispose() + controller.dispose() + }) + + it(`replays the desired window after collection cleanup`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + const unsubscribe = controller.subscribe(() => {}) + await lq.preload() + + await controller.fetchNextPage() + await flush() + expect(ids(controller.getSnapshot())).toEqual([`1`, `2`, `3`, `4`]) + + unsubscribe() + await lq.cleanup() + + controller.subscribe(() => {}) + await lq.preload() + await flush() + + expect(ids(controller.getSnapshot())).toEqual([`1`, `2`, `3`, `4`]) + expect(controller.getSnapshot().hasNextPage).toBe(true) + controller.dispose() + }) + + it(`establishes the desired window before preload`, async () => { + const source = makeSource() + const lq = createLiveQueryCollection({ + query: (q) => + q + .from({ r: source }) + .orderBy(({ r }) => r.n, `asc`) + .limit(2) + .select(({ r }) => ({ id: r.id, n: r.n })), + gcTime: 1, + }) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + + await controller.preload() + + expect(controller.getSnapshot().hasNextPage).toBe(true) + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + controller.dispose() + }) + + it(`coordinates the physical window across multiple controllers`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const larger = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + const smaller = createLiveQueryWindowController(lq as any, { + pageSize: 1, + }) + + larger.subscribe(() => {}) + smaller.subscribe(() => {}) + await lq.preload() + await flush() + + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + expect(ids(larger.getSnapshot())).toEqual([`1`, `2`]) + expect(larger.getSnapshot().hasNextPage).toBe(true) + + await larger.fetchNextPage() + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 5 }) + + await smaller.fetchNextPage() + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 5 }) + + larger.dispose() + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + smaller.dispose() + }) + + it(`rolls a failed lease request back to its committed window`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + + const failure = new Error(`window failed`) + vi.spyOn(lq.utils, `setWindow`).mockRejectedValueOnce(failure) + await expect(controller.fetchNextPage()).rejects.toBe(failure) + + const smaller = createLiveQueryWindowController(lq as any, { + pageSize: 1, + }) + smaller.subscribe(() => {}) + + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + smaller.dispose() + controller.dispose() + }) + + it(`restores the remaining lease after a pending larger lease is released`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const keeper = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + keeper.subscribe(() => {}) + await lq.preload() + await keeper.fetchNextPage() + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 5 }) + + const transient = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + transient.subscribe(() => {}) + await transient.fetchNextPage() + + const originalSetWindow = lq.utils.setWindow.bind(lq.utils) + let resolveExpansion!: () => void + vi.spyOn(lq.utils, `setWindow`).mockImplementation((options) => { + const result = originalSetWindow(options) + if (options.limit !== 7) return result + return new Promise((resolve) => { + resolveExpansion = resolve + }) + }) + + const expansion = transient.fetchNextPage() + transient.dispose() + + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 5 }) + resolveExpansion() + await expansion + keeper.dispose() + }) + + it(`repairs an externally moved physical window`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + + await lq.utils.setWindow({ offset: 1, limit: 3 }) + expect(lq.utils.getWindow()).toEqual({ offset: 1, limit: 3 }) + + await controller.preload() + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + controller.dispose() + }) + + it(`restores the query's initial window after the last lease is released`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + const unsubscribe = controller.subscribe(() => {}) + await lq.preload() + await controller.fetchNextPage() + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 5 }) + + unsubscribe() + controller.dispose() + await lq.cleanup() + await lq.preload() + + expect(lq.utils.getWindow()).toEqual({ offset: 0, limit: 3 }) + expect(lq.toArray).toHaveLength(3) + }) + + it(`ignores a failed attachment superseded by a new lease`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + await lq.preload() + + let rejectFirst!: (error: Error) => void + vi.spyOn(lq.utils, `setWindow`) + .mockReturnValueOnce( + new Promise((_, reject) => { + rejectFirst = reject + }), + ) + .mockReturnValue(true) + + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + const unsubscribe = controller.subscribe(() => {}) + unsubscribe() + controller.subscribe(() => {}) + + rejectFirst(new Error(`stale attachment failed`)) + await flush() + + expect(controller.getSnapshot().isError).toBe(false) + expect(controller.getSnapshot().error).toBeUndefined() + controller.dispose() + }) + + it(`does not notify synchronously while subscribing by default`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + await lq.preload() + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let subscribing = true + let notifiedWhileSubscribing = false + + const unsubscribe = controller.subscribe(() => { + if (subscribing) notifiedWhileSubscribing = true + }) + subscribing = false + + expect(notifiedWhileSubscribing).toBe(false) + unsubscribe() + controller.dispose() + }) + + it(`keeps duplicate callback subscriptions independent`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let notifications = 0 + const listener = () => notifications++ + const unsubscribeFirst = controller.subscribe(listener) + const unsubscribeSecond = controller.subscribe(listener) + await lq.preload() + await flush() + + notifications = 0 + unsubscribeFirst() + controller.fetchNextPage() + await flush() + + expect(notifications).toBeGreaterThan(0) + unsubscribeSecond() + controller.dispose() + }) + + it(`does not deliver an in-flight notification to a late subscriber`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let publishing = false + let lateNotifications = 0 + let unsubscribeLate: (() => void) | undefined + const unsubscribeFirst = controller.subscribe(() => { + if (publishing && !unsubscribeLate) { + unsubscribeLate = controller.subscribe(() => lateNotifications++) + } + }) + await lq.preload() + await flush() + vi.spyOn(lq.utils, `setWindow`).mockReturnValue(true) + + publishing = true + controller.fetchNextPage() + publishing = false + + expect(lateNotifications).toBe(0) + unsubscribeLate?.() + unsubscribeFirst() + controller.dispose() + }) + + it(`rejects every subscription after disposal`, () => { + const controller = createLiveQueryWindowController( + makeOrderedLiveQuery(makeSource(), 2) as any, + { pageSize: 2 }, + ) + controller.dispose() + + expect(() => controller.subscribe(() => {})).toThrow( + LiveQueryWindowControllerDisposedError, + ) + expect(() => controller.subscribe(() => {})).toThrow( + LiveQueryWindowControllerDisposedError, + ) + }) + + it(`releases subscriptions when disposed by a listener`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + + controller.subscribe(() => controller.dispose()) + await lq.preload() + await controller.fetchNextPage() + + expect(lq.subscriberCount).toBe(0) + }) + + it(`stops an in-flight publication when a listener disposes`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let publishing = false + let secondListenerNotifications = 0 + controller.subscribe(() => { + if (publishing) controller.dispose() + }) + controller.subscribe(() => { + if (publishing) secondListenerNotifications++ + }) + await lq.preload() + await flush() + vi.spyOn(lq.utils, `setWindow`).mockReturnValue(true) + + publishing = true + controller.fetchNextPage() + publishing = false + + expect(secondListenerNotifications).toBe(0) + }) + + it(`skips a listener unsubscribed during an in-flight publication`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + let publishing = false + let secondListenerNotifications = 0 + let unsubscribeSecond = () => {} + controller.subscribe(() => { + if (publishing) unsubscribeSecond() + }) + unsubscribeSecond = controller.subscribe(() => { + if (publishing) secondListenerNotifications++ + }) + await lq.preload() + await flush() + vi.spyOn(lq.utils, `setWindow`).mockReturnValue(true) + + publishing = true + await controller.fetchNextPage() + publishing = false + + expect(secondListenerNotifications).toBe(0) + controller.dispose() + }) + + it(`returns a stable snapshot identity when nothing changed`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + await flush() + expect(controller.getSnapshot()).toBe(controller.getSnapshot()) + controller.dispose() + }) + + it(`derives status flags from a pagination error status`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + await lq.preload() + + vi.spyOn(lq.utils, `setWindow`).mockRejectedValueOnce( + new Error(`window failed`), + ) + await expect(controller.fetchNextPage()).rejects.toThrow(`window failed`) + + expect(controller.getSnapshot()).toMatchObject({ + status: `error`, + isLoading: false, + isReady: false, + isIdle: false, + isError: true, + isCleanedUp: false, + }) + controller.dispose() + }) + + it(`normalizes a NaN initial page count to one page`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + initialPageCount: Number.NaN, + }) + controller.subscribe(() => {}) + await lq.preload() + + expect(controller.getSnapshot().pages).toHaveLength(1) + expect(ids(controller.getSnapshot())).toEqual([`1`, `2`]) + controller.dispose() + }) + + it(`represents a disabled controller (null collection)`, () => { + const controller = createLiveQueryWindowController(null) + const snap = controller.getSnapshot() + expect(snap.isEnabled).toBe(false) + expect(snap.data).toEqual([]) + expect(snap.hasNextPage).toBe(false) + expect(snap.pages).toEqual([]) + controller.dispose() + }) +}) diff --git a/packages/react-db/src/useLiveInfiniteQuery.ts b/packages/react-db/src/useLiveInfiniteQuery.ts index 99c77c739..c3d4c629c 100644 --- a/packages/react-db/src/useLiveInfiniteQuery.ts +++ b/packages/react-db/src/useLiveInfiniteQuery.ts @@ -1,23 +1,39 @@ -import { useCallback, useEffect, useMemo, useRef, useState } from 'react' -import { CollectionImpl } from '@tanstack/db' -import { useLiveQuery } from './useLiveQuery' +import { useCallback, useRef, useSyncExternalStore } from 'react' +import { + CollectionImpl, + createLiveQueryCollection, + createLiveQueryWindowController, + deepEquals, +} from '@tanstack/db' +// Type-only: used in `ReturnType` in UseLiveInfiniteQueryReturn. +import type { useLiveQuery } from './useLiveQuery' import type { Collection, Context, InferResultType, InitialQueryBuilder, - LiveQueryCollectionUtils, + LiveQueryWindowController, NonSingleResult, QueryBuilder, } from '@tanstack/db' -/** - * Type guard to check if utils object has setWindow method (LiveQueryCollectionUtils) - */ -function isLiveQueryCollectionUtils( - utils: unknown, -): utils is LiveQueryCollectionUtils { - return typeof (utils as any).setWindow === `function` +// Live queries created here are cleaned up immediately (0 disables GC). +const DEFAULT_GC_TIME_MS = 1 + +type WindowedCollection = Collection & { + utils: { + setWindow: (options: { + offset: number + limit: number + }) => true | Promise + } +} + +/** Type guard: does this collection expose `setWindow` (i.e. has an orderBy)? */ +function hasSetWindow( + collection: Collection, +): collection is WindowedCollection { + return typeof collection.utils?.setWindow === `function` } export type UseLiveInfiniteQueryConfig = { @@ -46,8 +62,13 @@ export type UseLiveInfiniteQueryReturn = Omit< fetchNextPage: () => void hasNextPage: boolean isFetchingNextPage: boolean + error: unknown } +type EnabledLiveQueryReturn = ReturnType< + typeof useLiveQuery +> + /** * Create an infinite query using a query function with live updates * @@ -151,180 +172,135 @@ export function useLiveInfiniteQuery( ) } - // Track how many pages have been loaded - const [loadedPageCount, setLoadedPageCount] = useState(1) - const [isFetchingNextPage, setIsFetchingNextPage] = useState(false) - - // Track collection instance and whether we've validated it (only for pre-created collections) - const collectionRef = useRef(isCollection ? queryFnOrCollection : null) - const hasValidatedCollectionRef = useRef(false) - - // Track deps for query functions (stringify for comparison) - let depsKey: string - try { - depsKey = JSON.stringify(deps) - } catch { - throw new Error( - `useLiveInfiniteQuery: dependency array contains values that cannot be serialized (e.g. circular references). ` + - `Ensure all dependency values are JSON-serializable.`, - ) - } - const prevDepsKeyRef = useRef(depsKey) - - // Reset pagination when inputs change - useEffect(() => { - let shouldReset = false - + const collectionRef = useRef | null>(null) + const controllerRef = useRef | null>(null) + const configRef = useRef(null) + const depsRef = useRef | null>(null) + const pageSizeRef = useRef(pageSize) + const initialPageParamRef = useRef(initialPageParam) + const validatedCollectionRef = useRef(null) + const inputKind = isCollection ? `collection` : `query` + const inputKindRef = useRef(null) + const previousInputKind = inputKindRef.current + + const dependenciesChanged = + !isCollection && + (depsRef.current === null || + depsRef.current.length !== deps.length || + depsRef.current.some((dep, index) => dep !== deps[index])) + const dependenciesStructurallyEqual = + !isCollection && + depsRef.current !== null && + deepEquals(depsRef.current, deps) + const needsNewCollection = + !collectionRef.current || + inputKindRef.current !== inputKind || + (isCollection && configRef.current !== queryFnOrCollection) || + dependenciesChanged + const pageShapeChanged = + pageSizeRef.current !== pageSize || + initialPageParamRef.current !== initialPageParam + const needsNewController = + !controllerRef.current || needsNewCollection || pageShapeChanged + + if (needsNewCollection) { + inputKindRef.current = inputKind if (isCollection) { - // Reset if collection instance changed - if (collectionRef.current !== queryFnOrCollection) { - collectionRef.current = queryFnOrCollection - hasValidatedCollectionRef.current = false - shouldReset = true - } - } else { - // Reset if deps changed (for query functions) - if (prevDepsKeyRef.current !== depsKey) { - prevDepsKeyRef.current = depsKey - shouldReset = true - } - } - - if (shouldReset) { - setLoadedPageCount(1) - } - }, [isCollection, queryFnOrCollection, depsKey]) - - // Create a live query with initial limit and offset - // Either pass collection directly or wrap query function - // Use pageSize + 1 for peek-ahead detection (to know if there are more pages) - const queryResult = isCollection - ? useLiveQuery(queryFnOrCollection) - : useLiveQuery( - (q) => - queryFnOrCollection(q) - .limit(pageSize + 1) - .offset(0), - deps, - ) - - // Adjust window when pagination changes - useEffect(() => { - const utils = queryResult.collection.utils - const expectedOffset = 0 - const expectedLimit = loadedPageCount * pageSize + 1 // +1 for peek ahead - - // Check if collection has orderBy (required for setWindow) - if (!isLiveQueryCollectionUtils(utils)) { - // For pre-created collections, throw an error if no orderBy - if (isCollection) { + const collection = queryFnOrCollection as Collection + if (!hasSetWindow(collection)) { throw new Error( `useLiveInfiniteQuery: Pre-created live query collection must have an orderBy clause for infinite pagination to work. ` + `Please add .orderBy() to your createLiveQueryCollection query.`, ) } - return - } - - // For pre-created collections, validate window on first check - if (isCollection && !hasValidatedCollectionRef.current) { - const currentWindow = utils.getWindow() - if ( - currentWindow && - (currentWindow.offset !== expectedOffset || - currentWindow.limit !== expectedLimit) - ) { - console.warn( - `useLiveInfiniteQuery: Pre-created collection has window {offset: ${currentWindow.offset}, limit: ${currentWindow.limit}} ` + - `but hook expects {offset: ${expectedOffset}, limit: ${expectedLimit}}. Adjusting window now.`, - ) + // Warn once per collection instance if its current window doesn't match + // the first page the hook is about to enforce. + if (validatedCollectionRef.current !== collection) { + validatedCollectionRef.current = collection + const currentWindow = collection.utils.getWindow?.() + if ( + currentWindow && + (currentWindow.offset !== 0 || currentWindow.limit !== pageSize + 1) + ) { + console.warn( + `useLiveInfiniteQuery: Pre-created collection has window {offset: ${currentWindow.offset}, limit: ${currentWindow.limit}} ` + + `but the hook expects {offset: 0, limit: ${pageSize + 1}}. Adjusting window now.`, + ) + } } - hasValidatedCollectionRef.current = true - } - - // For query functions, wait until collection is ready - if (!isCollection && !queryResult.isReady) return - - // Adjust the window - let cancelled = false - const result = utils.setWindow({ - offset: expectedOffset, - limit: expectedLimit, - }) - - if (result !== true) { - setIsFetchingNextPage(true) - result - .catch((error: unknown) => { - if (!cancelled) - console.error(`useLiveInfiniteQuery: setWindow failed:`, error) - }) - .finally(() => { - if (!cancelled) setIsFetchingNextPage(false) - }) + collectionRef.current = collection + configRef.current = queryFnOrCollection } else { - setIsFetchingNextPage(false) - } - - return () => { - cancelled = true - } - }, [ - isCollection, - queryResult.collection, - queryResult.isReady, - loadedPageCount, - pageSize, - ]) - - // Split the data array into pages and determine if there's a next page - const { pages, pageParams, hasNextPage, flatData } = useMemo(() => { - const dataArray = ( - Array.isArray(queryResult.data) ? queryResult.data : [] - ) as InferResultType - const totalItemsRequested = loadedPageCount * pageSize - - // Check if we have more data than requested (the peek ahead item) - const hasMore = dataArray.length > totalItemsRequested - - // Build pages array (without the peek ahead item) - const pagesResult: Array[number]>> = [] - const pageParamsResult: Array = [] - - for (let i = 0; i < loadedPageCount; i++) { - const pageData = dataArray.slice(i * pageSize, (i + 1) * pageSize) - pagesResult.push(pageData) - pageParamsResult.push(initialPageParam + i) + // Wrap the query with the first page's peek-ahead window; the controller + // grows the limit from here via setWindow. + collectionRef.current = createLiveQueryCollection({ + query: (q: InitialQueryBuilder) => + queryFnOrCollection(q) + .limit(pageSize + 1) + .offset(0), + // Construction happens during render. Synchronization starts only when + // useSyncExternalStore commits the controller subscription. + startSync: false, + gcTime: DEFAULT_GC_TIME_MS, + }) + depsRef.current = [...deps] } + } - // Flatten the pages for the data return (without peek ahead item) - const flatDataResult = dataArray.slice( - 0, - totalItemsRequested, - ) as InferResultType + if (needsNewController) { + const previousController = controllerRef.current + const canPreservePageCount = + previousController !== null && + (!needsNewCollection || + (previousInputKind === `query` && dependenciesStructurallyEqual)) + const initialPageCount = canPreservePageCount + ? Math.max(1, previousController.getSnapshot().pages.length) + : 1 + pageSizeRef.current = pageSize + initialPageParamRef.current = initialPageParam + controllerRef.current = createLiveQueryWindowController( + collectionRef.current, + { + pageSize, + initialPageParam, + initialPageCount, + }, + ) + } + const controller = controllerRef.current! - return { - pages: pagesResult, - pageParams: pageParamsResult, - hasNextPage: hasMore, - flatData: flatDataResult, - } - }, [queryResult.data, loadedPageCount, pageSize, initialPageParam]) + const subscribe = useCallback( + (onStoreChange: () => void) => controller.subscribe(onStoreChange), + [controller], + ) + const getSnapshot = useCallback(() => controller.getSnapshot(), [controller]) + const snapshot = useSyncExternalStore(subscribe, getSnapshot) - // Fetch next page const fetchNextPage = useCallback(() => { - if (!hasNextPage || isFetchingNextPage) return - - setLoadedPageCount((prev) => prev + 1) - }, [hasNextPage, isFetchingNextPage]) + void controller.fetchNextPage().catch(() => { + // Pagination errors are exposed through the controller snapshot. The + // hook's void callback has no promise error channel, so consume it here. + }) + }, [controller]) return { - ...queryResult, - data: flatData, - pages, - pageParams, + data: snapshot.data as InferResultType, + state: snapshot.state as EnabledLiveQueryReturn[`state`], + status: snapshot.status as EnabledLiveQueryReturn[`status`], + isLoading: snapshot.isLoading, + isReady: snapshot.isReady, + isIdle: snapshot.isIdle, + isError: snapshot.isError, + isCleanedUp: snapshot.isCleanedUp, + collection: + snapshot.collection as EnabledLiveQueryReturn[`collection`], + isEnabled: + snapshot.isEnabled as EnabledLiveQueryReturn[`isEnabled`], + pages: snapshot.pages as Array[number]>>, + pageParams: snapshot.pageParams as Array, fetchNextPage, - hasNextPage, - isFetchingNextPage, - } as UseLiveInfiniteQueryReturn + hasNextPage: snapshot.hasNextPage, + isFetchingNextPage: snapshot.isFetchingNextPage, + error: snapshot.error, + } } diff --git a/packages/react-db/tests/useLiveInfiniteQuery.test.tsx b/packages/react-db/tests/useLiveInfiniteQuery.test.tsx index 9aa63244e..77e33542f 100644 --- a/packages/react-db/tests/useLiveInfiniteQuery.test.tsx +++ b/packages/react-db/tests/useLiveInfiniteQuery.test.tsx @@ -1,11 +1,17 @@ -import { describe, expect, it } from 'vitest' -import { act, renderHook, waitFor } from '@testing-library/react' -import { createCollection, createLiveQueryCollection, eq } from '@tanstack/db' -import { BTreeIndex } from '@tanstack/db' +import { describe, expect, it, vi } from 'vitest' +import { act, render, renderHook, waitFor } from '@testing-library/react' +import { Suspense } from 'react' +import { + BTreeIndex, + createCollection, + createLiveQueryCollection, + eq, +} from '@tanstack/db' import { useLiveInfiniteQuery } from '../src/useLiveInfiniteQuery' import { mockSyncCollectionOptions } from '../../db/tests/utils' import { createFilterFunctionFromExpression } from '../../db/src/collection/change-events' -import type { LoadSubsetOptions } from '@tanstack/db' +import type { InitialQueryBuilder, LoadSubsetOptions } from '@tanstack/db' +import type { ReactNode } from 'react' type Post = { id: string @@ -104,6 +110,68 @@ function createOnDemandCollection(opts: OnDemandCollectionOptions) { } describe(`useLiveInfiniteQuery`, () => { + it(`does not activate a query-function collection for an abandoned render`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `abandoned-infinite-query`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const never = new Promise(() => {}) + + function AbandonedQuery(): ReactNode { + useLiveInfiniteQuery( + (q) => + q + .from({ post: source }) + .orderBy(({ post }) => post.createdAt, `desc`), + { pageSize: 3 }, + ) + throw never + } + + const rendered = render( + + + , + ) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(source.subscriberCount).toBe(0) + rendered.unmount() + }) + + it(`does not activate a supplied collection for an abandoned render`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `abandoned-supplied-infinite-query`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const liveQuery = createLiveQueryCollection({ + query: (q) => + q.from({ post: source }).orderBy(({ post }) => post.createdAt, `desc`), + }) + const never = new Promise(() => {}) + + function AbandonedQuery(): ReactNode { + useLiveInfiniteQuery(liveQuery, { pageSize: 3 }) + throw never + } + + const rendered = render( + + + , + ) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(source.subscriberCount).toBe(0) + rendered.unmount() + }) + it(`should fetch initial page of data`, async () => { const posts = createMockPosts(50) const collection = createCollection( @@ -695,6 +763,235 @@ describe(`useLiveInfiniteQuery`, () => { }) }) + it(`re-windows and re-slices when pageSize changes at runtime`, async () => { + const posts = createMockPosts(50) + const collection = createCollection( + mockSyncCollectionOptions({ + autoIndex: `eager`, + id: `pagesize-change-test`, + getKey: (post: Post) => post.id, + initialData: posts, + }), + ) + + const { result, rerender } = renderHook( + ({ pageSize }: { pageSize: number }) => + useLiveInfiniteQuery( + (q) => + q + .from({ posts: collection }) + .orderBy(({ posts: p }) => p.createdAt, `desc`), + { pageSize }, + ), + { initialProps: { pageSize: 5 } }, + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + expect(result.current.data).toHaveLength(5) + expect(result.current.pages[0]).toHaveLength(5) + + act(() => result.current.fetchNextPage()) + await waitFor(() => expect(result.current.pages).toHaveLength(2)) + act(() => result.current.fetchNextPage()) + await waitFor(() => expect(result.current.pages).toHaveLength(3)) + + // Grow the page size without discarding the committed page count. + act(() => { + rerender({ pageSize: 10 }) + }) + + await waitFor(() => expect(result.current.data).toHaveLength(30)) + expect(result.current.pages).toHaveLength(3) + expect(result.current.pages[0]).toHaveLength(10) + }) + + it(`compares dependencies by identity instead of serialization`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `infinite-query-map-deps`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const { result, rerender } = renderHook( + ({ filter }: { filter: Map }) => + useLiveInfiniteQuery( + (q) => + q + .from({ post: source }) + .where(({ post }) => eq(post.category, filter.get(`category`))) + .orderBy(({ post }) => post.createdAt, `desc`), + { pageSize: 3 }, + [filter], + ), + { + initialProps: { + filter: new Map([[`category`, `tech`]]), + }, + }, + ) + + await waitFor(() => { + expect(result.current.isReady).toBe(true) + expect(result.current.data.length).toBeGreaterThan(0) + expect( + result.current.data.every((post) => post.category === `tech`), + ).toBe(true) + }) + + rerender({ filter: new Map([[`category`, `life`]]) }) + + await waitFor(() => { + expect(result.current.data.length).toBeGreaterThan(0) + expect( + result.current.data.every((post) => post.category === `life`), + ).toBe(true) + }) + }) + + it(`preserves loaded pages when dependencies are structurally unchanged`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `infinite-query-structurally-equal-deps`, + getKey: (post) => post.id, + initialData: createMockPosts(20), + }), + ) + const { result, rerender } = renderHook( + ({ filter }: { filter: { category: string } }) => + useLiveInfiniteQuery( + (q) => + q + .from({ post: source }) + .where(({ post }) => eq(post.category, filter.category)) + .orderBy(({ post }) => post.createdAt, `desc`), + { pageSize: 2 }, + [filter], + ), + { initialProps: { filter: { category: `tech` } } }, + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + act(() => result.current.fetchNextPage()) + await waitFor(() => expect(result.current.pages).toHaveLength(2)) + + rerender({ filter: { category: `tech` } }) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + expect(result.current.pages).toHaveLength(2) + }) + + it(`releases a replaced controller through the external-store unsubscribe`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `infinite-query-controller-replacement`, + getKey: (post) => post.id, + initialData: createMockPosts(20), + }), + ) + const query = createLiveQueryCollection({ + query: (q) => + q.from({ post: source }).orderBy(({ post }) => post.createdAt, `desc`), + }) + const { result, rerender, unmount } = renderHook( + ({ pageSize }) => useLiveInfiniteQuery(query, { pageSize }), + { initialProps: { pageSize: 2 } }, + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + expect(query.subscriberCount).toBe(1) + + rerender({ pageSize: 3 }) + await waitFor(() => expect(result.current.pages[0]).toHaveLength(3)) + expect(query.subscriberCount).toBe(1) + + unmount() + expect(query.subscriberCount).toBe(0) + }) + + it(`binds fetchNextPage to the controller that returned it`, async () => { + const sourceA = createCollection( + mockSyncCollectionOptions({ + id: `infinite-query-generation-a`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const sourceB = createCollection( + mockSyncCollectionOptions({ + id: `infinite-query-generation-b`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const queryA = createLiveQueryCollection({ + query: (q) => + q.from({ post: sourceA }).orderBy(({ post }) => post.createdAt, `desc`), + }) + const queryB = createLiveQueryCollection({ + query: (q) => + q.from({ post: sourceB }).orderBy(({ post }) => post.createdAt, `desc`), + }) + const { result, rerender } = renderHook( + ({ query }) => useLiveInfiniteQuery(query, { pageSize: 2 }), + { initialProps: { query: queryA } }, + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + const fetchFromA = result.current.fetchNextPage + + rerender({ query: queryB }) + await waitFor(() => { + expect(result.current.collection).toBe(queryB) + expect(result.current.isReady).toBe(true) + expect(result.current.pages).toHaveLength(1) + }) + + act(() => fetchFromA()) + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 0)) + }) + expect(result.current.pages).toHaveLength(1) + + act(() => result.current.fetchNextPage()) + await waitFor(() => expect(result.current.pages).toHaveLength(2)) + }) + + it(`exposes pagination failures without an unhandled rejection`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `infinite-query-pagination-failure`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const query = createLiveQueryCollection({ + query: (q) => + q.from({ post: source }).orderBy(({ post }) => post.createdAt, `desc`), + }) + const { result } = renderHook(() => + useLiveInfiniteQuery(query, { pageSize: 2 }), + ) + + await waitFor(() => { + expect(result.current.isReady).toBe(true) + expect(result.current.hasNextPage).toBe(true) + }) + + const failure = new Error(`window load failed`) + vi.spyOn(query.utils, `setWindow`).mockRejectedValueOnce(failure) + + act(() => result.current.fetchNextPage()) + + await waitFor(() => { + expect(result.current.isError).toBe(true) + expect(result.current.error).toBe(failure) + expect(result.current.isFetchingNextPage).toBe(false) + }) + expect(result.current.pages).toHaveLength(1) + expect(result.current.hasNextPage).toBe(true) + }) + it(`should track pageParams correctly`, async () => { const posts = createMockPosts(30) const collection = createCollection( @@ -1738,6 +2035,43 @@ describe(`useLiveInfiniteQuery`, () => { expect(result.current.hasNextPage).toBe(true) }) + it(`warns when a pre-created collection's window differs from the first page`, async () => { + const posts = createMockPosts(50) + const collection = createCollection( + mockSyncCollectionOptions({ + autoIndex: `eager`, + id: `mismatched-window-warn-test`, + getKey: (post: Post) => post.id, + initialData: posts, + }), + ) + const liveQueryCollection = createLiveQueryCollection({ + query: (q) => + q + .from({ posts: collection }) + .orderBy(({ posts: p }) => p.createdAt, `desc`) + .limit(5) + .offset(0), + }) + await liveQueryCollection.preload() + + const warn = vi.spyOn(console, `warn`).mockImplementation(() => {}) + try { + renderHook(() => + useLiveInfiniteQuery(liveQueryCollection, { pageSize: 10 }), + ) + expect(warn).toHaveBeenCalledWith( + expect.stringContaining(`Pre-created collection has window`), + ) + expect(liveQueryCollection.utils.getWindow()).toEqual({ + offset: 0, + limit: 11, + }) + } finally { + warn.mockRestore() + } + }) + it(`should handle live updates with pre-created collection`, async () => { const posts = createMockPosts(30) const collection = createCollection( @@ -1867,7 +2201,7 @@ describe(`useLiveInfiniteQuery`, () => { }) }) - it(`throws a descriptive error when deps contain non-serializable values`, () => { + it(`accepts circular dependency values`, async () => { const posts = createMockPosts(10) const collection = createCollection( mockSyncCollectionOptions({ @@ -1881,21 +2215,119 @@ describe(`useLiveInfiniteQuery`, () => { const circular: Record = { a: 1 } circular.self = circular - expect(() => { - renderHook(() => { - return useLiveInfiniteQuery( - (q) => - q - .from({ posts: collection }) - .orderBy(({ posts: p }) => p.createdAt, `desc`), - { - pageSize: 5, - getNextPageParam: (lastPage) => - lastPage.length === 5 ? lastPage.length : undefined, - }, - [circular], - ) - }) - }).toThrow(/useLiveInfiniteQuery.*dependency/) + const { result } = renderHook(() => + useLiveInfiniteQuery( + (q) => + q + .from({ posts: collection }) + .orderBy(({ posts: p }) => p.createdAt, `desc`), + { pageSize: 5 }, + [circular], + ), + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + }) + + it(`recreates when switching collection to query function and back`, async () => { + const collectionSource = createCollection( + mockSyncCollectionOptions({ + id: `input-kind-collection-source`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const querySource = createCollection( + mockSyncCollectionOptions({ + id: `input-kind-query-source`, + getKey: (post) => post.id, + initialData: createMockPosts(10).map((post) => ({ + ...post, + id: `query-${post.id}`, + })), + }), + ) + const collectionInput = createLiveQueryCollection({ + query: (q) => + q + .from({ posts: collectionSource }) + .orderBy(({ posts }) => posts.createdAt, `desc`) + .limit(4), + }) + await collectionInput.preload() + const queryInput = (q: InitialQueryBuilder) => + q + .from({ posts: querySource }) + .orderBy(({ posts }) => posts.createdAt, `desc`) + + const { result, rerender } = renderHook( + ({ useCollection }: { useCollection: boolean }) => + useLiveInfiniteQuery( + (useCollection ? collectionInput : queryInput) as any, + { pageSize: 3 }, + ...((useCollection ? [] : [[]]) as [Array] | []), + ), + { initialProps: { useCollection: true } }, + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + expect((result.current.data[0] as Post).id).toBe(`1`) + rerender({ useCollection: false }) + await waitFor(() => + expect((result.current.data[0] as Post).id).toBe(`query-1`), + ) + rerender({ useCollection: true }) + await waitFor(() => expect((result.current.data[0] as Post).id).toBe(`1`)) + }) + + it(`recreates when switching query function to collection and back`, async () => { + const querySource = createCollection( + mockSyncCollectionOptions({ + id: `input-kind-query-first-source`, + getKey: (post) => post.id, + initialData: createMockPosts(10), + }), + ) + const collectionSource = createCollection( + mockSyncCollectionOptions({ + id: `input-kind-collection-second-source`, + getKey: (post) => post.id, + initialData: createMockPosts(10).map((post) => ({ + ...post, + id: `collection-${post.id}`, + })), + }), + ) + const queryInput = (q: InitialQueryBuilder) => + q + .from({ posts: querySource }) + .orderBy(({ posts }) => posts.createdAt, `desc`) + const collectionInput = createLiveQueryCollection({ + query: (q) => + q + .from({ posts: collectionSource }) + .orderBy(({ posts }) => posts.createdAt, `desc`) + .limit(4), + }) + await collectionInput.preload() + + const { result, rerender } = renderHook( + ({ useCollection }: { useCollection: boolean }) => + useLiveInfiniteQuery( + (useCollection ? collectionInput : queryInput) as any, + { pageSize: 3 }, + ...((useCollection ? [] : [[]]) as [Array] | []), + ), + { initialProps: { useCollection: false } }, + ) + + await waitFor(() => expect(result.current.isReady).toBe(true)) + expect((result.current.data[0] as Post).id).toBe(`1`) + rerender({ useCollection: true }) + await waitFor(() => + expect((result.current.data[0] as Post).id).toBe(`collection-1`), + ) + rerender({ useCollection: false }) + await waitFor(() => expect((result.current.data[0] as Post).id).toBe(`1`)) }) })