diff --git a/.changeset/report-incremental-subset-errors.md b/.changeset/report-incremental-subset-errors.md new file mode 100644 index 000000000..2d14c9f0f --- /dev/null +++ b/.changeset/report-incremental-subset-errors.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +Report incremental subset-load failures through subscriptions, live-query utilities, and effects while keeping cached source rows available. Recover cleanly from failed or overlapping must-refetch replays, collection cleanup, and effect teardown errors. diff --git a/docs/guides/error-handling.md b/docs/guides/error-handling.md index 0683eec7d..bff185c1a 100644 --- a/docs/guides/error-handling.md +++ b/docs/guides/error-handling.md @@ -119,6 +119,50 @@ Error tracking methods: - **`errorCount`**: Returns the number of consecutive sync failures. This counter is incremented only when queries fail completely (not per retry attempt) and is reset on successful queries: - **`clearError()`**: Clears the error state and triggers a refetch of the query. This method resets both `lastError` and `errorCount`: +## Incremental Subset Load Errors + +An incremental `loadSubset` failure does not discard rows that are already +available or put the shared source collection into `error`. The failure belongs +to the subscription that requested that subset: + +```ts +const subscription = todoCollection.subscribeChanges(handleChanges, { + includeInitialState: false, +}) + +subscription.on('loadSubset:error', ({ error, options }) => { + console.error('Subset failed', options, error) +}) + +subscription.requestSnapshot() + +// The most recent failure remains available for diagnostics. +console.log(subscription.lastError) +``` + +For ordered live queries, `utils.setWindow()` rejects with the same error. The +last failure is also available as `utils.lastSubsetError`, while the last +successful snapshot remains readable: + +```ts +try { + await liveTodos.utils.setWindow({ offset: 0, limit: 100 }) +} catch (error) { + console.error(liveTodos.utils.lastSubsetError) +} +``` + +Effects report subset failures through `onSourceError` and dispose because +their incremental result can no longer be kept complete. + +When a must-refetch truncate cannot reload every active subset, a subscription +keeps its last successful snapshot and reports the subset error. It discards +the incomplete replay batch, then resumes publishing ordinary source changes. +The next truncate retries every active subset. Overlapping truncates form one +atomic replay: all in-flight requests settle, the newest attempt decides the +result, and subscribers receive the replacement only when that attempt +succeeds. + ## Collection Status and Error States Collections track their status and transition between states: diff --git a/packages/db/skills/db-core/custom-adapter/SKILL.md b/packages/db/skills/db-core/custom-adapter/SKILL.md index 386b68f8a..212f3662c 100644 --- a/packages/db/skills/db-core/custom-adapter/SKILL.md +++ b/packages/db/skills/db-core/custom-adapter/SKILL.md @@ -198,6 +198,13 @@ return the fetched rows. `parseLoadSubsetOptions()` returns only `filters`, opaque backend cursor; translate or combine those expressions for your API. Return `unloadSubset` only when `loadSubset` creates an ongoing resource, such as a per-subset server subscription, that must be released. +Ownership transfers to core only when `loadSubset` returns `true` or a promise. +If it throws synchronously after partial setup, release that partial resource +before throwing; core will not call `unloadSubset` for a request that never +returned. A must-refetch can call `loadSubset` again with the same options. Each +successful return is a fresh acquisition: core releases the previous +acquisition when its replacement returns, then releases the current one when +the demand ends. ### Managing optimistic state duration diff --git a/packages/db/src/collection/changes.ts b/packages/db/src/collection/changes.ts index d51a0e799..00523a2f4 100644 --- a/packages/db/src/collection/changes.ts +++ b/packages/db/src/collection/changes.ts @@ -222,9 +222,6 @@ export class CollectionChangesManager< ) => void, options: SubscribeChangesOptions = {}, ): CollectionSubscription { - // Start sync and track subscriber - this.addSubscriber() - // Compile where callback to whereExpression if provided if (options.where && options.whereExpression) { throw new Error( @@ -240,37 +237,56 @@ export class CollectionChangesManager< whereExpression = toExpression(result) } - const subscription = new CollectionSubscription(this.collection, callback, { - ...opts, - whereExpression, - onUnsubscribe: () => { - this.removeSubscriber() - this.changeSubscriptions.delete(subscription) - }, - }) - - // Register status listener BEFORE requesting snapshot to avoid race condition. - // This ensures the listener catches all status transitions, even if the - // loadSubset promise resolves synchronously or very quickly. - if (options.onStatusChange) { - subscription.on(`status:change`, options.onStatusChange) - } + // Acquire ownership only after all fallible option validation and + // user-provided predicate compilation has completed. + this.addSubscriber() - if (options.includeInitialState) { - subscription.requestSnapshot({ - trackLoadSubsetPromise: false, - orderBy: options.orderBy, - limit: options.limit, - onLoadSubsetResult: options.onLoadSubsetResult, + let subscription: CollectionSubscription | undefined + try { + subscription = new CollectionSubscription(this.collection, callback, { + ...opts, + whereExpression, + onUnsubscribe: () => { + this.removeSubscriber() + if (subscription) this.changeSubscriptions.delete(subscription) + }, }) - } else if (options.includeInitialState === false) { - // When explicitly set to false (not just undefined), mark all state as "seen" - // so that all future changes (including deletes) pass through unfiltered. - subscription.markAllStateAsSeen() - } - // Add to batched listeners - this.changeSubscriptions.add(subscription) + // Register status listener BEFORE requesting snapshot to avoid race condition. + // This ensures the listener catches all status transitions, even if the + // loadSubset promise resolves synchronously or very quickly. + if (options.onStatusChange) { + subscription.on(`status:change`, options.onStatusChange) + } + + if (options.includeInitialState) { + subscription.requestSnapshot({ + trackLoadSubsetPromise: false, + orderBy: options.orderBy, + limit: options.limit, + onLoadSubsetResult: options.onLoadSubsetResult, + }) + } else if (options.includeInitialState === false) { + // When explicitly set to false (not just undefined), mark all state as "seen" + // so that all future changes (including deletes) pass through unfiltered. + subscription.markAllStateAsSeen() + } + + // Add to batched listeners + this.changeSubscriptions.add(subscription) + } catch (error) { + if (subscription) { + try { + subscription.unsubscribe() + } catch { + // Preserve the setup error. Cleanup still releases subscriber + // ownership and attempts every subset unload before it throws. + } + } else { + this.removeSubscriber() + } + throw error + } return subscription } @@ -283,12 +299,20 @@ export class CollectionChangesManager< this.activeSubscribersCount++ this.lifecycle.cancelGCTimer() - // Start sync if collection was cleaned up - if ( - this.lifecycle.status === `cleaned-up` || - this.lifecycle.status === `idle` - ) { - this.sync.startSync() + try { + // Start sync if collection was cleaned up + if ( + this.lifecycle.status === `cleaned-up` || + this.lifecycle.status === `idle` + ) { + this.sync.startSync() + } + } catch (error) { + this.activeSubscribersCount = previousSubscriberCount + if (this.activeSubscribersCount === 0) { + this.lifecycle.startGCTimer() + } + throw error } this.events.emitSubscribersChange( diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts index 87f3bbc51..0fe9cfe9f 100644 --- a/packages/db/src/collection/subscription.ts +++ b/packages/db/src/collection/subscription.ts @@ -4,6 +4,7 @@ import { PropRef, Value } from '../query/ir.js' import { EventEmitter } from '../event-emitter.js' import { compileExpression } from '../query/compiler/evaluators.js' import { buildCursor } from '../utils/cursor.js' +import { deepEquals } from '../utils.js' import { createFilterFunctionFromExpression, createFilteredCallback, @@ -15,6 +16,7 @@ import type { LoadSubsetOptions, Subscription, SubscriptionEvents, + SubscriptionLoadSubsetErrorEvent, SubscriptionStatus, SubscriptionUnsubscribedEvent, } from '../types.js' @@ -54,6 +56,34 @@ type CollectionSubscriptionOptions = { whereExpression?: BasicExpression /** Callback to call when the subscription is unsubscribed */ onUnsubscribe?: (event: SubscriptionUnsubscribedEvent) => void + /** Callback for subset-load failures scoped to this subscription. */ + onLoadSubsetError?: (event: SubscriptionLoadSubsetErrorEvent) => void +} + +type TruncatePublicationState = { + loadedInitialState: boolean + snapshotSent: boolean + sentKeys: Set + publishedRows: Map + limitedSnapshotRowCount: number + lastSentKey: string | number | undefined +} + +type SubsetDemand = { + options: LoadSubsetOptions +} + +type TruncateReplayAttempt = { + pending: Set> + failed: boolean + setupComplete: boolean +} + +type TruncateReplaySession = { + publicationState: TruncatePublicationState + buffer: Array>> + attempts: Set + currentAttempt: TruncateReplayAttempt } export class CollectionSubscription @@ -75,7 +105,7 @@ export class CollectionSubscription * Track all loadSubset calls made by this subscription so we can unload them on cleanup. * We store the exact LoadSubsetOptions we passed to loadSubset to ensure symmetric unload. */ - private loadedSubsets: Array = [] + private subsetDemands: Array = [] private readonly requestedSubsetWhere = new WeakMap< LoadSubsetOptions, BasicExpression @@ -83,6 +113,8 @@ export class CollectionSubscription // Keep track of the keys we've sent (needed for join and orderBy optimizations) private sentKeys = new Set() + private publishedRows = new Map() + private stalePublishedRows = new Map() // Track the count of rows sent via requestLimitedSnapshot for offset-based pagination private limitedSnapshotRowCount = 0 @@ -96,22 +128,24 @@ export class CollectionSubscription // Status tracking private _status: SubscriptionStatus = `ready` + private _lastError: unknown | undefined private pendingLoadSubsetPromises: Set> = new Set() // Cleanup function for truncate event listener private truncateCleanup: (() => void) | undefined - // Truncate buffering state - // When a truncate occurs, we buffer changes until all loadSubset refetches complete - // This prevents a flash of missing content between deletes and new inserts - private isBufferingForTruncate = false - private truncateBuffer: Array>> = [] - private pendingTruncateRefetches: Set> = new Set() + // One replay session owns the publication baseline, overlapping attempts, + // and buffered changes until every attempt settles. + private truncateReplaySession: TruncateReplaySession | undefined public get status(): SubscriptionStatus { return this._status } + public get lastError(): unknown | undefined { + return this._lastError + } + constructor( private collection: CollectionImpl, private callback: (changes: Array>) => void, @@ -119,7 +153,10 @@ export class CollectionSubscription ) { super() if (options.onUnsubscribe) { - this.on(`unsubscribed`, (event) => options.onUnsubscribe!(event)) + this.on(`unsubscribed`, options.onUnsubscribe) + } + if (options.onLoadSubsetError) { + this.on(`loadSubset:error`, options.onLoadSubsetError) } // Auto-index for where expressions if enabled @@ -131,6 +168,7 @@ export class CollectionSubscription changes: Array>, ) => { callback(changes) + this.trackPublishedRows(changes) this.trackSentKeys(changes) } @@ -157,11 +195,12 @@ export class CollectionSubscription * This is called when the sync layer receives a must-refetch and clears all data. * * To prevent a flash of missing content, we buffer all changes (deletes from truncate - * and inserts from refetch) until all loadSubset promises resolve, then emit them together. + * and inserts from refetch) until all loadSubset calls succeed, then emit them together. + * A failed replay keeps the last published snapshot, resumes ordinary deltas, + * and retains subset ownership so a later truncate can retry the replay. */ private handleTruncate() { - // Copy the loaded subsets before clearing (we'll re-request them) - const subsetsToReload = [...this.loadedSubsets] + const demandsToReload = [...this.subsetDemands] // Only buffer if there's an actual loadSubset handler that can do async work. // Without a loadSubset handler, there's nothing to re-request and no reason to buffer. @@ -169,95 +208,147 @@ export class CollectionSubscription const hasLoadSubsetHandler = this.collection._sync.syncLoadSubsetFn !== null // If there are no subsets to reload OR no loadSubset handler, just reset state - if (subsetsToReload.length === 0 || !hasLoadSubsetHandler) { + if (demandsToReload.length === 0 || !hasLoadSubsetHandler) { this.snapshotSent = false this.loadedInitialState = false this.limitedSnapshotRowCount = 0 this.lastSentKey = undefined - this.loadedSubsets = [] return } - // Start buffering BEFORE we receive the delete events from the truncate commit - // This ensures we capture both the deletes and subsequent inserts - this.isBufferingForTruncate = true - this.truncateBuffer = [] - this.pendingTruncateRefetches.clear() + const attempt: TruncateReplayAttempt = { + pending: new Set(), + failed: false, + setupComplete: false, + } + let session = this.truncateReplaySession + if (!session) { + session = { + publicationState: { + loadedInitialState: this.loadedInitialState, + snapshotSent: this.snapshotSent, + sentKeys: new Set(this.sentKeys), + publishedRows: new Map(this.publishedRows), + limitedSnapshotRowCount: this.limitedSnapshotRowCount, + lastSentKey: this.lastSentKey, + }, + buffer: [], + attempts: new Set(), + currentAttempt: attempt, + } + this.truncateReplaySession = session + } + session.attempts.add(attempt) + session.currentAttempt = attempt + + // Start buffering before the truncate commit publishes its deletes. Every + // overlapping attempt shares this one publication baseline and buffer. + this.stalePublishedRows.clear() - // Reset snapshot/pagination tracking state - // Note: We don't need to populate sentKeys here because filterAndFlipChanges - // will skip the delete filter when isBufferingForTruncate is true + // Reset snapshot/pagination tracking state for the replacement snapshot. this.snapshotSent = false this.loadedInitialState = false this.limitedSnapshotRowCount = 0 this.lastSentKey = undefined - // Clear the loadedSubsets array since we're re-requesting fresh - this.loadedSubsets = [] - - // Defer the loadSubset calls to a microtask so the truncate commit's delete events - // are buffered BEFORE the loadSubset calls potentially trigger nested commits. - // This ensures correct event ordering: deletes first, then inserts. + // Defer the requests so the truncate commit's deletes enter the session + // buffer before a synchronous adapter can publish replacement rows. queueMicrotask(() => { - // Check if we were unsubscribed while waiting - if (!this.isBufferingForTruncate) { - return - } - - // Re-request all previously loaded subsets and track their promises - for (const options of subsetsToReload) { - const syncResult = this.collection._sync.loadSubset(options) + if (this.truncateReplaySession !== session) return + + for (const demand of demandsToReload) { + if (!this.subsetDemands.includes(demand)) continue + + const isCurrentAttempt = () => + this.truncateReplaySession === session && + session.currentAttempt === attempt + let syncResult: Promise | true + try { + syncResult = this.loadSubset(demand.options, isCurrentAttempt) + this.replaceSubsetAcquisition(demand) + } catch { + attempt.failed = true + continue + } - // Track this loadSubset call so we can unload it later - this.loadedSubsets.push(options) - this.trackLoadSubsetPromise(syncResult) + this.observeLoadSubsetResult( + syncResult, + demand.options, + true, + isCurrentAttempt, + ) - // Track the promise for buffer flushing if (syncResult instanceof Promise) { - this.pendingTruncateRefetches.add(syncResult) - syncResult - .catch(() => { - // Ignore errors - we still want to flush the buffer even if some requests fail - }) - .finally(() => { - this.pendingTruncateRefetches.delete(syncResult) - this.checkTruncateRefetchComplete() - }) + attempt.pending.add(syncResult) + void syncResult.then( + () => this.settleTruncateReplay(session, attempt, syncResult), + () => { + attempt.failed = true + this.settleTruncateReplay(session, attempt, syncResult) + }, + ) } } - // If all loadSubset calls were synchronous (returned true), flush now - // At this point, delete events have already been buffered from the truncate commit - if (this.pendingTruncateRefetches.size === 0) { - this.flushTruncateBuffer() - } + attempt.setupComplete = true + this.checkTruncateReplayComplete(session) }) } - /** - * Check if all truncate refetch promises have completed and flush buffer if so - */ - private checkTruncateRefetchComplete() { - if ( - this.pendingTruncateRefetches.size === 0 && - this.isBufferingForTruncate - ) { - this.flushTruncateBuffer() + private settleTruncateReplay( + session: TruncateReplaySession, + attempt: TruncateReplayAttempt, + promise: Promise, + ): void { + if (this.truncateReplaySession !== session) return + attempt.pending.delete(promise) + this.checkTruncateReplayComplete(session) + } + + /** Publish only after every overlapping replay attempt has settled. */ + private checkTruncateReplayComplete(session: TruncateReplaySession): void { + if (this.truncateReplaySession !== session) return + for (const attempt of session.attempts) { + if (!attempt.setupComplete || attempt.pending.size > 0) return + } + + if (session.currentAttempt.failed) { + this.abandonTruncateReplay(session) + } else { + this.flushTruncateReplay(session) } } /** - * Flush the truncate buffer, emitting all buffered changes to the callback + * Discard an incomplete current replay and restore the last publication. + * Rows in that publication remain stale until a later source delta or replay + * reconciles them with the source collection. */ - private flushTruncateBuffer() { - this.isBufferingForTruncate = false + private abandonTruncateReplay(session: TruncateReplaySession): void { + if (this.truncateReplaySession !== session) return + const publicationState = session.publicationState + this.loadedInitialState = publicationState.loadedInitialState + this.snapshotSent = publicationState.snapshotSent + this.sentKeys = new Set(publicationState.sentKeys) + this.publishedRows = new Map(publicationState.publishedRows) + this.stalePublishedRows = new Map(publicationState.publishedRows) + this.limitedSnapshotRowCount = publicationState.limitedSnapshotRowCount + this.lastSentKey = publicationState.lastSentKey + this.truncateReplaySession = undefined + } + + /** Publish the complete buffered replacement as one subscriber batch. */ + private flushTruncateReplay(session: TruncateReplaySession): void { + if (this.truncateReplaySession !== session) return + this.truncateReplaySession = undefined + this.stalePublishedRows.clear() - // Flatten all buffered changes into a single array for atomic emission - // This ensures consumers see all truncate changes (deletes + inserts) in one callback - const merged = this.truncateBuffer.flat() + const merged = session.buffer.flat() if (merged.length > 0) this.filteredCallback(merged) + } - this.truncateBuffer = [] + private get isBufferingForTruncate(): boolean { + return this.truncateReplaySession !== undefined } setOrderByIndex(index: IndexInterface) { @@ -300,23 +391,67 @@ export class CollectionSubscription } as SubscriptionEvents[typeof eventKey]) } - /** - * Track a loadSubset promise and manage loading status - */ - private trackLoadSubsetPromise(syncResult: Promise | true) { - // Track the promise if it's actually a promise (async work) - if (syncResult instanceof Promise) { + /** Observe an asynchronous subset load and restore status on settlement. */ + private observeLoadSubsetResult( + syncResult: Promise | true, + options: LoadSubsetOptions, + trackStatus: boolean, + shouldReportError: () => boolean = () => true, + ) { + if (!(syncResult instanceof Promise)) return + + if (trackStatus) { this.pendingLoadSubsetPromises.add(syncResult) this.setStatus(`loadingSubset`) + } - const finish = () => { + const finish = () => { + if (trackStatus) { this.pendingLoadSubsetPromises.delete(syncResult) if (this.pendingLoadSubsetPromises.size === 0) { this.setStatus(`ready`) } } - void syncResult.then(finish, finish) } + + void syncResult.then(finish, (error: unknown) => { + if (shouldReportError()) this.recordLoadSubsetError(options, error) + finish() + }) + } + + private loadSubset( + options: LoadSubsetOptions, + shouldReportError: () => boolean = () => true, + ): Promise | true { + try { + return this.collection._sync.loadSubset(options) + } catch (error) { + if (shouldReportError()) this.recordLoadSubsetError(options, error) + throw error + } + } + + /** Replace the adapter lease held for one logical subset demand. */ + private replaceSubsetAcquisition(demand: SubsetDemand): void { + this.collection._sync.unloadSubset(demand.options) + } + + private recordLoadSubsetError( + options: LoadSubsetOptions, + error: unknown, + ): void { + // Aborted subset requests are obsolete demand, not load failures. The + // request may reject after its route has already been released. + if (options.signal?.aborted) return + + this._lastError = error + this.emitInner(`loadSubset:error`, { + type: `loadSubset:error`, + subscription: this, + options, + error, + }) } hasLoadedInitialState() { @@ -334,7 +469,7 @@ export class CollectionSubscription // Buffer the changes instead of emitting immediately // This prevents a flash of missing content during truncate/refetch if (newChanges.length > 0) { - this.truncateBuffer.push(newChanges) + this.truncateReplaySession!.buffer.push(newChanges) } return false } else { @@ -387,19 +522,22 @@ export class CollectionSubscription orderBy: opts?.orderBy, limit: opts?.limit, } - const syncResult = this.collection._sync.loadSubset(loadOptions) - // Pass the raw loadSubset result to the caller for external tracking - opts?.onLoadSubsetResult?.(syncResult) + const syncResult = this.loadSubset(loadOptions) - // Track this loadSubset call so we can unload it later - this.loadedSubsets.push(loadOptions) + // A returned result transfers ownership to the subscription. A throwing + // adapter retains responsibility for rolling back any partial setup. + this.subsetDemands.push({ options: loadOptions }) if (opts?.where) this.requestedSubsetWhere.set(loadOptions, opts.where) - const trackLoadSubsetPromise = opts?.trackLoadSubsetPromise ?? true - if (trackLoadSubsetPromise) { - this.trackLoadSubsetPromise(syncResult) - } + // Pass the raw loadSubset result to the caller for external tracking + opts?.onLoadSubsetResult?.(syncResult) + + this.observeLoadSubsetResult( + syncResult, + loadOptions, + opts?.trackLoadSubsetPromise ?? true, + ) // Also load data immediately from the collection let snapshot: Array> | void @@ -443,15 +581,15 @@ export class CollectionSubscription /** Release one exact subset request while keeping the subscription alive. */ releaseSnapshot(where: BasicExpression): void { - const index = this.loadedSubsets.findIndex( - (options) => - options.where === where || - this.requestedSubsetWhere.get(options) === where, + const index = this.subsetDemands.findIndex( + (demand) => + demand.options.where === where || + this.requestedSubsetWhere.get(demand.options) === where, ) if (index === -1) return - const [options] = this.loadedSubsets.splice(index, 1) - if (options) this.collection._sync.unloadSubset(options) + const [demand] = this.subsetDemands.splice(index, 1) + if (demand) this.collection._sync.unloadSubset(demand.options) } /** @@ -654,16 +792,20 @@ export class CollectionSubscription offset: offset ?? currentOffset, // Use provided offset, or auto-tracked offset subscription: this, } - const syncResult = this.collection._sync.loadSubset(loadOptions) + + const syncResult = this.loadSubset(loadOptions) + + // A returned result transfers ownership to the subscription. A throwing + // adapter retains responsibility for rolling back any partial setup. + this.subsetDemands.push({ options: loadOptions }) // Pass the raw loadSubset result to the caller for external tracking onLoadSubsetResult?.(syncResult) - - // Track this loadSubset call - this.loadedSubsets.push(loadOptions) - if (shouldTrackLoadSubsetPromise) { - this.trackLoadSubsetPromise(syncResult) - } + this.observeLoadSubsetResult( + syncResult, + loadOptions, + shouldTrackLoadSubsetPromise, + ) } // TODO: also add similar test but that checks that it can also load it from the collection's loadSubset function @@ -676,6 +818,8 @@ export class CollectionSubscription * Duplicate inserts are filtered out to prevent D2 multiplicity > 1. */ private filterAndFlipChanges(changes: Array>) { + changes = this.reconcileStalePublishedChanges(changes) + if (this.loadedInitialState || this.skipFiltering) { // We loaded the entire initial state or filtering is explicitly skipped // so no need to filter or flip changes @@ -725,6 +869,55 @@ export class CollectionSubscription return newChanges } + /** + * After a failed replay, the source collection is empty but subscribers still + * hold the last good publication. Reconcile the first later source delta for + * each retained key against that publication instead of treating it as a + * duplicate insert. + */ + private reconcileStalePublishedChanges( + changes: Array>, + ): Array> { + if (this.stalePublishedRows.size === 0) return changes + + const reconciled: Array> = [] + for (const change of changes) { + const previous = this.stalePublishedRows.get(change.key) + if (previous === undefined) { + reconciled.push(change) + continue + } + + this.stalePublishedRows.delete(change.key) + if (change.type === `delete`) { + reconciled.push({ + ...change, + value: previous, + previousValue: previous, + }) + } else if (!deepEquals(previous, change.value)) { + reconciled.push({ + ...change, + type: `update`, + previousValue: previous, + }) + } + } + return reconciled + } + + private trackPublishedRows( + changes: Array>, + ): void { + for (const change of changes) { + if (change.type === `delete`) { + this.publishedRows.delete(change.key) + } else { + this.publishedRows.set(change.key, change.value) + } + } + } + private trackSentKeys(changes: Array>) { if (this.loadedInitialState || this.skipFiltering) { // No need to track sent keys if we loaded the entire state or filtering is skipped. @@ -761,27 +954,42 @@ export class CollectionSubscription } unsubscribe() { + let firstCleanupError: unknown + // Clean up truncate event listener - this.truncateCleanup?.() + try { + this.truncateCleanup?.() + } catch (error) { + firstCleanupError = error + } this.truncateCleanup = undefined - // Clean up truncate buffer state - this.isBufferingForTruncate = false - this.truncateBuffer = [] - this.pendingTruncateRefetches.clear() + // Stop any buffered replay from publishing after unsubscription. + this.truncateReplaySession = undefined + this.stalePublishedRows.clear() - // Unload all subsets that this subscription loaded - // We pass the exact same LoadSubsetOptions we used for loadSubset - for (const options of this.loadedSubsets) { - this.collection._sync.unloadSubset(options) + // Release the current adapter acquisition for each logical subset demand. + for (const demand of this.subsetDemands) { + try { + this.collection._sync.unloadSubset(demand.options) + } catch (error) { + firstCleanupError ??= error + } } - this.loadedSubsets = [] + this.subsetDemands = [] - this.emitInner(`unsubscribed`, { - type: `unsubscribed`, - subscription: this, - }) - // Clear all event listeners to prevent memory leaks - this.clearListeners() + try { + this.emitInner(`unsubscribed`, { + type: `unsubscribed`, + subscription: this, + }) + } catch (error) { + firstCleanupError ??= error + } finally { + // Clear all event listeners to prevent memory leaks + this.clearListeners() + } + + if (firstCleanupError !== undefined) throw firstCleanupError } } diff --git a/packages/db/src/collection/sync.ts b/packages/db/src/collection/sync.ts index ce8f5729c..c3632d706 100644 --- a/packages/db/src/collection/sync.ts +++ b/packages/db/src/collection/sync.ts @@ -33,6 +33,15 @@ type DeferredLoadSubset = { deferred: Deferred } +type LoadSubsetOperation = { + pending: Set> + waiting: boolean + completed: boolean + hasError: boolean + error?: unknown + deferred?: Deferred +} + export class CollectionSyncManager< TOutput extends object = Record, TKey extends string | number = string | number, @@ -56,10 +65,12 @@ export class CollectionSyncManager< null private pendingLoadSubsetPromises: Set> = new Set() + private activeLoadSubsetOperation: LoadSubsetOperation | undefined private syncStartDeferred = false private syncStartRequested = false private deferredLoadSubsets: Array = [] private syncEpoch = 0 + private loadSubsetSession = 0 /** * Creates a new CollectionSyncManager instance @@ -579,6 +590,93 @@ export class CollectionSyncManager< return this.waitForPendingLoadSubset() } + /** @internal Observe subset requests caused by one imperative operation. */ + public beginLoadSubsetOperation(): { + wait: () => true | Promise + cancel: () => void + } { + const operation: LoadSubsetOperation = { + pending: new Set(), + waiting: false, + completed: false, + hasError: false, + } + // A new imperative operation owns future requests. Older operations keep + // waiting for the promises they already acquired, but cannot absorb work + // caused by a superseding physical window. + this.activeLoadSubsetOperation = operation + return { + wait: () => this.waitForLoadSubsetOperation(operation), + cancel: () => { + operation.completed = true + if (this.activeLoadSubsetOperation === operation) { + this.activeLoadSubsetOperation = undefined + } + }, + } + } + + private waitForLoadSubsetOperation( + operation: LoadSubsetOperation, + ): true | Promise { + operation.waiting = true + if (operation.pending.size === 0) { + operation.completed = true + if (this.activeLoadSubsetOperation === operation) { + this.activeLoadSubsetOperation = undefined + } + return operation.hasError ? Promise.reject(operation.error) : true + } + operation.deferred = createDeferred() + return operation.deferred.promise + } + + private settleLoadSubsetOperation( + operation: LoadSubsetOperation, + promise: Promise, + outcome: { ok: true } | { ok: false; error: unknown }, + ): void { + if (operation.completed) return + operation.pending.delete(promise) + if (!outcome.ok && !operation.hasError) { + operation.hasError = true + operation.error = outcome.error + } + if (!operation.waiting || operation.pending.size > 0) return + + // A resolved request can synchronously publish source rows that register + // follow-up loads. Let those registrations join this operation before it + // is considered complete. + queueMicrotask(() => { + if (operation.completed || operation.pending.size > 0) return + operation.completed = true + if (this.activeLoadSubsetOperation === operation) { + this.activeLoadSubsetOperation = undefined + } + if (operation.hasError) { + operation.deferred!.reject(operation.error) + } else { + operation.deferred!.resolve() + } + }) + } + + /** @internal Attach a relevant existing request to the active operation. */ + public trackLoadSubsetOperationPromise(promise: Promise): void { + const operation = this.activeLoadSubsetOperation + if (!operation || operation.pending.has(promise)) return + + operation.pending.add(promise) + void promise.then( + () => this.settleLoadSubsetOperation(operation, promise, { ok: true }), + (error) => + this.settleLoadSubsetOperation(operation, promise, { + ok: false, + error, + }), + ) + } + private async waitForPendingLoadSubset(): Promise { do { await Promise.all([...this.pendingLoadSubsetPromises]) @@ -590,8 +688,10 @@ export class CollectionSyncManager< * @internal This is for internal coordination (e.g., live-query glue code), not for general use. */ public trackLoadPromise(promise: Promise): void { + const loadSubsetSession = this.loadSubsetSession const loadingStarting = !this.isLoadingSubset this.pendingLoadSubsetPromises.add(promise) + this.trackLoadSubsetOperationPromise(promise) if (loadingStarting) { this._events.emit(`loadingSubset:change`, { @@ -604,6 +704,8 @@ export class CollectionSyncManager< } const finish = () => { + if (loadSubsetSession !== this.loadSubsetSession) return + const loadingEnding = this.pendingLoadSubsetPromises.size === 1 && this.pendingLoadSubsetPromises.has(promise) @@ -684,6 +786,7 @@ export class CollectionSyncManager< // Invalidate callbacks retained by asynchronous work from this session // before invoking adapter cleanup or allowing a new session to start. this.syncEpoch++ + this.loadSubsetSession++ try { if (this.syncCleanupFn) { this.syncCleanupFn() @@ -708,6 +811,24 @@ export class CollectionSyncManager< this.syncUnloadSubsetFn = null this.syncStartDeferred = false this.syncStartRequested = false + const wasLoadingSubset = this.pendingLoadSubsetPromises.size > 0 + this.pendingLoadSubsetPromises.clear() + if (wasLoadingSubset) { + this._events.emit(`loadingSubset:change`, { + type: `loadingSubset:change`, + collection: this.collection, + isLoadingSubset: false, + previousIsLoadingSubset: true, + loadingSubsetTransition: `end`, + }) + } + const activeOperation = this.activeLoadSubsetOperation + this.activeLoadSubsetOperation = undefined + if (activeOperation && !activeOperation.completed) { + activeOperation.completed = true + activeOperation.pending.clear() + activeOperation.deferred?.resolve() + } const deferredLoadSubsets = this.deferredLoadSubsets this.deferredLoadSubsets = [] for (const { deferred } of deferredLoadSubsets) { diff --git a/packages/db/src/query/effect.ts b/packages/db/src/query/effect.ts index 9991c7794..fabb31436 100644 --- a/packages/db/src/query/effect.ts +++ b/packages/db/src/query/effect.ts @@ -258,12 +258,19 @@ export function createEffect< abortController.abort() // Tear down the pipeline (unsubscribe from sources, etc.) - runner.dispose() + let cleanupError: unknown + try { + runner.dispose() + } catch (error) { + cleanupError = error + } // Wait for any in-flight async handlers to settle if (inFlightHandlers.size > 0) { await Promise.allSettled([...inFlightHandlers]) } + + if (cleanupError !== undefined) throw cleanupError } // Create and start the pipeline @@ -291,7 +298,12 @@ export function createEffect< dispose() }, }) - runner.start() + try { + runner.start() + } catch (error) { + runner.dispose() + throw error + } return { dispose, @@ -380,6 +392,7 @@ class EffectPipelineRunner { // Reentrance guard private isGraphRunning = false + private starting = false private disposed = false // When dispose() is called mid-graph-run, defer heavy cleanup until the run completes private deferredCleanup = false @@ -443,10 +456,16 @@ class EffectPipelineRunner { this.graph.finalize() } + private isDisposed(): boolean { + return this.disposed + } + /** Subscribe to source collections and start processing */ start(): void { + this.starting = true if (this.collectionSources.length === 0) { // Nothing to subscribe to + this.starting = false return } @@ -470,6 +489,11 @@ class EffectPipelineRunner { >() for (const source of this.collectionSources) { + if (this.isDisposed()) { + this.starting = false + return + } + const { sourceId, alias, collection } = source const collectionId = collection.id @@ -525,23 +549,40 @@ class EffectPipelineRunner { } } - // Determine subscription options based on ordered vs unordered path - const subscriptionOptions = this.buildSubscriptionOptions( - alias, - isLazy, - orderByInfo, - whereExpression, - ) - // Subscribe to source changes - const subscription = collection.subscribeChanges( - changeCallback, - subscriptionOptions, - ) + const subscription = collection.subscribeChanges(changeCallback, { + ...this.buildSubscriptionOptions( + alias, + isLazy, + orderByInfo, + whereExpression, + ), + onLoadSubsetError: ({ error }) => { + this.onSourceError(normaliseError(error)) + }, + }) // Store subscription immediately so the join compiler can find it this.subscriptions[sourceId] = subscription + const unsubscribe = () => { + subscription.unsubscribe() + delete this.subscriptions[sourceId] + } + + // subscribeChanges can synchronously report a source error and dispose + // the runner before returning the subscription. + if (this.isDisposed()) { + unsubscribe() + this.starting = false + return + } + + // Own the subscription before any ordered snapshot or lazy demand can + // throw. A partially started effect has no handle for its caller to + // dispose, so start() must be able to release every acquired source. + this.unsubscribeCallbacks.add(unsubscribe) + const lazyCallbacks = this.lazySourcesCallbacks[sourceId] if (lazyCallbacks) { lazyCallbacks.setDemand = (plan: LazyDemandPlan, keys: Set) => @@ -559,11 +600,6 @@ class EffectPipelineRunner { this.requestInitialOrderedSnapshot(alias, orderByInfo, subscription) } - this.unsubscribeCallbacks.add(() => { - subscription.unsubscribe() - delete this.subscriptions[sourceId] - }) - // Listen for status changes on source collections const statusUnsubscribe = collection.on(`status:change`, (event) => { if (this.disposed) return @@ -643,6 +679,7 @@ class EffectPipelineRunner { this.initialLoadComplete = true } } + this.starting = false } /** Handle incoming changes from a source collection */ @@ -659,13 +696,22 @@ class EffectPipelineRunner { plan: LazyDemandPlan, keys: Set, ): void { - const update = this.demand.setDemand(subscription, plan, keys) + let update + try { + update = this.demand.setDemand(subscription, plan, keys) + } catch (error) { + // The subscription error event already reports adapter failures and + // disposes this effect. Do not let that query-local failure escape the + // source commit, but keep unrelated graph errors visible. + if (subscription.lastError !== error) throw error + if (this.starting) throw error + return + } if (update.ready instanceof Promise) { - void update.ready.catch((error: unknown) => { - this.onSourceError( - error instanceof Error ? error : new Error(String(error)), - ) - }) + // Each segment reports its own failure through the subscription. Consume + // the aggregate rejection so Promise.all does not create a second, + // detached error channel. + void update.ready.then(undefined, () => {}) } } @@ -948,11 +994,12 @@ class EffectPipelineRunner { // Track in-flight load to prevent redundant concurrent requests if (loadResult instanceof Promise) { this.pendingOrderedLoadPromise = loadResult - loadResult.finally(() => { + const finish = () => { if (this.pendingOrderedLoadPromise === loadResult) { this.pendingOrderedLoadPromise = undefined } - }) + } + void loadResult.then(finish, finish) } }, }) @@ -986,8 +1033,15 @@ class EffectPipelineRunner { this.disposed = true this.subscribedToAllCollections = false - // Immediately unsubscribe from sources and clear cheap state - this.unsubscribeCallbacks.forEach((fn) => fn()) + // Immediately unsubscribe from every source, even if one release fails. + let firstCleanupError: unknown + for (const unsubscribe of this.unsubscribeCallbacks) { + try { + unsubscribe() + } catch (error) { + firstCleanupError ??= error + } + } this.unsubscribeCallbacks.clear() this.sentToD2KeysBySource.clear() this.pendingChanges.clear() @@ -1016,6 +1070,8 @@ class EffectPipelineRunner { } else { this.finalCleanup() } + + if (firstCleanupError !== undefined) throw firstCleanupError } /** Clear graph references — called after graph run completes or immediately from dispose */ @@ -1116,9 +1172,10 @@ function trackPromise( inFlightHandlers: Set>, ): void { inFlightHandlers.add(promise) - promise.finally(() => { + const finish = () => { inFlightHandlers.delete(promise) - }) + } + void promise.then(finish, finish) } /** Report an error to the onError callback or console */ @@ -1127,7 +1184,7 @@ function reportError( event: DeltaEvent, onError?: (error: Error, event: DeltaEvent) => void, ): void { - const normalised = error instanceof Error ? error : new Error(String(error)) + const normalised = normaliseError(error) if (onError) { try { onError(normalised, event) @@ -1140,3 +1197,7 @@ function reportError( console.error(`[Effect] Unhandled error in handler:`, normalised) } } + +function normaliseError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)) +} diff --git a/packages/db/src/query/live-query-collection.ts b/packages/db/src/query/live-query-collection.ts index 8649bc0bc..893c8f10a 100644 --- a/packages/db/src/query/live-query-collection.ts +++ b/packages/db/src/query/live-query-collection.ts @@ -190,9 +190,13 @@ export function createLiveQueryCollection< // been validated by the public signatures, but the branch loses that precision. const options = liveQueryCollectionOptions(config as any) - // Merge custom utils if provided, preserving the getBuilder() method for dependency tracking + // Merge custom utils without evaluating internal getters such as + // lastSubsetError into stale data properties. if (config.utils) { - options.utils = { ...options.utils, ...config.utils } + Object.defineProperties( + options.utils, + Object.getOwnPropertyDescriptors(config.utils), + ) } return bridgeToCreateCollection(options) as CollectionForContext< diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts index 9e13a6570..6fcaed06d 100644 --- a/packages/db/src/query/live/collection-config-builder.ts +++ b/packages/db/src/query/live/collection-config-builder.ts @@ -49,6 +49,8 @@ import type { AllCollectionEvents } from '../../collection/events.js' export type LiveQueryCollectionUtils = UtilsRecord & { getRunCount: () => number + /** Most recent subset-load failure observed by this live query. */ + readonly lastSubsetError: unknown | undefined /** * Sets the offset and limit of an ordered query. * Is a no-op if the query is not ordered. @@ -114,6 +116,7 @@ export class CollectionConfigBuilder< private isInErrorState = false private fatalQueryError = false private readonly erroredSourceIds = new Set() + private lastSubsetError: unknown | undefined // Reference to the live query collection for error state transitions public liveQueryCollection?: Collection @@ -121,6 +124,9 @@ export class CollectionConfigBuilder< private windowFn: ((options: WindowOptions) => void) | undefined private readonly initialWindow: WindowOptions | undefined private currentWindow: WindowOptions | undefined + private activeWindowOperation: + | { failed: boolean; error?: unknown } + | undefined private maybeRunGraphFn: (() => void) | undefined @@ -244,6 +250,7 @@ export class CollectionConfigBuilder< getConfig(): CollectionConfigSingleRowOption & { utils: LiveQueryCollectionUtils } { + const builder = this return { id: this.id, getKey: @@ -262,6 +269,9 @@ export class CollectionConfigBuilder< singleResult: this.query.singleResult, utils: { getRunCount: this.getRunCount.bind(this), + get lastSubsetError() { + return builder.lastSubsetError + }, setWindow: this.setWindow.bind(this), getWindow: this.getWindow.bind(this), [LIVE_QUERY_INTERNAL]: { @@ -279,10 +289,16 @@ export class CollectionConfigBuilder< throw new SetWindowRequiresOrderByError() } + const loadOperation = + this.liveQueryCollection?._sync.beginLoadSubsetOperation() const previousWindow = this.currentWindow ?? this.initialWindow + const previousOperation = this.activeWindowOperation + const operation: { failed: boolean; error?: unknown } = { failed: false } + this.activeWindowOperation = operation try { this.windowFn(options) this.maybeRunGraphFn?.() + if (operation.failed) throw operation.error this.currentWindow = options } catch (error) { if (previousWindow) { @@ -294,10 +310,13 @@ export class CollectionConfigBuilder< // window rather than replacing it with a rollback failure. } } + loadOperation?.cancel() throw error + } finally { + this.activeWindowOperation = previousOperation } - return this.liveQueryCollection?._sync.waitForCurrentLoadSubset() ?? true + return loadOperation?.wait() ?? true } getWindow(): { offset: number; limit: number } | undefined { @@ -356,10 +375,27 @@ export class CollectionConfigBuilder< failDemand(planId: string, generation: number, error: unknown): void { const demand = this.activeDemands.get(planId) if (!demand || demand.generation !== generation) return + this.recordSubsetError(error) + if (this.activeWindowOperation) { + this.activeWindowOperation.failed = true + this.activeWindowOperation.error = error + } const message = error instanceof Error ? error.message : String(error) this.transitionToError(`Subset demand '${planId}' failed: ${message}`) } + recordSubsetError(error: unknown): void { + this.lastSubsetError = error + } + + trackSubsetLoadPromise(promise: Promise): void { + this.liveQueryCollection!._sync.trackLoadPromise(promise) + } + + trackSubsetLoadOperationPromise(promise: Promise): void { + this.liveQueryCollection!._sync.trackLoadSubsetOperationPromise(promise) + } + retireDemand(planId: string): void { this.activeDemands.delete(planId) } @@ -632,6 +668,7 @@ export class CollectionConfigBuilder< this.isInErrorState = false this.fatalQueryError = false this.erroredSourceIds.clear() + this.lastSubsetError = undefined // Store config and syncState as instance properties for the duration of this sync session this.currentSyncConfig = config @@ -641,49 +678,20 @@ export class CollectionConfigBuilder< unsubscribeCallbacks: new Set<() => void>(), } - // Extend the pipeline such that it applies the incoming changes to the collection - const fullSyncState = this.extendPipelineWithChangeProcessing( - config, - syncState, - ) - this.currentSyncState = fullSyncState + let tornDown = false + const teardown = () => { + if (tornDown) return + tornDown = true - // Listen for scheduler context clears to clean up our pending state - // Re-register on each sync start so the listener is active for the sync session's lifetime - this.unsubscribeFromSchedulerClears = transactionScopedScheduler.onClear( - (contextId) => { - this.clearPendingGraphRun(contextId) - }, - ) - - // Listen for loadingSubset changes on the live query collection BEFORE subscribing. - // This ensures we don't miss the event if subset loading completes synchronously. - // When isLoadingSubset becomes false, we may need to mark the collection as ready - // (if all source collections are already ready but we were waiting for subset load to complete) - const loadingSubsetUnsubscribe = config.collection.on( - `loadingSubset:change`, - (event) => { - if (!event.isLoadingSubset) { - // Subset loading finished, check if we can now mark ready - this.updateLiveQueryStatus(config) + let firstCleanupError: unknown + for (const unsubscribe of syncState.unsubscribeCallbacks) { + try { + unsubscribe() + } catch (error) { + firstCleanupError ??= error } - }, - ) - syncState.unsubscribeCallbacks.add(loadingSubsetUnsubscribe) - - const loadSubsetDataCallbacks = this.subscribeToAllCollections( - config, - fullSyncState, - ) - - this.maybeRunGraphFn = () => this.scheduleGraphRun(loadSubsetDataCallbacks) - - // Initial run with callback to load more data if needed - this.scheduleGraphRun(loadSubsetDataCallbacks) - - // Return the unsubscribe function - return () => { - syncState.unsubscribeCallbacks.forEach((unsubscribe) => unsubscribe()) + } + syncState.unsubscribeCallbacks.clear() // Clear current sync session state this.currentSyncConfig = undefined @@ -724,7 +732,61 @@ export class CollectionConfigBuilder< // The scheduler's listener Set would otherwise keep a strong reference to this builder this.unsubscribeFromSchedulerClears?.() this.unsubscribeFromSchedulerClears = undefined + + if (firstCleanupError !== undefined) throw firstCleanupError + } + + try { + // Extend the pipeline such that it applies the incoming changes to the collection + const fullSyncState = this.extendPipelineWithChangeProcessing( + config, + syncState, + ) + this.currentSyncState = fullSyncState + + // Listen for scheduler context clears to clean up our pending state + // Re-register on each sync start so the listener is active for the sync session's lifetime + this.unsubscribeFromSchedulerClears = transactionScopedScheduler.onClear( + (contextId) => { + this.clearPendingGraphRun(contextId) + }, + ) + + // Listen for loadingSubset changes on the live query collection BEFORE subscribing. + // This ensures we don't miss the event if subset loading completes synchronously. + // When isLoadingSubset becomes false, we may need to mark the collection as ready + // (if all source collections are already ready but we were waiting for subset load to complete) + const loadingSubsetUnsubscribe = config.collection.on( + `loadingSubset:change`, + (event) => { + if (!event.isLoadingSubset) { + // Subset loading finished, check if we can now mark ready + this.updateLiveQueryStatus(config) + } + }, + ) + syncState.unsubscribeCallbacks.add(loadingSubsetUnsubscribe) + + const loadSubsetDataCallbacks = this.subscribeToAllCollections( + config, + fullSyncState, + ) + + this.maybeRunGraphFn = () => + this.scheduleGraphRun(loadSubsetDataCallbacks) + + // Initial run with callback to load more data if needed + this.scheduleGraphRun(loadSubsetDataCallbacks) + } catch (error) { + try { + teardown() + } catch { + // Preserve the setup failure. It is the error the caller can act on. + } + throw error } + + return teardown } /** diff --git a/packages/db/src/query/live/collection-subscriber.ts b/packages/db/src/query/live/collection-subscriber.ts index 66279737c..d6f555f4a 100644 --- a/packages/db/src/query/live/collection-subscriber.ts +++ b/packages/db/src/query/live/collection-subscriber.ts @@ -14,6 +14,7 @@ import { SubsetDemandController } from './subset-demand-controller.js' import type { Collection } from '../../collection/index.js' import type { ChangeMessage, + SubscriptionLoadSubsetErrorEvent, SubscriptionStatusChangeEvent, } from '../../types.js' import type { Context, GetResult } from '../builder/types.js' @@ -82,9 +83,7 @@ export class CollectionSubscriber< // can break under microtask timing (e.g., queueMicrotask in TanStack Query). const trackLoadResult = (result: Promise | true) => { if (result instanceof Promise) { - this.collectionConfigBuilder.liveQueryCollection!._sync.trackLoadPromise( - result, - ) + this.collectionConfigBuilder.trackSubsetLoadPromise(result) } } @@ -106,6 +105,9 @@ export class CollectionSubscriber< } } } + const onLoadSubsetError = (event: SubscriptionLoadSubsetErrorEvent) => { + this.collectionConfigBuilder.recordSubsetError(event.error) + } // Create subscription with onStatusChange - listener is registered before any async work let subscription: CollectionSubscription @@ -115,6 +117,7 @@ export class CollectionSubscriber< orderByInfo, onStatusChange, trackLoadResult, + onLoadSubsetError, ) } else { // Lazy sources load only the subsets demanded by the compiled graph. @@ -126,7 +129,9 @@ export class CollectionSubscriber< whereExpression, includeInitialState, onStatusChange, + onLoadSubsetError, ) + this.registerSubscriptionCleanup(subscription) } // Check current status after subscribing - if status is 'loadingSubset', track it. @@ -138,6 +143,12 @@ export class CollectionSubscriber< this.ensureLoadingPromise(subscription) } + return subscription + } + + private registerSubscriptionCleanup( + subscription: CollectionSubscription, + ): void { const unsubscribe = () => { // If subscription has a pending promise, resolve it before unsubscribing const deferred = this.subscriptionLoadingPromises.get(subscription) @@ -154,7 +165,6 @@ export class CollectionSubscriber< this.collectionConfigBuilder.currentSyncState!.unsubscribeCallbacks.add( unsubscribe, ) - return subscription } setDemand( @@ -162,7 +172,22 @@ export class CollectionSubscriber< plan: LazyDemandPlan, keys: Set, ): void { - const update = this.demand.setDemand(subscription, plan, keys) + let update + try { + update = this.demand.setDemand(subscription, plan, keys) + } catch (error) { + // CollectionSubscription reports adapter failures before rethrowing. + // Convert that synchronous form to the same query-local fatal demand + // state as a rejected load, without letting it escape the source commit. + // Preserve unrelated graph/programming errors as throws. + if (subscription.lastError !== error) throw error + const isInitialSync = + this.collectionConfigBuilder.liveQueryCollection?.status === `loading` + const generation = this.collectionConfigBuilder.beginDemand(plan.id) + this.collectionConfigBuilder.failDemand(plan.id, generation, error) + if (isInitialSync) throw error + return + } if (!update.changed) return if (update.empty) { @@ -172,6 +197,7 @@ export class CollectionSubscriber< const generation = this.collectionConfigBuilder.beginDemand(plan.id) if (update.ready instanceof Promise) { + this.collectionConfigBuilder.trackSubsetLoadOperationPromise(update.ready) void update.ready.then( () => this.collectionConfigBuilder.settleDemand(plan.id, generation), (error) => @@ -215,6 +241,7 @@ export class CollectionSubscriber< whereExpression: BasicExpression | undefined, includeInitialState: boolean, onStatusChange: (event: SubscriptionStatusChangeEvent) => void, + onLoadSubsetError: (event: SubscriptionLoadSubsetErrorEvent) => void, ): CollectionSubscription { const sendChanges = ( changes: Array>, @@ -234,9 +261,7 @@ export class CollectionSubscriber< const onLoadSubsetResult = includeInitialState ? (result: Promise | true) => { if (result instanceof Promise) { - this.collectionConfigBuilder.liveQueryCollection!._sync.trackLoadPromise( - result, - ) + this.collectionConfigBuilder.trackSubsetLoadPromise(result) } } : undefined @@ -245,6 +270,7 @@ export class CollectionSubscriber< ...(includeInitialState && { includeInitialState }), whereExpression, onStatusChange, + onLoadSubsetError, orderBy: hints.orderBy, limit: hints.limit, onLoadSubsetResult, @@ -258,6 +284,7 @@ export class CollectionSubscriber< orderByInfo: OrderByOptimizationInfo, onStatusChange: (event: SubscriptionStatusChangeEvent) => void, onLoadSubsetResult: (result: Promise | true) => void, + onLoadSubsetError: (event: SubscriptionLoadSubsetErrorEvent) => void, ): CollectionSubscription { const { orderBy, offset, limit, index } = orderByInfo @@ -302,8 +329,10 @@ export class CollectionSubscriber< const subscription = this.collection.subscribeChanges(sendChangesInRange, { whereExpression, onStatusChange, + onLoadSubsetError, }) subscriptionHolder.current = subscription + this.registerSubscriptionCleanup(subscription) // Listen for truncate events to reset cursor tracking state and sentToD2Keys // This ensures that after a must-refetch/truncate, we don't use stale cursor data @@ -371,15 +400,19 @@ export class CollectionSubscriber< return true } - if (this.pendingOrderedLoadPromise) { - // Wait for in-flight ordered loads to resolve before issuing another request. - return true - } - // `dataNeeded` probes the orderBy operator to see if it needs more data // if it needs more data, it returns the number of items it needs const n = dataNeeded() if (n > 0) { + if (this.pendingOrderedLoadPromise) { + // The current window still needs the in-flight coverage. Attach it to + // this operation without making an unrelated or superseded request a + // dependency of every window change. + this.collectionConfigBuilder.trackSubsetLoadOperationPromise( + this.pendingOrderedLoadPromise, + ) + return true + } this.loadNextItems(n, subscription) } return true @@ -430,18 +463,35 @@ export class CollectionSubscriber< ) if (!cursor) return // Duplicate request — skip - this.lastLoadRequestKey = cursor.loadRequestKey + const loadRequestKey = cursor.loadRequestKey + this.lastLoadRequestKey = loadRequestKey // Take the `n` items after the biggest sent value // Omit offset so requestLimitedSnapshot can advance based on // the number of rows already loaded (supports offset-based backends). - subscription.requestLimitedSnapshot({ - orderBy: cursor.normalizedOrderBy, - limit: n, - minValues: cursor.minValues, - trackLoadSubsetPromise: false, - onLoadSubsetResult: this.orderedLoadSubsetResult, - }) + try { + subscription.requestLimitedSnapshot({ + orderBy: cursor.normalizedOrderBy, + limit: n, + minValues: cursor.minValues, + trackLoadSubsetPromise: false, + onLoadSubsetResult: (result) => { + if (result instanceof Promise) { + void result.then(undefined, () => { + if (this.lastLoadRequestKey === loadRequestKey) { + this.lastLoadRequestKey = undefined + } + }) + } + this.orderedLoadSubsetResult?.(result) + }, + }) + } catch (error) { + if (this.lastLoadRequestKey === loadRequestKey) { + this.lastLoadRequestKey = undefined + } + throw error + } } private getWhereClause(): BasicExpression | undefined { @@ -491,8 +541,6 @@ export class CollectionSubscriber< this.subscriptionLoadingPromises.set(subscription, { resolve: resolve!, }) - this.collectionConfigBuilder.liveQueryCollection!._sync.trackLoadPromise( - promise, - ) + this.collectionConfigBuilder.trackSubsetLoadPromise(promise) } } diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 73f1115db..45f0ddd04 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -229,6 +229,14 @@ export interface SubscriptionStatusEvent { status: T } +/** Event emitted when a subset requested by this subscription fails to load. */ +export interface SubscriptionLoadSubsetErrorEvent { + type: `loadSubset:error` + subscription: Subscription + options: LoadSubsetOptions + error: unknown +} + /** * Event emitted when subscription is unsubscribed */ @@ -244,6 +252,7 @@ export type SubscriptionEvents = { 'status:change': SubscriptionStatusChangeEvent 'status:ready': SubscriptionStatusEvent<`ready`> 'status:loadingSubset': SubscriptionStatusEvent<`loadingSubset`> + 'loadSubset:error': SubscriptionLoadSubsetErrorEvent unsubscribed: SubscriptionUnsubscribedEvent } @@ -254,6 +263,8 @@ export type SubscriptionEvents = { export interface Subscription extends EventEmitter { /** Current status of the subscription */ readonly status: SubscriptionStatus + /** Most recent subset-load failure observed by this subscription. */ + readonly lastError: unknown | undefined } /** @@ -319,6 +330,11 @@ export type LoadSubsetOptions = { subscription?: Subscription } +/** + * Loads one subset and transfers its ongoing resource ownership only after + * returning `true` or a promise. An implementation that throws synchronously + * must release any partially acquired resource before throwing. + */ export type LoadSubsetFn = (options: LoadSubsetOptions) => true | Promise export type UnloadSubsetFn = (options: LoadSubsetOptions) => void @@ -893,6 +909,8 @@ export interface SubscribeChangesOptions< * @internal */ onLoadSubsetResult?: (result: Promise | true) => void + /** Receives subset-load failures scoped to this subscription. @internal */ + onLoadSubsetError?: (event: SubscriptionLoadSubsetErrorEvent) => void } export interface SubscribeChangesSnapshotOptions< diff --git a/packages/db/tests/collection-subscribe-changes.test.ts b/packages/db/tests/collection-subscribe-changes.test.ts index 4f851f08a..810acdcd1 100644 --- a/packages/db/tests/collection-subscribe-changes.test.ts +++ b/packages/db/tests/collection-subscribe-changes.test.ts @@ -2151,6 +2151,74 @@ describe(`Collection.subscribeChanges`, () => { whereExpression: eq(new PropRef([`status`]), `active`), }) }).toThrow(`Cannot specify both 'where' and 'whereExpression' options`) + expect(collection.subscriberCount).toBe(0) + }) + + it(`releases subscriber ownership when a where callback throws`, () => { + const failure = new Error(`where callback failed`) + const collection = createCollection<{ id: number; status: string }>({ + id: `where-callback-error-test`, + getKey: (item) => item.id, + sync: { sync: () => {} }, + }) + + expect(() => + collection.subscribeChanges(() => {}, { + where: () => { + throw failure + }, + }), + ).toThrow(failure) + expect(collection.subscriberCount).toBe(0) + }) + + it(`rolls back subscriber ownership when starting sync throws`, () => { + const failure = new Error(`sync setup failed`) + const collection = createCollection<{ id: number }>({ + id: `subscriber-start-sync-error-test`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: () => { + throw failure + }, + }, + }) + + expect(() => collection.subscribeChanges(() => {})).toThrow(failure) + expect(collection.subscriberCount).toBe(0) + expect(collection.status).toBe(`error`) + }) + + it(`preserves setup failure when subscription cleanup also throws`, () => { + const loadFailure = new Error(`initial subset failed`) + const unloadFailure = new Error(`subset cleanup failed`) + const collection = createCollection<{ id: number }>({ + id: `subscriber-load-and-unload-error-test`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => true, + unloadSubset: () => { + throw unloadFailure + }, + } + }, + }, + }) + + expect(() => + collection.subscribeChanges(() => {}, { + includeInitialState: true, + onLoadSubsetResult: () => { + throw loadFailure + }, + }), + ).toThrow(loadFailure) + expect(collection.subscriberCount).toBe(0) }) }) diff --git a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts new file mode 100644 index 000000000..07b401ef8 --- /dev/null +++ b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts @@ -0,0 +1,394 @@ +import { fc, test as fcTest } from '@fast-check/vitest' +import { describe, expect, it } from 'vitest' +import { createCollection } from '../src/collection/index.js' +import { createDeferred } from '../src/deferred.js' +import { oracleRandomParameters, readOracleRunConfig } from './oracle-config.js' +import { flushPromises } from './utils.js' +import type { Collection } from '../src/collection/index.js' +import type { ChangeMessageOrDeleteKeyMessage } from '../src/types.js' + +type ReplayRow = { + id: `one` | `two` + value: number +} + +type ReplayLoad = { + rows: ReadonlyArray + outcome: `resolve` | `reject` +} + +type ReplayAttempt = { + loads: ReadonlyArray +} + +type SourceAction = + | { type: `put`; row: ReplayRow } + | { type: `delete`; id: ReplayRow[`id`] } + +type ReplayScenario = { + initialRows: ReadonlyArray + demandCount: number + attempts: ReadonlyArray + settlementOrder: ReadonlyArray + afterSettlement: ReadonlyArray +} + +type PendingReplay = { + attemptIndex: number + load: ReplayLoad + deferred: ReturnType> + error: Error + settled: boolean +} + +const rowArbitrary: fc.Arbitrary = fc.record({ + id: fc.constantFrom(`one` as const, `two` as const), + value: fc.integer({ min: -2, max: 2 }), +}) + +const rowsArbitrary = fc.uniqueArray(rowArbitrary, { + minLength: 0, + maxLength: 2, + selector: ({ id }) => id, +}) + +const replayLoadArbitrary: fc.Arbitrary = fc.record({ + rows: rowsArbitrary, + outcome: fc.constantFrom(`resolve` as const, `reject` as const), +}) + +const sourceActionArbitrary: fc.Arbitrary = fc.oneof( + rowArbitrary.map((row) => ({ type: `put` as const, row })), + fc + .constantFrom(`one`, `two`) + .map((id) => ({ type: `delete` as const, id })), +) + +const replayScenarioArbitrary: fc.Arbitrary = fc + .integer({ min: 1, max: 2 }) + .chain((demandCount) => + fc + .record({ + initialRows: rowsArbitrary, + attempts: fc.array( + fc.record({ + loads: fc.array(replayLoadArbitrary, { + minLength: demandCount, + maxLength: demandCount, + }), + }), + { minLength: 1, maxLength: 3 }, + ), + afterSettlement: fc.array(sourceActionArbitrary, { + minLength: 0, + maxLength: 3, + }), + }) + .chain(({ initialRows, attempts, afterSettlement }) => { + const replayCount = attempts.length * demandCount + return fc + .shuffledSubarray( + Array.from({ length: replayCount }, (_, index) => index), + { minLength: replayCount, maxLength: replayCount }, + ) + .map((settlementOrder) => ({ + initialRows, + demandCount, + attempts, + settlementOrder, + afterSettlement, + })) + }), + ) + +function rowsById( + rows: ReadonlyArray, +): Map { + return new Map(rows.map((row) => [row.id, { ...row }])) +} + +function sortedRows( + rows: ReadonlyMap, +): Array { + return [...rows.values()].sort((left, right) => + left.id.localeCompare(right.id), + ) +} + +async function runReplayScenario(scenario: ReplayScenario): Promise { + let begin!: () => void + let write!: ( + message: ChangeMessageOrDeleteKeyMessage, + ) => void + let commit!: () => void + let truncate!: () => void + let loadCount = 0 + let unloadCount = 0 + const queuedLoads: Array<{ attemptIndex: number; load: ReplayLoad }> = [] + const pendingReplays: Array = [] + + const applyRows = (rows: ReadonlyArray) => { + if (rows.length === 0) return + begin() + for (const row of rows) { + write({ + type: collection.get(row.id) === undefined ? `insert` : `update`, + value: { ...row }, + }) + } + commit() + } + + const collection: Collection = + createCollection({ + id: `subscription-replay-oracle`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: (params) => { + begin = params.begin + write = params.write + commit = params.commit + truncate = params.truncate + params.markReady() + return { + loadSubset: () => { + loadCount++ + if (loadCount <= scenario.demandCount) { + if (loadCount === 1) applyRows(scenario.initialRows) + return true + } + + const queued = queuedLoads.shift() + if (!queued) throw new Error(`Replay load was not queued`) + const pending: PendingReplay = { + attemptIndex: queued.attemptIndex, + load: queued.load, + deferred: createDeferred(), + error: new Error(`Replay rejected`), + settled: false, + } + pendingReplays.push(pending) + return pending.deferred.promise + }, + unloadSubset: () => { + unloadCount++ + }, + } + }, + }, + }) + + const visible = new Map() + let publicationCount = 0 + const subscription = collection.subscribeChanges((changes) => { + publicationCount++ + for (const change of changes) { + if (change.type === `delete`) visible.delete(change.key) + else { + visible.set(change.key, { + id: change.value.id, + value: change.value.value, + }) + } + } + }) + let unsubscribed = false + + const assertPublished = ( + expected: ReadonlyMap, + ) => { + expect(sortedRows(visible)).toEqual(sortedRows(expected)) + } + + const applySourceAction = (action: SourceAction): boolean => { + if (action.type === `delete`) { + const previous = collection.get(action.id) + if (!previous) return false + begin() + write({ type: `delete`, key: action.id }) + commit() + return true + } + + const previous = collection.get(action.row.id) + if (previous?.value === action.row.value) return false + applyRows([action.row]) + return true + } + + try { + for (let demand = 0; demand < scenario.demandCount; demand++) { + subscription.requestSnapshot({ optimizedOnly: false }) + } + const expectedPublished = rowsById(scenario.initialRows) + const expectedSource = rowsById(scenario.initialRows) + assertPublished(expectedPublished) + const publicationCountBeforeReplay = publicationCount + + for (const [attemptIndex, attempt] of scenario.attempts.entries()) { + for (const load of attempt.loads) { + queuedLoads.push({ attemptIndex, load }) + } + begin() + truncate() + commit() + expectedSource.clear() + await flushPromises() + assertPublished(expectedPublished) + expect(publicationCount).toBe(publicationCountBeforeReplay) + } + + const replayBaseline = new Map(expectedPublished) + const currentAttemptIndex = scenario.attempts.length - 1 + const currentAttempt = scenario.attempts[currentAttemptIndex]! + const currentAttemptSucceeds = currentAttempt.loads.every( + ({ outcome }) => outcome === `resolve`, + ) + let lastReportedError: Error | undefined + for (const replayIndex of scenario.settlementOrder) { + const pending = pendingReplays[replayIndex]! + const load = pending.load + pending.settled = true + if (load.outcome === `resolve`) { + applyRows(load.rows) + for (const row of load.rows) { + expectedSource.set(row.id, { ...row }) + } + pending.deferred.resolve() + } else { + if (pending.attemptIndex === currentAttemptIndex) { + lastReportedError = pending.error + } + pending.deferred.reject(pending.error) + } + await flushPromises() + + const allSettled = pendingReplays.every((replay) => replay.settled) + if (allSettled && currentAttemptSucceeds) { + expectedPublished.clear() + for (const [id, row] of expectedSource) { + expectedPublished.set(id, { ...row }) + } + } else if (allSettled) { + expectedPublished.clear() + for (const [id, row] of replayBaseline) { + expectedPublished.set(id, { ...row }) + } + } + assertPublished(expectedPublished) + if (!allSettled || !currentAttemptSucceeds) { + expect(publicationCount).toBe(publicationCountBeforeReplay) + } else { + expect( + publicationCount - publicationCountBeforeReplay, + ).toBeLessThanOrEqual(1) + } + expect(subscription.lastError).toBe(lastReportedError) + } + + for (const action of scenario.afterSettlement) { + const countBeforeAction = publicationCount + const applied = applySourceAction(action) + if (applied && action.type === `delete`) { + expectedSource.delete(action.id) + expectedPublished.delete(action.id) + } else if (applied && action.type === `put`) { + expectedSource.set(action.row.id, { ...action.row }) + expectedPublished.set(action.row.id, { ...action.row }) + } + assertPublished(expectedPublished) + expect(publicationCount).toBe(countBeforeAction + Number(applied)) + } + + subscription.unsubscribe() + unsubscribed = true + expect(unloadCount).toBe(loadCount) + } finally { + for (const replay of pendingReplays) { + if (!replay.settled) replay.deferred.resolve() + } + await flushPromises() + if (!unsubscribed) subscription.unsubscribe() + await collection.cleanup() + } +} + +const { multiplier, replaySeed } = readOracleRunConfig() +const generatedRuns = 30 * multiplier + +describe(`CollectionSubscription replay oracle`, () => { + it(`publishes a same-key replacement after a failed replay`, async () => { + await runReplayScenario({ + initialRows: [{ id: `one`, value: 1 }], + demandCount: 1, + attempts: [{ loads: [{ rows: [], outcome: `reject` }] }], + settlementOrder: [0], + afterSettlement: [{ type: `put`, row: { id: `one`, value: 2 } }], + }) + }) + + it(`lets the newest successful replay replace an older failed replay`, async () => { + await runReplayScenario({ + initialRows: [{ id: `one`, value: 1 }], + demandCount: 1, + attempts: [ + { loads: [{ rows: [], outcome: `reject` }] }, + { + loads: [{ rows: [{ id: `one`, value: 2 }], outcome: `resolve` }], + }, + ], + settlementOrder: [1, 0], + afterSettlement: [], + }) + }) + + it(`releases every successful overlapping replay acquisition`, async () => { + await runReplayScenario({ + initialRows: [], + demandCount: 1, + attempts: [ + { loads: [{ rows: [], outcome: `resolve` }] }, + { loads: [{ rows: [], outcome: `resolve` }] }, + ], + settlementOrder: [1, 0], + afterSettlement: [], + }) + }) + + it(`uses the newest complete multi-demand replay`, async () => { + await runReplayScenario({ + initialRows: [{ id: `one`, value: 1 }], + demandCount: 2, + attempts: [ + { + loads: [ + { rows: [{ id: `one`, value: 2 }], outcome: `resolve` }, + { rows: [], outcome: `reject` }, + ], + }, + { + loads: [ + { rows: [{ id: `one`, value: 3 }], outcome: `resolve` }, + { rows: [{ id: `two`, value: 4 }], outcome: `resolve` }, + ], + }, + ], + settlementOrder: [2, 3, 0, 1], + afterSettlement: [], + }) + }) + + fcTest.prop([replayScenarioArbitrary], { + numRuns: generatedRuns, + seed: 1756, + })(`matches replay and ownership laws for a fixed seed`, runReplayScenario) + + fcTest.prop( + [replayScenarioArbitrary], + oracleRandomParameters(generatedRuns, replaySeed), + )( + `matches replay and ownership laws for a random or replayed seed`, + runReplayScenario, + ) +}) diff --git a/packages/db/tests/collection-subscription.test.ts b/packages/db/tests/collection-subscription.test.ts index 65465a6a8..f8c1fc23e 100644 --- a/packages/db/tests/collection-subscription.test.ts +++ b/packages/db/tests/collection-subscription.test.ts @@ -246,6 +246,485 @@ describe(`CollectionSubscription status tracking`, () => { subscription.unsubscribe() }) + it(`records the last rejected subset load without hiding ready data`, async () => { + const error = new Error(`incremental subset failed`) + const collection = createCollection<{ id: string; value: string }>({ + id: `subset-error-recording`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ + type: `insert`, + value: { id: `cached`, value: `available` }, + }) + commit() + markReady() + return { + loadSubset: () => Promise.reject(error), + } + }, + }, + }) + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + const failures: Array = [] + subscription.on(`loadSubset:error`, (event) => failures.push(event.error)) + + subscription.requestSnapshot({ optimizedOnly: false }) + await flushPromises() + + expect(subscription.status).toBe(`ready`) + expect(collection.get(`cached`)).toMatchObject({ value: `available` }) + expect(subscription.lastError).toBe(error) + expect(failures).toEqual([error]) + + subscription.unsubscribe() + await collection.cleanup() + }) + + it(`records a synchronously thrown subset failure`, async () => { + const error = new Error(`synchronous subset failure`) + const collection = createCollection<{ id: string }>({ + id: `synchronous-subset-error`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw error + }, + } + }, + }, + }) + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + const failures: Array = [] + subscription.on(`loadSubset:error`, (event) => failures.push(event.error)) + + expect(() => + subscription.requestSnapshot({ optimizedOnly: false }), + ).toThrow(error) + expect(subscription.lastError).toBe(error) + expect(failures).toEqual([error]) + + subscription.unsubscribe() + await collection.cleanup() + }) + + it(`does not unload a subset when loadSubset throws before acquisition`, async () => { + const failure = new Error(`subset failed before acquisition`) + const unloadedOptions: Array = [] + const collection = createCollection<{ id: string }>({ + id: `failed-subset-acquisition`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + unloadSubset: (options) => unloadedOptions.push(options), + } + }, + }, + }) + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + + expect(() => + subscription.requestSnapshot({ optimizedOnly: false }), + ).toThrow(failure) + subscription.unsubscribe() + + expect(unloadedOptions).toEqual([]) + await collection.cleanup() + }) + + it(`releases a subset when its load-result observer throws`, async () => { + const failure = new Error(`load-result observer failed`) + let acquiredOptions: unknown + const unloadedOptions: Array = [] + const collection = createCollection<{ id: string }>({ + id: `subset-observer-failure`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: (options) => { + acquiredOptions = options + return true + }, + unloadSubset: (options) => unloadedOptions.push(options), + } + }, + }, + }) + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + + expect(() => + subscription.requestSnapshot({ + optimizedOnly: false, + onLoadSubsetResult: () => { + throw failure + }, + }), + ).toThrow(failure) + subscription.unsubscribe() + + expect(unloadedOptions).toEqual([acquiredOptions]) + await collection.cleanup() + }) + + it(`reports a rejected subset replay after truncate`, async () => { + const error = new Error(`truncate replay failed`) + let truncateSource: () => void = () => { + throw new Error(`source has not started`) + } + let loadCount = 0 + const collection = createCollection<{ id: string }>({ + id: `truncate-subset-error`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, commit, markReady, truncate }) => { + markReady() + truncateSource = () => { + begin() + truncate() + commit() + } + return { + loadSubset: () => { + loadCount++ + return loadCount === 1 ? Promise.resolve() : Promise.reject(error) + }, + } + }, + }, + }) + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + const failures: Array = [] + subscription.on(`loadSubset:error`, (event) => failures.push(event.error)) + + subscription.requestSnapshot({ optimizedOnly: false }) + await flushPromises() + truncateSource() + await flushPromises() + + expect(subscription.status).toBe(`ready`) + expect(subscription.lastError).toBe(error) + expect(failures).toEqual([error]) + + subscription.unsubscribe() + await collection.cleanup() + }) + + it(`retains a subset after a synchronous truncate replay failure`, async () => { + const error = new Error(`synchronous truncate replay failed`) + let truncateSource: () => void = () => { + throw new Error(`source has not started`) + } + let loadCount = 0 + let unloadCount = 0 + const collection = createCollection<{ id: string }>({ + id: `synchronous-truncate-subset-error`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, commit, markReady, truncate }) => { + markReady() + truncateSource = () => { + begin() + truncate() + commit() + } + return { + loadSubset: () => { + loadCount++ + if (loadCount === 2) throw error + return true + }, + unloadSubset: () => { + unloadCount++ + }, + } + }, + }, + }) + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + + subscription.requestSnapshot({ optimizedOnly: false }) + truncateSource() + await flushPromises() + truncateSource() + await flushPromises() + + expect(loadCount).toBe(3) + expect(subscription.lastError).toBe(error) + + subscription.unsubscribe() + // The initial load and the later successful replay each acquired a lease. + expect(unloadCount).toBe(2) + await collection.cleanup() + }) + + it.each([`throw`, `reject`] as const)( + `keeps the last published snapshot when truncate replay fails ($0)`, + async (delivery) => { + type Row = { id: string } + const error = new Error(`truncate replay failed before replacement`) + let begin!: () => void + let write!: (message: { type: `insert`; value: Row }) => void + let commit!: () => void + let truncate!: () => void + let loadCount = 0 + let failReplay = true + const collection = createCollection({ + id: `truncate-replay-preserves-snapshot`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: (params) => { + begin = params.begin + write = params.write + commit = params.commit + truncate = params.truncate + params.markReady() + return { + loadSubset: () => { + loadCount++ + if (loadCount > 1 && failReplay) { + if (delivery === `throw`) throw error + return Promise.reject(error) + } + begin() + write({ type: `insert`, value: { id: `one` } }) + commit() + return true + }, + } + }, + }, + }) + const visible = new Map() + const subscription = collection.subscribeChanges( + (changes) => { + for (const change of changes) { + if (change.type === `delete`) visible.delete(change.key) + else visible.set(change.key, change.value) + } + }, + { includeInitialState: false }, + ) + + subscription.requestSnapshot({ optimizedOnly: false }) + expect([...visible.keys()]).toEqual([`one`]) + + begin() + truncate() + commit() + await flushPromises() + + expect(subscription.lastError).toBe(error) + expect([...visible.keys()]).toEqual([`one`]) + + begin() + write({ type: `insert`, value: { id: `two` } }) + commit() + await flushPromises() + + expect([...visible.keys()].sort()).toEqual([`one`, `two`]) + + failReplay = false + begin() + truncate() + commit() + await flushPromises() + + expect([...visible.keys()]).toEqual([`one`]) + + subscription.unsubscribe() + await collection.cleanup() + }, + ) + + it(`publishes one coherent snapshot after overlapping truncate replays`, async () => { + type Row = { id: string } + let begin!: () => void + let write!: (message: { type: `insert`; value: Row }) => void + let commit!: () => void + let truncate!: () => void + let loadCount = 0 + const resolveReplays: Array<() => void> = [] + const collection = createCollection({ + id: `overlapping-truncate-replays`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: (params) => { + begin = params.begin + write = params.write + commit = params.commit + truncate = params.truncate + params.markReady() + return { + loadSubset: () => { + loadCount++ + if (loadCount === 1) { + begin() + write({ type: `insert`, value: { id: `old` } }) + commit() + return true + } + if (loadCount === 3) { + begin() + write({ type: `insert`, value: { id: `new` } }) + commit() + } + return new Promise((resolve) => + resolveReplays.push(resolve), + ) + }, + } + }, + }, + }) + const visible = new Map() + const subscription = collection.subscribeChanges( + (changes) => { + for (const change of changes) { + if (change.type === `delete`) visible.delete(change.key) + else visible.set(change.key, change.value) + } + }, + { includeInitialState: false }, + ) + + subscription.requestSnapshot({ optimizedOnly: false }) + expect([...visible.keys()]).toEqual([`old`]) + + begin() + truncate() + commit() + await flushPromises() + + begin() + truncate() + commit() + await flushPromises() + + resolveReplays[1]!() + await flushPromises() + expect([...visible.keys()]).toEqual([`old`]) + + resolveReplays[0]!() + await flushPromises() + expect([...visible.keys()]).toEqual([`new`]) + + subscription.unsubscribe() + await collection.cleanup() + }) + + it(`scopes a subset failure to the subscription that requested it`, async () => { + const error = new Error(`first subscription failed`) + let loadCount = 0 + const collection = createCollection<{ id: string }>({ + id: `scoped-subset-error`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + loadCount++ + return loadCount === 1 ? Promise.reject(error) : Promise.resolve() + }, + } + }, + }, + }) + const failing = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + const healthy = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + + failing.requestSnapshot({ optimizedOnly: false }) + healthy.requestSnapshot({ optimizedOnly: false }) + await flushPromises() + + expect(collection.status).toBe(`ready`) + expect(failing.lastError).toBe(error) + expect(healthy.lastError).toBeUndefined() + + failing.unsubscribe() + healthy.unsubscribe() + await collection.cleanup() + }) + + it(`does not report an aborted subset request as a failure`, async () => { + const cancellation = new Error(`obsolete subset request`) + cancellation.name = `AbortError` + const collection = createCollection<{ id: string }>({ + id: `aborted-subset-request`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: ({ signal }) => + new Promise((_resolve, reject) => { + signal?.addEventListener(`abort`, () => reject(cancellation), { + once: true, + }) + }), + } + }, + }, + }) + const controller = new AbortController() + const failures: Array = [] + const subscription = collection.subscribeChanges(() => {}, { + includeInitialState: false, + }) + subscription.on(`loadSubset:error`, (event) => failures.push(event.error)) + + subscription.requestSnapshot({ + optimizedOnly: false, + signal: controller.signal, + }) + controller.abort() + await flushPromises() + + expect(subscription.status).toBe(`ready`) + expect(subscription.lastError).toBeUndefined() + expect(failures).toEqual([]) + + subscription.unsubscribe() + await collection.cleanup() + }) + it(`unsubscribe clears event listeners`, () => { const collection = createCollection<{ id: string; value: string }>({ id: `test`, diff --git a/packages/db/tests/collection.test.ts b/packages/db/tests/collection.test.ts index e96994db6..3ff8ede81 100644 --- a/packages/db/tests/collection.test.ts +++ b/packages/db/tests/collection.test.ts @@ -2182,6 +2182,45 @@ describe(`Collection isLoadingSubset property`, () => { expect(collection.isLoadingSubset).toBe(false) }) + it(`cleanup isolates subset loading state from a later sync session`, async () => { + const resolveLoads: Array<() => void> = [] + const collection = createCollection<{ id: string; value: string }>({ + id: `cleanup-isolates-subset-loading`, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => + new Promise((resolve) => resolveLoads.push(resolve)), + } + }, + }, + }) + + collection._sync.loadSubset({}) + expect(collection.isLoadingSubset).toBe(true) + + await collection.cleanup() + expect(collection.isLoadingSubset).toBe(false) + + collection.startSyncImmediate() + collection._sync.loadSubset({}) + expect(collection.isLoadingSubset).toBe(true) + + resolveLoads[0]!() + await flushPromises() + expect(collection.isLoadingSubset).toBe(true) + + resolveLoads[1]!() + await flushPromises() + expect(collection.isLoadingSubset).toBe(false) + + await collection.cleanup() + }) + it(`emits loadingSubset:change event`, async () => { let resolveLoadSubset: () => void const loadSubsetPromise = new Promise((resolve) => { diff --git a/packages/db/tests/effect.test.ts b/packages/db/tests/effect.test.ts index 154b35382..98b9b94bc 100644 --- a/packages/db/tests/effect.test.ts +++ b/packages/db/tests/effect.test.ts @@ -6,7 +6,10 @@ import { mockSyncCollectionOptions, mockSyncCollectionOptionsNoInitialState, } from './utils.js' -import type { DeltaEvent } from '../src/index.js' +import type { + DeltaEvent, + SubscriptionLoadSubsetErrorEvent, +} from '../src/index.js' // --------------------------------------------------------------------------- // Test types and helpers @@ -1507,6 +1510,325 @@ describe(`createEffect`, () => { }) describe(`source error handling`, () => { + it(`does not subscribe later sources after startup disposes the effect`, async () => { + const failure = new Error(`synchronous source failure`) + const users = createUsersCollection([sampleUsers[0]!]) + const issues = createIssuesCollection([sampleIssues[0]!]) + const subscribeChanges = users.subscribeChanges.bind(users) + + vi.spyOn(users, `subscribeChanges`).mockImplementation((( + callback, + options, + ) => { + const subscription = subscribeChanges(callback, { + ...options, + includeInitialState: false, + }) + const errorEvent: SubscriptionLoadSubsetErrorEvent = { + type: `loadSubset:error`, + subscription, + options: { subscription }, + error: failure, + } + options?.onLoadSubsetError?.(errorEvent) + return subscription + }) as typeof users.subscribeChanges) + + const sourceErrors: Array = [] + const effect = createEffect({ + query: (q) => + q + .from({ user: users }) + .leftJoin({ issue: issues }, ({ user, issue }) => + eq(user.id, issue.userId), + ) + .select(({ user, issue }) => ({ + id: user.id, + issueId: issue.id, + })), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }) + + expect(sourceErrors).toEqual([failure]) + expect(effect.disposed).toBe(true) + expect(users.subscriberCount).toBe(0) + expect(issues.subscriberCount).toBe(0) + await Promise.all([users.cleanup(), issues.cleanup()]) + }) + + it(`releases every source when one unsubscriber throws`, async () => { + const failure = new Error(`first source unload failed`) + const createSource = (id: string, unloadSubset: () => void) => + createCollection<{ id: number }>({ + id, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => true, + unloadSubset, + } + }, + }, + }) + const left = createSource(`effect-cleanup-left`, () => { + throw failure + }) + const right = createSource(`effect-cleanup-right`, () => {}) + const effect = createEffect({ + query: (q) => + q + .from({ left }) + .leftJoin({ right }, ({ left: leftRow, right: rightRow }) => + eq(leftRow.id, rightRow.id), + ), + onBatch: () => {}, + }) + + expect(left.subscriberCount).toBe(1) + expect(right.subscriberCount).toBe(1) + + await expect(effect.dispose()).rejects.toBe(failure) + expect(left.subscriberCount).toBe(0) + expect(right.subscriberCount).toBe(0) + + await Promise.all([left.cleanup(), right.cleanup()]) + }) + + it(`releases source ownership when the automatic subset load throws`, async () => { + const failure = new Error(`automatic subset failed`) + const users = createCollection({ + id: `effect-synchronous-subset-error`, + getKey: (user) => user.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const sourceErrors: Array = [] + + expect(() => + createEffect({ + query: (q) => q.from({ user: users }), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }), + ).toThrow(failure) + + expect(sourceErrors).toEqual([failure]) + expect(users.subscriberCount).toBe(0) + await users.cleanup() + }) + + it(`releases an ordered source when its initial subset load throws`, async () => { + const failure = new Error(`initial ordered subset failed`) + const users = createCollection({ + id: `effect-initial-ordered-subset-error`, + getKey: (user) => user.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const sourceErrors: Array = [] + + expect(() => + createEffect({ + query: (q) => + q + .from({ user: users }) + .orderBy(({ user }) => user.name, `asc`) + .limit(1), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }), + ).toThrow(failure) + + expect(sourceErrors).toEqual([failure]) + expect(users.subscriberCount).toBe(0) + await users.cleanup() + }) + + it(`releases every source when initial lazy demand throws`, async () => { + const failure = new Error(`initial lazy demand failed`) + const users = createUsersCollection([sampleUsers[0]!]) + const issues = createCollection({ + id: `effect-initial-lazy-subset-error`, + getKey: (issue) => issue.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const sourceErrors: Array = [] + + expect(() => + createEffect({ + query: (q) => + q + .from({ user: users }) + .leftJoin({ issue: issues }, ({ user, issue }) => + eq(user.id, issue.userId), + ) + .select(({ user, issue }) => ({ + id: user.id, + issueId: issue.id, + })), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }), + ).toThrow(failure) + + expect(sourceErrors).toEqual([failure]) + expect(users.subscriberCount).toBe(0) + expect(issues.subscriberCount).toBe(0) + await Promise.all([users.cleanup(), issues.cleanup()]) + }) + + it(`isolates synchronous lazy-demand failure from an established source commit`, async () => { + const failure = new Error(`incremental effect lazy demand failed`) + const users = createCollection( + mockSyncCollectionOptions({ + id: `incremental-effect-users`, + getKey: (user) => user.id, + initialData: [], + }), + ) + const issues = createCollection({ + id: `incremental-effect-issues`, + getKey: (issue) => issue.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const sourceErrors: Array = [] + const effect = createEffect({ + query: (q) => + q + .from({ user: users }) + .leftJoin({ issue: issues }, ({ user, issue }) => + eq(user.id, issue.userId), + ) + .select(({ user, issue }) => ({ + id: user.id, + issueId: issue.id, + })), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }) + + try { + let commitError: unknown + try { + users.utils.begin() + users.utils.write({ type: `insert`, value: sampleUsers[0]! }) + users.utils.commit() + } catch (error) { + commitError = error + } + + expect(commitError).toBeUndefined() + expect(sourceErrors).toEqual([failure]) + expect(effect.disposed).toBe(true) + } finally { + await effect.dispose() + await Promise.all([users.cleanup(), issues.cleanup()]) + } + }) + + it(`reports a rejected ordered subset load and disposes the effect`, async () => { + const failure = new Error(`ordered subset failed`) + let loadCount = 0 + let removeVisibleRow: () => void = () => { + throw new Error(`source has not started`) + } + const users = createCollection({ + id: `effect-rejected-ordered-users`, + getKey: (user) => user.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + removeVisibleRow = () => { + begin() + write({ type: `delete`, value: sampleUsers[0]! }) + commit() + } + return { + loadSubset: () => { + loadCount++ + if (loadCount > 1) return Promise.reject(failure) + begin() + write({ type: `insert`, value: sampleUsers[0]! }) + commit() + return Promise.resolve() + }, + } + }, + }, + }) + const sourceErrors: Array = [] + const effect = createEffect({ + query: (q) => + q + .from({ user: users }) + .orderBy(({ user }) => user.name, `asc`) + .limit(1), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }) + + try { + await flushPromises() + expect(sourceErrors).toEqual([]) + + removeVisibleRow() + await flushPromises() + + expect(sourceErrors).toEqual([failure]) + expect(effect.disposed).toBe(true) + } finally { + await effect.dispose() + await users.cleanup() + } + }) + it(`reports a rejected lazy subset load and disposes the effect`, async () => { const users = createUsersCollection([sampleUsers[0]!]) const issues = createCollection({ @@ -1549,6 +1871,69 @@ describe(`createEffect`, () => { } }) + it(`keeps the effect alive when obsolete lazy demand is aborted`, async () => { + const users = createUsersCollection([sampleUsers[0]!]) + const cancellation = new Error(`obsolete lazy demand`) + cancellation.name = `AbortError` + let capturedSignal: AbortSignal | undefined + const issues = createCollection({ + id: `effect-aborted-lazy-issues`, + getKey: (issue) => issue.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: ({ signal }) => { + capturedSignal = signal + return new Promise((_resolve, reject) => { + signal?.addEventListener( + `abort`, + () => reject(cancellation), + { once: true }, + ) + }) + }, + } + }, + }, + }) + const sourceErrors: Array = [] + const effect = createEffect({ + query: (q) => + q + .from({ user: users }) + .leftJoin({ issue: issues }, ({ user, issue }) => + eq(user.id, issue.userId), + ) + .select(({ user, issue }) => ({ + id: user.id, + issueId: issue.id, + })), + onBatch: () => {}, + onSourceError: (error) => sourceErrors.push(error), + }) + + try { + await flushPromises() + expect(capturedSignal?.aborted).toBe(false) + + users.utils.begin() + users.utils.write({ type: `delete`, value: sampleUsers[0]! }) + users.utils.commit() + await flushPromises() + + expect(capturedSignal?.aborted).toBe(true) + expect(sourceErrors).toEqual([]) + expect(effect.disposed).toBe(false) + } finally { + await effect.dispose() + await Promise.all([users.cleanup(), issues.cleanup()]) + } + }) + it(`should auto-dispose when source collection is cleaned up`, async () => { const users = createUsersCollection() const events: Array> = [] diff --git a/packages/db/tests/live-query-window-controller.test.ts b/packages/db/tests/live-query-window-controller.test.ts index 1bca17d3a..e35b29725 100644 --- a/packages/db/tests/live-query-window-controller.test.ts +++ b/packages/db/tests/live-query-window-controller.test.ts @@ -476,7 +476,20 @@ describe(`createLiveQueryWindowController`, () => { }, }, }) - const lq = makeOrderedLiveQuery(source, 2) + const lq = createLiveQueryCollection({ + query: (q) => + q + .from({ r: source }) + .orderBy(({ r }) => r.n, `asc`) + .limit(3) + .offset(0) + .select(({ r }) => ({ id: r.id, n: r.n })), + startSync: true, + gcTime: 1, + utils: { + customUtility: () => `custom`, + }, + }) const controller = createLiveQueryWindowController(lq as any, { pageSize: 2, }) @@ -488,6 +501,8 @@ describe(`createLiveQueryWindowController`, () => { await expect(controller.fetchNextPage()).rejects.toBe(failure) expect(controller.getSnapshot().pages).toHaveLength(1) expect(controller.getSnapshot().error).toBe(failure) + expect(lq.utils.lastSubsetError).toBe(failure) + expect(lq.utils.customUtility()).toBe(`custom`) controller.dispose() }) @@ -524,6 +539,95 @@ describe(`createLiveQueryWindowController`, () => { controller.dispose() }) + it(`reset does not inherit a superseded expansion failure`, async () => { + const failure = new Error(`superseded expansion failed`) + let loadCount = 0 + const rejectLoads = new Map void>() + const loaded = new Set() + const source = createCollection({ + id: `window-reset-real-source-${seq++}`, + getKey: (row) => row.id, + syncMode: `on-demand`, + startSync: true, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: (options) => { + loadCount++ + if (loadCount === 2) { + return new Promise((_resolve, reject) => { + rejectLoads.set(loadCount, reject) + }) + } + begin() + ROWS.slice(0, options.limit).forEach((row) => { + if (loaded.has(row.id)) return + loaded.add(row.id) + write({ type: `insert`, value: row }) + }) + commit() + return Promise.resolve() + }, + } + }, + }, + }) + const lq = makeOrderedLiveQuery(source, 2) + const controller = createLiveQueryWindowController(lq as any, { + pageSize: 2, + }) + controller.subscribe(() => {}) + + try { + await controller.preload() + const expansion = Promise.resolve(controller.fetchNextPage()) + expect(loadCount).toBe(2) + const rejectExpansion = rejectLoads.get(2) + expect(rejectExpansion).toBeDefined() + const reset = Promise.resolve(controller.reset()) + void expansion.catch(() => undefined) + void reset.catch(() => undefined) + + rejectExpansion!(failure) + + await expect(reset).resolves.toBeUndefined() + await expect(expansion).rejects.toBe(failure) + expect(controller.getSnapshot().pages).toHaveLength(1) + } finally { + controller.dispose() + await Promise.all([lq.cleanup(), source.cleanup()]) + } + }) + + it(`cleanup settles the active load operation before another sync session`, async () => { + const lq = makeOrderedLiveQuery(makeSource(), 2) + await lq.preload() + + let resolveLoad!: () => void + const load = new Promise((resolve) => { + resolveLoad = resolve + }) + const operation = lq._sync.beginLoadSubsetOperation() + lq._sync.trackLoadPromise(load) + const waiting = Promise.resolve(operation.wait()) + let settled = false + void waiting.then(() => { + settled = true + }) + + lq._sync.cleanup() + await Promise.resolve() + + expect(settled).toBe(true) + + resolveLoad() + await waiting + await lq.cleanup() + }) + it(`retains an unsubscribed lease until overlapping requests settle`, async () => { const lq = makeOrderedLiveQuery(makeSource(), 2) await lq.preload() diff --git a/packages/db/tests/query/live-query-collection.test.ts b/packages/db/tests/query/live-query-collection.test.ts index 85478e95c..c6fc4be39 100644 --- a/packages/db/tests/query/live-query-collection.test.ts +++ b/packages/db/tests/query/live-query-collection.test.ts @@ -1435,6 +1435,300 @@ describe(`createLiveQueryCollection`, () => { expect(liveQuery.isLoadingSubset).toBe(false) }) + it(`releases an ordered source when initial live-query loading throws`, async () => { + const failure = new Error(`initial ordered live-query load failed`) + const source = createCollection({ + id: `initial-ordered-live-query-error`, + getKey: (user) => user.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const live = createLiveQueryCollection((q) => + q + .from({ user: source }) + .orderBy(({ user }) => user.name, `asc`) + .limit(1), + ) + + await expect(Promise.resolve().then(() => live.preload())).rejects.toBe( + failure, + ) + expect(source.subscriberCount).toBe(0) + + await Promise.all([live.cleanup(), source.cleanup()]) + }) + + it(`releases earlier live-query sources when initial lazy demand throws`, async () => { + type Issue = { id: number; userId: number } + const failure = new Error(`initial live-query lazy demand failed`) + const users = createCollection( + mockSyncCollectionOptions({ + id: `partial-live-query-users`, + getKey: (user) => user.id, + initialData: [sampleUsers[0]!], + }), + ) + const issues = createCollection({ + id: `partial-live-query-issues`, + getKey: (issue) => issue.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const live = createLiveQueryCollection((q) => + q + .from({ user: users }) + .leftJoin({ issue: issues }, ({ user, issue }) => + eq(user.id, issue.userId), + ) + .select(({ user, issue }) => ({ + id: user.id, + issueId: issue.id, + })), + ) + + await expect(Promise.resolve().then(() => live.preload())).rejects.toBe( + failure, + ) + expect(users.subscriberCount).toBe(0) + expect(issues.subscriberCount).toBe(0) + + await Promise.all([live.cleanup(), users.cleanup(), issues.cleanup()]) + }) + + it(`isolates synchronous lazy-demand failure from an established source commit`, async () => { + type Issue = { id: number; userId: number } + const failure = new Error(`incremental live-query lazy demand failed`) + const users = createCollection( + mockSyncCollectionOptions({ + id: `incremental-live-query-users`, + getKey: (user) => user.id, + initialData: [], + }), + ) + const issues = createCollection({ + id: `incremental-live-query-issues`, + getKey: (issue) => issue.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + throw failure + }, + } + }, + }, + }) + const live = createLiveQueryCollection((q) => + q + .from({ user: users }) + .leftJoin({ issue: issues }, ({ user, issue }) => + eq(user.id, issue.userId), + ) + .select(({ user, issue }) => ({ + id: user.id, + issueId: issue.id, + })), + ) + + try { + await live.preload() + expect(live.status).toBe(`ready`) + + let commitError: unknown + try { + users.utils.begin() + users.utils.write({ type: `insert`, value: sampleUsers[0]! }) + users.utils.commit() + } catch (error) { + commitError = error + } + + expect(commitError).toBeUndefined() + expect(live.status).toBe(`error`) + expect(live.utils.lastSubsetError).toBe(failure) + } finally { + await Promise.all([live.cleanup(), users.cleanup(), issues.cleanup()]) + } + }) + + it.each([`throw`, `reject`] as const)( + `propagates lazy child demand failure from a window change ($0)`, + async (delivery) => { + type Parent = { id: number; rank: number } + type Child = { id: number; parentId: number } + const failure = new Error(`window child demand failed`) + const loadedParents = new Set() + let parentLoadCount = 0 + const parents = createCollection({ + id: `window-lazy-demand-parents`, + getKey: (parent) => parent.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: () => { + parentLoadCount++ + begin() + const candidates: Array = [ + { id: 1, rank: 1 }, + { id: 2, rank: 2 }, + ] + candidates.slice(0, parentLoadCount).forEach((parent) => { + if (loadedParents.has(parent.id)) return + loadedParents.add(parent.id) + write({ type: `insert`, value: parent }) + }) + commit() + return Promise.resolve() + }, + } + }, + }, + }) + let childLoadCount = 0 + const children = createCollection({ + id: `window-lazy-demand-children`, + getKey: (child) => child.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: () => { + childLoadCount++ + if (childLoadCount > 1) { + if (delivery === `throw`) throw failure + return Promise.reject(failure) + } + begin() + write({ type: `insert`, value: { id: 10, parentId: 1 } }) + commit() + return Promise.resolve() + }, + } + }, + }, + }) + const live = createLiveQueryCollection((q) => + q + .from({ parent: parents }) + .leftJoin({ child: children }, ({ parent, child }) => + eq(parent.id, child.parentId), + ) + .orderBy(({ parent }) => parent.rank, `asc`) + .limit(1) + .select(({ parent, child }) => ({ + id: parent.id, + childId: child.id, + })), + ) + + try { + await live.preload() + expect(live.status).toBe(`ready`) + + const setWindow = async () => { + const result = live.utils.setWindow({ offset: 0, limit: 2 }) + if (result !== true) await result + } + await expect(setWindow()).rejects.toBe(failure) + expect(live.utils.lastSubsetError).toBe(failure) + } finally { + await Promise.all([ + live.cleanup(), + parents.cleanup(), + children.cleanup(), + ]) + } + }, + ) + + it(`retries the same ordered refill after a transient rejection`, async () => { + type Row = { id: number; rank: number } + const failure = new Error(`ordered refill failed`) + let loadCount = 0 + const source = createCollection({ + id: `ordered-refill-retry-source`, + getKey: (row) => row.id, + syncMode: `on-demand`, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: () => { + loadCount++ + if (loadCount === 2) return Promise.reject(failure) + const deliver = () => { + begin() + write({ + type: `insert`, + value: { id: loadCount, rank: loadCount }, + }) + commit() + } + if (loadCount === 1) { + deliver() + return true + } + return Promise.resolve().then(deliver) + }, + } + }, + }, + }) + const live = createLiveQueryCollection((q) => + q + .from({ row: source }) + .orderBy(({ row }) => row.rank, `asc`) + .limit(2), + ) + + try { + await live.preload() + await flushPromises() + expect(loadCount).toBe(2) + expect(live.utils.lastSubsetError).toBe(failure) + + const retry = live.utils.setWindow({ offset: 0, limit: 2 }) + if (retry instanceof Promise) await retry + await flushPromises() + + expect(loadCount).toBe(3) + expect(Array.from(live.values(), ({ id }) => id)).toEqual([1, 3]) + } finally { + await Promise.all([live.cleanup(), source.cleanup()]) + } + }) + it(`concurrent live queries should each track loading state independently`, async () => { // This tests the fix for the !wasLoadingBefore bug: // When multiple live queries subscribe to the same source collection, @@ -2054,6 +2348,37 @@ describe(`createLiveQueryCollection`, () => { expect(result).toBe(true) }) + it(`does not wait for subset work that predates the window operation`, async () => { + const source = createCollection( + mockSyncCollectionOptions({ + id: `window-with-unrelated-load`, + getKey: (user) => user.id, + initialData: sampleUsers, + autoIndex: `eager`, + }), + ) + const live = createLiveQueryCollection((q) => + q + .from({ user: source }) + .orderBy(({ user }) => user.name, `asc`) + .limit(1), + ) + let resolveUnrelated: () => void + const unrelated = new Promise((resolve) => { + resolveUnrelated = resolve + }) + + try { + await live.preload() + live._sync.trackLoadPromise(unrelated) + + expect(live.utils.setWindow({ offset: 0, limit: 2 })).toBe(true) + } finally { + resolveUnrelated!() + await Promise.all([live.cleanup(), source.cleanup()]) + } + }) + it(`setWindow returns and resolves a Promise when async loading is triggered`, async () => { // This is an integration test that validates the full async flow: // 1. setWindow triggers loading more data