From 62e65f7d39720864967d4452de037a40bd761c8e Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 12:30:25 -0600 Subject: [PATCH 1/2] fix(db): finish review recovery and draft follow-ups --- .changeset/finish-review-recovery.md | 11 + docs/guides/error-handling.md | 25 ++ docs/guides/mutations.md | 7 + .../src/persisted.ts | 43 ++-- .../tests/persisted.test.ts | 97 +++++++- packages/db/package.json | 2 +- packages/db/src/collection/state.ts | 9 + packages/db/src/proxy.ts | 50 +++- packages/db/src/query/live/ARCHITECTURE.md | 18 ++ .../src/query/live/collection-subscriber.ts | 3 + .../src/query/live/ordered-source-loader.ts | 154 ++++++++++-- packages/db/src/transactions.ts | 18 +- packages/db/src/utils.ts | 5 +- .../collection-subscribe-changes.test.ts | 116 +++++---- .../tests/proxy-detachment-contract.test.ts | 231 ++++++++++++++++++ .../tests/query/ordered-default-work.test.ts | 231 ++++++++++++++++++ packages/db/tests/transactions.test.ts | 55 +++++ .../powersync-db-collection/src/powersync.ts | 3 + .../tests/on-demand-sync.test.ts | 109 +++++++-- 19 files changed, 1040 insertions(+), 147 deletions(-) create mode 100644 .changeset/finish-review-recovery.md create mode 100644 packages/db/tests/proxy-detachment-contract.test.ts create mode 100644 packages/db/tests/query/ordered-default-work.test.ts diff --git a/.changeset/finish-review-recovery.md b/.changeset/finish-review-recovery.md new file mode 100644 index 0000000000..692dee167d --- /dev/null +++ b/.changeset/finish-review-recovery.md @@ -0,0 +1,11 @@ +--- +'@tanstack/db': patch +'@tanstack/db-sqlite-persistence-core': patch +'@tanstack/powersync-db-collection': patch +--- + +Preserve native values, arbitrary class references, and draft cycles during mutation detachment; keep transaction persistence receipts settled after publication errors and avoid restoring an acknowledged direct insert over its server row. + +Retire replaced ordered prefixes and retry automatic ordered repair at most twice while retaining stale results and exposing the error; cleanup cancels retries and explicit window retry remains available. + +Keep persisted acquisitions independent, reject upstream load failures without discarding cached rows, and restore PowerSync readiness after a successful tracking rebuild and applied baseline. diff --git a/docs/guides/error-handling.md b/docs/guides/error-handling.md index 6f20d4aea1..955a8b7937 100644 --- a/docs/guides/error-handling.md +++ b/docs/guides/error-handling.md @@ -155,6 +155,19 @@ try { Effects report subset failures through `onSourceError` and dispose because their incremental result can no longer be kept complete. +When a source change invalidates an ordered window, automatic full-source +repair keeps the last complete snapshot visible. A failed repair exposes +`utils.lastSubsetError` and retries at most twice, after 250 ms and 500 ms. +Exhausting those retries does not put an already-ready query into a terminal +error state or clear its rows. The app can show the error and explicitly retry +with `setWindow()`. Cleanup or truncate cancels the old repair timer. Failed +imperative window moves and initial loads do not use this background retry. + +For SQLite-persisted on-demand collections, a failed upstream `loadSubset` +rejects even when hydration succeeded. Cached rows remain readable; their +availability does not mean the remote request succeeded. Background coordinator +retry, where supported, does not change the failed caller's outcome. + 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 and keeps later source changes private because they @@ -323,6 +336,18 @@ try { } ``` +Explicit cancellation is different from a mutation failure. If you call +`tx.rollback()` while `mutationFn` is pending, the rollback settles +`tx.isPersisted.promise` as rejected. A later result or rejection from that +mutation function is ignored: the outstanding `commit()` call resolves and +`tx.error` is not populated by that late rejection. Observe `isPersisted.promise` +when you need the transaction's outcome, including explicit cancellation. + +After the mutation function succeeds, a publication listener can still throw +while the completed transaction updates its collections. In that case +`commit()` rejects with the listener error, but `isPersisted.promise` resolves +and the transaction remains completed. This is not a persistence failure. + ## Collection Operation Errors ### Invalid Collection State diff --git a/docs/guides/mutations.md b/docs/guides/mutations.md index 1a535f8600..8b1aa3ed2e 100644 --- a/docs/guides/mutations.md +++ b/docs/guides/mutations.md @@ -364,6 +364,13 @@ need to keep a new caller-owned object unchanged during the callback, insert your own copy. A thrown callback does not roll back edits to that caller-owned object; it leaves existing collection data unchanged. +Arbitrary class instances are an exception: newly assigned instances stay by +reference so their methods, prototypes, and private fields remain intact. +Later changes to such an instance can therefore affect stored data without a +new update or notification. Treat those instances as immutable, or convert them +to plain data before assignment when you need isolation. Supported native values +such as `URL`, `Date`, `RegExp`, and typed arrays are copied instead. + ### Delete Remove items from a collection: diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index dc2084d39b..7db554ec82 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -707,18 +707,6 @@ function stableSerialize(value: unknown): string { return JSON.stringify(toStableSerializable(value) ?? null) } -function normalizeSubsetOptionsForKey( - options: LoadSubsetOptions, -): Record { - return { - where: toStableSerializable(options.where), - orderBy: toStableSerializable(options.orderBy), - limit: options.limit, - cursor: toStableSerializable(options.cursor), - offset: options.offset, - } -} - function normalizeSyncFnResult(result: void | (() => void) | SyncConfigRes) { if (typeof result === `function`) { return { cleanup: result } satisfies SyncConfigRes @@ -802,7 +790,7 @@ class PersistedCollectionRuntime< BufferedSyncTransaction > = [] private readonly queuedTxCommitted: Array = [] - private readonly subscriptionIds = new WeakMap() + private readonly requestIds = new WeakMap() private collection: Collection | null = null @@ -824,7 +812,7 @@ class PersistedCollectionRuntime< private indexAddedUnsubscribe: (() => void) | null = null private indexRemovedUnsubscribe: (() => void) | null = null private remoteEnsureRetryTimer: ReturnType | null = null - private nextSubscriptionId = 0 + private nextRequestId = 0 private latestTerm = 0 private latestSeq = 0 @@ -938,7 +926,8 @@ class PersistedCollectionRuntime< await this.bootstrapPersistedIndexes(indexBootstrapSnapshot) if (this.syncMode !== `on-demand`) { - this.activeSubsets.set(this.getSubsetKey({}), {}) + const initialSubset = {} + this.activeSubsets.set(this.getSubsetKey(initialSubset), initialSubset) const appliedCursor = this.appliedReceiptSequence await this.applyMutex.run(() => this.hydrateSubsetUnsafe({}, { requestRemoteEnsure: false }), @@ -1044,6 +1033,8 @@ class PersistedCollectionRuntime< } console.warn(`Failed to trigger remote subset load:`, error) this.queueRemoteSubsetEnsure(options) + // Hydration remains readable, but it does not satisfy remote demand. + throw error } } } @@ -1808,20 +1799,14 @@ class PersistedCollectionRuntime< } private getSubsetKey(options: LoadSubsetOptions): string { - const subscription = options.subscription as object | undefined - if (subscription && typeof subscription === `object`) { - const existingId = this.subscriptionIds.get(subscription) - if (existingId) { - return existingId - } - - this.nextSubscriptionId++ - const id = `sub:${this.nextSubscriptionId}` - this.subscriptionIds.set(subscription, id) - return id - } - - return `opts:${stableSerialize(normalizeSubsetOptionsForKey(options))}` + // A subscription can own several independent acquisitions, including + // identical requests. Only releasing this options object ends its lease. + let id = this.requestIds.get(options) + if (id === undefined) { + id = `request:${++this.nextRequestId}` + this.requestIds.set(options, id) + } + return id } private queueRemoteSubsetEnsure(options: LoadSubsetOptions): void { diff --git a/packages/db-sqlite-persistence-core/tests/persisted.test.ts b/packages/db-sqlite-persistence-core/tests/persisted.test.ts index 89f9fef44b..d90dadccb6 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted.test.ts @@ -1744,6 +1744,101 @@ describe(`persistedCollectionOptions`, () => { expect(collection.get(`2`)).toBeUndefined() }) + it.each( + [false, true].flatMap((sharedSubscription) => + [false, true].map((identical) => ({ sharedSubscription, identical })), + ), + )( + `keeps sibling requests owned after one release: %j`, + async ({ sharedSubscription, identical }) => { + const adapter = createRecordingAdapter([{ id: `1`, title: `Before` }]) + const coordinator = createCoordinatorHarness() + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + const subscription = collection.subscribeChanges(() => {}) + const owner = sharedSubscription ? { subscription } : {} + const page: LoadSubsetOptions = { + ...owner, + ...(identical ? {} : { limit: 1 }), + } + const all: LoadSubsetOptions = { ...owner } + try { + await collection._sync.loadSubset(page) + await collection._sync.loadSubset(all) + collection._sync.unloadSubset(page) + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `sibling-update`, + latestRowVersion: 1, + requiresFullReload: false, + changedRows: [{ key: `1`, value: { id: `1`, title: `After` } }], + deletedKeys: [], + }) + await flushAsyncWork() + expect(stripVirtualProps(collection.get(`1`))).toEqual({ + id: `1`, + title: `After`, + }) + } finally { + subscription.unsubscribe() + await collection.cleanup() + } + }, + ) + + it(`reports a non-abort upstream failure rather than treating hydration as remote success`, async () => { + const failure = new Error(`remote acquisition failed`) + const warn = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const collection = createCollection( + persistedCollectionOptions({ + id: `remote-acquisition-failure`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => Promise.reject(failure), + } + }, + }, + persistence: { + adapter: createRecordingAdapter([ + { id: `cached`, title: `Last known row` }, + ]), + }, + }), + ) + collection.startSyncImmediate() + try { + await expect( + Promise.resolve(collection._sync.loadSubset({})), + ).rejects.toBe(failure) + expect(stripVirtualProps(collection.get(`cached`))).toEqual({ + id: `cached`, + title: `Last known row`, + }) + } finally { + warn.mockRestore() + await collection.cleanup() + } + }) + it(`does not release or acquire an upstream lease cancelled during hydration`, async () => { const adapter = createRecordingAdapter() const hydrate = adapter.loadSubset @@ -1861,7 +1956,7 @@ describe(`persistedCollectionOptions`, () => { const callsBeforeRetry = ensure.mock.calls.length await vi.advanceTimersByTimeAsync(200) if (action === `offline`) { - expect(result).toBe(`ready`) + expect(result).toBe(failure) expect(ensure.mock.calls.length).toBeGreaterThan(callsBeforeRetry) } else { if (action === `abort`) expect(result).toBe(failure) diff --git a/packages/db/package.json b/packages/db/package.json index 5f619e5461..5dba5f322e 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -22,7 +22,7 @@ "lint": "eslint . --fix", "test": "vitest --run", "test:facade-retention": "node --expose-gc --import tsx tests/facade-retention.probe.ts", - "test:oracles": "vitest --run tests/collection-cleanup-restart-oracle.test.ts tests/effect-disposal-oracle.test.ts tests/collection-metadata-publication-oracle.property.test.ts tests/collection-state-retention-oracle.property.test.ts tests/collection-subscription-lifecycle-history.property.test.ts tests/collection-subscription-lifecycle-oracle.test.ts tests/collection-subscription-lifecycle-publication.property.test.ts tests/collection-subscription-replay-oracle.property.test.ts tests/d2-source-reconciliation-oracle.property.test.ts tests/query/includes-collection-oracle.property.test.ts tests/query/includes-functional-projection-oracle.test.ts tests/query/includes-functional-input-boundary.test.ts tests/query/includes-context-transport-oracle.test.ts tests/query/includes-cross-formulation-oracle.property.test.ts tests/query/includes-optimistic-oracle.property.test.ts tests/query/includes-oracle.property.test.ts tests/query/includes-publication-oracle.test.ts tests/query/includes-query-shape-oracle.test.ts tests/query/includes-temporal-oracle.test.ts tests/query/includes-work-counter-oracle.test.ts tests/query/load-subset-oracle.property.test.ts tests/query/load-subset-replay-refinement-oracle.test.ts tests/query/load-subset-source-readiness-refinement-oracle.test.ts tests/query/load-subset-transaction-refinement-oracle.test.ts tests/query/ordered-source-loader-state.test.ts tests/query/ordered-demand-retirement.test.ts tests/query/ordered-lifecycle-oracle.property.test.ts tests/query/ordered-work-oracle.property.test.ts tests/query/pagination-oracle.property.test.ts" + "test:oracles": "vitest --run tests/collection-cleanup-restart-oracle.test.ts tests/effect-disposal-oracle.test.ts tests/collection-metadata-publication-oracle.property.test.ts tests/collection-state-retention-oracle.property.test.ts tests/collection-subscription-lifecycle-history.property.test.ts tests/collection-subscription-lifecycle-oracle.test.ts tests/collection-subscription-lifecycle-publication.property.test.ts tests/collection-subscription-replay-oracle.property.test.ts tests/d2-source-reconciliation-oracle.property.test.ts tests/query/includes-collection-oracle.property.test.ts tests/query/includes-functional-projection-oracle.test.ts tests/query/includes-functional-input-boundary.test.ts tests/query/includes-context-transport-oracle.test.ts tests/query/includes-cross-formulation-oracle.property.test.ts tests/query/includes-optimistic-oracle.property.test.ts tests/query/includes-oracle.property.test.ts tests/query/includes-publication-oracle.test.ts tests/query/includes-query-shape-oracle.test.ts tests/query/includes-temporal-oracle.test.ts tests/query/includes-work-counter-oracle.test.ts tests/query/load-subset-oracle.property.test.ts tests/query/load-subset-replay-refinement-oracle.test.ts tests/query/load-subset-source-readiness-refinement-oracle.test.ts tests/query/load-subset-transaction-refinement-oracle.test.ts tests/query/ordered-source-loader-state.test.ts tests/query/ordered-demand-retirement.test.ts tests/query/ordered-default-work.test.ts tests/query/ordered-lifecycle-oracle.property.test.ts tests/query/ordered-work-oracle.property.test.ts tests/query/pagination-oracle.property.test.ts" }, "type": "module", "main": "dist/cjs/index.cjs", diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index f0702e5aab..4fae0900a7 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -518,6 +518,15 @@ export class CollectionStateManager< if (!this.isThisCollection(mutation.collection)) { continue } + // A truncate/immediate sync can acknowledge an insert before its + // persistence callback settles. Do not resurrect that client row. + if ( + isDirectTransaction && + mutation.type === `insert` && + this.syncedData.has(mutation.key) + ) { + continue + } this.pendingLocalOrigins.add(mutation.key) if (!mutation.optimistic) { continue diff --git a/packages/db/src/proxy.ts b/packages/db/src/proxy.ts index 0f0fe5d38a..cecc2222a6 100644 --- a/packages/db/src/proxy.ts +++ b/packages/db/src/proxy.ts @@ -3,7 +3,7 @@ * and provides a way to retrieve those changes. */ -import { deepEquals, isTemporal } from './utils' +import { deepEquals, deepEqualsInternal, isTemporal } from './utils' // Resolve draft handles before calling native Map/Set membership methods. const draftCopies = new WeakMap() @@ -343,7 +343,10 @@ interface ChangeTracker { function deepClone( obj: T, visited = new WeakMap(), + detach = false, ): T { + // A draft handle and its underlying copy must share one cycle identity. + obj = unwrapDraft(obj) as T // Handle null and undefined if (obj === null || obj === undefined) { return obj @@ -360,18 +363,29 @@ function deepClone( } if (obj instanceof Date) { - return new Date(obj.getTime()) as unknown as T + const clone = new Date(obj.getTime()) + visited.set(obj, clone) + return clone as T } if (obj instanceof RegExp) { - return new RegExp(obj.source, obj.flags) as unknown as T + const clone = new RegExp(obj.source, obj.flags) + clone.lastIndex = obj.lastIndex + visited.set(obj, clone) + return clone as T + } + + if (obj instanceof URL) { + const clone = new URL(obj.href) + visited.set(obj, clone) + return clone as T } if (Array.isArray(obj)) { const arrayClone = [] as Array visited.set(obj as object, arrayClone) obj.forEach((item, index) => { - arrayClone[index] = deepClone(item, visited) + arrayClone[index] = deepClone(item, visited, detach) }) return arrayClone as unknown as T } @@ -397,7 +411,7 @@ function deepClone( const clone = new Map() as Map visited.set(obj as object, clone) obj.forEach((value, key) => { - clone.set(key, deepClone(value, visited)) + clone.set(key, deepClone(value, visited, detach)) }) return clone as unknown as T } @@ -406,7 +420,7 @@ function deepClone( const clone = new Set() visited.set(obj as object, clone) obj.forEach((value) => { - clone.add(deepClone(value, visited)) + clone.add(deepClone(value, visited, detach)) }) return clone as unknown as T } @@ -418,6 +432,13 @@ function deepClone( return obj } + // Arbitrary instances may carry private/native state we cannot reconstruct. + // Keep them by reference at publication, rather than silently flattening them. + if (detach) { + const prototype = Object.getPrototypeOf(obj) + if (prototype !== Object.prototype && prototype !== null) return obj + } + const clone = {} as Record visited.set(obj as object, clone) @@ -426,6 +447,7 @@ function deepClone( clone[key] = deepClone( (obj as Record)[key], visited, + detach, ) } } @@ -435,6 +457,7 @@ function deepClone( clone[sym] = deepClone( (obj as Record)[sym], visited, + detach, ) } @@ -897,20 +920,23 @@ export function createChangeProxy< const mayHaveChangedAliases = Object.keys(changeTracker.assigned_).some( (key) => typeof changeTracker.copy_[key] === `object`, ) + const pairedRoots = new Map([ + [changeTracker.copy_, changeTracker.originalObject], + ]) // Iterate through keys in keyObj for (const key in changeTracker.copy_) { const value: unknown = changeTracker.copy_[key] const original: unknown = changeTracker.originalObject[key] - // Include sibling aliases changed through another route. An untouched - // link back to this row does not add a field to its sparse change set. + // Compare child contents, stopping only at paired root backedges. A + // child's own changes still count even when it also points to this row. if ( (changeTracker.assigned_[key] === true || (mayHaveChangedAliases && - changeTracker.copy_[key] !== changeTracker.copy_ && - !deepEquals( + !deepEqualsInternal( value instanceof Set ? Array.from(value) : value, original instanceof Set ? Array.from(original) : original, + pairedRoots, ))) && key in changeTracker.copy_ ) { @@ -959,7 +985,7 @@ export function withChangeTracking( callback(proxy) - return deepClone(getChanges()) + return deepClone(getChanges(), undefined, true) } /** @@ -978,5 +1004,5 @@ export function withArrayChangeTracking( callback(proxies) - return deepClone(getChanges()) + return deepClone(getChanges(), undefined, true) } diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index 418b612890..918e2e9815 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -810,6 +810,24 @@ cannot advance it merely by entering D2. This relies on the adapter fulfilling the exact ordered request, not just resolving after an arbitrary partial write. An empty range does not invent a boundary or prove source exhaustion. +For no-index and multi-column prefix loading, an unrelated new key does not +reacquire an already full window. An explicit window move, an underfilled +window, or a settled prefix smaller than a window widened during that request +still requires acquisition. A full local window alone does not prove that the +provider fulfilled a concurrent window change. + +A successful larger prefix retires settled smaller prefix acquisitions from +the same ordered source plan, after the replacement has applied. It does not +retire cursor suffixes, ties, unfinished work, or another subscription's leases. +Adapter eviction must still preserve rows owned by the replacement or peers. + +Automatic full-source repair after an established window fails can retry twice, +after 250 ms and 500 ms. Every retry releases failed acquisitions before starting +the replacement. It uses the same publication barrier; stale rows stay public +and the last error stays observable if the budget is exhausted. Initial loads +and explicit window failures do not auto-retry. Cleanup, truncate, and explicit +retry supersede queued repair work. A successful repair resets the budget. + An explicit window move counts current rows at or before that boundary in the requested prefix. It acquires only the missing portion, with both cursor and offset derived from that confirmed range, not from all observed rows. These diff --git a/packages/db/src/query/live/collection-subscriber.ts b/packages/db/src/query/live/collection-subscriber.ts index a4e35b8af6..4b1bcb9c72 100644 --- a/packages/db/src/query/live/collection-subscriber.ts +++ b/packages/db/src/query/live/collection-subscriber.ts @@ -353,6 +353,9 @@ export class CollectionSubscriber< } onLoadSubsetResult(result) }, + () => + this.collectionConfigBuilder.liveQueryCollection?.status === `ready` && + !this.collectionConfigBuilder.hasActiveWindowOperation(), ) this.orderedLoader.start() diff --git a/packages/db/src/query/live/ordered-source-loader.ts b/packages/db/src/query/live/ordered-source-loader.ts index 9883df13e7..cd8e26bfe5 100644 --- a/packages/db/src/query/live/ordered-source-loader.ts +++ b/packages/db/src/query/live/ordered-source-loader.ts @@ -35,7 +35,10 @@ export class OrderedSourceLoader { private fullSource: `none` | `held` | `complete` | `failed` = `none` // Keep callbacks, not copied requests or rows. Successful full-source work // subsumes these logical owners; unfinished transports remain observed. - private settledFiniteAcquisitions = new Set() + private settledFiniteAcquisitions = new Map< + ReleaseLoadSubset, + number | undefined + >() // The record's presence blocks automatic retry, including initial requests // that have no explicit window-operation generation. private failedRequest: @@ -47,6 +50,8 @@ export class OrderedSourceLoader { private lastPage: { count: number; boundary: unknown } | undefined private lastPrefixCount: number | undefined private lastBoundary: unknown + private repairRetries = 0 + private repairTimer: ReturnType | undefined constructor( private readonly info: OrderByOptimizationInfo, @@ -56,6 +61,7 @@ export class OrderedSourceLoader { result: LoadSubsetRequestResult, holdPublication: boolean, ) => void = () => {}, + private readonly canRetryRepair: () => boolean = () => false, ) { this.info.isRequesting = () => this.requesting } @@ -111,24 +117,17 @@ export class OrderedSourceLoader { (this.failedRequest || this.failedAcquisitions.size > 0) && windowOperationGeneration !== undefined ) { + this.cancelRepairRetry() + this.repairRetries = 0 // Move ownership to the explicit replacement before releasing the old // lease. Adapter cleanup may reenter the loader. if (this.failedRequest) { this.failedRequest.windowOperationGeneration = windowOperationGeneration } - const failedAcquisitions = this.failedAcquisitions - this.failedAcquisitions = new Map() - if (failedAcquisitions.size > 0) { - this.requesting = true - try { - runAllCallbacks(failedAcquisitions.keys()) - } finally { - this.requesting = false - } - // Adapter cleanup can synchronously tear down this loader. - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (!this.active) return - } + this.releaseFailedAcquisitions() + // Adapter cleanup can synchronously tear down this loader. + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (!this.active) return } if (this.fullSource === `failed`) this.fullSource = `none` else if (this.fullSource !== `none`) return this.pending @@ -137,10 +136,17 @@ export class OrderedSourceLoader { return this.pending } if (!this.info.index || this.info.orderBy.length !== 1) { - this.loadPrefix( - this.info.offset + this.info.limit, - windowOperationGeneration, - ) + if ( + windowOperationGeneration !== undefined || + (this.info.dataNeeded?.() ?? 0) > 0 || + (this.lastPrefixCount !== undefined && + this.lastPrefixCount < this.info.offset + this.info.limit) + ) { + this.loadPrefix( + this.info.offset + this.info.limit, + windowOperationGeneration, + ) + } return this.pending } if (!this.info.dataNeeded || this.pending) return this.pending @@ -202,6 +208,8 @@ export class OrderedSourceLoader { } resetCursor(): void { + this.cancelRepairRetry() + this.repairRetries = 0 this.generation++ if (this.fullSource === `complete`) this.fullSource = `held` this.pending = undefined @@ -226,16 +234,27 @@ export class OrderedSourceLoader { } } - private retireSettledFiniteAcquisitions(): void { + private retireSettledFiniteAcquisitions(prefix?: { + release: ReleaseLoadSubset + count: number + }): void { if ( - this.fullSource !== `complete` || + (this.fullSource !== `complete` && !prefix) || this.subscription.hasPendingTruncateReplacement ) return const generation = this.generation runAllCallbacks( - Array.from(this.settledFiniteAcquisitions, (release) => () => { + Array.from(this.settledFiniteAcquisitions, ([release, count]) => () => { if (!this.active || generation !== this.generation) return + if ( + this.fullSource !== `complete` && + (!prefix || + release === prefix.release || + count === undefined || + count > prefix.count) + ) + return this.settledFiniteAcquisitions.delete(release) release() }), @@ -324,6 +343,11 @@ export class OrderedSourceLoader { options?: LoadSubsetOptions, ): Promise { const isFullSource = kind === `full-source` + const retryRepair = + isFullSource && + this.hasSettledSourceRequest && + this.needsFullSourceRecovery && + windowOperationGeneration === undefined const generation = this.generation const complete = (): void => { if (this.pending === tracked) this.pending = undefined @@ -331,8 +355,16 @@ export class OrderedSourceLoader { if (!isFullSource) { // A replay can replace the physical lease while this older transport // finishes. Retire its logical owner only outside the replay barrier. - this.settledFiniteAcquisitions.add(releaseAcquisition) + const prefixCount = + options?.orderBy && !options.cursor ? options.limit : undefined + this.settledFiniteAcquisitions.set(releaseAcquisition, prefixCount) this.retireSettledFiniteAcquisitions() + if (generation === this.generation && prefixCount !== undefined) { + this.retireSettledFiniteAcquisitions({ + release: releaseAcquisition, + count: prefixCount, + }) + } } if (generation !== this.generation) return // A finite request may finish behind an authoritative repair. It cannot @@ -357,6 +389,8 @@ export class OrderedSourceLoader { } } if (isFullSource) { + this.cancelRepairRetry() + this.repairRetries = 0 this.needsFullSourceRecovery = false this.fullSource = `complete` this.retireSettledFiniteAcquisitions() @@ -385,6 +419,7 @@ export class OrderedSourceLoader { if (isFullSource) this.fullSource = `failed` this.recordRequestFailure(windowOperationGeneration) this.failedAcquisitions.set(releaseAcquisition, kind) + if (retryRepair) this.scheduleRepairRetry() throw error } const tracked = request.then(complete, fail) @@ -440,6 +475,64 @@ export class OrderedSourceLoader { this.needsFullSourceRecovery = true } + private cancelRepairRetry(): void { + clearTimeout(this.repairTimer) + this.repairTimer = undefined + } + + private releaseFailedAcquisitions(): void { + const failed = this.failedAcquisitions + this.failedAcquisitions = new Map() + this.requesting = true + try { + runAllCallbacks(failed.keys()) + } finally { + this.requesting = false + } + } + + private scheduleRepairRetry(): void { + if ( + !this.active || + !this.canRetryRepair() || + this.repairTimer !== undefined || + this.repairRetries >= 2 + ) + return + const generation = this.generation + const failedRequest = this.failedRequest + this.repairTimer = setTimeout( + () => { + this.repairTimer = undefined + const retry = Promise.resolve().then(() => { + if ( + !this.active || + !this.canRetryRepair() || + generation !== this.generation || + this.failedRequest !== failedRequest || + this.failedRequest?.windowOperationGeneration !== undefined + ) + return + this.releaseFailedAcquisitions() + // Release may dispose or start a new replay; neither belongs to this retry. + if ( + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- release callbacks can dispose the loader + !this.active || + generation !== this.generation || + this.subscription.hasPendingTruncateReplacement + ) + return + this.failedRequest = undefined + this.fullSource = `none` + this.loadFullSource() + }) + void retry.catch(() => {}) + this.onResult(retry, true) + }, + 250 * 2 ** this.repairRetries++, + ) + } + private failRequest( observed: | { @@ -465,6 +558,13 @@ export class OrderedSourceLoader { } catch { // Cleanup is attempted once and must not replace the request failure. } + if ( + isFullSource && + this.hasSettledSourceRequest && + windowOperationGeneration === undefined + ) { + this.scheduleRepairRetry() + } return error } @@ -519,13 +619,21 @@ export class OrderedSourceLoader { // Keep refinement blocked until failure and release finish unwinding. this.requesting = true try { - throw this.failRequest( + const failure = this.failRequest( observed, normalizeError(error), isFullSource, windowOperationGeneration, observing, ) + if (!observing && isFullSource && this.hasSettledSourceRequest) { + // A synchronous repair failure has no transport promise, but must + // still close publication before the triggering graph turn returns. + const rejected = Promise.reject(failure) + void rejected.catch(() => {}) + this.onResult(rejected, true) + } + throw failure } finally { this.requesting = false } diff --git a/packages/db/src/transactions.ts b/packages/db/src/transactions.ts index 34461569a0..738ec3cedb 100644 --- a/packages/db/src/transactions.ts +++ b/packages/db/src/transactions.ts @@ -637,13 +637,6 @@ class Transaction> { await this.mutationFn({ transaction: this as unknown as TransactionWithMutations, }) - - if ((this.state as TransactionState) !== `persisting`) return this - - this.setState(`completed`) - this.touchCollection() - - this.isPersisted.resolve(this) } catch (error) { if ((this.state as TransactionState) !== `persisting`) return this @@ -663,6 +656,17 @@ class Transaction> { throw originalError } + if ((this.state as TransactionState) !== `persisting`) return this + + this.setState(`completed`) + // Publication errors cannot undo persistence or leave its receipt pending. + // Keep normal publication queued before callers resume from the receipt. + try { + this.touchCollection() + } finally { + this.isPersisted.resolve(this) + } + return this } diff --git a/packages/db/src/utils.ts b/packages/db/src/utils.ts index c5eb9aef3e..d54c6e3e5c 100644 --- a/packages/db/src/utils.ts +++ b/packages/db/src/utils.ts @@ -39,9 +39,10 @@ function enumerableOwnKeys(value: object): Array { } /** - * Internal implementation with cycle detection to prevent infinite recursion + * Internal implementation with cycle detection to prevent infinite recursion. + * Internal callers can seed already-paired roots when comparing their children. */ -function deepEqualsInternal( +export function deepEqualsInternal( a: any, b: any, visited: Map, diff --git a/packages/db/tests/collection-subscribe-changes.test.ts b/packages/db/tests/collection-subscribe-changes.test.ts index 000e025555..7297da8e59 100644 --- a/packages/db/tests/collection-subscribe-changes.test.ts +++ b/packages/db/tests/collection-subscribe-changes.test.ts @@ -2671,57 +2671,79 @@ describe(`Virtual properties`, () => { expect(collection.state.get(`row-1`)?.$origin).toBe(`remote`) }) - it(`replaces a completed direct mutation with an authoritative truncate row`, async () => { - let syncFns: - | { - begin: () => void - write: (change: { - type: `insert` - value: { id: string; value: string } - }) => void - commit: () => true | Promise - truncate: () => void - } - | undefined + it.each([`before`, `after`] as const)( + `replaces a direct mutation settling %s truncate with its authoritative row`, + async (settlement) => { + let finishMutation!: () => void + const mutation = new Promise((resolve) => { + finishMutation = resolve + }) + let syncFns: + | { + begin: () => void + write: (change: { + type: `insert` + value: { id: string; value: string } + }) => void + commit: () => true | Promise + truncate: () => void + } + | undefined - const collection = createCollection<{ id: string; value: string }, string>({ - id: `truncate-replaces-completed-direct-mutation`, - getKey: (item) => item.id, - startSync: true, - sync: { - sync: ({ begin, write, commit, truncate, markReady }) => { - syncFns = { begin, write, commit, truncate } - markReady() + const collection = createCollection< + { id: string; value: string }, + string + >({ + id: `truncate-replaces-completed-direct-mutation`, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ begin, write, commit, truncate, markReady }) => { + syncFns = { begin, write, commit, truncate } + markReady() + }, }, - }, - onInsert: () => Promise.resolve(), - }) - - await collection.stateWhenReady() - const transaction = collection.insert({ id: `row-1`, value: `client` }) - await transaction.isPersisted.promise - expect(collection.get(`row-1`)).toMatchObject({ - id: `row-1`, - value: `client`, - }) + onInsert: () => mutation, + }) - if (!syncFns) throw new Error(`Sync not ready`) - syncFns.begin() - syncFns.truncate() - syncFns.write({ - type: `insert`, - value: { id: `row-1`, value: `server` }, - }) - const applied = syncFns.commit() - if (applied !== true) await applied - await waitForChanges() + await collection.stateWhenReady() + const transaction = collection.insert({ id: `row-1`, value: `client` }) + if (settlement === `before`) { + finishMutation() + await transaction.isPersisted.promise + } + expect(collection.get(`row-1`)).toMatchObject({ + id: `row-1`, + value: `client`, + }) - expect(collection.get(`row-1`)).toMatchObject({ - id: `row-1`, - value: `server`, - }) - expect(collection.state.get(`row-1`)?.$origin).toBe(`remote`) - }) + if (!syncFns) throw new Error(`Sync not ready`) + syncFns.begin() + syncFns.truncate() + syncFns.write({ + type: `insert`, + value: { id: `row-1`, value: `server` }, + }) + const applied = syncFns.commit() + if (settlement === `after`) { + finishMutation() + await transaction.isPersisted.promise + } + if (applied !== true) await applied + await waitForChanges() + + expect(collection.get(`row-1`)).toMatchObject({ + id: `row-1`, + value: `server`, + }) + expect(collection.state.get(`row-1`)?.$synced).toBe(true) + // A same-key sync during the active mutation is a local acknowledgement; + // after completion, truncate is an independent remote replacement. + expect(collection.state.get(`row-1`)?.$origin).toBe( + settlement === `after` ? `local` : `remote`, + ) + }, + ) it(`should preserve local origin for rows confirmed in the same truncate batch`, async () => { let syncFns: diff --git a/packages/db/tests/proxy-detachment-contract.test.ts b/packages/db/tests/proxy-detachment-contract.test.ts new file mode 100644 index 0000000000..80b1d80577 --- /dev/null +++ b/packages/db/tests/proxy-detachment-contract.test.ts @@ -0,0 +1,231 @@ +import { describe, expect, it } from 'vitest' +import { createCollection } from '../src/collection/index.js' +import { withChangeTracking } from '../src/proxy.js' + +class Label { + #text: string + constructor(text: string) { + this.#text = text + } + read() { + return this.#text + } + rename(text: string) { + this.#text = text + } +} + +function storedRow(row: T & { id: number }) { + return createCollection({ + getKey: (value) => value.id, + startSync: true, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ type: `insert`, value: row }) + commit() + markReady() + }, + }, + onUpdate: () => Promise.resolve(), + }) +} + +describe(`Mutation result detachment`, () => { + it(`keeps arbitrary class instances by reference as an explicit isolation exception`, async () => { + const label = new Label(`before`) + const collection = storedRow({ id: 1, value: undefined as unknown }) + try { + const tx = collection.update(1, (draft) => { + draft.value = label + }) + expect(collection.get(1)!.value).toBe(label) + label.rename(`after`) + expect((collection.get(1)!.value as Label).read()).toBe(`after`) + await tx.isPersisted.promise + } finally { + await collection.cleanup() + } + }) + + it(`preserves and detaches a regular expression's matching position`, () => { + const expression = /x/g + expression.lastIndex = 2 + const changes = withChangeTracking( + { value: undefined as unknown }, + (draft) => { + draft.value = expression + }, + ) + expect((changes.value as RegExp).lastIndex).toBe(2) + expression.lastIndex = 0 + expect((changes.value as RegExp).lastIndex).toBe(2) + }) + + it.each([`URL`, `class`, `Date`, `RegExp`, `typed-array`] as const)( + `preserves a newly assigned %s in the stored row`, + async (kind) => { + const value = + kind === `URL` + ? new URL(`https://example.com/path`) + : kind === `class` + ? new Label(`saved`) + : kind === `Date` + ? new Date(`2026-01-01T00:00:00Z`) + : kind === `RegExp` + ? /saved/gi + : new Uint8Array([1, 2]) + const collection = storedRow({ id: 1, value: undefined as unknown }) + try { + const tx = collection.update(1, (draft) => { + draft.value = value + }) + const saved = collection.get(1)!.value + expect(Object.getPrototypeOf(saved)).toBe(Object.getPrototypeOf(value)) + if (value instanceof URL) expect((saved as URL).href).toBe(value.href) + else if (value instanceof Label) + expect((saved as Label).read()).toBe(`saved`) + else expect(saved).toEqual(value) + await tx.isPersisted.promise + } finally { + await collection.cleanup() + } + }, + ) + + it.each([`Date`, `typed-array`] as const)( + `detaches a known mutable %s after callback return`, + (kind) => { + const value = kind === `Date` ? new Date(0) : new Uint8Array([1]) + const changes = withChangeTracking( + { value: undefined as unknown }, + (draft) => { + draft.value = value + }, + ) + if (value instanceof Date) { + value.setTime(1000) + expect((changes.value as Date).getTime()).toBe(0) + } else { + value[0] = 2 + expect((changes.value as Uint8Array)[0]).toBe(1) + } + }, + ) + + it(`keeps a stored URL unchanged when the caller later changes its URL`, async () => { + const value = new URL(`https://example.com/before`) + const collection = storedRow({ id: 1, value: undefined as unknown }) + try { + const tx = collection.update(1, (draft) => { + draft.value = value + }) + value.pathname = `/after` + expect((collection.get(1)!.value as URL).pathname).toBe(`/before`) + await tx.isPersisted.promise + } finally { + await collection.cleanup() + } + }) + + it.each([`Set`, `array`] as const)( + `can commit a new %s member holding a draft handle`, + async (kind) => { + type Row = { + id: number + count: number + s: Set<{ back: Row }> + arr: Array<{ owner: Row }> + } + const collection = storedRow({ + id: 1, + count: 0, + s: new Set(), + arr: [], + }) + try { + const tx = collection.update(1, (draft) => { + draft.count = 1 + if (kind === `Set`) draft.s.add({ back: draft }) + else draft.arr.push({ owner: draft }) + }) + const saved = collection.get(1)! + const back = + kind === `Set` + ? saved.s.values().next().value!.back + : saved.arr[0]!.owner + expect(back.count).toBe(1) + await tx.isPersisted.promise + } finally { + await collection.cleanup() + } + }, + ) + + it.each( + ([`scalar`, `object`] as const).flatMap((kind) => + ([`object`, `array`, `Map`, `Set`] as const).map((path) => ({ + kind, + path, + })), + ), + )( + `omits an untouched $path back-reference on a $kind-only edit`, + ({ kind, path }) => { + type Row = { + count: number + value: { x: number } + child?: { + name: string + back: Row | Array | Map | Set + } + } + const row: Row = { count: 0, value: { x: 0 } } + row.child = { + name: `before`, + back: + path === `array` + ? [row] + : path === `Map` + ? new Map([[`row`, row]]) + : path === `Set` + ? new Set([row]) + : row, + } + const changes = withChangeTracking(row, (draft) => { + if (kind === `scalar`) draft.count = 1 + else draft.value = { x: 1 } + }) + expect(Object.keys(changes)).toEqual([ + kind === `scalar` ? `count` : `value`, + ]) + }, + ) + + it(`keeps a real nested edit even when that child also reaches the row`, () => { + type Row = { count: number; child?: { name: string; back: Row } } + const row: Row = { count: 0 } + row.child = { name: `before`, back: row } + const changes = withChangeTracking(row, (draft) => { + draft.count = 1 + draft.child!.name = `after` + }) + expect((changes.child as NonNullable).name).toBe(`after`) + expect(row.child.name).toBe(`before`) + }) + + it(`publishes a changed sibling alias even when it has a nested row back-reference`, () => { + type Child = { name: string; back?: Row } + type Row = { child: Child; alias: Child } + const child: Child = { name: `before` } + const row: Row = { child, alias: child } + child.back = row + const changes = withChangeTracking(row, (draft) => { + draft.alias.name = `after` + }) + const saved = { ...row, ...changes } + expect(saved.child.name).toBe(`after`) + expect(saved.child).toBe(saved.alias) + expect(child.name).toBe(`before`) + }) +}) diff --git a/packages/db/tests/query/ordered-default-work.test.ts b/packages/db/tests/query/ordered-default-work.test.ts new file mode 100644 index 0000000000..d346e21197 --- /dev/null +++ b/packages/db/tests/query/ordered-default-work.test.ts @@ -0,0 +1,231 @@ +import { describe, expect, it, vi } from 'vitest' +import { createCollection } from '../../src/collection/index.js' +import { BTreeIndex } from '../../src/indexes/btree-index.js' +import { createLiveQueryCollection } from '../../src/query/index.js' +import { evaluateReferenceExpression } from '../reference-expression.js' +import { flushPromises } from '../utils.js' +import type { LoadSubsetOptions, SyncConfig } from '../../src/types.js' + +type Row = { id: number; rank: number } + +async function setup( + indexed: boolean, + multi: boolean, + evict = false, + syncFailure = false, +) { + const truth = Array.from({ length: 20 }, (_, id) => ({ + id: id + 1, + rank: id + 1, + })) + const installed = new Map() + const calls: Array = [] + const active = new Set() + const owned = new Map>() + let sync!: Parameters[`sync`]>[0] + let failFull = 0 + const failure = new Error(`transient full-source failure`) + const source = createCollection({ + getKey: (row) => row.id, + syncMode: `on-demand`, + ...(indexed + ? { autoIndex: `eager` as const, defaultIndexType: BTreeIndex } + : {}), + sync: { + sync: (operations) => { + sync = operations + sync.markReady() + return { + loadSubset: (options) => { + calls.push(options) + if (failFull && !options.orderBy && !options.where) { + failFull-- + if (syncFailure) throw failure + active.add(options) + return Promise.reject(failure) + } + active.add(options) + const rows = truth + .filter( + (row) => + (!options.where || + evaluateReferenceExpression(options.where, row) === true) && + (!options.cursor || + evaluateReferenceExpression( + options.cursor.whereFrom, + row, + ) === true), + ) + .sort((a, b) => a.rank - b.rank || a.id - b.id) + const offset = options.cursor ? 0 : (options.offset ?? 0) + const selected = rows.slice( + offset, + options.limit === undefined ? undefined : offset + options.limit, + ) + owned.set(options, new Set(selected.map((row) => row.id))) + sync.begin() + for (const row of selected) { + if (installed.get(row.id) === row) continue + sync.write({ + type: installed.has(row.id) ? `update` : `insert`, + value: row, + }) + installed.set(row.id, row) + } + return Promise.resolve(sync.commit()).then(() => {}) + }, + unloadSubset: (options) => { + expect(active.delete(options)).toBe(true) + const released = owned.get(options) + owned.delete(options) + if (!evict || !released) return + sync.begin() + for (const id of released) { + const row = installed.get(id) + if (row && ![...owned.values()].some((keys) => keys.has(id))) { + installed.delete(id) + sync.write({ type: `delete`, value: row }) + } + } + void sync.commit() + }, + } + }, + }, + }) + const createQuery = () => + createLiveQueryCollection((q) => { + const sorted = q.from({ row: source }).orderBy(({ row }) => row.rank) + return (multi ? sorted.orderBy(({ row }) => row.id) : sorted) + .limit(2) + .select(({ row }) => ({ id: row.id, rank: row.rank })) + }) + const live = createQuery() + await live.preload() + return { + live, + createQuery, + source, + calls, + active, + failure, + failNextFull: (count = 1) => { + failFull = count + }, + insert: (row: Row) => { + truth.push(row) + installed.set(row.id, row) + sync.begin() + sync.write({ type: `insert`, value: row }) + return sync.commit() + }, + remove: (id: number) => { + const index = truth.findIndex((row) => row.id === id) + const [row] = truth.splice(index, 1) + installed.delete(id) + sync.begin() + sync.write({ type: `delete`, value: row! }) + return sync.commit() + }, + cleanup: async () => { + await live.cleanup() + await source.cleanup() + }, + } +} + +describe(`Ordered source work across default and indexed plans`, () => { + it.each( + [false, true].flatMap((indexed) => + [false, true].map((multi) => ({ indexed, multi })), + ), + )( + `does not reacquire a full window for out-of-window inserts: %j`, + async ({ indexed, multi }) => { + const h = await setup(indexed, multi) + try { + const calls = h.calls.length + const active = h.active.size + for (let id = 100; id < 103; id++) { + await h.insert({ id, rank: id }) + await flushPromises() + } + expect(h.live.toArray.map((row) => row.id)).toEqual([1, 2]) + expect.soft(h.calls.length - calls).toBe(0) + expect(h.active.size).toBe(active) + } finally { + await h.cleanup() + } + }, + ) + + it.each( + [false, true].flatMap((multi) => + [false, true].map((peer) => ({ multi, peer })), + ), + )( + `retires a replaced prefix without evicting current or peer rows: %j`, + async ({ multi, peer }) => { + const h = await setup(false, multi, true) + const other = peer ? h.createQuery() : undefined + try { + await other?.preload() + for (const limit of [3, 4, 5]) await h.live.utils.setWindow({ limit }) + expect(h.live.toArray.map((row) => row.id)).toEqual([1, 2, 3, 4, 5]) + expect([...h.active].filter((options) => options.orderBy)).toHaveLength( + peer ? 2 : 1, + ) + if (other) expect(other.toArray.map((row) => row.id)).toEqual([1, 2]) + } finally { + await other?.cleanup() + await h.cleanup() + } + }, + ) + + it.each( + [`success`, `exhausted`, `cleanup`].flatMap((outcome) => + [false, true].map((syncFailure) => ({ outcome, syncFailure })), + ), + )( + `bounds automatic repair with stale rows retained: %j`, + async ({ outcome, syncFailure }) => { + const h = await setup(true, false, false, syncFailure) + vi.useFakeTimers({ toFake: [`setTimeout`, `clearTimeout`] }) + try { + h.failNextFull(outcome === `success` ? 1 : 3) + await h.remove(1) + await vi.advanceTimersByTimeAsync(0) + const afterFailure = h.calls.length + expect(h.live.utils.lastSubsetError).toBe(h.failure) + expect(h.live.status).toBe(`ready`) + expect(h.live.toArray.map((row) => row.id)).toEqual([1, 2]) + if (outcome === `cleanup`) await h.live.cleanup() + await vi.advanceTimersByTimeAsync(249) + expect(h.calls).toHaveLength(afterFailure) + await vi.advanceTimersByTimeAsync(751) + expect(h.calls.length - afterFailure).toBe( + outcome === `cleanup` ? 0 : outcome === `success` ? 1 : 2, + ) + if (outcome === `cleanup`) return + expect(h.live.status).toBe(`ready`) + expect(h.live.toArray.map((row) => row.id)).toEqual( + outcome === `success` ? [2, 3] : [1, 2], + ) + const settledCalls = h.calls.length + await h.insert({ id: 50, rank: 0 }) + await vi.advanceTimersByTimeAsync(0) + await vi.advanceTimersByTimeAsync(10000) + expect(h.calls).toHaveLength(settledCalls) + if (outcome === `exhausted`) { + expect(h.live.toArray.map((row) => row.id)).toEqual([1, 2]) + await h.live.utils.setWindow({ limit: 2 }) + } + expect(h.live.toArray.map((row) => row.id)).toEqual([50, 2]) + } finally { + await h.cleanup() + vi.useRealTimers() + } + }, + ) +}) diff --git a/packages/db/tests/transactions.test.ts b/packages/db/tests/transactions.test.ts index d0f70a890a..33373d4ade 100644 --- a/packages/db/tests/transactions.test.ts +++ b/packages/db/tests/transactions.test.ts @@ -2,14 +2,69 @@ import { describe, expect, it } from 'vitest' import { DbClient, collectionOptions } from '../src/client.js' import { createTransaction } from '../src/transactions' import { createCollection } from '../src/collection/index.js' +import { createDeferred } from '../src/deferred.js' import { MissingMutationFunctionError, TransactionAlreadyCompletedRollbackError, TransactionNotPendingCommitError, TransactionNotPendingMutateError, } from '../src/errors' +import { flushPromises } from './utils.js' +import type { SyncConfig } from '../src/types.js' describe(`Transactions`, () => { + it(`settles persistence and reports a listener error while draining its parked echo`, async () => { + type Row = { id: number; value: string } + let sync!: Parameters[`sync`]>[0] + const gate = createDeferred() + const collection = createCollection({ + getKey: (row) => row.id, + startSync: true, + sync: { + sync: (operations) => { + sync = operations + sync.markReady() + }, + }, + }) + const tx = createTransaction({ + autoCommit: false, + mutationFn: () => gate.promise, + }) + const failure = new Error(`echo listener failed`) + const subscription = collection.subscribeChanges((batch) => { + if (batch.some((change) => change.value.value === `server`)) throw failure + }) + let persisted = false + const receipt = tx.isPersisted.promise.then( + () => { + persisted = true + }, + () => {}, + ) + try { + tx.mutate(() => collection.insert({ id: 1, value: `client` })) + const outcome = tx.commit().then( + () => ({ ok: true as const }), + (error: unknown) => ({ ok: false as const, error }), + ) + sync.begin() + sync.write({ type: `insert`, value: { id: 1, value: `server` } }) + const echo = sync.commit() + if (echo !== true) void echo.catch(() => {}) + gate.resolve() + const result = await outcome + await flushPromises() + expect.soft(result).toEqual({ ok: false, error: failure }) + expect.soft(tx.state).toBe(`completed`) + expect(persisted).toBe(true) + await receipt + } finally { + subscription.unsubscribe() + gate.resolve() + await collection.cleanup() + } + }) it.each([ { name: `Error`, diff --git a/packages/powersync-db-collection/src/powersync.ts b/packages/powersync-db-collection/src/powersync.ts index 5d41661e91..6efe6b7f63 100644 --- a/packages/powersync-db-collection/src/powersync.ts +++ b/packages/powersync-db-collection/src/powersync.ts @@ -698,6 +698,9 @@ function createPowerSyncCollectionConfig< await Promise.all(appliedReceipts) if (isCurrent()) { reconciledTrackingRevision = revision + // Replacing the trigger alone is not recovery: its baseline + // writes must also be applied before the source is ready again. + if (collection.status === `error`) markReady() } } } diff --git a/packages/powersync-db-collection/tests/on-demand-sync.test.ts b/packages/powersync-db-collection/tests/on-demand-sync.test.ts index 44dcf3d171..0d093fa928 100644 --- a/packages/powersync-db-collection/tests/on-demand-sync.test.ts +++ b/packages/powersync-db-collection/tests/on-demand-sync.test.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto' import { tmpdir } from 'node:os' import { PowerSyncDatabase, Schema, Table, column } from '@powersync/node' import { + IR, and, createCollection, createLiveQueryCollection, @@ -2212,6 +2213,12 @@ describe(`On-Demand Sync Mode`, () => { }) describe(`Tracking lifecycle`, () => { + const categoryEquals = (category: string) => + new IR.Func(`eq`, [ + new IR.PropRef([`category`]), + new IR.Value(category), + ]) + // The sync handler catches its own errors and surfaces them only through the // logger, so captured errors are how these tests assert it stayed healthy. function captureSyncErrors(db: PowerSyncDatabase) { @@ -2309,13 +2316,13 @@ describe(`On-Demand Sync Mode`, () => { try { const provisional = loadSubset({ - where: eq(`category`, `electronics`), + where: categoryEquals(`electronics`), }) await vi.waitFor(() => expect(onLoadSubset).toHaveBeenCalledOnce()) await expect( - loadSubset({ where: eq(`category`, `outdoors`) }), + loadSubset({ where: categoryEquals(`outdoors`) }), ).rejects.toBe(hookFailure) - await loadSubset({ where: eq(`category`, `clothing`) }) + await loadSubset({ where: categoryEquals(`clothing`) }) const when = createDiffTrigger.mock.calls.at(-1)?.[0].when expect(when?.INSERT).toContain(`clothing`) @@ -2339,7 +2346,7 @@ describe(`On-Demand Sync Mode`, () => { }) const controller = new AbortController() const request = { - where: eq(`category`, `electronics`), + where: categoryEquals(`electronics`), signal: controller.signal, } @@ -2381,13 +2388,13 @@ describe(`On-Demand Sync Mode`, () => { let secondSettled = false const first = Promise.resolve( - loadSubset({ where: eq(`category`, `electronics`) }), + loadSubset({ where: categoryEquals(`electronics`) }), ).then(() => { firstSettled = true }) await vi.waitFor(() => expect(locks).toHaveLength(1)) const second = Promise.resolve( - loadSubset({ where: eq(`category`, `clothing`) }), + loadSubset({ where: categoryEquals(`clothing`) }), ).then(() => { secondSettled = true }) @@ -2439,7 +2446,7 @@ describe(`On-Demand Sync Mode`, () => { }) const { sync, loadSubset } = startOnDemandSync(db, {}, { commit }) const first = Promise.resolve( - loadSubset({ where: eq(`category`, `electronics`) }), + loadSubset({ where: categoryEquals(`electronics`) }), ) let second: Promise | undefined try { @@ -2450,12 +2457,21 @@ describe(`On-Demand Sync Mode`, () => { applied.resolve() for (let i = 0; i < turn; i++) await Promise.resolve() second = Promise.resolve( - loadSubset({ where: eq(`category`, `clothing`) }), + loadSubset({ where: categoryEquals(`clothing`) }), ) await second const when = createDiffTrigger.mock.calls.at(-1)?.[0].when expect(when?.INSERT).toContain(`electronics`) expect(when?.INSERT).toContain(`clothing`) + // Literal names in SQL are not proof of a working column filter. + // Execute the exact trigger clause against matching and excluded rows. + for (const category of [`electronics`, `clothing`, `outdoors`]) { + const row = await db.get<{ matches: number }>( + `SELECT CASE WHEN (${when!.INSERT}) THEN 1 ELSE 0 END AS matches FROM (SELECT ? AS data) AS NEW`, + [JSON.stringify({ category })], + ) + expect(row.matches).toBe(category === `outdoors` ? 0 : 1) + } await first } finally { applied.resolve() @@ -2481,13 +2497,13 @@ describe(`On-Demand Sync Mode`, () => { .mockResolvedValueOnce(currentDispose) const { sync, loadSubset } = startOnDemandSync(db) const first = Promise.resolve( - loadSubset({ where: eq(`category`, `electronics`) }), + loadSubset({ where: categoryEquals(`electronics`) }), ) try { await triggerStarted.promise const second = Promise.resolve( - loadSubset({ where: eq(`category`, `clothing`) }), + loadSubset({ where: categoryEquals(`clothing`) }), ) finishTrigger.resolve() await Promise.all([first, second]) @@ -2530,7 +2546,7 @@ describe(`On-Demand Sync Mode`, () => { ) let settled = false const load = Promise.resolve( - loadSubset({ where: eq(`category`, `electronics`) }), + loadSubset({ where: categoryEquals(`electronics`) }), ).then(() => { settled = true }) @@ -2575,7 +2591,7 @@ describe(`On-Demand Sync Mode`, () => { .mockResolvedValue(vi.fn()) const { sync, loadSubset } = startOnDemandSync(db) - const load = loadSubset({ where: eq(`category`, `electronics`) }) + const load = loadSubset({ where: categoryEquals(`electronics`) }) await lockQueued.promise sync.cleanup?.() await runLock() @@ -2587,8 +2603,8 @@ describe(`On-Demand Sync Mode`, () => { it(`cleans each acquired subset at most once during reentrant cleanup`, async () => { const db = await createDatabase() vi.spyOn(db.triggers, `createDiffTrigger`).mockResolvedValue(vi.fn()) - const first = { where: eq(`category`, `electronics`) } - const second = { where: eq(`category`, `clothing`) } + const first = { where: categoryEquals(`electronics`) } + const second = { where: categoryEquals(`clothing`) } const firstCleanup = vi.fn() const secondCleanup = vi.fn(() => started.unloadSubset(first)) const onLoadSubset = vi.fn((options: LoadSubsetOptions) => @@ -2607,8 +2623,8 @@ describe(`On-Demand Sync Mode`, () => { const db = await createDatabase() vi.spyOn(db.triggers, `createDiffTrigger`).mockResolvedValue(vi.fn()) const getAll = vi.spyOn(db, `getAll`).mockResolvedValue([]) - const first = { where: eq(`category`, `electronics`) } - const second = { where: eq(`category`, `clothing`) } + const first = { where: categoryEquals(`electronics`) } + const second = { where: categoryEquals(`clothing`) } const onLoadSubset = vi.fn((options: LoadSubsetOptions) => options === first ? () => started.unloadSubset(second) : undefined, ) @@ -2738,7 +2754,7 @@ describe(`On-Demand Sync Mode`, () => { const failure = new Error(`trigger installation failed`) try { await collection._sync.loadSubset({ - where: eq(`category`, `electronics`), + where: categoryEquals(`electronics`), }) expect(collection.status).toBe(`ready`) vi.spyOn(db.triggers, `createDiffTrigger`).mockRejectedValueOnce( @@ -2746,7 +2762,7 @@ describe(`On-Demand Sync Mode`, () => { ) await expect( Promise.resolve( - collection._sync.loadSubset({ where: eq(`category`, `clothing`) }), + collection._sync.loadSubset({ where: categoryEquals(`clothing`) }), ), ).rejects.toBe(failure) expect(collection.status).toBe(`error`) @@ -2755,6 +2771,49 @@ describe(`On-Demand Sync Mode`, () => { } }) + it(`does not silently stay errored after a release rebuild retries successfully`, async () => { + vi.useFakeTimers() + const db = await createDatabase() + await db.execute( + `INSERT INTO products (id, name, price, category) VALUES ('retained', 'Before', 10, 'clothing')`, + ) + vi.spyOn(db.logger, `error`).mockImplementation(() => {}) + const collection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.products, + syncMode: `on-demand`, + }), + ) + const first = { where: categoryEquals(`electronics`) } + const second = { where: categoryEquals(`clothing`) } + try { + await collection._sync.loadSubset(first) + await collection._sync.loadSubset(second) + const trigger = vi + .spyOn(db.triggers, `createDiffTrigger`) + .mockRejectedValueOnce(new Error(`release rebuild failed`)) + collection._sync.unloadSubset(first) + await vi.waitFor(() => expect(collection.status).toBe(`error`)) + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => + expect(trigger.mock.calls.length).toBeGreaterThan(1), + ) + await vi.waitFor(() => expect(collection.status).toBe(`ready`)) + expect(collection.get(`retained`)?.name).toBe(`Before`) + await db.execute( + `UPDATE products SET name = 'After' WHERE id = 'retained'`, + ) + await vi.waitFor(() => + expect(collection.get(`retained`)?.name).toBe(`After`), + ) + } finally { + await collection.cleanup() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) + it(`retries a failed physical release`, async () => { vi.useFakeTimers() const db = await createDatabase() @@ -2765,7 +2824,7 @@ describe(`On-Demand Sync Mode`, () => { .mockRejectedValueOnce(new Error(`transient eviction failure`)) .mockResolvedValueOnce([]) const { sync, loadSubset, unloadSubset } = startOnDemandSync(db) - const request = { where: eq(`category`, `electronics`) } + const request = { where: categoryEquals(`electronics`) } try { await loadSubset(request) @@ -2793,8 +2852,8 @@ describe(`On-Demand Sync Mode`, () => { : Promise.resolve([]), ) const { sync, loadSubset, unloadSubset } = startOnDemandSync(db) - const failing = { where: eq(`category`, `electronics`) } - const succeeding = { where: eq(`category`, `clothing`) } + const failing = { where: categoryEquals(`electronics`) } + const succeeding = { where: categoryEquals(`clothing`) } try { await Promise.all([loadSubset(failing), loadSubset(succeeding)]) @@ -2829,8 +2888,8 @@ describe(`On-Demand Sync Mode`, () => { : Promise.resolve([]), ) const { sync, loadSubset, unloadSubset } = startOnDemandSync(db) - const first = { where: eq(`category`, `electronics`) } - const second = { where: eq(`category`, `clothing`) } + const first = { where: categoryEquals(`electronics`) } + const second = { where: categoryEquals(`clothing`) } try { await Promise.all([loadSubset(first), loadSubset(second)]) unloadSubset(first) @@ -2865,14 +2924,14 @@ describe(`On-Demand Sync Mode`, () => { {}, { write }, ) - const departing = { where: eq(`category`, `electronics`) } + const departing = { where: categoryEquals(`electronics`) } try { await loadSubset(departing) unloadSubset(departing) await vi.waitFor(() => expect(getAll).toHaveBeenCalledOnce()) - await loadSubset({ where: eq(`category`, `clothing`) }) + await loadSubset({ where: categoryEquals(`clothing`) }) firstEviction.resolve([{ id: `now-owned` }]) await vi.waitFor(() => expect(getAll).toHaveBeenCalledTimes(2)) From bb6e55824782ef814b50f35fc14be1e65271d3fb Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 14:47:14 -0600 Subject: [PATCH 2/2] fix: preserve acknowledgment and recovery invariants Distinguish acknowledged inserts from stale synced keys, reconcile negative rows after PowerSync tracking outages, and keep one-shot persistence refreshes out of permanent demand ownership. Finish successful ordered-prefix bookkeeping even when retiring older leases throws. Add red/green regressions for each boundary and strengthen peer-delivery and detached-cycle laws. --- .changeset/finish-review-recovery.md | 6 +- .../src/persisted.ts | 5 +- .../tests/persisted.test.ts | 100 ++++++++++++++++ packages/db/src/collection/state.ts | 27 ++--- packages/db/src/query/live/ARCHITECTURE.md | 3 + .../src/query/live/ordered-source-loader.ts | 109 ++++++++++-------- .../collection-subscribe-changes.test.ts | 43 +++++++ .../tests/proxy-detachment-contract.test.ts | 10 ++ .../tests/query/ordered-source-loader.test.ts | 69 ++++++++++- .../powersync-db-collection/src/powersync.ts | 94 ++++++++------- .../tests/on-demand-sync.test.ts | 102 +++++++++------- 11 files changed, 423 insertions(+), 145 deletions(-) diff --git a/.changeset/finish-review-recovery.md b/.changeset/finish-review-recovery.md index 692dee167d..16d050dae5 100644 --- a/.changeset/finish-review-recovery.md +++ b/.changeset/finish-review-recovery.md @@ -4,8 +4,8 @@ '@tanstack/powersync-db-collection': patch --- -Preserve native values, arbitrary class references, and draft cycles during mutation detachment; keep transaction persistence receipts settled after publication errors and avoid restoring an acknowledged direct insert over its server row. +Preserve native values, arbitrary class references, and draft cycles during mutation detachment; keep transaction persistence receipts settled after publication errors and avoid restoring an acknowledged direct insert over its server row. Keep a delete/reinsert visible when the old synced row has not yet been replaced. -Retire replaced ordered prefixes and retry automatic ordered repair at most twice while retaining stale results and exposing the error; cleanup cancels retries and explicit window retry remains available. +Retire replaced ordered prefixes without interrupting successful-load bookkeeping if release throws. Retry automatic ordered repair at most twice while retaining stale results and exposing the error; cleanup cancels retries and explicit window retry remains available. -Keep persisted acquisitions independent, reject upstream load failures without discarding cached rows, and restore PowerSync readiness after a successful tracking rebuild and applied baseline. +Keep persisted acquisitions independent, avoid retaining one-shot refreshes as permanent demand, and reject upstream load failures without discarding cached rows. Restore PowerSync readiness only after the recovered baseline also removes rows deleted or moved outside active filters during the tracking outage. diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 7db554ec82..e2592a5baf 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -352,6 +352,7 @@ export interface PersistedCollectionUtils extends UtilsRecord { mutations: Array>> }) => Promise | void getLeadershipState?: () => PersistedCollectionLeadershipState + /** Hydrate once without acquiring a new ongoing subset lease. */ forceReloadSubset?: (options: LoadSubsetOptions) => Promise | void } @@ -1049,7 +1050,7 @@ class PersistedCollectionRuntime< } async forceReloadSubset(options: LoadSubsetOptions): Promise { - this.activeSubsets.set(this.getSubsetKey(options), options) + // A one-shot refresh does not acquire an enduring subscription lease. await this.applyMutex.run(() => this.hydrateSubsetUnsafe(options, { requestRemoteEnsure: false }), ) @@ -2590,6 +2591,8 @@ function createWrappedSyncConfig< if (!resolvedSourceResult.loadSubset) return true acquisition.forwarded = true try { + // Returning a promise transfers its lease even if it rejects. + // Only a synchronous throw leaves no upstream lease to release. return resolvedSourceResult.loadSubset(loadOptions) } catch (error) { acquisition.forwarded = false diff --git a/packages/db-sqlite-persistence-core/tests/persisted.test.ts b/packages/db-sqlite-persistence-core/tests/persisted.test.ts index d90dadccb6..bc3ad4a9be 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted.test.ts @@ -1801,6 +1801,39 @@ describe(`persistedCollectionOptions`, () => { }, ) + it(`does not retain refresh history as permanent subset demand`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `Before` }]) + const coordinator = createCoordinatorHarness() + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (row) => row.id, + syncMode: `on-demand`, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + try { + await collection._sync.loadSubset({ limit: 1 }) + for (let i = 0; i < 20; i++) + await collection.utils.forceReloadSubset!({ limit: 1 }) + const before = adapter.loadSubsetCalls.length + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `refresh-invalidation`, + latestRowVersion: 1, + requiresFullReload: true, + }) + await flushAsyncWork() + await flushAsyncWork() + expect(adapter.loadSubsetCalls.length - before).toBe(1) + } finally { + await collection.cleanup() + } + }) + it(`reports a non-abort upstream failure rather than treating hydration as remote success`, async () => { const failure = new Error(`remote acquisition failed`) const warn = vi.spyOn(console, `warn`).mockImplementation(() => {}) @@ -1839,6 +1872,73 @@ describe(`persistedCollectionOptions`, () => { } }) + it.each([`throw`, `reject`] as const)( + `releases only transferred upstream ownership after a load %s`, + async (mode) => { + const failure = new Error(`failed upstream load`) + const warn = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const peer: LoadSubsetOptions = { limit: 1 } + const failed: LoadSubsetOptions = { limit: 1 } + const leases = new Set() + let publish!: (title: string) => Promise + const unload = vi.fn((options: LoadSubsetOptions) => { + leases.delete(options) + }) + const collection = createCollection( + persistedCollectionOptions({ + id: `failed-load-ownership-${mode}`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + publish = async (title) => { + if (!leases.has(peer)) return + begin() + write({ + type: collection.has(`live`) ? `update` : `insert`, + value: { id: `live`, title }, + }) + await commit() + } + markReady() + return { + loadSubset: (options) => { + if (options === failed && mode === `throw`) throw failure + // Returning a promise transfers the ongoing lease, even if + // fetching its initial snapshot subsequently fails. + leases.add(options) + return options === failed ? Promise.reject(failure) : true + }, + unloadSubset: unload, + } + }, + }, + persistence: { adapter: createRecordingAdapter() }, + }), + ) + collection.startSyncImmediate() + try { + await collection._sync.loadSubset(peer) + await expect(collection._sync.loadSubset(failed)).rejects.toBe(failure) + expect(leases.has(failed)).toBe(mode === `reject`) + collection._sync.unloadSubset(failed) + expect(unload.mock.calls.map(([options]) => options)).toEqual( + mode === `reject` ? [failed] : [], + ) + expect(leases).toEqual(new Set([peer])) + await publish(`Peer still live`) + expect(collection.get(`live`)?.title).toBe(`Peer still live`) + collection._sync.unloadSubset(peer) + expect(leases.size).toBe(0) + await publish(`Must not arrive`) + expect(collection.get(`live`)?.title).toBe(`Peer still live`) + } finally { + warn.mockRestore() + await collection.cleanup() + } + }, + ) + it(`does not release or acquire an upstream lease cancelled during hydration`, async () => { const adapter = createRecordingAdapter() const hydrate = adapter.loadSubset diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index 4fae0900a7..a74bb261c8 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -92,6 +92,7 @@ export class CollectionStateManager< public pendingOptimisticDeletes = new Set() public pendingOptimisticDirectUpserts = new Set() public pendingOptimisticDirectDeletes = new Set() + private acknowledgedInserts = new WeakSet() /** * Tracks the origin of confirmed changes for each row. @@ -518,12 +519,12 @@ export class CollectionStateManager< if (!this.isThisCollection(mutation.collection)) { continue } - // A truncate/immediate sync can acknowledge an insert before its - // persistence callback settles. Do not resurrect that client row. + // Only a sync write during this insertion acknowledges it. A stale + // base row can also exist after an optimistic delete and reinsert. if ( isDirectTransaction && mutation.type === `insert` && - this.syncedData.has(mutation.key) + this.acknowledgedInserts.has(mutation) ) { continue } @@ -948,9 +949,12 @@ export class CollectionStateManager< // First collect all keys that will be affected by sync operations const changedKeys = new Set() + const syncedInsertedOrUpdatedKeys = new Set() for (const transaction of committedSyncedTransactions) { for (const operation of transaction.operations) { changedKeys.add(operation.key as TKey) + if (operation.type !== `delete`) + syncedInsertedOrUpdatedKeys.add(operation.key as TKey) } for (const [key] of transaction.rowMetadataWrites) { changedKeys.add(key) @@ -1154,16 +1158,6 @@ export class CollectionStateManager< // the UI preserves local intent while respecting server rebuild semantics. // Ordering: deletes (above) -> server ops (just applied) -> optimistic upserts. if (hasTruncateSync) { - // Avoid duplicating keys that were inserted/updated by synced operations in this commit - const syncedInsertedOrUpdatedKeys = new Set() - for (const t of committedSyncedTransactions) { - for (const op of t.operations) { - if (op.type === `insert` || op.type === `update`) { - syncedInsertedOrUpdatedKeys.add(op.key as TKey) - } - } - } - // Build re-apply sets from the snapshot taken at the start of this function. // This prevents losing optimistic state if transactions complete during truncate processing. const reapplyUpserts = new Map( @@ -1251,6 +1245,13 @@ export class CollectionStateManager< for (const transaction of this.transactions.values()) { if (![`completed`, `failed`].includes(transaction.state)) { for (const mutation of transaction.mutations) { + if ( + this.isThisCollection(mutation.collection) && + mutation.type === `insert` && + syncedInsertedOrUpdatedKeys.has(mutation.key) + ) { + this.acknowledgedInserts.add(mutation) + } if ( this.isThisCollection(mutation.collection) && mutation.optimistic diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index 918e2e9815..4e5e8e9e73 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -820,6 +820,9 @@ A successful larger prefix retires settled smaller prefix acquisitions from the same ordered source plan, after the replacement has applied. It does not retire cursor suffixes, ties, unfinished work, or another subscription's leases. Adapter eviction must still preserve rows owned by the replacement or peers. +If an older prefix's release throws, the successful replacement still finishes +its boundary and continuation bookkeeping before surfacing the cleanup error. +Cleanup failure does not turn the successful acquisition into a failed load. Automatic full-source repair after an established window fails can retry twice, after 250 ms and 500 ms. Every retry releases failed acquisitions before starting diff --git a/packages/db/src/query/live/ordered-source-loader.ts b/packages/db/src/query/live/ordered-source-loader.ts index cd8e26bfe5..af680ec560 100644 --- a/packages/db/src/query/live/ordered-source-loader.ts +++ b/packages/db/src/query/live/ordered-source-loader.ts @@ -352,56 +352,67 @@ export class OrderedSourceLoader { const complete = (): void => { if (this.pending === tracked) this.pending = undefined if (!this.active) return - if (!isFullSource) { - // A replay can replace the physical lease while this older transport - // finishes. Retire its logical owner only outside the replay barrier. - const prefixCount = - options?.orderBy && !options.cursor ? options.limit : undefined - this.settledFiniteAcquisitions.set(releaseAcquisition, prefixCount) - this.retireSettledFiniteAcquisitions() - if (generation === this.generation && prefixCount !== undefined) { - this.retireSettledFiniteAcquisitions({ - release: releaseAcquisition, - count: prefixCount, - }) - } - } - if (generation !== this.generation) return - // A finite request may finish behind an authoritative repair. It cannot - // clear that repair's failure or resume finite refinement around it. - if (!isFullSource && (this.failedRequest || this.fullSource !== `none`)) - return - this.failedRequest = undefined - if (kind !== `boundary`) { - this.hasSettledSourceRequest = true - // Source delivery can invalidate the in-flight prefix marker. - if (options?.orderBy && !options.cursor) { - this.lastPrefixCount = options.limit - } - if (!isFullSource && options?.orderBy) { - try { - this.settledSourceBoundary = - this.subscription.readOrderedSnapshot(options).at(-1)?.value ?? - this.settledSourceBoundary - } catch (error) { - fail(error) + // Retirement failure does not undo a successful acquisition. Finish its + // boundary and continuation, then report the first cleanup error. + runAllCallbacks([ + () => { + if (!isFullSource) { + // A replay can replace the physical lease while this older transport + // finishes. Retire its logical owner only outside the replay barrier. + const prefixCount = + options?.orderBy && !options.cursor ? options.limit : undefined + this.settledFiniteAcquisitions.set(releaseAcquisition, prefixCount) + this.retireSettledFiniteAcquisitions() + if (generation === this.generation && prefixCount !== undefined) { + this.retireSettledFiniteAcquisitions({ + release: releaseAcquisition, + count: prefixCount, + }) + } } - } - } - if (isFullSource) { - this.cancelRepairRetry() - this.repairRetries = 0 - this.needsFullSourceRecovery = false - this.fullSource = `complete` - this.retireSettledFiniteAcquisitions() - } - if (kind === `ordered`) { - this.loadBoundary(windowOperationGeneration) - return - } - // A boundary request may add tied rows without filling the query's - // window. Resume forward loading once it settles. - this.loadMore() + }, + () => { + if (generation !== this.generation) return + // A finite request may finish behind an authoritative repair. It cannot + // clear that repair's failure or resume finite refinement around it. + if ( + !isFullSource && + (this.failedRequest || this.fullSource !== `none`) + ) + return + this.failedRequest = undefined + if (kind !== `boundary`) { + this.hasSettledSourceRequest = true + // Source delivery can invalidate the in-flight prefix marker. + if (options?.orderBy && !options.cursor) { + this.lastPrefixCount = options.limit + } + if (!isFullSource && options?.orderBy) { + try { + this.settledSourceBoundary = + this.subscription.readOrderedSnapshot(options).at(-1) + ?.value ?? this.settledSourceBoundary + } catch (error) { + fail(error) + } + } + } + if (isFullSource) { + this.cancelRepairRetry() + this.repairRetries = 0 + this.needsFullSourceRecovery = false + this.fullSource = `complete` + this.retireSettledFiniteAcquisitions() + } + if (kind === `ordered`) { + this.loadBoundary(windowOperationGeneration) + return + } + // A boundary request may add tied rows without filling the query's + // window. Resume forward loading once it settles. + this.loadMore() + }, + ]) } const settlesAsync = result instanceof Promise const request = settlesAsync ? result : Promise.resolve() diff --git a/packages/db/tests/collection-subscribe-changes.test.ts b/packages/db/tests/collection-subscribe-changes.test.ts index 7297da8e59..3c99ef1f97 100644 --- a/packages/db/tests/collection-subscribe-changes.test.ts +++ b/packages/db/tests/collection-subscribe-changes.test.ts @@ -2671,6 +2671,45 @@ describe(`Virtual properties`, () => { expect(collection.state.get(`row-1`)?.$origin).toBe(`remote`) }) + it.each([false, true])( + `keeps a completed reinsert visible before its sync echo (delete echoed: %s)`, + async (deleteEchoed) => { + let echoDelete!: () => void + const collection = createCollection< + { id: string; value: string }, + string + >({ + getKey: (row) => row.id, + startSync: true, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ type: `insert`, value: { id: `row`, value: `original` } }) + commit() + markReady() + echoDelete = () => { + begin() + write({ type: `delete`, value: { id: `row`, value: `original` } }) + commit() + } + }, + }, + onDelete: () => Promise.resolve(), + onInsert: () => Promise.resolve(), + }) + try { + await collection.delete(`row`).isPersisted.promise + expect(collection.has(`row`)).toBe(false) + if (deleteEchoed) echoDelete() + await collection.insert({ id: `row`, value: `replacement` }).isPersisted + .promise + expect(collection.get(`row`)?.value).toBe(`replacement`) + } finally { + await collection.cleanup() + } + }, + ) + it.each([`before`, `after`] as const)( `replaces a direct mutation settling %s truncate with its authoritative row`, async (settlement) => { @@ -2726,8 +2765,12 @@ describe(`Virtual properties`, () => { }) const applied = syncFns.commit() if (settlement === `after`) { + // An unrelated mutation recomputes the optimistic overlay before the + // acknowledged insertion completes; it must not erase that evidence. + const peer = collection.insert({ id: `other`, value: `peer` }) finishMutation() await transaction.isPersisted.promise + await peer.isPersisted.promise } if (applied !== true) await applied await waitForChanges() diff --git a/packages/db/tests/proxy-detachment-contract.test.ts b/packages/db/tests/proxy-detachment-contract.test.ts index 80b1d80577..0b88283d50 100644 --- a/packages/db/tests/proxy-detachment-contract.test.ts +++ b/packages/db/tests/proxy-detachment-contract.test.ts @@ -155,6 +155,16 @@ describe(`Mutation result detachment`, () => { ? saved.s.values().next().value!.back : saved.arr[0]!.owner expect(back.count).toBe(1) + // The draft becomes a detached snapshot, not the published row wrapper. + // Its containers must still lead back to that same snapshot. + expect(kind === `Set` ? back.s : back.arr).toBe( + kind === `Set` ? saved.s : saved.arr, + ) + const cycle = + kind === `Set` + ? back.s.values().next().value!.back + : back.arr[0]!.owner + expect(cycle).toBe(back) await tx.isPersisted.promise } finally { await collection.cleanup() diff --git a/packages/db/tests/query/ordered-source-loader.test.ts b/packages/db/tests/query/ordered-source-loader.test.ts index 7015dc19d5..c08093a0db 100644 --- a/packages/db/tests/query/ordered-source-loader.test.ts +++ b/packages/db/tests/query/ordered-source-loader.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { createCollection } from '../../src/collection/index.js' import { OrderedSourceLoader } from '../../src/query/live/ordered-source-loader.js' import { Func, PropRef, Value } from '../../src/query/ir.js' @@ -63,6 +63,73 @@ function createOrderByInfo( } describe(`OrderedSourceLoader`, () => { + it(`settles a larger prefix after an older lease release throws`, async () => { + const failure = new Error(`old prefix release failed`) + const requests: Array = [] + const source = createCollection<{ id: number; rank: number }>({ + getKey: (row) => row.id, + syncMode: `on-demand`, + startSync: true, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + for (let id = 1; id <= 3; id++) + write({ type: `insert`, value: { id, rank: id } }) + commit() + markReady() + return { + loadSubset: (options) => { + requests.push(options) + return Promise.resolve() + }, + unloadSubset: (options) => { + if (options === requests[0]) throw failure + }, + } + }, + }, + }) + const subscription = source.subscribeChanges(() => {}, { + includeInitialState: false, + }) + const reads = vi.spyOn(subscription, `readOrderedSnapshot`) + const info = createOrderByInfo({ index: undefined, dataNeeded: () => 0 }) + const loader = new OrderedSourceLoader( + info, + subscription as unknown as CollectionSubscription, + `row`, + ) + try { + loader.start() + await pendingPromise(loader) + reads.mockClear() + info.limit = 2 + await expect(loader.loadMore(1)).rejects.toBe(failure) + expect(reads).toHaveBeenCalledWith(expect.objectContaining({ limit: 2 })) + expect(subscription.lastError).toBe(failure) + info.limit = 3 + await loader.loadMore(2) + expect(requests.map((request) => request.limit)).toEqual([ + 1, + undefined, + 2, + undefined, + 3, + undefined, + ]) + expect( + requests + .filter((request) => request.limit === undefined) + .every((request) => request.where !== undefined), + ).toBe(true) + expect(source.size).toBe(3) + } finally { + loader.dispose() + subscription.unsubscribe() + await source.cleanup() + } + }) + const syncRouteCells = ( [`page`, `prefix`, `boundary`, `full-source`] as const ).flatMap((route) => diff --git a/packages/powersync-db-collection/src/powersync.ts b/packages/powersync-db-collection/src/powersync.ts index 6efe6b7f63..786bb817c8 100644 --- a/packages/powersync-db-collection/src/powersync.ts +++ b/packages/powersync-db-collection/src/powersync.ts @@ -653,47 +653,63 @@ function createPowerSyncCollectionConfig< if (!isCurrent()) return const active = activeWhereExpressions() - if (active.length === 0) return - const combinedWhere = - active.length === 1 - ? active[0] - : or(active[0], active[1], ...active.slice(2)) - const compiledNewData = compileSQLite( - { where: combinedWhere }, - { jsonColumn: 'NEW.data' }, - ) - const compiledOldData = compileSQLite( - { where: combinedWhere }, - { jsonColumn: 'OLD.data' }, - ) - const compiledView = compileSQLite({ where: combinedWhere }) - const newDataWhenClause = toInlinedWhereClause(compiledNewData) - const oldDataWhenClause = toInlinedWhereClause(compiledOldData) - const viewWhereClause = toInlinedWhereClause(compiledView) - - await establishTracking( - { - setupContext: ctx, - when: { - [DiffTriggerOperation.INSERT]: newDataWhenClause, - [DiffTriggerOperation.UPDATE]: `(${newDataWhenClause}) OR (${oldDataWhenClause})`, - [DiffTriggerOperation.DELETE]: oldDataWhenClause, + // Tracking was absent during an error. Positive baseline rows + // alone cannot reveal deletes or predicate exits from that gap. + const missing = + collection.status === `error` + ? new Set(collection.keys()) + : undefined + if (active.length > 0) { + const combinedWhere = + active.length === 1 + ? active[0] + : or(active[0], active[1], ...active.slice(2)) + const compiledNewData = compileSQLite( + { where: combinedWhere }, + { jsonColumn: 'NEW.data' }, + ) + const compiledOldData = compileSQLite( + { where: combinedWhere }, + { jsonColumn: 'OLD.data' }, + ) + const compiledView = compileSQLite({ where: combinedWhere }) + const newDataWhenClause = toInlinedWhereClause(compiledNewData) + const oldDataWhenClause = toInlinedWhereClause(compiledOldData) + const viewWhereClause = toInlinedWhereClause(compiledView) + await establishTracking( + { + setupContext: ctx, + when: { + [DiffTriggerOperation.INSERT]: newDataWhenClause, + [DiffTriggerOperation.UPDATE]: `(${newDataWhenClause}) OR (${oldDataWhenClause})`, + [DiffTriggerOperation.DELETE]: oldDataWhenClause, + }, + writeType: (rowId: string) => + collection.has(rowId) ? `update` : `insert`, + batchQuery: ( + lockContext: LockContext, + batchSize: number, + cursor: number, + ) => + lockContext + .getAll( + `SELECT * FROM ${viewName} WHERE ${viewWhereClause} LIMIT ? OFFSET ?`, + [batchSize, cursor], + ) + .then((rows) => { + for (const row of rows) missing?.delete(row.id) + return rows + }), }, - writeType: (rowId: string) => - collection.has(rowId) ? `update` : `insert`, - batchQuery: ( - lockContext: LockContext, - batchSize: number, - cursor: number, - ) => - lockContext.getAll( - `SELECT * FROM ${viewName} WHERE ${viewWhereClause} LIMIT ? OFFSET ?`, - [batchSize, cursor], - ), - }, - appliedReceipts, - ) + appliedReceipts, + ) + } if (!isCurrent()) await safelyDisposeTracking(ctx) + else if (missing?.size) { + begin() + for (const key of missing) write({ type: `delete`, key }) + appliedReceipts.push(commit()) + } }) await Promise.all(appliedReceipts) if (isCurrent()) { diff --git a/packages/powersync-db-collection/tests/on-demand-sync.test.ts b/packages/powersync-db-collection/tests/on-demand-sync.test.ts index 0d093fa928..f6676d5e3f 100644 --- a/packages/powersync-db-collection/tests/on-demand-sync.test.ts +++ b/packages/powersync-db-collection/tests/on-demand-sync.test.ts @@ -2771,48 +2771,72 @@ describe(`On-Demand Sync Mode`, () => { } }) - it(`does not silently stay errored after a release rebuild retries successfully`, async () => { - vi.useFakeTimers() - const db = await createDatabase() - await db.execute( - `INSERT INTO products (id, name, price, category) VALUES ('retained', 'Before', 10, 'clothing')`, - ) - vi.spyOn(db.logger, `error`).mockImplementation(() => {}) - const collection = createCollection( - powerSyncCollectionOptions({ - database: db, - table: APP_SCHEMA.props.products, - syncMode: `on-demand`, - }), - ) - const first = { where: categoryEquals(`electronics`) } - const second = { where: categoryEquals(`clothing`) } - try { - await collection._sync.loadSubset(first) - await collection._sync.loadSubset(second) - const trigger = vi - .spyOn(db.triggers, `createDiffTrigger`) - .mockRejectedValueOnce(new Error(`release rebuild failed`)) - collection._sync.unloadSubset(first) - await vi.waitFor(() => expect(collection.status).toBe(`error`)) - await vi.advanceTimersByTimeAsync(1_000) - await vi.waitFor(() => - expect(trigger.mock.calls.length).toBeGreaterThan(1), - ) - await vi.waitFor(() => expect(collection.status).toBe(`ready`)) - expect(collection.get(`retained`)?.name).toBe(`Before`) + it.each([`unchanged`, `delete`, `predicate exit`, `release-last`] as const)( + `reconciles rows after a release rebuild outage with %s`, + async (change) => { + vi.useFakeTimers() + const db = await createDatabase() await db.execute( - `UPDATE products SET name = 'After' WHERE id = 'retained'`, + `INSERT INTO products (id, name, price, category) VALUES ('retained', 'Before', 10, 'clothing')`, ) - await vi.waitFor(() => - expect(collection.get(`retained`)?.name).toBe(`After`), + vi.spyOn(db.logger, `error`).mockImplementation(() => {}) + const collection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.products, + syncMode: `on-demand`, + }), ) - } finally { - await collection.cleanup() - await vi.runOnlyPendingTimersAsync() - vi.useRealTimers() - } - }) + const first = { where: categoryEquals(`electronics`) } + const second = { where: categoryEquals(`clothing`) } + try { + await collection._sync.loadSubset(first) + await collection._sync.loadSubset(second) + const trigger = vi + .spyOn(db.triggers, `createDiffTrigger`) + .mockRejectedValueOnce(new Error(`release rebuild failed`)) + collection._sync.unloadSubset(first) + await vi.waitFor(() => expect(collection.status).toBe(`error`)) + if (change === `delete` || change === `release-last`) { + await db.execute(`DELETE FROM products WHERE id = 'retained'`) + } else if (change === `predicate exit`) { + await db.execute( + `UPDATE products SET category = 'outdoors' WHERE id = 'retained'`, + ) + } + if (change === `release-last`) collection._sync.unloadSubset(second) + await vi.advanceTimersByTimeAsync(1_000) + if (change !== `release-last`) + await vi.waitFor(() => + expect(trigger.mock.calls.length).toBeGreaterThan(1), + ) + await vi.waitFor(() => expect(collection.status).toBe(`ready`)) + expect([...collection.keys()]).toEqual( + change === `unchanged` ? [`retained`] : [], + ) + if (change === `release-last`) + await collection._sync.loadSubset({ ...second }) + if (change !== `unchanged`) { + await db.execute( + `INSERT OR REPLACE INTO products (id, name, price, category) VALUES ('retained', 'Before', 10, 'clothing')`, + ) + await vi.waitFor(() => + expect(collection.get(`retained`)?.name).toBe(`Before`), + ) + } + await db.execute( + `UPDATE products SET name = 'After' WHERE id = 'retained'`, + ) + await vi.waitFor(() => + expect(collection.get(`retained`)?.name).toBe(`After`), + ) + } finally { + await collection.cleanup() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }, + ) it(`retries a failed physical release`, async () => { vi.useFakeTimers()