diff --git a/.changeset/fix-on-demand-readiness.md b/.changeset/fix-on-demand-readiness.md new file mode 100644 index 0000000000..7097036ee1 --- /dev/null +++ b/.changeset/fix-on-demand-readiness.md @@ -0,0 +1,8 @@ +--- +'@tanstack/db': patch +'@tanstack/react-db': patch +'@tanstack/db-sqlite-persistence-core': patch +'@tanstack/electric-db-collection': patch +--- + +Restore synchronous readiness for warm on-demand queries, retain retry-stable Suspense resources, and preserve persisted demand readiness across lifecycle transitions. Avoid redundant Electric refreshes when requesting subset snapshots, and declare React 18 as the minimum supported React version. diff --git a/docs/installation.md b/docs/installation.md index 2d643373fd..72cf4a26e7 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -11,7 +11,7 @@ Each supported framework comes with its own package. Each framework package re-e npm install @tanstack/react-db ``` -TanStack DB is compatible with React v16.8+ +TanStack DB is compatible with React v18+ ## Solid diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 0f1e4112ac..8a9edba0ad 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -1,6 +1,7 @@ import { SyncTransactionAbortedError, compileSingleRowExpression, + getLoadSubsetDemandKey, safeRandomUUID, toBooleanPredicate, withCollectionConfigFactory, @@ -792,6 +793,7 @@ class PersistedCollectionRuntime< > = [] private readonly queuedTxCommitted: Array = [] private readonly requestIds = new WeakMap() + private readonly hydratedDemands = new Set() private collection: Collection | null = null @@ -816,6 +818,8 @@ class PersistedCollectionRuntime< private indexRemovedUnsubscribe: (() => void) | null = null private remoteEnsureRetryTimer: ReturnType | null = null private nextRequestId = 0 + private startupSettled = false + private sourceTruncateGeneration = 0 private latestTerm = 0 private latestSeq = 0 @@ -928,6 +932,8 @@ class PersistedCollectionRuntime< private async hydrateBaseline(lifecycleGeneration: number): Promise { if (lifecycleGeneration !== this.lifecycleGeneration) return + // The baseline shares the unconstrained demand key. Its lease is never + // released, and every reload rereads it, so that coverage stays valid. const baseline = {} this.activeSubsets.set(this.getSubsetKey(baseline), baseline) const appliedCursor = this.appliedReceiptSequence @@ -971,6 +977,10 @@ class PersistedCollectionRuntime< if (this.syncMode !== `on-demand`) { await this.hydrateBaseline(lifecycleGeneration) } + + if (lifecycleGeneration === this.lifecycleGeneration) { + this.startupSettled = true + } } private async loadStartupMetadataInternal( @@ -1045,7 +1055,9 @@ class PersistedCollectionRuntime< upstreamLoadSubset?: LoadSubsetFn, ): Promise { const lifecycleGeneration = this.lifecycleGeneration - this.activeSubsets.set(this.getSubsetKey(options), options) + const subsetKey = this.getSubsetKey(options) + const truncateGeneration = this.sourceTruncateGeneration + this.activeSubsets.set(subsetKey, options) const appliedCursor = this.appliedReceiptSequence await this.applyMutex.run(() => @@ -1057,34 +1069,94 @@ class PersistedCollectionRuntime< if (lifecycleGeneration !== this.lifecycleGeneration) return await this.waitForAppliedReceiptsAfter(appliedCursor) + if ( + truncateGeneration === this.sourceTruncateGeneration && + this.activeSubsets.get(subsetKey) === options + ) { + this.hydratedDemands.add(getLoadSubsetDemandKey(options)) + } + if (upstreamLoadSubset) { try { await upstreamLoadSubset(options) } catch (error) { - if ( - options.signal?.aborted || - (typeof error === `object` && - error !== null && - `name` in error && - error.name === `AbortError`) - ) { - this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options)) - throw error - } - console.warn(`Failed to trigger remote subset load:`, error) - this.queueRemoteSubsetEnsure(options) - // Hydration remains readable, but it does not satisfy remote demand. + this.noteUpstreamLoadFailure(options, error) throw error } } } + loadHydratedSubset( + options: LoadSubsetOptions, + upstreamLoadSubset?: LoadSubsetFn, + ): true | Promise | undefined { + if ( + !this.startupSettled || + this.isHydratingNow() || + !this.hydratedDemands.has(getLoadSubsetDemandKey(options)) + ) { + return + } + + this.activeSubsets.set(this.getSubsetKey(options), options) + let result: true | Promise + try { + result = upstreamLoadSubset?.(options) ?? true + } catch (error) { + this.noteUpstreamLoadFailure(options, error) + return Promise.reject(error) + } + if (result === true) { + this.queueRemoteSubsetEnsure(options) + return true + } + + return result.then( + () => { + this.queueRemoteSubsetEnsure(options) + }, + (error: unknown) => { + this.noteUpstreamLoadFailure(options, error) + throw error + }, + ) + } + + private noteUpstreamLoadFailure( + options: LoadSubsetOptions, + error: unknown, + ): void { + if ( + options.signal?.aborted || + (typeof error === `object` && + error !== null && + `name` in error && + error.name === `AbortError`) + ) { + this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options)) + return + } + console.warn(`Failed to trigger remote subset load:`, error) + this.queueRemoteSubsetEnsure(options) + } + + noteSourceTruncate(): void { + this.sourceTruncateGeneration++ + this.hydratedDemands.clear() + } + unloadSubset( options: LoadSubsetOptions, upstreamUnloadSubset?: (options: LoadSubsetOptions) => void, ): void { - this.activeSubsets.delete(this.getSubsetKey(options)) - this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options)) + const subsetKey = this.getSubsetKey(options) + this.activeSubsets.delete(subsetKey) + this.pendingRemoteSubsetEnsures.delete(subsetKey) + const demandKey = getLoadSubsetDemandKey(options) + const stillActive = Array.from(this.activeSubsets.values()).some( + (active) => getLoadSubsetDemandKey(active) === demandKey, + ) + if (!stillActive) this.hydratedDemands.delete(demandKey) upstreamUnloadSubset?.(options) } @@ -1232,6 +1304,7 @@ class PersistedCollectionRuntime< this.pendingRemoteSubsetEnsures.clear() this.activeSubsets.clear() + this.hydratedDemands.clear() for (const transaction of this.queuedHydrationTransactions) { transaction.rejectApplied?.(new SyncTransactionAbortedError()) } @@ -1244,6 +1317,8 @@ class PersistedCollectionRuntime< private advanceLifecycle(): void { this.lifecycleGeneration++ this.started = false + this.startupSettled = false + this.hydratedDemands.clear() this.startupMetadataPromise = null this.startPromise = null this.resumeBaselinePromise = null @@ -2083,6 +2158,10 @@ class PersistedCollectionRuntime< } private async truncateAndReloadUnsafe(): Promise { + // Revoke synchronous coverage before publishing the empty collection. + // A subscriber may reacquire reentrantly from the truncate commit, and + // the following persistence read can fail. + this.hydratedDemands.clear() if (this.syncControls.begin && this.syncControls.commit) { this.withInternalApply(() => { this.syncControls.begin?.({ immediate: true }) @@ -2181,11 +2260,13 @@ class PersistedCollectionRuntime< private async reloadActiveSubsetsUnsafe(): Promise { const lifecycleGeneration = this.lifecycleGeneration + const truncateGeneration = this.sourceTruncateGeneration const activeSubsetOptions = this.activeSubsets.size > 0 ? Array.from(this.activeSubsets.values()) : [{}] + this.hydratedDemands.clear() this.hydratingGeneration = lifecycleGeneration try { const mergedRows = new Map() @@ -2210,14 +2291,25 @@ class PersistedCollectionRuntime< })), collectionMetadata, ) + + // Buffered source transactions and coordinator commits are part of the + // hydrated baseline. Keep the fast path closed until both queues have + // applied, or a reentrant acquisition can observe only the snapshot. + await this.flushQueuedHydrationTransactionsUnsafe() + await this.flushQueuedTxCommittedUnsafe() + + if (truncateGeneration === this.sourceTruncateGeneration) { + for (const options of activeSubsetOptions) { + if (this.activeSubsets.get(this.getSubsetKey(options)) === options) { + this.hydratedDemands.add(getLoadSubsetDemandKey(options)) + } + } + } } finally { if (this.hydratingGeneration === lifecycleGeneration) { this.hydratingGeneration = null } } - - await this.flushQueuedHydrationTransactionsUnsafe() - await this.flushQueuedTxCommittedUnsafe() } private attachIndexLifecycleListeners(): void { @@ -2556,6 +2648,7 @@ function createWrappedSyncConfig< : undefined, truncate: () => { if (startupState.cleanedUp) return + runtime.noteSourceTruncate() const openTransaction = getOpenTransaction() if (!openTransaction) { params.truncate() @@ -2629,17 +2722,20 @@ function createWrappedSyncConfig< } let sourceResult: SyncConfigRes = {} + let sourceResultSettled = false fullStartPromise = runtime.ensureStarted() const sourceResultPromise = (async () => { await runtime.ensureStartupMetadataLoaded() if (startupState.cleanedUp) { + sourceResultSettled = true return sourceResult } sourceResult = normalizeSyncFnResult( sourceSyncConfig.sync(wrappedParams), ) + sourceResultSettled = true return sourceResult })() @@ -2651,18 +2747,13 @@ function createWrappedSyncConfig< runtime.cleanup() runtime.clearSyncControls() }, - loadSubset: async (options: LoadSubsetOptions) => { + loadSubset: (options: LoadSubsetOptions): true | Promise => { const acquisition = { forwarded: false } acquisitions.set(options, acquisition) - await fullStartPromise - const resolvedSourceResult = await sourceResultPromise - if ( - startupState.cleanedUp || - acquisitions.get(options) !== acquisition - ) { - return - } - return runtime.loadSubset(options, (loadOptions) => { + const forwardUpstream = ( + resolvedSourceResult: SyncConfigRes, + loadOptions: LoadSubsetOptions, + ): true | Promise => { // Hydration is another async boundary. A release before this // point owns no upstream lease and must not start one later. if ( @@ -2681,7 +2772,30 @@ function createWrappedSyncConfig< acquisition.forwarded = false throw error } - }) + } + + const hydrated = sourceResultSettled + ? runtime.loadHydratedSubset(options, (loadOptions) => + forwardUpstream(sourceResult, loadOptions), + ) + : undefined + if (hydrated !== undefined) { + return hydrated + } + + return (async () => { + await fullStartPromise + const resolvedSourceResult = await sourceResultPromise + if ( + startupState.cleanedUp || + acquisitions.get(options) !== acquisition + ) { + return + } + return runtime.loadSubset(options, (loadOptions) => + forwardUpstream(resolvedSourceResult, loadOptions), + ) + })() }, unloadSubset: (options: LoadSubsetOptions) => { const acquisition = acquisitions.get(options) @@ -2728,7 +2842,8 @@ function createLoopbackSyncConfig< runtime.cleanup() runtime.clearSyncControls() }, - loadSubset: (options: LoadSubsetOptions) => runtime.loadSubset(options), + loadSubset: (options: LoadSubsetOptions): true | Promise => + runtime.loadHydratedSubset(options) ?? runtime.loadSubset(options), unloadSubset: (options: LoadSubsetOptions) => runtime.unloadSubset(options), } diff --git a/packages/db-sqlite-persistence-core/tests/persisted.test.ts b/packages/db-sqlite-persistence-core/tests/persisted.test.ts index 0895e52375..e0da866907 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted.test.ts @@ -1,12 +1,16 @@ import { describe, expect, it, vi } from 'vitest' +import { fc, test as fcTest } from '@fast-check/vitest' import { BasicIndex, DbClient, IR, collectionOptions, createCollection, + createLiveQueryCollection, createTransaction, + eq, } from '@tanstack/db' +import { oraclePropertyOptions } from '../../db/tests/oracle-config.js' import { InvalidPersistedCollectionCoordinatorError, InvalidPersistedStorageKeyEncodingError, @@ -19,6 +23,7 @@ import { persistedCollectionOptions, } from '../src' import type { + CollectionReset, PersistedCollectionCoordinator, PersistedCollectionPersistence, PersistedSyncWrappedOptions, @@ -184,7 +189,7 @@ function createNoopAdapter(): PersistenceAdapter { } type CoordinatorHarness = PersistedCollectionCoordinator & { - emit: (payload: TxCommitted, senderId?: string) => void + emit: (payload: TxCommitted | CollectionReset, senderId?: string) => void pullSinceCalls: number setPullSinceResponse: (response: PullSinceResponse) => void } @@ -1235,7 +1240,6 @@ describe(`persistedCollectionOptions`, () => { }, }), ) - const readyPromise = collection.stateWhenReady() for (let attempt = 0; attempt < 20 && !resolveLoadSubset; attempt++) { await flushAsyncWork() @@ -1977,6 +1981,426 @@ describe(`persistedCollectionOptions`, () => { }, ) + it(`answers only retained exact demands synchronously from hydrated rows`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + ]) + const loadRows = adapter.loadSubset + adapter.loadSubset = async (...args) => { + const rows = await loadRows(...args) + return rows.slice(0, args[1].limit) + } + const coordinator = createCoordinatorHarness() + const upstreamLoads: Array = [] + const upstreamUnloads: Array = [] + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: (options) => { + upstreamLoads.push(options) + return true + }, + unloadSubset: (options) => upstreamUnloads.push(options), + } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + const first: LoadSubsetOptions = { limit: 2 } + const sibling: LoadSubsetOptions = { limit: 2 } + try { + await collection._sync.loadSubset(first) + const persistedReads = adapter.loadSubsetCalls.length + + expect(collection._sync.loadSubset(sibling)).toBe(true) + expect(adapter.loadSubsetCalls).toHaveLength(persistedReads) + expect(upstreamLoads).toEqual([first, sibling]) + + collection._sync.unloadSubset(first) + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `retained-sibling`, + latestRowVersion: 1, + requiresFullReload: true, + }) + await flushAsyncWork() + await flushAsyncWork() + expect(collection.size).toBe(2) + + collection._sync.unloadSubset(sibling) + const narrow: LoadSubsetOptions = { limit: 1 } + await collection._sync.loadSubset(narrow) + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 2, + txId: `drop-released-demand`, + latestRowVersion: 2, + requiresFullReload: true, + }) + await flushAsyncWork() + await flushAsyncWork() + expect(collection.size).toBe(1) + + const reacquired = collection._sync.loadSubset({ limit: 2 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(collection.size).toBe(2) + expect(upstreamUnloads).toEqual([first, sibling]) + } finally { + await collection.cleanup() + } + }) + + it(`makes a sibling live query synchronously ready from a retained hydrated demand`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `cached` }]) + const source = createCollection( + persistedCollectionOptions({ + id: `hydrated-demand-live-query`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + source.startSyncImmediate() + const query = () => + createLiveQueryCollection({ + query: (q) => + q.from({ todo: source }).where(({ todo }) => eq(todo.id, `1`)), + startSync: true, + }) + const owner = query() + let sibling: ReturnType | undefined + try { + await owner.preload() + sibling = query() + expect(sibling.status).toBe(`ready`) + expect(sibling.toArray.map(({ id }) => id)).toEqual([`1`]) + } finally { + await sibling?.cleanup() + await owner.cleanup() + await source.cleanup() + } + }) + + it(`makes a loopback sibling synchronously ready from a retained hydrated demand`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `cached` }]) + const source = createCollection( + persistedCollectionOptions({ + id: `loopback-hydrated-demand-live-query`, + getKey: (row) => row.id, + syncMode: `on-demand`, + persistence: { adapter }, + }), + ) + source.startSyncImmediate() + const query = () => + createLiveQueryCollection({ + query: (q) => + q.from({ todo: source }).where(({ todo }) => eq(todo.id, `1`)), + startSync: true, + }) + const owner = query() + let sibling: ReturnType | undefined + try { + await owner.preload() + sibling = query() + expect(sibling.status).toBe(`ready`) + expect(sibling.toArray.map(({ id }) => id)).toEqual([`1`]) + } finally { + await sibling?.cleanup() + await owner.cleanup() + await source.cleanup() + } + }) + + fcTest.prop( + [ + fc + .array( + fc + .record({ + type: fc.constantFrom(`acquire` as const, `release` as const), + demand: fc.constantFrom(`one` as const, `two` as const), + }) + .map( + (operation) => + operation as + | { type: `acquire`; demand: `one` | `two` } + | { type: `release`; demand: `one` | `two` }, + ), + { minLength: 1, maxLength: 20 }, + ) + .chain((operations) => + fc + .array(fc.integer({ min: 0, max: operations.length }), { + maxLength: 4, + }) + .map((truncatePositions) => { + const positions = new Set(truncatePositions) + return operations.flatMap((operation, index) => + positions.has(index) + ? ([{ type: `truncate` as const }, operation] as const) + : [operation], + ) + }), + ), + ], + oraclePropertyOptions(50, `persistence.retained-demand`), + )( + `matches the retained exact-demand model across acquire, release, and truncate histories`, + async (history) => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + ]) + const loadRows = adapter.loadSubset + adapter.loadSubset = async (...args) => { + const rows = await loadRows(...args) + return rows.slice(0, args[1].limit) + } + let truncateSource: (() => void) | undefined + const collection = createCollection( + persistedCollectionOptions({ + id: `hydrated-demand-history`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, truncate, commit, markReady }) => { + truncateSource = () => { + begin() + truncate() + commit() + } + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + for (let attempt = 0; attempt < 20 && !truncateSource; attempt++) { + await flushAsyncWork() + } + expect(truncateSource).toBeTypeOf(`function`) + + const active: Record<`one` | `two`, Array> = { + one: [], + two: [], + } + const hydrated = new Set<`one` | `two`>() + const optionsFor = (demand: `one` | `two`): LoadSubsetOptions => ({ + limit: demand === `one` ? 1 : 2, + }) + + try { + for (const operation of history) { + if (operation.type === `truncate`) { + truncateSource?.() + hydrated.clear() + continue + } + + const demand = operation.demand + if (operation.type === `release`) { + const options = active[demand].pop() + if (!options) continue + collection._sync.unloadSubset(options) + if (active[demand].length === 0) hydrated.delete(demand) + continue + } + + const options = optionsFor(demand) + const result = collection._sync.loadSubset(options) + expect(result === true).toBe(hydrated.has(demand)) + if (result !== true) await result + active[demand].push(options) + hydrated.add(demand) + } + } finally { + await collection.cleanup() + } + }, + ) + + it(`invalidates hydrated demand when the source truncates`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `cached` }]) + let truncateSource!: () => void + const collection = createCollection( + persistedCollectionOptions({ + id: `hydrated-demand-truncate`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, truncate, commit, markReady }) => { + truncateSource = () => { + begin() + truncate() + commit() + } + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + try { + await collection._sync.loadSubset({ limit: 1 }) + expect(collection.size).toBe(1) + truncateSource() + expect(collection.size).toBe(0) + + const persistedReads = adapter.loadSubsetCalls.length + const reacquired = collection._sync.loadSubset({ limit: 1 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(adapter.loadSubsetCalls).toHaveLength(persistedReads + 1) + } finally { + await collection.cleanup() + } + }) + + it(`does not restore hydrated coverage after a truncate overtakes a reload`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `cached` }]) + const readSubset = adapter.loadSubset + let subsetReads = 0 + let enterReload!: () => void + let releaseReload!: () => void + const reloadEntered = new Promise((resolve) => { + enterReload = resolve + }) + const reloadGate = new Promise((resolve) => { + releaseReload = resolve + }) + adapter.loadSubset = async (...args) => { + subsetReads++ + if (subsetReads === 2) { + enterReload() + await reloadGate + } + return readSubset(...args) + } + const coordinator = createCoordinatorHarness() + let truncateSource!: () => void + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, truncate, commit, markReady }) => { + truncateSource = () => { + begin() + truncate() + commit() + } + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + try { + await collection._sync.loadSubset({ limit: 1 }) + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `reload-before-truncate`, + latestRowVersion: 1, + requiresFullReload: true, + }) + await reloadEntered + truncateSource() + releaseReload() + await flushAsyncWork() + await flushAsyncWork() + expect(collection.size).toBe(0) + + const readsBeforeReacquire = adapter.loadSubsetCalls.length + const reacquired = collection._sync.loadSubset({ limit: 1 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(adapter.loadSubsetCalls).toHaveLength(readsBeforeReacquire + 1) + } finally { + releaseReload() + await collection.cleanup() + } + }) + + it(`rereads a retained demand after a reset reload fails`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `cached` }]) + const readSubset = adapter.loadSubset + const failure = new Error(`reset reload failed`) + let failNextRead = false + adapter.loadSubset = async (...args) => { + if (failNextRead) { + failNextRead = false + throw failure + } + return readSubset(...args) + } + const coordinator = createCoordinatorHarness() + const warn = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { sync: ({ markReady }) => (markReady(), {}) }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + try { + const retained: LoadSubsetOptions = { limit: 1 } + await collection._sync.loadSubset(retained) + expect(collection.size).toBe(1) + + failNextRead = true + coordinator.emit({ + type: `collection:reset`, + schemaVersion: 1, + resetEpoch: 1, + }) + await flushAsyncWork() + await flushAsyncWork() + expect(collection.size).toBe(0) + + const readsBeforeReacquire = adapter.loadSubsetCalls.length + const reacquired = collection._sync.loadSubset({ limit: 1 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(adapter.loadSubsetCalls).toHaveLength(readsBeforeReacquire + 1) + expect(collection.size).toBe(1) + } finally { + warn.mockRestore() + await collection.cleanup() + } + }) + it(`does not retain refresh history as permanent subset demand`, async () => { const adapter = createRecordingAdapter([{ id: `1`, title: `Before` }]) const coordinator = createCoordinatorHarness() @@ -2115,6 +2539,182 @@ describe(`persistedCollectionOptions`, () => { }, ) + it(`does not mark a demand hydrated after its last acquisition leaves mid-read`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + { id: `3`, title: `three` }, + ]) + const readSubset = adapter.loadSubset + let releaseRead!: () => void + let enterRead!: () => void + const readGate = new Promise((resolve) => { + releaseRead = resolve + }) + const readEntered = new Promise((resolve) => { + enterRead = resolve + }) + let blockNextRead = true + adapter.loadSubset = async (...args) => { + if (blockNextRead) { + blockNextRead = false + enterRead() + await readGate + } + return readSubset(...args) + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `released-mid-read`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true, unloadSubset: () => {} } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + const first: LoadSubsetOptions = { limit: 3 } + const pending = collection._sync.loadSubset(first) + await readEntered + collection._sync.unloadSubset(first) + releaseRead() + await pending + + const reacquired = collection._sync.loadSubset({ limit: 3 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(collection.get(`3`)).toBeDefined() + } finally { + releaseRead() + await collection.cleanup() + } + }) + + it(`does not answer a registered but unread demand synchronously`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `one` }]) + const readSubset = adapter.loadSubset + let releaseRead!: () => void + let enterRead!: () => void + const readGate = new Promise((resolve) => { + releaseRead = resolve + }) + const readEntered = new Promise((resolve) => { + enterRead = resolve + }) + let blockNextRead = true + adapter.loadSubset = async (...args) => { + if (blockNextRead) { + blockNextRead = false + enterRead() + await readGate + } + return readSubset(...args) + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `registered-before-read`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true, unloadSubset: () => {} } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + const first = collection._sync.loadSubset({ limit: 1 }) + await readEntered + const second = collection._sync.loadSubset({ limit: 1 }) + expect(second).not.toBe(true) + releaseRead() + await Promise.all([first, second]) + expect(collection.get(`1`)).toBeDefined() + } finally { + releaseRead() + await collection.cleanup() + } + }) + + it(`stops ensuring an aborted fast-path acquisition`, async () => { + vi.useFakeTimers() + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const abortError = Object.assign(new Error(`abort`), { name: `AbortError` }) + const ensured: Array = [] + const first: LoadSubsetOptions = { limit: 1 } + const second: LoadSubsetOptions = { limit: 1 } + const coordinator: PersistedCollectionCoordinator = { + getNodeId: () => `fast-path-abort`, + subscribe: () => () => {}, + publish: () => {}, + isLeader: () => true, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + requestEnsureRemoteSubset: (_id, options) => { + ensured.push(options) + return Promise.reject(new Error(`offline`)) + }, + } + const collection = createCollection( + persistedCollectionOptions({ + id: `fast-path-abort`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: (options) => + options === second ? Promise.reject(abortError) : true, + unloadSubset: () => {}, + } + }, + }, + persistence: { + adapter: createRecordingAdapter([{ id: `1`, title: `one` }]), + coordinator, + }, + }), + ) + + try { + collection.startSyncImmediate() + await collection._sync.loadSubset(first) + const result = await Promise.resolve( + collection._sync.loadSubset(second), + ).then( + () => `ready`, + (error: unknown) => error, + ) + expect(result).toBe(abortError) + + const callsBeforeRetry = ensured.filter( + (options) => options === second, + ).length + await vi.advanceTimersByTimeAsync(500) + expect(ensured.filter((options) => options === second)).toHaveLength( + callsBeforeRetry, + ) + } finally { + await collection.cleanup() + warning.mockRestore() + vi.useRealTimers() + } + }) + it(`does not release or acquire an upstream lease cancelled during hydration`, async () => { const adapter = createRecordingAdapter() const hydrate = adapter.loadSubset @@ -2161,7 +2761,7 @@ describe(`persistedCollectionOptions`, () => { ) collection.startSyncImmediate() const first: LoadSubsetOptions = { limit: 1 } - const second: LoadSubsetOptions = { limit: 1 } + const second: LoadSubsetOptions = { limit: 2 } try { await collection._sync.loadSubset(first) expect(leases).toBe(1) diff --git a/packages/db/src/query/live/ordered-source-loader.ts b/packages/db/src/query/live/ordered-source-loader.ts index af680ec560..f2a7e41f95 100644 --- a/packages/db/src/query/live/ordered-source-loader.ts +++ b/packages/db/src/query/live/ordered-source-loader.ts @@ -52,6 +52,7 @@ export class OrderedSourceLoader { private lastBoundary: unknown private repairRetries = 0 private repairTimer: ReturnType | undefined + private synchronousCompletionDepth = 0 constructor( private readonly info: OrderByOptimizationInfo, @@ -153,7 +154,9 @@ export class OrderedSourceLoader { // A recorded failure always carries recovery debt, so it cannot reach this // finite path; only the first request needs the whole prefix here. let count = Math.max( - this.info.dataNeeded(), + this.synchronousCompletionDepth > 0 + ? this.directSourceRowsNeeded() + : this.info.dataNeeded(), this.hasSettledSourceRequest ? 0 : this.info.offset + this.info.limit, ) if ( @@ -187,7 +190,11 @@ export class OrderedSourceLoader { private loadPrefix(count: number, windowOperationGeneration?: number): void { if (!this.active || this.pending) return if (this.lastPrefixCount === count) { - if ((this.info.dataNeeded?.() ?? 0) > 0) { + const needsRows = + this.synchronousCompletionDepth > 0 + ? this.directSourceRowsNeeded() > 0 + : (this.info.dataNeeded?.() ?? 0) > 0 + if (needsRows) { this.loadFullSource(windowOperationGeneration) } return @@ -290,6 +297,16 @@ export class OrderedSourceLoader { ).length } + /** Read direct-source demand without waiting for D2's next graph turn. */ + private directSourceRowsNeeded(): number { + const needed = this.info.offset + this.info.limit + const available = this.subscription.readOrderedSnapshot({ + orderBy: normalizeOrderByPaths(this.info.orderBy, this.alias), + limit: needed, + }).length + return Math.max(0, needed - available) + } + private loadPage(count: number, windowOperationGeneration?: number): void { if (!this.active || this.pending) return // Rows observed before the first provider request do not prove ordered @@ -341,7 +358,7 @@ export class OrderedSourceLoader { kind: OrderedRequestKind, windowOperationGeneration?: number, options?: LoadSubsetOptions, - ): Promise { + ): Promise | undefined { const isFullSource = kind === `full-source` const retryRepair = isFullSource && @@ -349,8 +366,9 @@ export class OrderedSourceLoader { this.needsFullSourceRecovery && windowOperationGeneration === undefined const generation = this.generation - const complete = (): void => { - if (this.pending === tracked) this.pending = undefined + const complete = (settledRequest?: Promise): void => { + if (settledRequest && this.pending === settledRequest) + this.pending = undefined if (!this.active) return // Retirement failure does not undo a successful acquisition. Finish its // boundary and continuation, then report the first cleanup error. @@ -393,7 +411,7 @@ export class OrderedSourceLoader { this.subscription.readOrderedSnapshot(options).at(-1) ?.value ?? this.settledSourceBoundary } catch (error) { - fail(error) + fail(error, settledRequest) } } } @@ -414,11 +432,10 @@ export class OrderedSourceLoader { }, ]) } - const settlesAsync = result instanceof Promise - const request = settlesAsync ? result : Promise.resolve() - const fail = (error: unknown) => { + const fail = (error: unknown, settledRequest?: Promise) => { this.settledFiniteAcquisitions.delete(releaseAcquisition) - if (this.pending === tracked) this.pending = undefined + if (settledRequest && this.pending === settledRequest) + this.pending = undefined if (!this.active) return // A failed request may already have written only part of its result. // None of those rows is a safe continuation boundary. @@ -433,16 +450,49 @@ export class OrderedSourceLoader { if (retryRepair) this.scheduleRepairRetry() throw error } - const tracked = request.then(complete, fail) + if (result === true) { + let completionError: unknown + this.synchronousCompletionDepth++ + try { + complete() + } catch (error) { + completionError = error + } finally { + this.synchronousCompletionDepth-- + } + if (completionError !== undefined) { + // Acquisition succeeded even if its completion bookkeeping reports an + // error (for example, retiring an older lease). Preserve that state + // and surface the error through the same operation channel as an + // asynchronous completion callback would. + const rejected = Promise.reject(completionError) + // Completion may have synchronously started the next request before a + // cleanup callback's error was rethrown. Keep that newer request as the + // active refinement participant. + if (this.pending === undefined) this.pending = rejected + void rejected.catch(() => {}) + void rejected.then( + () => {}, + () => { + if (this.pending === rejected) this.pending = undefined + }, + ) + this.onResult(rejected, false) + return rejected + } + this.onResult(true, false) + return + } + const tracked: Promise = result.then( + () => complete(tracked), + (error) => fail(error, tracked), + ) this.pending = tracked void tracked.catch(() => {}) // Register each request separately. The operation tracker observes the // next request before this promise settles, so the logical chain remains // pending without retaining every ancestor promise until the final page. - this.onResult( - tracked, - settlesAsync && isFullSource && this.needsFullSourceRecovery, - ) + this.onResult(tracked, isFullSource && this.needsFullSourceRecovery) return tracked } diff --git a/packages/db/tests/live-query-window-controller.test.ts b/packages/db/tests/live-query-window-controller.test.ts index 6526a2cafd..1c48a3a561 100644 --- a/packages/db/tests/live-query-window-controller.test.ts +++ b/packages/db/tests/live-query-window-controller.test.ts @@ -30,6 +30,28 @@ function makeSource(initialData: Array = ROWS) { ) } +function makePendingSource(initialData: Array) { + let resolveLoad!: () => void + const pendingLoad = new Promise((resolve) => { + resolveLoad = resolve + }) + const source = createCollection({ + id: `window-ctrl-${seq++}`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + for (const row of initialData) write({ type: `insert`, value: row }) + commit() + markReady() + return { loadSubset: () => pendingLoad } + }, + }, + }) + return { source, resolveLoad } +} + /** Ordered live query with page 1's peek-ahead window baked in, as the React adapter builds it. */ function makeOrderedLiveQuery(source: Collection, pageSize: number) { return createLiveQueryCollection({ @@ -57,7 +79,7 @@ describe(`createLiveQueryWindowController`, () => { )( `handles $action during initial loading with $rowCount rows`, async ({ rowCount, action }) => { - const source = makeSource(ROWS.slice(0, rowCount)) + const { source, resolveLoad } = makePendingSource(ROWS.slice(0, rowCount)) const lq = makeOrderedLiveQuery(source, 2) const controller = createLiveQueryWindowController(lq, { pageSize: 2, @@ -67,8 +89,10 @@ describe(`createLiveQueryWindowController`, () => { expect(controller.getSnapshot().isLoading).toBe(true) const fetch = controller.fetchNextPage() expect(controller.fetchNextPage()).toBe(fetch) - if (action === `reset`) await controller.reset() + const reset = action === `reset` ? controller.reset() : undefined if (action === `dispose`) controller.dispose() + resolveLoad() + await reset await fetch const visibleCount = action === `fetch` ? 4 : 2 expect(ids(controller.getSnapshot())).toEqual( diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index d9b4d3749b..d0b6f13a67 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -107,6 +107,7 @@ const staticOracleProperties = [ `pagination.pending-history`, `pagination.pending-mutation`, `pagination.window-transition`, + `persistence.retained-demand`, `predicate-subtraction.duplicate-terms`, `predicate-subtraction.finite-world`, `predicate-subtraction.unbounded`, diff --git a/packages/db/tests/oracle-replay-manifest.ts b/packages/db/tests/oracle-replay-manifest.ts index 191c46663b..369d863ab0 100644 --- a/packages/db/tests/oracle-replay-manifest.ts +++ b/packages/db/tests/oracle-replay-manifest.ts @@ -141,6 +141,11 @@ const ownerGroups: ReadonlyArray = [ `pagination`, `multi-order nullable-cursor pending-mutation pending-history ordered-window window-transition async-cursor`, ], + [ + `db-sqlite-persistence-core/tests/persisted.test.ts`, + `persistence`, + `retained-demand`, + ], ] const owners = new Map() diff --git a/packages/db/tests/query/live-query-collection.test.ts b/packages/db/tests/query/live-query-collection.test.ts index 48727bedb5..8728489631 100644 --- a/packages/db/tests/query/live-query-collection.test.ts +++ b/packages/db/tests/query/live-query-collection.test.ts @@ -1712,6 +1712,38 @@ describe(`createLiveQueryCollection`, () => { expect(liveQuery.size).toBeGreaterThan(0) }) + it(`makes a warm ordered window ready before construction returns when every acquisition is synchronous`, () => { + const sourceCollection = createCollection<{ id: number; value: number }>({ + id: `source-fully-sync-subset`, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + sync: { + sync: ({ markReady, begin, write, commit }) => { + begin() + write({ type: `insert`, value: { id: 1, value: 10 } }) + write({ type: `insert`, value: { id: 2, value: 20 } }) + commit() + markReady() + return { loadSubset: () => true } + }, + }, + }) + + const liveQuery = createLiveQueryCollection({ + query: (q) => + q + .from({ item: sourceCollection }) + .orderBy(({ item }) => item.value, `asc`) + .limit(1), + startSync: true, + }) + + expect(liveQuery.isLoadingSubset).toBe(false) + expect(liveQuery.status).toBe(`ready`) + expect(liveQuery.toArray.map(({ id }) => id)).toEqual([1]) + }) + it(`live query result collection has isLoadingSubset property`, async () => { const sourceCollection = createCollection<{ id: string; value: string }>({ id: `source`, @@ -2811,6 +2843,8 @@ describe(`createLiveQueryCollection`, () => { type Row = { id: number; rank: number } const gate = createDeferred() let loadCount = 0 + let afterPreload = false + const loaded = new Set() const source = createCollection({ id: `ordered-cleanup-window-source`, getKey: (row) => row.id, @@ -2819,14 +2853,24 @@ describe(`createLiveQueryCollection`, () => { defaultIndexType: BTreeIndex, sync: { sync: ({ begin, write, commit, markReady }) => { - begin() - write({ type: `insert`, value: { id: 1, rank: 1 } }) - commit() markReady() return { - loadSubset: () => { + loadSubset: (options) => { loadCount++ - return loadCount === 3 ? gate.promise : true + const start = options.cursor ? (options.offset ?? 0) : 0 + const end = start + (options.limit ?? 0) + const rows = [1, 2, 3].filter( + (id) => id > start && id <= end && !loaded.has(id), + ) + if (rows.length > 0) { + begin() + for (const id of rows) { + loaded.add(id) + write({ type: `insert`, value: { id, rank: id } }) + } + commit(options.signal) + } + return afterPreload ? gate.promise : true }, } }, @@ -2855,7 +2899,8 @@ describe(`createLiveQueryCollection`, () => { return withHistoryCleanup( async () => { await live.preload() - const move = live.utils.setWindow({ offset: 0, limit: 2 }) + afterPreload = true + const move = live.utils.setWindow({ offset: 2, limit: 1 }) observedMove = Promise.resolve(move).then( () => {}, () => {}, @@ -2895,7 +2940,9 @@ describe(`createLiveQueryCollection`, () => { type Row = { id: number; rank: number } const oldGate = createDeferred() const newGate = createDeferred() - let limitFourCalls = 0 + const loaded = new Set() + let windowAttempts = 0 + let initialPreloadComplete = false const source = createCollection({ id: `ordered-window-restart-generation-source`, getKey: (row) => row.id, @@ -2904,26 +2951,47 @@ describe(`createLiveQueryCollection`, () => { defaultIndexType: BTreeIndex, sync: { sync: (operations) => { - operations.begin() - for (let id = 1; id <= 6; id++) { - operations.write({ type: `insert`, value: { id, rank: id } }) - } - operations.commit() operations.markReady() return { loadSubset: (options) => { - if (options.where || options.limit !== 4) return true - limitFourCalls++ - if (limitFourCalls === 1) { - options.signal?.addEventListener( - `abort`, - () => - oldGate.reject(new DOMException(`aborted`, `AbortError`)), - { once: true }, + const start = options.cursor ? (options.offset ?? 0) : 0 + const end = start + (options.limit ?? 0) + const deliver = () => { + const rows = [1, 2, 3, 4].filter( + (id) => id > start && id <= end && !loaded.has(id), ) - return oldGate.promise + if (rows.length === 0) return + operations.begin() + for (const id of rows) { + loaded.add(id) + operations.write({ + type: `insert`, + value: { id, rank: id }, + }) + } + operations.commit(options.signal) } - return newGate.promise + if ( + initialPreloadComplete && + options.cursor && + options.offset === 1 + ) { + windowAttempts++ + if (windowAttempts === 1) { + options.signal?.addEventListener( + `abort`, + () => + oldGate.reject( + new DOMException(`aborted`, `AbortError`), + ), + { once: true }, + ) + return oldGate.promise + } + return newGate.promise.then(deliver) + } + deliver() + return true }, } }, @@ -2938,6 +3006,7 @@ describe(`createLiveQueryCollection`, () => { try { await live.preload() + initialPreloadComplete = true const abandoned = live.utils.setWindow({ offset: 2, limit: 2 }) expect(abandoned).toBeInstanceOf(Promise) const abandonedRejection = expect(abandoned).rejects.toMatchObject({ @@ -2945,7 +3014,9 @@ describe(`createLiveQueryCollection`, () => { }) const cleanup = live.cleanup() + initialPreloadComplete = false const preload = live.preload() + initialPreloadComplete = true const replacement = live.utils.setWindow({ offset: 2, limit: 2 }) expect(replacement).toBeInstanceOf(Promise) await Promise.all([cleanup, preload, abandonedRejection]) @@ -3027,7 +3098,8 @@ describe(`createLiveQueryCollection`, () => { it(`settles a superseding window only after that window is visible`, async () => { type Row = { id: number; rank: number } const gate = createDeferred() - let loadCount = 0 + const loaded = new Set() + let initialPreloadComplete = false const source = createCollection({ id: `ordered-superseding-window-source`, getKey: (row) => row.id, @@ -3036,15 +3108,32 @@ describe(`createLiveQueryCollection`, () => { defaultIndexType: BTreeIndex, sync: { sync: ({ begin, write, commit, markReady }) => { - begin() - write({ type: `insert`, value: { id: 1, rank: 1 } }) - write({ type: `insert`, value: { id: 2, rank: 2 } }) - commit() markReady() return { - loadSubset: () => { - loadCount++ - return loadCount <= 2 ? true : gate.promise + loadSubset: (options) => { + const start = options.cursor ? (options.offset ?? 0) : 0 + const end = start + (options.limit ?? 0) + const deliver = () => { + const rows = [1, 2, 3].filter( + (id) => id > start && id <= end && !loaded.has(id), + ) + if (rows.length === 0) return + begin() + for (const id of rows) { + loaded.add(id) + write({ type: `insert`, value: { id, rank: id } }) + } + commit(options.signal) + } + if ( + initialPreloadComplete && + options.cursor && + options.offset === 1 + ) { + return gate.promise.then(deliver) + } + deliver() + return true }, } }, @@ -3059,6 +3148,7 @@ describe(`createLiveQueryCollection`, () => { try { await live.preload() + initialPreloadComplete = true const first = live.utils.setWindow({ offset: 0, limit: 3 }) expect(first).toBeInstanceOf(Promise) const second = live.utils.setWindow({ offset: 1, limit: 1 }) @@ -3085,6 +3175,7 @@ describe(`createLiveQueryCollection`, () => { type Row = { id: number; rank: number } const failure = new Error(`restarted ordered page failed`) let failPage = false + const loaded = new Set() const source = createCollection({ id: `ordered-window-restart-source`, getKey: (row) => row.id, @@ -3093,14 +3184,25 @@ describe(`createLiveQueryCollection`, () => { defaultIndexType: BTreeIndex, sync: { sync: ({ begin, write, commit, markReady }) => { - begin() - write({ type: `insert`, value: { id: 1, rank: 1 } }) - write({ type: `insert`, value: { id: 2, rank: 2 } }) - commit() markReady() return { loadSubset: (options) => { - if (failPage && !options.where) throw failure + if (failPage && options.cursor) { + throw failure + } + const start = options.cursor ? (options.offset ?? 0) : 0 + const end = start + (options.limit ?? 0) + const rows = [1, 2, 3].filter( + (id) => id > start && id <= end && !loaded.has(id), + ) + if (rows.length > 0) { + begin() + for (const id of rows) { + loaded.add(id) + write({ type: `insert`, value: { id, rank: id } }) + } + commit(options.signal) + } return true }, } @@ -3126,7 +3228,7 @@ describe(`createLiveQueryCollection`, () => { failPage = true await expect( Promise.resolve().then(() => - live.utils.setWindow({ offset: 0, limit: 3 }), + live.utils.setWindow({ offset: 50, limit: 1 }), ), ).rejects.toBe(failure) expect(Array.from(live.values(), ({ id }) => id)).toEqual([1]) @@ -3964,80 +4066,67 @@ describe(`createLiveQueryCollection`, () => { // 3. The Promise waits for loading to complete // 4. The Promise resolves once loading is done - vi.useFakeTimers() - - try { - let loadSubsetCallCount = 0 - - const sourceCollection = createCollection<{ - id: number - value: number - }>({ - id: `source-async-subset-loading`, - getKey: (item) => item.id, - syncMode: `on-demand`, - startSync: true, - autoIndex: `eager`, // Enable auto-indexing for orderBy optimization - defaultIndexType: BTreeIndex, - sync: { - sync: ({ markReady, begin, write, commit }) => { - // Provide minimal initial data - begin() - write({ type: `insert`, value: { id: 1, value: 1 } }) - write({ type: `insert`, value: { id: 2, value: 2 } }) - write({ type: `insert`, value: { id: 3, value: 3 } }) - commit() - markReady() - - return { - loadSubset: () => { - loadSubsetCallCount++ - - // First call is for the initial window request - if (loadSubsetCallCount === 1) { - return true - } + const gate = createDeferred() + let initialPreloadComplete = false + let moveStarted = false + let loadSubsetCallCount = 0 - // The second call closes the initial ordered boundary. - if (loadSubsetCallCount === 2) return true - - // The later call triggered by setWindow returns a promise. - const loadPromise = new Promise((resolve) => { - // Simulate async data loading with a delay - setTimeout(() => { - begin() - // Load additional items that would be needed for the new window - write({ type: `insert`, value: { id: 4, value: 4 } }) - write({ type: `insert`, value: { id: 5, value: 5 } }) - write({ type: `insert`, value: { id: 6, value: 6 } }) - commit() - resolve() - }, 50) - }) + const sourceCollection = createCollection<{ + id: number + value: number + }>({ + id: `source-async-subset-loading`, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + autoIndex: `eager`, // Enable auto-indexing for orderBy optimization + defaultIndexType: BTreeIndex, + sync: { + sync: ({ markReady, begin, write, commit }) => { + // Provide minimal initial data + begin() + write({ type: `insert`, value: { id: 1, value: 1 } }) + write({ type: `insert`, value: { id: 2, value: 2 } }) + write({ type: `insert`, value: { id: 3, value: 3 } }) + commit() + markReady() - return loadPromise - }, - } - }, + return { + loadSubset: (options) => { + loadSubsetCallCount++ + if (!initialPreloadComplete || moveStarted) return true + moveStarted = true + return gate.promise.then(() => { + begin() + write({ type: `insert`, value: { id: 4, value: 4 } }) + write({ type: `insert`, value: { id: 5, value: 5 } }) + write({ type: `insert`, value: { id: 6, value: 6 } }) + commit(options.signal) + }) + }, + } }, - }) + }, + }) - const liveQuery = createLiveQueryCollection({ - query: (q) => - q - .from({ item: sourceCollection }) - .orderBy(({ item }) => item.value, `asc`) - .limit(2) - .offset(0), - startSync: true, - }) + const liveQuery = createLiveQueryCollection({ + query: (q) => + q + .from({ item: sourceCollection }) + .orderBy(({ item }) => item.value, `asc`) + .limit(1) + .offset(0), + startSync: true, + }) + try { await liveQuery.preload() + initialPreloadComplete = true - // Initial state: should have 2 items (values 1, 2) - expect(liveQuery.size).toBe(2) + // Initial state: one visible row, with later rows already resident. + expect(liveQuery.size).toBe(1) expect(liveQuery.isLoadingSubset).toBe(false) - expect(loadSubsetCallCount).toBe(2) + const initialLoadCount = loadSubsetCallCount // Move window to offset 3, which requires loading more data // This should trigger loadSubset and return a Promise @@ -4047,11 +4136,8 @@ describe(`createLiveQueryCollection`, () => { expect(result).toBeInstanceOf(Promise) expect(result).not.toBe(true) - // Advance just a bit to let the scheduler execute and trigger loadSubset - await vi.advanceTimersByTimeAsync(1) - // Verify that loading was triggered and is in progress - expect(loadSubsetCallCount).toBeGreaterThan(1) + expect(loadSubsetCallCount).toBeGreaterThan(initialLoadCount) expect(liveQuery.isLoadingSubset).toBe(true) // Track when the promise resolves @@ -4063,20 +4149,11 @@ describe(`createLiveQueryCollection`, () => { } // Promise should NOT be resolved yet because loading is still in progress - await vi.advanceTimersByTimeAsync(10) - expect(promiseResolved).toBe(false) - expect(liveQuery.isLoadingSubset).toBe(true) - - // Complete the page request. The operation must remain pending while - // the loader closes the ordering boundary so equal sort values cannot - // be omitted from later window moves. - await vi.advanceTimersByTimeAsync(40) - expect(loadSubsetCallCount).toBe(4) + await flushPromises() expect(promiseResolved).toBe(false) expect(liveQuery.isLoadingSubset).toBe(true) - // Complete the boundary request as well. - await vi.advanceTimersByTimeAsync(50) + gate.resolve() // Wait for the promise to resolve if (result !== true) { @@ -4092,7 +4169,8 @@ describe(`createLiveQueryCollection`, () => { const items = liveQuery.toArray expect(items.map((i) => i.value)).toEqual([4, 5]) } finally { - vi.useRealTimers() + gate.resolve() + await Promise.all([liveQuery.cleanup(), sourceCollection.cleanup()]) } }) diff --git a/packages/db/tests/query/ordered-lifecycle-oracle.property.test.ts b/packages/db/tests/query/ordered-lifecycle-oracle.property.test.ts index fb52e99c8c..aff13980e7 100644 --- a/packages/db/tests/query/ordered-lifecycle-oracle.property.test.ts +++ b/packages/db/tests/query/ordered-lifecycle-oracle.property.test.ts @@ -688,6 +688,180 @@ describe(`ordered lifecycle product`, () => { ) }) +describe(`warm ordered readiness oracle`, () => { + it.each([ + { limit: 0, expectedRequests: [] }, + { + limit: 50, + expectedRequests: [ + { ordered: true, filtered: false, limit: 50 }, + { ordered: false, filtered: false, limit: undefined }, + ], + }, + ])( + `makes a retained $limit-row window synchronously ready after an async cold prime`, + async ({ limit, expectedRequests }) => { + type DatedRow = { id: number; createdAt: Date | null } + const remote: Array = [ + { id: 1, createdAt: new Date(`2026-01-03T00:00:00.000Z`) }, + { id: 2, createdAt: null }, + { id: 3, createdAt: new Date(`2026-01-01T00:00:00.000Z`) }, + { id: 4, createdAt: new Date(`2026-01-02T00:00:00.000Z`) }, + ] + const installed = new Set() + const warmRequests: Array<{ + ordered: boolean + filtered: boolean + limit: number | undefined + }> = [] + let cold = true + const source = createCollection({ + id: `warm-ordered-readiness-${limit}`, + getKey: ({ id }) => id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: (options) => { + if (!cold) { + warmRequests.push({ + ordered: options.orderBy !== undefined, + filtered: options.where !== undefined, + limit: options.limit, + }) + } + const missing = remote.filter(({ id }) => !installed.has(id)) + if (missing.length > 0) { + begin() + for (const row of missing) { + installed.add(row.id) + write({ type: `insert`, value: row }) + } + commit() + } + return cold ? Promise.resolve() : true + }, + } + }, + }, + }) + const query = (queryLimit: number) => + createLiveQueryCollection({ + query: (q) => + q + .from({ row: source }) + .orderBy(({ row }) => row.createdAt, { + direction: `asc`, + nulls: `last`, + }) + .limit(queryLimit), + startSync: true, + }) + const coldQuery = query(50) + try { + await coldQuery.preload() + cold = false + + const warmQuery = query(limit) + try { + expect(warmQuery.status).toBe(`ready`) + expect(warmQuery.isLoadingSubset).toBe(false) + expect(warmQuery.toArray.map(({ id }) => id)).toEqual( + limit === 0 ? [] : [3, 4, 1, 2], + ) + expect(warmRequests).toEqual(expectedRequests) + } finally { + await warmQuery.cleanup() + } + } finally { + await coldQuery.cleanup() + await source.cleanup() + } + }, + ) + + it(`makes a retained nonzero-offset prefix and boundary synchronously ready`, async () => { + type WarmRow = { id: number; rank: number; tie: number } + const remote: Array = [ + { id: 1, rank: 1, tie: 1 }, + { id: 2, rank: 2, tie: 1 }, + { id: 3, rank: 3, tie: 1 }, + { id: 4, rank: 4, tie: 1 }, + ] + const installed = new Set() + const warmRequests: Array<{ + ordered: boolean + filtered: boolean + limit: number | undefined + }> = [] + let cold = true + const source = createCollection({ + id: `warm-offset-prefix-readiness`, + getKey: ({ id }) => id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: (options) => { + if (!cold) { + warmRequests.push({ + ordered: options.orderBy !== undefined, + filtered: options.where !== undefined, + limit: options.limit, + }) + } + const missing = remote.filter(({ id }) => !installed.has(id)) + if (missing.length > 0) { + begin() + for (const row of missing) { + installed.add(row.id) + write({ type: `insert`, value: row }) + } + commit() + } + return cold ? Promise.resolve() : true + }, + } + }, + }, + }) + const query = () => + createLiveQueryCollection({ + query: (q) => + q + .from({ row: source }) + .orderBy(({ row }) => row.rank, `asc`) + .orderBy(({ row }) => row.tie, `asc`) + .offset(1) + .limit(2), + startSync: true, + }) + const owner = query() + try { + await owner.preload() + cold = false + const sibling = query() + try { + expect(sibling.status).toBe(`ready`) + expect(sibling.toArray.map(({ id }) => id)).toEqual([2, 3]) + expect(warmRequests).toEqual([ + { ordered: true, filtered: false, limit: 3 }, + { ordered: false, filtered: true, limit: undefined }, + ]) + } finally { + await sibling.cleanup() + } + } finally { + await owner.cleanup() + await source.cleanup() + } + }) +}) + describe(`nullable multi-term lifecycle product`, () => { const cells: Array = ([`first`, `last`] as const) .flatMap((primaryNulls) => diff --git a/packages/db/tests/query/ordered-source-loader-state.test.ts b/packages/db/tests/query/ordered-source-loader-state.test.ts index 4f1cc78e11..5272bce032 100644 --- a/packages/db/tests/query/ordered-source-loader-state.test.ts +++ b/packages/db/tests/query/ordered-source-loader-state.test.ts @@ -98,6 +98,92 @@ function fakeSubscription( } describe(`Ordered source request ownership`, () => { + it(`keeps nested refinement pending when successful cleanup reports an error`, async () => { + const requests: Array = [] + const releases: Array = [] + const participants: Array> = [] + const participantErrors = new Map, unknown>() + const cleanupError = new Error(`older prefix cleanup failed`) + let boundary = 1 + const request = (method: Observed[`method`], options: RequestOptions) => { + const index = requests.length + const acquisition: LoadSubsetOptions = { + orderBy: options.orderBy, + limit: options.limit, + where: options.where, + } + const deferred = createDeferred() + requests.push({ method, options, acquisition, deferred }) + options.onLoadSubsetResult?.( + index === 2 ? true : deferred.promise, + acquisition, + () => { + releases.push(index) + if (index === 0) throw cleanupError + }, + ) + } + const subscription = { + readOrderedSnapshot: () => [ + { + type: `insert`, + key: boundary, + value: { id: boundary, rank: boundary }, + }, + ], + setOrderByIndex: () => {}, + requestLimitedSnapshot: (options: RequestOptions) => + request(`limited`, options), + requestSnapshot: (options: RequestOptions) => + request(`snapshot`, options), + } as unknown as CollectionSubscription + const loader = new OrderedSourceLoader( + createOrderByInfo(), + subscription, + `row`, + (result) => { + if (result instanceof Promise) { + participants.push(result) + void result.catch((error) => participantErrors.set(result, error)) + } + }, + ) + + try { + loader.start() + requests[0]!.deferred.resolve() + await participants[0] + requests[1]!.deferred.resolve() + await participants[1] + + boundary = 2 + loader.invalidateCursor() + const result = loader.loadMore(1)! + expect(requests).toHaveLength(4) + expect(participants).toHaveLength(4) + await flushPromises() + const recent = participants.slice(2) + const cleanup = recent.find( + (participant) => participantErrors.get(participant) === cleanupError, + )! + const nested = recent.find((participant) => participant !== cleanup)! + const pending = ( + loader as unknown as { pending: Promise | undefined } + ).pending + expect(result).toBe(pending) + expect(pending).toBe(nested) + await expect(cleanup).rejects.toBe(cleanupError) + + expect(loader.loadMore()).toBe(nested) + expect(requests).toHaveLength(4) + expect(releases).toEqual([0]) + } finally { + loader.dispose() + for (const observed of requests) observed.deferred.resolve() + await Promise.allSettled(participants) + } + }) + it.each([`success`, `failure`] as const)( `keeps a failed public window private when its older tie request ends in %s`, async (olderOutcome) => { diff --git a/packages/db/tests/query/ordered-source-loader.test.ts b/packages/db/tests/query/ordered-source-loader.test.ts index c08093a0db..4872f89163 100644 --- a/packages/db/tests/query/ordered-source-loader.test.ts +++ b/packages/db/tests/query/ordered-source-loader.test.ts @@ -192,8 +192,14 @@ describe(`OrderedSourceLoader`, () => { expect(() => loader.start()).toThrow(failure) } else { loader.start() - if (outcome === `success`) await pendingPromise(loader) - else await expect(pendingPromise(loader)).rejects.toBe(failure) + if (outcome === `success`) { + const pending = pendingPromise(loader) + if (route === `page` || route === `prefix`) + expect(pending).toBeInstanceOf(Promise) + else expect(pending).toBeUndefined() + } else { + await expect(pendingPromise(loader)).rejects.toBe(failure) + } } // Drain the synchronous boundary's own settlement as well as its parent. await Promise.resolve() @@ -209,7 +215,9 @@ describe(`OrderedSourceLoader`, () => { if (outcome === `success`) { // Ordered loads establish a cursor and refine ties; neither a tie // load nor a full-source load may restart that refinement step. - expect(boundaryReads).toBe(route === `full-source` ? 0 : 1) + expect(boundaryReads).toBe( + route === `full-source` ? 0 : route === `boundary` ? 2 : 1, + ) expect(requests).toHaveLength(route === `full-source` ? 1 : 2) if (route === `page` || route === `prefix`) { expect(requests[1]!.options.where).toBeDefined() @@ -235,6 +243,47 @@ describe(`OrderedSourceLoader`, () => { }, ) + it.each([`page`, `prefix`, `full-source`] as const)( + `settles a fully synchronous $route acquisition before start returns`, + (route) => { + const requests: Array = [] + const request = (method: string, options: RequestOptions): void => { + requests.push(method) + options.onLoadSubsetResult?.(true, options, () => {}) + } + const subscription = { + setOrderByIndex: () => {}, + readOrderedSnapshot: () => [{ value: { rank: 1 } }], + requestLimitedSnapshot: (options: RequestOptions) => + request(`limited`, options), + requestSnapshot: (options: RequestOptions) => + request(`snapshot`, options), + } + const loader = new OrderedSourceLoader( + createOrderByInfo({ + dataNeeded: () => 0, + ...(route === `prefix` ? { index: undefined } : {}), + requiresFullSource: route === `full-source`, + }), + subscription as unknown as CollectionSubscription, + `row`, + ) + try { + loader.start() + expect(pendingPromise(loader)).toBeUndefined() + expect(requests).toEqual( + route === `full-source` + ? [`snapshot`] + : route === `prefix` + ? [`snapshot`, `snapshot`] + : [`limited`, `snapshot`], + ) + } finally { + loader.dispose() + } + }, + ) + it(`recovers authoritatively when reading a settled boundary fails`, async () => { const failure = new Error(`boundary read failed`) const requests: Array<{ method: string; options: RequestOptions }> = [] diff --git a/packages/db/tests/query/pagination-oracle.property.test.ts b/packages/db/tests/query/pagination-oracle.property.test.ts index 91ffdd45ab..568492bfbc 100644 --- a/packages/db/tests/query/pagination-oracle.property.test.ts +++ b/packages/db/tests/query/pagination-oracle.property.test.ts @@ -1930,7 +1930,6 @@ async function runRejectedCursorRetryAfterMutation(): Promise { query .from({ row: source }) .orderBy(({ row }) => row.rank, `asc`) - .orderBy(({ row }) => row.id, `asc`) .limit(1), ) diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index dd1d2d5848..f474ab6eba 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -80,8 +80,6 @@ export { isChangeMessage, isControlMessage } from '@electric-sql/client' const debug = DebugModule.debug(`ts/db:electric`) -const FORCE_DISCONNECT_AND_REFRESH_TIMEOUT_MS = 250 - /** * Symbol for internal test hooks (hidden from public API) */ @@ -758,56 +756,6 @@ function createLoadSubsetDedupe>({ const { cursor, where, orderBy, limit } = opts - // When the stream is already up-to-date, it may be in a long-poll wait. - // Forcing a disconnect-and-refresh ensures requestSnapshot gets a response - // from a fresh server round-trip rather than waiting for the current poll to end. - // Some native fetch implementations (notably React Native/Expo) may not abort - // long-poll requests promptly. Bound the wait so on-demand live queries don't - // remain loading until the long-poll naturally times out. - // If the refresh fails or times out, we fall through to requestSnapshot which - // still works. - if (stream.isUpToDate) { - let timeoutId: ReturnType | undefined - const abortSignals = [signal, opts.signal].filter( - (candidate): candidate is AbortSignal => candidate !== undefined, - ) - let rejectAbort: (reason: unknown) => void = () => {} - const aborted = new Promise((_resolve, reject) => { - rejectAbort = reject - }) - const abort = (event: Event) => - rejectAbort(abortReason(event.currentTarget as AbortSignal)) - for (const abortSignal of abortSignals) { - abortSignal.addEventListener(`abort`, abort, { once: true }) - } - try { - await Promise.race([ - stream.forceDisconnectAndRefresh(), - new Promise((resolve) => { - timeoutId = setTimeout( - resolve, - FORCE_DISCONNECT_AND_REFRESH_TIMEOUT_MS, - ) - }), - aborted, - ]) - } catch (error) { - if (signal.aborted || opts.signal?.aborted) throw error - if (handleSnapshotError(error, `forceDisconnectAndRefresh`)) { - return - } - debug( - `${logPrefix}forceDisconnectAndRefresh failed, proceeding to requestSnapshot: %o`, - error, - ) - } finally { - clearTimeout(timeoutId) - for (const abortSignal of abortSignals) { - abortSignal.removeEventListener(`abort`, abort) - } - } - } - throwIfAborted() // Upstream limitation: ShapeStream.requestSnapshot() publishes its rows diff --git a/packages/electric-db-collection/tests/electric-live-query.test.ts b/packages/electric-db-collection/tests/electric-live-query.test.ts index 19468bfc78..e8112ecd46 100644 --- a/packages/electric-db-collection/tests/electric-live-query.test.ts +++ b/packages/electric-db-collection/tests/electric-live-query.test.ts @@ -772,6 +772,40 @@ describe(`Electric Collection with Live Query - syncMode integration`, () => { expect(liveQuery.size).toBeGreaterThan(2) }) + it(`loads a prefix while publishing only the requested offset window`, async () => { + const electricCollection = createElectricCollectionWithSyncMode(`on-demand`) + + simulateInitialSync([]) + mockRequestSnapshot.mockResolvedValueOnce({ + data: sampleUsers.map((user) => ({ + headers: { operation: `insert` }, + key: user.id, + value: user, + })), + }) + + const liveQuery = createLiveQueryCollection({ + id: `offset-live-query`, + startSync: true, + query: (q) => + q + .from({ user: electricCollection }) + .orderBy(({ user }) => user.id, `asc`) + .limit(2) + .offset(2), + }) + + await vi.waitFor(() => expect(liveQuery.status).toBe(`ready`)) + + expect(mockRequestSnapshot.mock.calls[0]?.[0]).toMatchObject({ + limit: 4, + orderBy: `"id" NULLS FIRST`, + params: {}, + }) + expect(mockRequestSnapshot.mock.calls[0]?.[0]).not.toHaveProperty(`offset`) + expect(liveQuery.toArray.map((user) => user.id)).toEqual([3, 4]) + }) + it(`should trigger fetchSnapshot in progressive mode when live query needs more data`, async () => { const electricCollection = createElectricCollectionWithSyncMode(`progressive`) diff --git a/packages/electric-db-collection/tests/electric-sdk-delivery.property.test.ts b/packages/electric-db-collection/tests/electric-sdk-delivery.property.test.ts index 6ad1de7ff3..d03a2eae6f 100644 --- a/packages/electric-db-collection/tests/electric-sdk-delivery.property.test.ts +++ b/packages/electric-db-collection/tests/electric-sdk-delivery.property.test.ts @@ -24,7 +24,7 @@ type Request = { url: URL; respond: (response: Response) => void } function controlledHttp() { const queued: Array = [] const waiting: Array<{ - snapshot: boolean + snapshot: boolean | undefined resolve: (request: Request) => void reject: (error: unknown) => void }> = [] @@ -68,7 +68,9 @@ function controlledHttp() { signal?.addEventListener(`abort`, abort, { once: true }) active.add(abort) const index = waiting.findIndex( - (waiter) => waiter.snapshot === isSnapshot(request), + (waiter) => + waiter.snapshot === undefined || + waiter.snapshot === isSnapshot(request), ) if (index >= 0) waiting.splice(index, 1)[0]!.resolve(request) else queued.push(request) @@ -85,6 +87,13 @@ function controlledHttp() { waiting.push({ snapshot, resolve, reject }), ) }, + takeAny: (): Promise => { + if (closed) return Promise.reject(closedError) + if (queued.length > 0) return Promise.resolve(queued.shift()!) + return new Promise((resolve, reject) => + waiting.push({ snapshot: undefined, resolve, reject }), + ) + }, close: () => { closed = true for (const abort of [...active]) abort() @@ -259,6 +268,92 @@ it(`rejects a snapshot driver that delivers boundaries but drops response rows`, }) }) +it(`lets requestSnapshot own the warm-stream transport transition`, async () => { + const http = controlledHttp() + let delivery = deferred() + const subscribe = ShapeStream.prototype.subscribe + const spy = vi + .spyOn(ShapeStream.prototype, `subscribe`) + .mockImplementation(function (this: ShapeStream, callback, onError) { + return subscribe.call( + this, + (messages) => { + const result = callback(messages) + delivery.resolve() + return result + }, + onError, + ) + }) + const collection = createCollection( + electricCollectionOptions({ + id: `sdk-warm-snapshot-${++sequence}`, + shapeOptions: { + url: `http://test-url/warm-snapshot-${sequence}`, + params: { table: `rows` }, + fetchClient: http.fetchClient, + }, + syncMode: `on-demand`, + startSync: true, + getKey: (row) => row.id, + }), + ) + + return withElectricCleanup(async () => { + const initial = await atCheckpoint(http.take(), `initial live request`) + initial.respond( + new Response( + JSON.stringify([ + { + headers: { + control: `up-to-date`, + global_last_seen_lsn: `1`, + }, + }, + ]), + { headers: headers(1) }, + ), + ) + await atCheckpoint(delivery.promise, `initial up-to-date delivery`) + + // Hold the next live poll. The installed SDK's requestSnapshot owns the + // pause/abort transition from this poll to the subset request. + await atCheckpoint(http.take(), `warm live poll`) + delivery = deferred() + const loading = collection._sync.loadSubset({ + where: new IR.Func(`gte`, [new IR.PropRef([`id`]), new IR.Value(1)]), + }) + const request = await atCheckpoint( + http.takeAny(), + `first request after warm loadSubset`, + ) + expect(request.url.searchParams.has(`subset__where`)).toBe(true) + expect(http.activeCount()).toBe(1) + request.respond( + new Response( + JSON.stringify({ + metadata: { + xmin: `10`, + xmax: `20`, + xip_list: [], + database_lsn: `10`, + snapshot_mark: 2, + }, + data: [], + }), + { headers: headers(2) }, + ), + ) + await atCheckpoint(delivery.promise, `warm snapshot delivery`) + await atCheckpoint(Promise.resolve(loading), `warm snapshot completion`) + }, [ + () => collection.cleanup(), + () => http.close(), + () => spy.mockRestore(), + () => expect(http.activeCount()).toBe(0), + ]) +}) + async function checkMembership( leftWidth: number, rightWidth: number, diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index bcb7528bd2..25cc70705c 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -3128,94 +3128,11 @@ describe(`Electric Integration`, () => { }, ) - it.each([true, false])( - `settles reasonless cancellation after refresh with DOMException available %s`, - async (hasDOMException) => { - const originalDOMException = globalThis.DOMException - const controller = new NativeAbortController() - const refresh = createDeferred() - mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) - const testCollection = createOnDemandCollection( - `reasonless-refresh-abort`, - ) - try { - const load = testCollection._sync.loadSubset({ - limit: 10, - signal: controller.signal, - }) - const outcome = Promise.resolve(load).then( - () => undefined, - (error: unknown) => error, - ) - await Promise.resolve() - // Model a platform signal without reason; no event is required for - // the post-refresh cancellation check to observe its terminal state. - Object.defineProperty(controller.signal, `aborted`, { value: true }) - Object.defineProperty(controller.signal, `reason`, { - value: undefined, - }) - if (!hasDOMException) vi.stubGlobal(`DOMException`, undefined) - refresh.resolve() - await expect(outcome).resolves.toMatchObject({ name: `AbortError` }) - expect(mockRequestSnapshot).not.toHaveBeenCalled() - } finally { - vi.stubGlobal(`DOMException`, originalDOMException) - refresh.resolve() - await testCollection.cleanup() - } - }, - ) - - it(`cancels a pending refresh wait when the collection is cleaned up`, async () => { - vi.useFakeTimers() - const schedule = vi.spyOn(globalThis, `setTimeout`) - const cancel = vi.spyOn(globalThis, `clearTimeout`) - const refresh = createDeferred() - mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) - const testCollection = createOnDemandCollection( - `on-demand-refresh-cleanup-test`, - ) - try { - const load = Promise.resolve( - testCollection._sync.loadSubset({ limit: 10 }), - ) - const loadError = load.then( - () => undefined, - (error: unknown) => error, - ) - const refreshTimerIndex = schedule.mock.calls.findIndex( - ([, delay]) => delay === 250, - ) - expect(refreshTimerIndex).toBeGreaterThanOrEqual(0) - const refreshTimer = schedule.mock.results[refreshTimerIndex]!.value - - await Promise.resolve() - await testCollection.cleanup() - await expect(loadError).resolves.toMatchObject({ name: `AbortError` }) - expect(mockRequestSnapshot).not.toHaveBeenCalled() - // The shared collection GC timer may still exist; this wait must not. - expect(cancel).toHaveBeenCalledWith(refreshTimer) - - refresh.resolve() - await refresh.promise - await load.catch(() => undefined) - expect(mockRequestSnapshot).not.toHaveBeenCalled() - } finally { - refresh.resolve() - await testCollection.cleanup() - schedule.mockRestore() - cancel.mockRestore() - vi.useRealTimers() - } - }) - - it(`should refresh the stream before requesting on-demand snapshots when already up-to-date`, async () => { + it(`lets requestSnapshot transition an up-to-date stream to a subset request`, async () => { vi.clearAllMocks() const config = { - id: `on-demand-refresh-before-snapshot-test`, + id: `on-demand-snapshot-up-to-date-test`, shapeOptions: { url: `http://test-url`, params: { @@ -3233,132 +3150,47 @@ describe(`Electric Integration`, () => { await testCollection._sync.loadSubset({ limit: 10 }) - expect(mockForceDisconnectAndRefresh).toHaveBeenCalledTimes(1) - expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) - const refreshCall = - mockForceDisconnectAndRefresh.mock.invocationCallOrder[0]! - const snapshotCall = mockRequestSnapshot.mock.invocationCallOrder[0]! - expect(refreshCall).toBeLessThan(snapshotCall) - }) - - it(`should fall through to requestSnapshot when forceDisconnectAndRefresh fails`, async () => { - vi.clearAllMocks() - - const config = { - id: `on-demand-refresh-fallthrough-test`, - shapeOptions: { - url: `http://test-url`, - params: { - table: `test_table`, - }, - }, - syncMode: `on-demand` as const, - getKey: (item: Row) => item.id as number, - startSync: true, - } - - const testCollection = createCollection(electricCollectionOptions(config)) - - mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockImplementationOnce(() => { - return Promise.reject(new Error(`PauseLock held`)) - }) - - await testCollection._sync.loadSubset({ limit: 10 }) - - expect(mockForceDisconnectAndRefresh).toHaveBeenCalledTimes(1) + expect(mockForceDisconnectAndRefresh).not.toHaveBeenCalled() expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) + await testCollection.cleanup() }) - it(`should request the snapshot after the refresh timeout and ignore late fulfillment`, async () => { - vi.useFakeTimers() - const refresh = createDeferred() - mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh.promise) + it(`issues one snapshot and no ordinary refresh for each distinct demand`, async () => { const testCollection = createOnDemandCollection( - `on-demand-refresh-timeout-fulfillment-test`, + `on-demand-distinct-snapshot-count-test`, ) - try { - let loadSettled = false - const load = Promise.resolve( - testCollection._sync.loadSubset({ limit: 10 }), - ).then(() => { - loadSettled = true - }) - - await vi.advanceTimersByTimeAsync(249) - expect(mockRequestSnapshot).not.toHaveBeenCalled() - expect(loadSettled).toBe(false) - - await vi.advanceTimersByTimeAsync(1) - await load - expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) - expect(loadSettled).toBe(true) - - refresh.resolve() - await refresh.promise - await Promise.resolve() - expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) - } finally { - refresh.resolve() - await testCollection.cleanup() - vi.useRealTimers() - } - }) - it(`should handle late refresh rejection after requesting the snapshot`, async () => { - vi.useFakeTimers() - let rejectRefresh: (error: Error) => void = () => {} - let resolveRefresh: () => void = () => {} - const refresh = new Promise((resolve, reject) => { - resolveRefresh = resolve - rejectRefresh = reject - }) - mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockReturnValueOnce(refresh) - const testCollection = createOnDemandCollection( - `on-demand-refresh-timeout-rejection-test`, - ) try { - const load = testCollection._sync.loadSubset({ limit: 10 }) - await vi.advanceTimersByTimeAsync(250) - await load + for (let limit = 1; limit <= 10; limit++) { + await testCollection._sync.loadSubset({ limit }) + } - rejectRefresh(new Error(`late refresh failure`)) - await expect(refresh).rejects.toThrow(`late refresh failure`) - await Promise.resolve() - expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) + expect(mockRequestSnapshot).toHaveBeenCalledTimes(10) + expect(mockForceDisconnectAndRefresh).not.toHaveBeenCalled() } finally { - resolveRefresh() await testCollection.cleanup() - vi.useRealTimers() } }) - it(`should clear the refresh timeout when refresh settles early`, async () => { - vi.useFakeTimers() - const schedule = vi.spyOn(globalThis, `setTimeout`) - const cancel = vi.spyOn(globalThis, `clearTimeout`) - mockStream.isUpToDate = true - mockForceDisconnectAndRefresh.mockResolvedValueOnce(undefined) + it(`retries a warm snapshot after its previous request rejects`, async () => { + const failure = new Error(`snapshot failed`) + mockRequestSnapshot + .mockRejectedValueOnce(failure) + .mockResolvedValueOnce(undefined) const testCollection = createOnDemandCollection( - `on-demand-refresh-clears-timeout-test`, + `on-demand-snapshot-retry-test`, ) + try { + await expect( + Promise.resolve(testCollection._sync.loadSubset({ limit: 10 })), + ).rejects.toBe(failure) await testCollection._sync.loadSubset({ limit: 10 }) - const refreshTimerIndex = schedule.mock.calls.findIndex( - ([, delay]) => delay === 250, - ) - expect(refreshTimerIndex).toBeGreaterThanOrEqual(0) - const refreshTimer = schedule.mock.results[refreshTimerIndex]!.value - expect(mockRequestSnapshot).toHaveBeenCalledTimes(1) - expect(cancel).toHaveBeenCalledWith(refreshTimer) + expect(mockRequestSnapshot).toHaveBeenCalledTimes(2) + expect(mockForceDisconnectAndRefresh).not.toHaveBeenCalled() } finally { await testCollection.cleanup() - schedule.mockRestore() - cancel.mockRestore() - vi.useRealTimers() } }) diff --git a/packages/react-db/package.json b/packages/react-db/package.json index 6fc3109fe7..2f107c4c4f 100644 --- a/packages/react-db/package.json +++ b/packages/react-db/package.json @@ -47,18 +47,16 @@ "skills" ], "dependencies": { - "@tanstack/db": "workspace:*", - "use-sync-external-store": "^1.6.0" + "@tanstack/db": "workspace:*" }, "peerDependencies": { - "react": ">=16.8.0" + "react": ">=18.0.0" }, "devDependencies": { "@electric-sql/client": "^1.5.15", "@testing-library/react": "^16.3.2", "@types/react": "^19.2.13", "@types/react-dom": "^19.2.3", - "@types/use-sync-external-store": "^1.5.0", "@vitest/coverage-istanbul": "^3.2.4", "react": "^19.2.4", "react-dom": "^19.2.4" diff --git a/packages/react-db/src/useLiveQuery.ts b/packages/react-db/src/useLiveQuery.ts index c9c0126465..568009f21d 100644 --- a/packages/react-db/src/useLiveQuery.ts +++ b/packages/react-db/src/useLiveQuery.ts @@ -3,6 +3,7 @@ import { useRef, useSyncExternalStore } from 'react' import { BaseQueryBuilder, + IR, UnhashableQueryIRError, createLiveQueryCollection, createLiveQueryObserver, @@ -31,6 +32,9 @@ import type { } from '@tanstack/db' const DEFAULT_GC_TIME_MS = 1 // Live queries created by useLiveQuery are cleaned up immediately (0 disables GC) +// Suspense renders can be abandoned before React subscribes. Keep their +// collection briefly so a nearby retry can commit without starting over. +const DEFAULT_SUSPENSE_GC_TIME_MS = 5000 const DERIVED_IDENTITY_SINGLE_RENDER_WARN_MS = 16 const DERIVED_IDENTITY_RENDER_COUNT_WARN_THRESHOLD = 10 const DERIVED_IDENTITY_TOTAL_WARN_MS = 50 @@ -39,6 +43,72 @@ const warnedDerivedIdentityCallsites = new Set() const warnedUnhashableIdentityCallsites = new Set() const unpreparedQueryValue = Symbol(`unpreparedQueryValue`) +type SuspenseCollection = Collection +type SuspenseCollectionEntry = { + collection: SuspenseCollection + removeStatusListeners: () => void +} + +const unscopedSuspenseCollections = new Map() +const suspenseCollectionsByClient = new WeakMap< + DbClient, + Map +>() +const suspenseSourceIds = new WeakMap() +let nextSuspenseSourceId = 0 + +function getSuspenseSourceId(source: object): number { + let id = suspenseSourceIds.get(source) + if (id === undefined) { + id = ++nextSuspenseSourceId + suspenseSourceIds.set(source, id) + } + return id +} + +function getUnscopedSuspenseKey( + preparedValue: unknown, + queryHash: string, +): string { + const query = + preparedValue instanceof BaseQueryBuilder + ? preparedValue + : preparedValue && + typeof preparedValue === `object` && + `query` in preparedValue && + preparedValue.query instanceof BaseQueryBuilder + ? preparedValue.query + : undefined + if (!query) return queryHash + const sourceIds = IR.collectCollectionSources(query._getQuery()).map( + ({ collection }) => getSuspenseSourceId(collection), + ) + return `${sourceIds.join(`,`)}:${queryHash}` +} + +function getSuspenseCollections( + client: DbClient | undefined, +): Map { + if (!client) return unscopedSuspenseCollections + let collections = suspenseCollectionsByClient.get(client) + if (!collections) { + collections = new Map() + suspenseCollectionsByClient.set(client, collections) + } + return collections +} + +function releaseSuspenseCollection( + collections: Map, + queryHash: string, + collection: SuspenseCollection, +): void { + const entry = collections.get(queryHash) + if (entry?.collection !== collection) return + collections.delete(queryHash) + entry.removeStatusListeners() +} + export type DerivedIdentityProfiler = { renderCount: number totalMs: number @@ -120,7 +190,16 @@ function getCurrentTime(): number { function getWarningCallsite(stackIndex: number): string { const stack = new Error().stack ?? `unknown` - return stack.split(`\n`)[stackIndex]?.trim() ?? stack + const lines = stack.split(`\n`) + const userFrame = lines + .slice(1) + .find( + (line) => + !line.includes(`useLiveQuery.ts`) && + !line.includes(`useLiveSuspenseQuery.ts`) && + !line.includes(`useLiveInfiniteQuery.ts`), + ) + return userFrame?.trim() ?? lines[stackIndex]?.trim() ?? stack } function warnDerivedIdentityHotPath( @@ -256,7 +335,10 @@ export function warnUnhashableDerivedIdentity( ) } -function createCollectionFromPreparedQuery(value: unknown) { +function createCollectionFromPreparedQuery( + value: unknown, + defaultGcTime = DEFAULT_GC_TIME_MS, +) { if (value === undefined || value === null) { return null } @@ -270,14 +352,14 @@ function createCollectionFromPreparedQuery(value: unknown) { return createLiveQueryCollection({ query: value, startSync: true, - gcTime: DEFAULT_GC_TIME_MS, + gcTime: defaultGcTime, }) } if (typeof value === `object`) { return createLiveQueryCollection({ startSync: true, - gcTime: DEFAULT_GC_TIME_MS, + gcTime: defaultGcTime, ...(value as LiveQueryCollectionConfig), }) } @@ -643,6 +725,22 @@ export function useLiveQuery< export function useLiveQuery( configOrQueryOrCollection: any, deps?: Array, +) { + return useLiveQueryImpl(configOrQueryOrCollection, deps) +} + +/** @internal Shared implementation for the Suspense wrapper. */ +export function useLiveQueryForSuspense( + configOrQueryOrCollection: any, + deps: Array | undefined, +) { + return useLiveQueryImpl(configOrQueryOrCollection, deps, true) +} + +function useLiveQueryImpl( + configOrQueryOrCollection: any, + deps: Array | undefined, + forSuspense = false, ) { const contextDbClient = useOptionalDbClient() // Check if it's already a collection @@ -676,6 +774,7 @@ export function useLiveQuery( null, ) const queryHashRef = useRef(undefined) + const suspenseKeyRef = useRef(undefined) const identityErrorRef = useRef(undefined) const queryKey = !inputIsCollection @@ -747,6 +846,47 @@ export function useLiveQuery( warnDeprecatedDepsArray() } + const canReuseSuspenseKey = + forSuspense && + !inputIsCollection && + queryHash !== undefined && + !dbClient && + collectionRef.current !== null && + clientRef.current === dbClient && + queryHashRef.current === queryHash && + suspenseKeyRef.current !== undefined + + if ( + forSuspense && + !inputIsCollection && + queryHash && + !dbClient && + !canReuseSuspenseKey && + preparedQueryValue === unpreparedQueryValue + ) { + preparedQueryValue = prepareQueryValue( + configOrQueryOrCollection, + dbClient, + deferredCollectionsRef.current, + ) + } + + const suspenseKey = + queryHash && !dbClient + ? canReuseSuspenseKey + ? suspenseKeyRef.current + : getUnscopedSuspenseKey(preparedQueryValue, queryHash) + : queryHash + + const suspenseCollections = + forSuspense && !inputIsCollection && suspenseKey + ? getSuspenseCollections(dbClient) + : undefined + const suspenseEntry = suspenseKey + ? suspenseCollections?.get(suspenseKey) + : undefined + const suspenseCollection = suspenseEntry?.collection + const identityChanged = depsRef.current === null || (deps !== undefined @@ -792,21 +932,59 @@ export function useLiveQuery( collectionRef.current = configOrQueryOrCollection configRef.current = configOrQueryOrCollection } else { - if (preparedQueryValue === unpreparedQueryValue) { - preparedQueryValue = prepareQueryValue( - configOrQueryOrCollection, - dbClient, - deferredCollectionsRef.current, - ) + if (suspenseCollection) { + collectionRef.current = suspenseCollection + } else { + if (preparedQueryValue === unpreparedQueryValue) { + preparedQueryValue = prepareQueryValue( + configOrQueryOrCollection, + dbClient, + deferredCollectionsRef.current, + ) + } + collectionRef.current = createCollectionFromPreparedQuery( + preparedQueryValue, + forSuspense ? DEFAULT_SUSPENSE_GC_TIME_MS : DEFAULT_GC_TIME_MS, + ) as SuspenseCollection | null + if (suspenseCollections && suspenseKey && collectionRef.current) { + const collection = collectionRef.current + const removeCleanupListener = collection.on(`status:cleaned-up`, () => + releaseSuspenseCollection( + suspenseCollections, + suspenseKey, + collection, + ), + ) + const removeErrorListener = collection.on(`status:error`, () => { + // Keep the failed collection through React's immediate retry so + // the hook can throw its actual load error to an ErrorBoundary. + // Retire it afterward so a later boundary reset starts fresh. + setTimeout( + () => + releaseSuspenseCollection( + suspenseCollections, + suspenseKey, + collection, + ), + 0, + ) + }) + const entry: SuspenseCollectionEntry = { + collection, + removeStatusListeners: () => { + removeCleanupListener() + removeErrorListener() + }, + } + suspenseCollections.set(suspenseKey, entry) + } } - collectionRef.current = createCollectionFromPreparedQuery( - preparedQueryValue, - ) as Collection configRef.current = configOrQueryOrCollection depsRef.current = [...identityDeps] } clientRef.current = dbClient queryHashRef.current = queryHash + suspenseKeyRef.current = suspenseKey identityErrorRef.current = identityError } diff --git a/packages/react-db/src/useLiveSuspenseQuery.ts b/packages/react-db/src/useLiveSuspenseQuery.ts index a4e8dec5b1..69de509e43 100644 --- a/packages/react-db/src/useLiveSuspenseQuery.ts +++ b/packages/react-db/src/useLiveSuspenseQuery.ts @@ -1,7 +1,7 @@ 'use client' import { useRef } from 'react' -import { useLiveQuery } from './useLiveQuery' +import { useLiveQueryForSuspense } from './useLiveQuery' import { getLiveQueryResultInfo } from './live-query-internals' import type { UseLiveQueryConfig } from './useLiveQuery' import type { @@ -173,19 +173,9 @@ export function useLiveSuspenseQuery( // Use useLiveQuery to handle collection management and reactivity const result = deps === undefined - ? useLiveQuery(configOrQueryOrCollection) - : useLiveQuery(configOrQueryOrCollection, deps) - const queryInfo = getLiveQueryResultInfo(result) + ? useLiveQueryForSuspense(configOrQueryOrCollection, undefined) + : useLiveQueryForSuspense(configOrQueryOrCollection, deps) - // Reset promise and ready state when query identity changes - if (collectionRef.current !== result.collection) { - promiseRef.current = null - collectionRef.current = result.collection - hasBeenReadyRef.current = false - } - - // SUSPENSE LOGIC: Throw promise or error based on collection status - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition if (!result.isEnabled) { // Suspense queries cannot be disabled - this matches TanStack Query's useSuspenseQuery behavior throw new Error( @@ -197,6 +187,17 @@ export function useLiveSuspenseQuery( ) } + const queryInfo = getLiveQueryResultInfo(result) + + // Reset promise and ready state when query identity changes + if (collectionRef.current !== result.collection) { + promiseRef.current = null + collectionRef.current = result.collection + hasBeenReadyRef.current = false + } + + // SUSPENSE LOGIC: Throw promise or error based on collection status + const collectionStatus = result.collection.status // Track when we reach ready state @@ -229,14 +230,11 @@ export function useLiveSuspenseQuery( `Cannot stream this live query during SSR because ${reason}. Provide an explicit serializable queryKey.`, ) } - // Create or reuse promise for current collection if (!promiseRef.current) { promiseRef.current = queryInfo.observer.preload() } - // THROW PROMISE - React Suspense catches this (React 18+ required) - // Note: We don't check React version here. In React <18, this will be caught - // by an Error Boundary, which provides a reasonable failure mode. + // React Suspense catches this promise and retries after preload settles. throw promiseRef.current } diff --git a/packages/react-db/tests/useLiveQuery.uncommitted-render.test.tsx b/packages/react-db/tests/useLiveQuery.uncommitted-render.test.tsx index 7b1126cb2a..9e4378829a 100644 --- a/packages/react-db/tests/useLiveQuery.uncommitted-render.test.tsx +++ b/packages/react-db/tests/useLiveQuery.uncommitted-render.test.tsx @@ -1,13 +1,14 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { Component, Suspense } from 'react' import { act, cleanup, render } from '@testing-library/react' -import { createCollection, createLiveQueryCollection } from '@tanstack/db' +import { createCollection, createLiveQueryCollection, eq } from '@tanstack/db' import { useLiveQuery } from '../src/useLiveQuery' import { useLiveSuspenseQuery } from '../src/useLiveSuspenseQuery' import { mockSyncCollectionOptions, resetCleanupQueue, } from '../../db/tests/utils' +import type { InitialQueryBuilder } from '@tanstack/db' import type { ReactNode } from 'react' type Person = { id: string; name: string } @@ -189,4 +190,475 @@ describe(`live queries across uncommitted renders`, () => { await advanceTime(2) expect(source.subscriberCount).toBe(0) }) + + it(`reuses an asynchronous on-demand preload across the pre-mount suspense retry`, async () => { + let resolveLoad!: () => void + let loadSettled = false + let loadCount = 0 + let deliver = () => {} + const load = new Promise((resolve) => { + resolveLoad = resolve + }) + const never = new Promise(() => {}) + const source = createCollection({ + id: `uncommitted-async-on-demand`, + getKey: (person) => person.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + deliver = () => { + begin() + write({ type: `insert`, value: { id: `1`, name: `Alice` } }) + commit() + } + return { + loadSubset: () => { + loadCount++ + return loadSettled ? never : load + }, + } + }, + }, + }) + collections.push(source) + + const People = () => { + const { data } = useLiveSuspenseQuery({ + query: (q) => + q.from({ person: source }).where(({ person }) => eq(person.id, `1`)), + }) + return
{data.map((person) => person.name).join(`, `)}
+ } + + const view = render( + Loading}> + + , + ) + expect(view.getByText(`Loading`)).toBeDefined() + expect(loadCount).toBe(1) + + await act(async () => { + deliver() + loadSettled = true + resolveLoad() + await Promise.resolve() + }) + await advanceTime(1) + + expect(view.getByText(`Alice`)).toBeDefined() + expect(loadCount).toBe(1) + }) + + it(`retains a shared precommit collection until every suspense consumer commits`, async () => { + let resolveLoad!: () => void + let releaseSecondRender!: () => void + let loadSettled = false + let secondRenderBlocked = true + let loadCount = 0 + let deliver = () => {} + const load = new Promise((resolve) => { + resolveLoad = resolve + }) + const secondRender = new Promise((resolve) => { + releaseSecondRender = resolve + }) + const never = new Promise(() => {}) + const source = createCollection({ + id: `two-uncommitted-async-on-demand-consumers`, + getKey: (person) => person.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + deliver = () => { + begin() + write({ type: `insert`, value: { id: `1`, name: `Alice` } }) + commit() + } + return { + loadSubset: () => { + loadCount++ + return loadSettled ? never : load + }, + } + }, + }, + }) + collections.push(source) + const config = { + query: (q: InitialQueryBuilder) => + q.from({ person: source }).where(({ person }) => eq(person.id, `1`)), + } + const First = () => { + const { data } = useLiveSuspenseQuery(config) + return
First: {data[0]?.name}
+ } + const Second = () => { + const { data } = useLiveSuspenseQuery(config) + if (secondRenderBlocked) throw secondRender + return
Second: {data[0]?.name}
+ } + + const view = render( + <> + First loading}> + + + Second loading}> + + + , + ) + expect(loadCount).toBe(1) + + await act(async () => { + deliver() + loadSettled = true + resolveLoad() + await Promise.resolve() + }) + await advanceTime(1) + expect(view.getByText(`First: Alice`)).toBeDefined() + expect(view.getByText(`Second loading`)).toBeDefined() + + await act(async () => { + secondRenderBlocked = false + releaseSecondRender() + await Promise.resolve() + }) + await advanceTime(1) + expect(view.getByText(`Second: Alice`)).toBeDefined() + expect(loadCount).toBe(1) + }) + + it(`shares an active suspense collection and retires it after cleanup`, async () => { + let releaseSecondRender!: () => void + let secondRenderBlocked = true + const secondRender = new Promise((resolve) => { + releaseSecondRender = resolve + }) + const source = makeSource(`committed-rerender-precommit-ownership`) + const config = { + query: (q: InitialQueryBuilder) => + q.from({ person: source }).where(({ person }) => eq(person.id, `1`)), + } + const liveCollections = new Map() + const People = ({ label, tick = 0 }: { label: string; tick?: number }) => { + const result = useLiveSuspenseQuery(config) + liveCollections.set(label, result.collection) + if (label === `Second` && secondRenderBlocked) throw secondRender + return ( +
+ {label}: {result.data[0]?.name} {tick} +
+ ) + } + + const firstView = render( + First loading}> + + , + ) + await advanceTime(1) + expect(firstView.getByText(`First: A 0`)).toBeDefined() + const firstCollection = liveCollections.get(`First`) + + const secondView = render( + Second loading}> + + , + ) + expect(secondView.getByText(`Second loading`)).toBeDefined() + + firstView.rerender( + First loading}> + + , + ) + expect(liveCollections.get(`First`)).toBe(firstCollection) + await act(async () => { + secondRenderBlocked = false + releaseSecondRender() + await Promise.resolve() + }) + await advanceTime(1) + expect(secondView.getByText(`Second: A 0`)).toBeDefined() + + const secondCollection = liveCollections.get(`Second`) + const thirdView = render( + Third loading}> + + , + ) + await advanceTime(1) + + expect(thirdView.getByText(`Third: A 0`)).toBeDefined() + expect(liveCollections.get(`Third`)).toBe(secondCollection) + + firstView.unmount() + secondView.unmount() + thirdView.unmount() + await advanceTime(5100) + + const fourthView = render( + Fourth loading}> + + , + ) + await advanceTime(1) + + expect(fourthView.getByText(`Fourth: A 0`)).toBeDefined() + expect(liveCollections.get(`Fourth`)).not.toBe(secondCollection) + }) + + it(`retains a ready collection while its next suspense consumer is pending commit`, async () => { + let releaseSecondRender!: () => void + let secondRenderBlocked = true + const secondRender = new Promise((resolve) => { + releaseSecondRender = resolve + }) + const source = makeSource(`pending-ready-suspense-consumer`) + const config = { + query: (q: InitialQueryBuilder) => + q.from({ person: source }).where(({ person }) => eq(person.id, `1`)), + } + const collectionsByLabel = new Map() + const People = ({ label }: { label: string }) => { + const result = useLiveSuspenseQuery(config) + collectionsByLabel.set(label, result.collection) + if (label === `Second` && secondRenderBlocked) throw secondRender + return ( +
+ {label}: {result.data[0]?.name} +
+ ) + } + + const firstView = render( + First loading}> + + , + ) + await advanceTime(1) + expect(firstView.getByText(`First: A`)).toBeDefined() + const firstCollection = collectionsByLabel.get(`First`) + + const secondView = render( + Second loading}> + + , + ) + expect(secondView.getByText(`Second loading`)).toBeDefined() + expect(collectionsByLabel.get(`Second`)).toBe(firstCollection) + + firstView.unmount() + await advanceTime(2) + + await act(async () => { + secondRenderBlocked = false + releaseSecondRender() + await Promise.resolve() + }) + await advanceTime(1) + + expect(secondView.getByText(`Second: A`)).toBeDefined() + expect(collectionsByLabel.get(`Second`)).toBe(firstCollection) + }) + + it(`does not share precommit queries across distinct source objects with the same id`, async () => { + const createSource = (name: string) => { + let resolveLoad!: () => void + let deliver = () => {} + let loadCount = 0 + const load = new Promise((resolve) => { + resolveLoad = resolve + }) + const source = createCollection({ + id: `same-source-id`, + getKey: (person) => person.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + deliver = () => { + begin() + write({ type: `insert`, value: { id: name, name } }) + commit() + } + return { + loadSubset: () => { + loadCount++ + return load + }, + } + }, + }, + }) + collections.push(source) + return { + source, + deliver: () => deliver(), + resolveLoad: () => resolveLoad(), + getLoadCount: () => loadCount, + } + } + const alice = createSource(`Alice`) + const bob = createSource(`Bob`) + const People = ({ source }: { source: typeof alice.source }) => { + const { data } = useLiveSuspenseQuery((q) => + q.from({ person: source }).select(({ person }) => person), + ) + return
{data.map(({ name }) => name).join(`, `)}
+ } + + const aliceView = render( + Alice loading}> + + , + ) + const bobView = render( + Bob loading}> + + , + ) + expect(alice.getLoadCount()).toBe(1) + expect(bob.getLoadCount()).toBe(1) + + await act(async () => { + alice.deliver() + alice.resolveLoad() + await Promise.resolve() + }) + await advanceTime(1) + expect(aliceView.getByText(`Alice`)).toBeDefined() + expect(bobView.getByText(`Bob loading`)).toBeDefined() + + await act(async () => { + bob.deliver() + bob.resolveLoad() + await Promise.resolve() + }) + await advanceTime(1) + expect(bobView.getByText(`Bob`)).toBeDefined() + }) + + it(`releases a rejected precommit query so an error-boundary retry can reload`, async () => { + let rejectFirst!: (error: Error) => void + let resolveSecond!: () => void + let deliverSecond = () => {} + let loadCount = 0 + const first = new Promise((_, reject) => { + rejectFirst = reject + }) + const second = new Promise((resolve) => { + resolveSecond = resolve + }) + const source = createCollection({ + id: `rejected-precommit-retry`, + getKey: (person) => person.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + deliverSecond = () => { + begin() + write({ type: `insert`, value: { id: `1`, name: `Recovered` } }) + commit() + } + return { + loadSubset: () => (++loadCount === 1 ? first : second), + } + }, + }, + }) + collections.push(source) + const People = () => { + const { data } = useLiveSuspenseQuery((q) => + q.from({ person: source }).select(({ person }) => person), + ) + return
{data.map(({ name }) => name).join(`, `)}
+ } + const logError = console.error.bind(console) + vi.spyOn(console, `error`).mockImplementation((...args: Array) => { + if (args.some((arg) => arg instanceof Error)) return + logError(...args) + }) + + const firstView = render( + + Loading}> + + + , + ) + await act(async () => { + rejectFirst(new Error(`first load failed`)) + await Promise.resolve() + }) + await advanceTime(2) + expect(firstView.getByText(`Failed`)).toBeDefined() + firstView.unmount() + await advanceTime(2) + + const secondView = render( + + Retry loading}> + + + , + ) + expect(loadCount).toBe(2) + await act(async () => { + deliverSecond() + resolveSecond() + await Promise.resolve() + }) + await advanceTime(1) + expect(secondView.getByText(`Recovered`)).toBeDefined() + }) + + it(`reclaims a precommit query abandoned before its load settles`, async () => { + let resolveLoad!: () => void + let deliver = () => {} + const load = new Promise((resolve) => { + resolveLoad = resolve + }) + const source = createCollection({ + id: `abandoned-precommit-query`, + getKey: (person) => person.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + deliver = () => { + begin() + write({ type: `insert`, value: { id: `1`, name: `Alice` } }) + commit() + } + return { loadSubset: () => load } + }, + }, + }) + collections.push(source) + const People = () => { + useLiveSuspenseQuery((q) => q.from({ person: source })) + return
Ready
+ } + const view = render( + Loading}> + + , + ) + expect(source.subscriberCount).toBeGreaterThan(0) + view.unmount() + await act(async () => { + deliver() + resolveLoad() + await Promise.resolve() + }) + await advanceTime(5100) + expect(source.subscriberCount).toBe(0) + }) }) diff --git a/packages/react-db/tests/useLiveSuspenseQuery.test.tsx b/packages/react-db/tests/useLiveSuspenseQuery.test.tsx index ef2c3fe49b..b2f811a5ff 100644 --- a/packages/react-db/tests/useLiveSuspenseQuery.test.tsx +++ b/packages/react-db/tests/useLiveSuspenseQuery.test.tsx @@ -467,6 +467,42 @@ describe(`useLiveSuspenseQuery`, () => { }).toThrow(/does not support disabled queries/) }) + it(`does not rebuild an explicitly keyed query after it commits`, async () => { + const collection = createCollection( + mockSyncCollectionOptions({ + id: `stable-keyed-suspense-query`, + getKey: (person: Person) => person.id, + initialData: initialPersons, + }), + ) + let queryExecutions = 0 + + const { result, rerender } = renderHook( + ({ render }: { render: number }) => { + void render + return useLiveSuspenseQuery({ + queryKey: [collection.id, `stable`], + query: (q) => { + queryExecutions += 1 + return q.from({ people: collection }) + }, + }) + }, + { + initialProps: { render: 0 }, + wrapper: SuspenseWrapper, + }, + ) + + await waitFor(() => expect(result.current.data).toHaveLength(3)) + const executionsAfterCommit = queryExecutions + + rerender({ render: 1 }) + + expect(result.current.data).toHaveLength(3) + expect(queryExecutions).toBe(executionsAfterCommit) + }) + it(`should work with config object`, async () => { const collection = createCollection( mockSyncCollectionOptions({ @@ -572,7 +608,11 @@ describe(`useLiveSuspenseQuery`, () => { ) const { result, unmount } = renderHook( - () => useLiveSuspenseQuery((q) => q.from({ persons: collection })), + () => + useLiveSuspenseQuery({ + query: (q) => q.from({ persons: collection }), + gcTime: 1, + }), { wrapper: SuspenseWrapper, }, @@ -587,7 +627,7 @@ describe(`useLiveSuspenseQuery`, () => { unmount() - // Collection should eventually be cleaned up (gcTime is 1ms) + // An explicit short gcTime still opts out of the Suspense grace period. await waitFor( () => { expect(liveQueryCollection.status).toBe(`cleaned-up`) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 328d4fe5ad..01fd6b0b2a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1460,9 +1460,6 @@ importers: '@tanstack/db': specifier: workspace:* version: link:../db - use-sync-external-store: - specifier: ^1.6.0 - version: 1.6.0(react@19.2.4) devDependencies: '@electric-sql/client': specifier: ^1.5.15 @@ -1476,9 +1473,6 @@ importers: '@types/react-dom': specifier: ^19.2.3 version: 19.2.3(@types/react@19.2.13) - '@types/use-sync-external-store': - specifier: ^1.5.0 - version: 1.5.0 '@vitest/coverage-istanbul': specifier: ^3.2.4 version: 3.2.4(vitest@3.2.4)