From 8f51dddcc2054ef13850d2e99f1005c17e01dea8 Mon Sep 17 00:00:00 2001 From: Muhammad Amin Saffari Taheri Date: Sun, 20 Sep 2026 07:42:00 +0330 Subject: [PATCH 1/3] fix(db-sqlite-persistence-core): answer a repeat subset acquisition synchronously MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `createWrappedSyncConfig` returned its `loadSubset` as an `async` function, so it was always a Promise — even when startup had settled, the rows were already in the collection, and the wrapped sync answered with the synchronous `true` that `LoadSubsetFn` is declared to allow. `CollectionSyncManager.loadSubset` treats a Promise as outstanding work, so a persisted `on-demand` collection was never synchronously ready and `useLiveSuspenseQuery` re-suspended without limit (13 778 renders in 3s in Chromium, never leaving the fallback). `createLoopbackSyncConfig` had the same defect, so local-only persisted collections were affected too. A repeat acquisition of an already-hydrated subset now returns `true` without touching the store. It is gated on startup having settled, no hydration being in flight, and some acquisition still holding the demand — tracked with a refcount, because `activeSubsets` is keyed per options object while "are these rows here?" is a question about the demand. A synchronous upstream throw is converted to a rejection so the failure shape does not depend on cache state. Concurrent acquisitions of the same subset still read the store once each. Co-Authored-By: Claude Opus 5 (1M context) --- .../persisted-loadsubset-sync-fast-path.md | 5 + .../src/persisted.ts | 244 +++++++++++++++-- .../tests/persisted.test.ts | 216 ++++++++++++++- .../tests/subset-fast-path.test.ts | 245 ++++++++++++++++++ 4 files changed, 681 insertions(+), 29 deletions(-) create mode 100644 .changeset/persisted-loadsubset-sync-fast-path.md create mode 100644 packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts diff --git a/.changeset/persisted-loadsubset-sync-fast-path.md b/.changeset/persisted-loadsubset-sync-fast-path.md new file mode 100644 index 0000000000..cac64b35ad --- /dev/null +++ b/.changeset/persisted-loadsubset-sync-fast-path.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db-sqlite-persistence-core': patch +--- + +Answer a repeat acquisition of an already-loaded subset synchronously. The persisted sync wrapper declared its `loadSubset` `async`, so it returned a promise even when the rows were already in the collection and the wrapped sync had answered `true`; a live query over such a subset was therefore never ready at construction, which left `useLiveSuspenseQuery` re-suspending without limit. Repeat acquisitions now skip the redundant store read, take a lease of their own so releasing a sibling cannot end one still in use, and fall back to the asynchronous path whenever the rows could have gone away — after a source truncate, after the last acquisition is released, during a reload, and before startup settles. Concurrent acquisitions of the same subset still read the store once each; sharing those is a separate change. diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 0f1e4112ac..159f143778 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -1,6 +1,7 @@ import { SyncTransactionAbortedError, compileSingleRowExpression, + getLoadSubsetDemandKey, safeRandomUUID, toBooleanPredicate, withCollectionConfigFactory, @@ -792,6 +793,31 @@ class PersistedCollectionRuntime< > = [] private readonly queuedTxCommitted: Array = [] private readonly requestIds = new WeakMap() + /** + * How many live acquisitions currently cover each demand, and which demands + * are already hydrated into the collection. + * + * `activeSubsets` is keyed per *options object* so that releasing one + * acquisition cannot end another's lease. Answering "are these rows already + * here?" is a different question, asked per *demand*, so it needs its own + * index. Coverage is what keeps the two in step: a demand stays hydrated + * for exactly as long as some acquisition still holds it. + */ + private readonly demandCoverage = new Map< + ReturnType, + number + >() + private readonly hydratedDemands = new Set< + ReturnType + >() + private startupSettled = false + /** + * Bumped whenever the source truncates. A truncate that lands while a + * hydration is in flight is buffered and applied later, so clearing + * `hydratedDemands` outright would be undone by the in-flight hydration + * marking its demand hydrated again. Comparing generations cannot be. + */ + private sourceTruncateGeneration = 0 private collection: Collection | null = null @@ -929,7 +955,7 @@ class PersistedCollectionRuntime< if (lifecycleGeneration !== this.lifecycleGeneration) return const baseline = {} - this.activeSubsets.set(this.getSubsetKey(baseline), baseline) + this.registerSubsetAcquisition(baseline) const appliedCursor = this.appliedReceiptSequence await this.applyMutex.run(async () => { if (lifecycleGeneration !== this.lifecycleGeneration) return @@ -971,6 +997,10 @@ class PersistedCollectionRuntime< if (this.syncMode !== `on-demand`) { await this.hydrateBaseline(lifecycleGeneration) } + + if (lifecycleGeneration === this.lifecycleGeneration) { + this.startupSettled = true + } } private async loadStartupMetadataInternal( @@ -1045,7 +1075,8 @@ class PersistedCollectionRuntime< upstreamLoadSubset?: LoadSubsetFn, ): Promise { const lifecycleGeneration = this.lifecycleGeneration - this.activeSubsets.set(this.getSubsetKey(options), options) + const sourceTruncateGeneration = this.sourceTruncateGeneration + this.registerSubsetAcquisition(options) const appliedCursor = this.appliedReceiptSequence await this.applyMutex.run(() => @@ -1057,6 +1088,17 @@ class PersistedCollectionRuntime< if (lifecycleGeneration !== this.lifecycleGeneration) return await this.waitForAppliedReceiptsAfter(appliedCursor) + // Only claim the rows are here if nothing invalidated them while we were + // reading: a restart bumps the lifecycle, a source truncate bumps its own + // generation, and either means this hydration no longer describes the + // collection. + if ( + lifecycleGeneration === this.lifecycleGeneration && + sourceTruncateGeneration === this.sourceTruncateGeneration + ) { + this.hydratedDemands.add(getLoadSubsetDemandKey(options)) + } + if (upstreamLoadSubset) { try { await upstreamLoadSubset(options) @@ -1083,11 +1125,100 @@ class PersistedCollectionRuntime< options: LoadSubsetOptions, upstreamUnloadSubset?: (options: LoadSubsetOptions) => void, ): void { - this.activeSubsets.delete(this.getSubsetKey(options)) - this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options)) + const subsetKey = this.getSubsetKey(options) + if (this.activeSubsets.has(subsetKey)) this.releaseDemand(options) + this.activeSubsets.delete(subsetKey) + this.pendingRemoteSubsetEnsures.delete(subsetKey) upstreamUnloadSubset?.(options) } + /** + * Record one acquisition of a subset. + * + * **Idempotent per options object, because `activeSubsets` is.** Its key is + * a per-object `request:N`, so acquiring the *same* object twice replaces + * one entry rather than adding a second — and a second `retainDemand` would + * then leave coverage claiming a holder that no `unloadSubset` can ever + * release. Coverage has to count entries, not calls, or a demand can stay + * marked hydrated after the last subset covering it is gone. + */ + private registerSubsetAcquisition(options: LoadSubsetOptions): void { + const subsetKey = this.getSubsetKey(options) + + if (this.activeSubsets.has(subsetKey)) return + + this.activeSubsets.set(subsetKey, options) + this.retainDemand(options) + } + + /** + * Take this demand for one more acquisition. + */ + private retainDemand(options: LoadSubsetOptions): void { + const key = getLoadSubsetDemandKey(options) + this.demandCoverage.set(key, (this.demandCoverage.get(key) ?? 0) + 1) + } + + /** + * Give one acquisition's hold back, and forget the demand once nothing + * holds it: its rows are no longer guaranteed to be in the collection. + */ + private releaseDemand(options: LoadSubsetOptions): void { + const key = getLoadSubsetDemandKey(options) + const remaining = (this.demandCoverage.get(key) ?? 1) - 1 + + if (remaining > 0) { + this.demandCoverage.set(key, remaining) + return + } + + this.demandCoverage.delete(key) + this.hydratedDemands.delete(key) + } + + /** + * Acquire a subset whose rows are already in the collection, without + * touching the store — reporting whether that was possible. + * + * `LoadSubsetFn` is declared `(options) => true | Promise` precisely so + * an implementation can say "already here, nothing to await", and + * `CollectionSyncManager.loadSubset` only treats a subset as outstanding + * work when it gets a promise. Re-reading the store to answer a question we + * already know the answer to costs a promise, and that promise is what makes + * a live query `loading` when it should be `ready`. + * + * Three conditions, and each rules out a way the rows could be absent: + * startup must have finished; no hydration may be in flight (there is a + * window inside `truncateAndReloadUnsafe` where the collection is + * deliberately empty); and some acquisition must still hold this demand, + * because every in-generation invalidation path preserves the rows of + * demands that are still active and only those. + */ + tryAcquireHydratedSubset(options: LoadSubsetOptions): boolean { + if (!this.startupSettled || this.isHydratingNow()) return false + + const key = getLoadSubsetDemandKey(options) + + if (!this.hydratedDemands.has(key)) return false + if ((this.demandCoverage.get(key) ?? 0) <= 0) return false + + // The same registration the slow path performs, so this acquisition owns + // a lease and the invalidation paths keep treating its rows as wanted. + this.registerSubsetAcquisition(options) + this.queueRemoteSubsetEnsure(options) + + return true + } + + /** + * Record that the source truncated, so no hydration that started before it + * may report its demand as still present. + */ + noteSourceTruncate(): void { + this.sourceTruncateGeneration++ + this.hydratedDemands.clear() + } + async forceReloadSubset(options: LoadSubsetOptions): Promise { const lifecycleGeneration = this.lifecycleGeneration // A one-shot refresh does not acquire an enduring subscription lease. @@ -1232,6 +1363,8 @@ class PersistedCollectionRuntime< this.pendingRemoteSubsetEnsures.clear() this.activeSubsets.clear() + this.demandCoverage.clear() + this.hydratedDemands.clear() for (const transaction of this.queuedHydrationTransactions) { transaction.rejectApplied?.(new SyncTransactionAbortedError()) } @@ -1242,8 +1375,12 @@ class PersistedCollectionRuntime< } private advanceLifecycle(): void { + this.lifecycleGeneration++ this.started = false + this.startupSettled = false + this.demandCoverage.clear() + this.hydratedDemands.clear() this.startupMetadataPromise = null this.startPromise = null this.resumeBaselinePromise = null @@ -2181,6 +2318,7 @@ class PersistedCollectionRuntime< private async reloadActiveSubsetsUnsafe(): Promise { const lifecycleGeneration = this.lifecycleGeneration + const sourceTruncateGeneration = this.sourceTruncateGeneration const activeSubsetOptions = this.activeSubsets.size > 0 ? Array.from(this.activeSubsets.values()) @@ -2210,6 +2348,23 @@ class PersistedCollectionRuntime< })), collectionMetadata, ) + + // Everything active was just re-read, so those demands are hydrated + // again even though they did not go through `loadSubset`. When nothing + // was active this reloaded the bare baseline instead, which no + // acquisition holds and must not be recorded as a demand. + // + // A source truncate that lands *while* this reload is in flight empties + // the collection after these rows were read, so re-marking them here + // would undo the invalidation the truncate just performed. + if ( + this.activeSubsets.size > 0 && + sourceTruncateGeneration === this.sourceTruncateGeneration + ) { + for (const options of activeSubsetOptions) { + this.hydratedDemands.add(getLoadSubsetDemandKey(options)) + } + } } finally { if (this.hydratingGeneration === lifecycleGeneration) { this.hydratingGeneration = null @@ -2556,6 +2711,7 @@ function createWrappedSyncConfig< : undefined, truncate: () => { if (startupState.cleanedUp) return + runtime.noteSourceTruncate() const openTransaction = getOpenTransaction() if (!openTransaction) { params.truncate() @@ -2629,17 +2785,20 @@ function createWrappedSyncConfig< } let sourceResult: SyncConfigRes = {} + let sourceResultSettled = false fullStartPromise = runtime.ensureStarted() const sourceResultPromise = (async () => { await runtime.ensureStartupMetadataLoaded() if (startupState.cleanedUp) { + sourceResultSettled = true return sourceResult } sourceResult = normalizeSyncFnResult( sourceSyncConfig.sync(wrappedParams), ) + sourceResultSettled = true return sourceResult })() @@ -2651,37 +2810,63 @@ function createWrappedSyncConfig< runtime.cleanup() runtime.clearSyncControls() }, - loadSubset: async (options: LoadSubsetOptions) => { + loadSubset: (options: LoadSubsetOptions): true | Promise => { const acquisition = { forwarded: false } acquisitions.set(options, acquisition) - await fullStartPromise - const resolvedSourceResult = await sourceResultPromise - if ( - startupState.cleanedUp || - acquisitions.get(options) !== acquisition - ) { - return + + // Hydration is another async boundary. A release before this point + // owns no upstream lease and must not start one later. + const forwardUpstream = + (resolved: SyncConfigRes) => + (loadOptions: LoadSubsetOptions): true | Promise => { + if ( + startupState.cleanedUp || + acquisitions.get(options) !== acquisition + ) { + return true + } + if (!resolved.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 resolved.loadSubset(loadOptions) + } catch (error) { + acquisition.forwarded = false + throw error + } + } + + // Nothing to await: the source is up and these rows are already in + // the collection, so the only work left is the upstream forward. + // Staying synchronous here is what lets the live query be `ready` at + // construction rather than `loading` for ever under Suspense. + if (sourceResultSettled && runtime.tryAcquireHydratedSubset(options)) { + try { + return forwardUpstream(sourceResult)(options) + } catch (error) { + // The async path below turns a synchronous upstream throw into a + // rejection. Callers must not see a different failure shape + // depending on whether the rows happened to be cached. + return Promise.reject(error) + } } - return runtime.loadSubset(options, (loadOptions) => { - // Hydration is another async boundary. A release before this - // point owns no upstream lease and must not start one later. + + return (async () => { + await fullStartPromise + const resolvedSourceResult = await sourceResultPromise if ( startupState.cleanedUp || acquisitions.get(options) !== acquisition ) { - return true + return } - 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 - throw error - } - }) + return runtime.loadSubset( + options, + forwardUpstream(resolvedSourceResult), + ) + })() }, unloadSubset: (options: LoadSubsetOptions) => { const acquisition = acquisitions.get(options) @@ -2728,7 +2913,10 @@ function createLoopbackSyncConfig< runtime.cleanup() runtime.clearSyncControls() }, - loadSubset: (options: LoadSubsetOptions) => runtime.loadSubset(options), + loadSubset: (options: LoadSubsetOptions): true | Promise => + runtime.tryAcquireHydratedSubset(options) + ? true + : runtime.loadSubset(options), unloadSubset: (options: LoadSubsetOptions) => runtime.unloadSubset(options), } diff --git a/packages/db-sqlite-persistence-core/tests/persisted.test.ts b/packages/db-sqlite-persistence-core/tests/persisted.test.ts index 0895e52375..ad9d2dab73 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted.test.ts @@ -2161,7 +2161,10 @@ describe(`persistedCollectionOptions`, () => { ) collection.startSyncImmediate() const first: LoadSubsetOptions = { limit: 1 } - const second: LoadSubsetOptions = { limit: 1 } + // A *different* demand, so the second load reaches the store and can be + // blocked mid-hydration. Two acquisitions of the same demand are answered + // from the rows already in the collection and never get there. + const second: LoadSubsetOptions = { limit: 2 } try { await collection._sync.loadSubset(first) expect(leases).toBe(1) @@ -2181,6 +2184,217 @@ describe(`persistedCollectionOptions`, () => { } }) + + it(`answers a repeat acquisition of a loaded subset synchronously`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `first` }, + ]) + const collection = createCollection( + persistedCollectionOptions({ + id: `repeat-acquisition-is-synchronous`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + await collection._sync.loadSubset({ limit: 1 }) + const storeReads = adapter.loadSubsetCalls.length + + // The same demand, a second acquisition. `LoadSubsetFn` is declared + // `true | Promise` so this can be answered without a promise, and + // a live query built on it is `ready` at construction rather than + // `loading`. + expect(collection._sync.loadSubset({ limit: 1 })).toBe(true) + expect(adapter.loadSubsetCalls.length).toBe(storeReads) + } finally { + await collection.cleanup() + } + }) + + it(`keeps the upstream lease of a sibling acquisition when one is released`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `first` }, + ]) + let leases = 0 + const released: Array = [] + let publish!: (row: Todo) => void + const collection = createCollection( + persistedCollectionOptions({ + id: `sibling-acquisition-lease`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + publish = (row) => { + begin() + write({ type: `insert`, value: row }) + commit() + } + return { + loadSubset: () => { + leases++ + return true + }, + unloadSubset: (options) => { + leases-- + released.push(options) + }, + } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + const first: LoadSubsetOptions = { limit: 1 } + const second: LoadSubsetOptions = { limit: 1 } + + try { + await collection._sync.loadSubset(first) + // The fast path must still take a lease of its own, or releasing the + // first acquisition would end one the second still depends on. + expect(collection._sync.loadSubset(second)).toBe(true) + expect(leases).toBe(2) + + collection._sync.unloadSubset(first) + expect(released).toEqual([first]) + expect(leases).toBe(1) + + // Still loaded for the surviving acquisition. + expect(collection._sync.loadSubset(second)).toBe(true) + + // And the surviving acquisition is still a live demand: a later source + // write must reach the collection rather than be discarded as matching + // no active subset. + publish({ id: `2`, title: `second` }) + await flushAsyncWork() + expect(collection.get(`2`)).toMatchObject({ id: `2`, title: `second` }) + } finally { + await collection.cleanup() + } + }) + + it(`re-reads a subset once its last acquisition is released`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `first` }, + ]) + const collection = createCollection( + persistedCollectionOptions({ + id: `released-subset-is-reread`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + const only: LoadSubsetOptions = { limit: 1 } + + try { + await collection._sync.loadSubset(only) + const storeReads = adapter.loadSubsetCalls.length + + collection._sync.unloadSubset(only) + + // Nothing holds the demand any more, so its rows are no longer + // guaranteed to be in the collection and the next request must read. + const reacquired = collection._sync.loadSubset({ limit: 1 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(adapter.loadSubsetCalls.length).toBeGreaterThan(storeReads) + } finally { + await collection.cleanup() + } + }) + + it(`re-reads a subset after the source truncates`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `first` }, + ]) + let truncateSource!: () => void + const collection = createCollection( + persistedCollectionOptions({ + id: `truncated-subset-is-reread`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ begin, commit, markReady, truncate }) => { + markReady() + truncateSource = () => { + begin() + truncate() + commit() + } + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + await collection._sync.loadSubset({ limit: 1 }) + const storeReads = adapter.loadSubsetCalls.length + + truncateSource() + await flushAsyncWork() + + // The rows are gone, so a repeat request must not be answered from the + // belief that they are still there. + const afterTruncate = collection._sync.loadSubset({ limit: 1 }) + expect(afterTruncate).not.toBe(true) + await afterTruncate + expect(adapter.loadSubsetCalls.length).toBeGreaterThan(storeReads) + } finally { + await collection.cleanup() + } + }) + + it(`answers a repeat acquisition synchronously for a local-only collection`, async () => { + const adapter = createRecordingAdapter([ + { id: `1`, title: `first` }, + ]) + const collection = createCollection( + persistedCollectionOptions({ + id: `local-only-repeat-acquisition`, + syncMode: `on-demand`, + getKey: (row) => row.id, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + await collection._sync.loadSubset({ limit: 1 }) + const storeReads = adapter.loadSubsetCalls.length + + // The loopback sync config has no source to forward to, so it is + // unconditionally synchronous once the rows are in hand. + expect(collection._sync.loadSubset({ limit: 1 })).toBe(true) + expect(adapter.loadSubsetCalls.length).toBe(storeReads) + } finally { + await collection.cleanup() + } + }) + it.each([`abort`, `release`, `offline`] as const)( `handles remote ensure after %s without resurrecting cancelled demand`, async (action) => { diff --git a/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts new file mode 100644 index 0000000000..13dc05b87c --- /dev/null +++ b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts @@ -0,0 +1,245 @@ +import { describe, expect, it } from 'vitest' +import { createCollection } from '@tanstack/db' +import { persistedCollectionOptions } from '../src' +import type { + PersistedCollectionCoordinator, + PersistenceAdapter, + ProtocolEnvelope, + PullSinceResponse, + TxCommitted, +} from '../src' +import type { LoadSubsetOptions } from '@tanstack/db' + +type Todo = { id: string; title: string } + +function createLimitAdapter(initial: Array) { + const rows = new Map(initial.map((r) => [r.id, r])) + const calls: Array = [] + const adapter: PersistenceAdapter & { + rows: Map + calls: Array + } = { + rows, + calls, + loadSubset: (_id, options) => { + calls.push(options) + const all = Array.from(rows.values()).map((value) => ({ + key: value.id, + value, + })) + const limited = + options.limit === undefined ? all : all.slice(0, options.limit) + return Promise.resolve(limited) + }, + loadCollectionMetadata: () => Promise.resolve([]), + applyCommittedTx: (_id, tx) => { + if (tx.truncate) rows.clear() + for (const m of tx.mutations) { + if (m.type === `delete`) rows.delete(m.key as string) + else rows.set(m.key as string, m.value as Todo) + } + return Promise.resolve() + }, + ensureIndex: () => Promise.resolve(), + } + return adapter +} + +type CoordinatorHarness = PersistedCollectionCoordinator & { + emit: (payload: TxCommitted, senderId?: string) => void +} + +function createCoordinatorHarness(collectionId: string): CoordinatorHarness { + let subscriber: ((message: ProtocolEnvelope) => void) | undefined + const pullSinceResponse: PullSinceResponse = { + type: `rpc:pullSince:res`, + rpcId: `pull-0`, + ok: true, + latestTerm: 1, + latestSeq: 0, + latestRowVersion: 0, + requiresFullReload: false, + changedKeys: [], + deletedKeys: [], + } + const harness: CoordinatorHarness = { + getNodeId: () => `coordinator-node`, + subscribe: (_collectionId, onMessage) => { + subscriber = onMessage + return () => { + subscriber = undefined + } + }, + publish: () => {}, + isLeader: () => true, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + requestEnsureRemoteSubset: async () => {}, + pullSince: () => Promise.resolve(pullSinceResponse), + emit: (payload, senderId = `remote-node`) => { + subscriber?.({ + v: 1, + dbName: `test-db`, + collectionId, + senderId, + ts: Date.now(), + payload, + }) + }, + } + return harness +} + +async function flush(times = 6): Promise { + for (let i = 0; i < times; i++) { + await new Promise((resolve) => setTimeout(resolve, 0)) + } +} + +/** + * The two ways a repeat acquisition could be answered from rows that are no + * longer in the collection. Both are about `hydratedDemands` outliving what it + * describes, and both produce a `true` where a promise is owed. + */ +describe(`subset fast path`, () => { + it(`does not leak coverage when one options object is acquired twice`, async () => { + const adapter = createLimitAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + { id: `3`, title: `three` }, + ]) + const coordinator = createCoordinatorHarness(`sync-present`) + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + syncMode: `on-demand`, + getKey: (r) => r.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true, unloadSubset: () => {} } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + + const wide: LoadSubsetOptions = { limit: 3 } + await collection._sync.loadSubset(wide) + expect(collection.size).toBe(3) + + // Same options object, loaded again (no unload in between). + expect(collection._sync.loadSubset(wide)).toBe(true) + // One release for what the runtime recorded as two retains. + collection._sync.unloadSubset(wide) + + // A narrow demand is now the only live acquisition. + await collection._sync.loadSubset({ limit: 1 }) + + // Full reload: rebuilt from active subsets only -> rows 2 and 3 drop out. + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `tx-full-reload`, + latestRowVersion: 1, + requiresFullReload: true, + }) + await flush() + expect(collection.size).toBe(1) + expect(adapter.rows.size).toBe(3) // the store still has all three + + // A fresh acquisition of the wide demand. Its rows are NOT in the + // collection, so this must not be answered synchronously. + const reacquired = collection._sync.loadSubset({ limit: 3 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(collection.size).toBe(3) + + await collection.cleanup() + }) + + it(`does not re-mark a demand hydrated when a truncate lands during a reload`, async () => { + const adapter = createLimitAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + { id: `3`, title: `three` }, + ]) + // A store whose truncate write fails: the collection is emptied, the store + // is not. + const applyCommittedTx = adapter.applyCommittedTx.bind(adapter) + adapter.applyCommittedTx = (id, tx) => { + if (tx.truncate) return Promise.reject(new Error(`store write failed`)) + return applyCommittedTx(id, tx) + } + const coordinator = createCoordinatorHarness(`sync-present`) + const loadSubset = adapter.loadSubset.bind(adapter) + let calls = 0 + let releaseReload!: () => void + let reloadEntered!: () => void + const reloadGate = new Promise((r) => { + releaseReload = r + }) + const entered = new Promise((r) => { + reloadEntered = r + }) + adapter.loadSubset = async (...args) => { + calls++ + if (calls === 2) { + reloadEntered() + await reloadGate + } + return loadSubset(...args) + } + + let truncateSource!: () => void + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + syncMode: `on-demand`, + getKey: (r) => r.id, + sync: { + sync: ({ begin, commit, truncate, markReady }) => { + markReady() + truncateSource = () => { + begin() + truncate() + commit() + } + return { loadSubset: () => true, unloadSubset: () => {} } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + + await collection._sync.loadSubset({ limit: 3 }) + expect(collection.size).toBe(3) + + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `tx-full-reload`, + latestRowVersion: 1, + requiresFullReload: true, + }) + await entered + // The source truncates while the reload is in flight. + truncateSource() + releaseReload() + await flush() + + expect(collection.size).toBe(0) + expect(adapter.rows.size).toBe(3) + + const reacquired = collection._sync.loadSubset({ limit: 3 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(collection.size).toBe(3) + + await collection.cleanup() + }) +}) From e91bded8be2bc125a86b29f9dfda40a8d649c18c Mon Sep 17 00:00:00 2001 From: Muhammad Amin Saffari Taheri Date: Sun, 20 Sep 2026 08:00:57 +0330 Subject: [PATCH 2/3] fix(db-sqlite-persistence-core): close two holes in the subset fast path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An adversarial pass over the previous commit found two ways it could still be wrong, both confirmed by running code. **A released demand could be marked hydrated after the fact.** `loadSubset` recorded its demand on completion having checked only the lifecycle and source truncate generations, so a demand whose every acquisition was released while its read was in flight was recorded with nobody holding it — an entry no release can ever remove, because only a release-to-zero clears one. The record is now also gated on coverage. **Registration is synchronous; the read is not.** `ApplyMutex.run` always defers, so there was a window where a demand was covered, nothing was hydrating, and no row had been read. Demands with a read in flight are now tracked and excluded, which is what makes the documented invariant true rather than nearly true. **An aborted fast-path acquisition was ensured remotely for ever.** The slow path deletes a cancelled demand from the pending remote ensures and re-queues any other failure; the fast path forwarded upstream without any of it, so an aborted acquisition was retried against the coordinator indefinitely — the invariant `handles remote ensure after abort without resurrecting cancelled demand` already asserts for the slow path. That handling is extracted and both paths use it. Adds `tests/subset-fast-path.test.ts`. Four of its five cases fail on the previous commit; the fifth asserts the released-mid-read invariant directly, because provoking the wrong answer it prevents needs the apply mutex held by non-hydrating work and no public API exposes that. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/persisted.ts | 132 ++++++++--- .../tests/subset-fast-path.test.ts | 205 +++++++++++++++++- 2 files changed, 304 insertions(+), 33 deletions(-) diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 159f143778..6c6f3d0c32 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -810,6 +810,16 @@ class PersistedCollectionRuntime< private readonly hydratedDemands = new Set< ReturnType >() + /** + * Demands with a store read in flight. Registration is synchronous but the + * read is not (`ApplyMutex.run` always defers), so without this there is a + * window where a demand is covered, nothing is hydrating, and no row has + * been read. + */ + private readonly loadsInFlight = new Map< + ReturnType, + number + >() private startupSettled = false /** * Bumped whenever the source truncates. A truncate that lands while a @@ -954,6 +964,11 @@ class PersistedCollectionRuntime< private async hydrateBaseline(lifecycleGeneration: number): Promise { if (lifecycleGeneration !== this.lifecycleGeneration) return + // `getLoadSubsetDemandKey({})` is `undefined`, the same key a genuinely + // unconstrained subset request has. That is safe only because this + // acquisition is never released and its entry stays in `activeSubsets` for + // the life of the collection, so every reload re-reads it and the rows + // really are present — a different argument from every other demand's. const baseline = {} this.registerSubsetAcquisition(baseline) const appliedCursor = this.appliedReceiptSequence @@ -1076,51 +1091,99 @@ class PersistedCollectionRuntime< ): Promise { const lifecycleGeneration = this.lifecycleGeneration const sourceTruncateGeneration = this.sourceTruncateGeneration + const demandKey = getLoadSubsetDemandKey(options) this.registerSubsetAcquisition(options) + this.enterLoad(demandKey) const appliedCursor = this.appliedReceiptSequence - await this.applyMutex.run(() => - this.hydrateSubsetUnsafe(options, { - requestRemoteEnsure: this.mode === `sync-present`, - lifecycleGeneration, - }), - ) - if (lifecycleGeneration !== this.lifecycleGeneration) return - await this.waitForAppliedReceiptsAfter(appliedCursor) + try { + await this.applyMutex.run(() => + this.hydrateSubsetUnsafe(options, { + requestRemoteEnsure: this.mode === `sync-present`, + lifecycleGeneration, + }), + ) + if (lifecycleGeneration !== this.lifecycleGeneration) return + await this.waitForAppliedReceiptsAfter(appliedCursor) + } finally { + this.leaveLoad(demandKey) + } // Only claim the rows are here if nothing invalidated them while we were // reading: a restart bumps the lifecycle, a source truncate bumps its own // generation, and either means this hydration no longer describes the // collection. + // ...and only if some acquisition still wants them. Every acquisition of + // this demand may have been released while the read was in flight, and + // recording a demand nothing holds would leave an entry that no release + // can ever remove — a `true` owed to rows the next reload will drop. if ( lifecycleGeneration === this.lifecycleGeneration && - sourceTruncateGeneration === this.sourceTruncateGeneration + sourceTruncateGeneration === this.sourceTruncateGeneration && + (this.demandCoverage.get(demandKey) ?? 0) > 0 ) { - this.hydratedDemands.add(getLoadSubsetDemandKey(options)) + this.hydratedDemands.add(demandKey) } if (upstreamLoadSubset) { try { await upstreamLoadSubset(options) } catch (error) { - if ( - options.signal?.aborted || - (typeof error === `object` && - error !== null && - `name` in error && - error.name === `AbortError`) - ) { - this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options)) - throw error - } - console.warn(`Failed to trigger remote subset load:`, error) - this.queueRemoteSubsetEnsure(options) - // Hydration remains readable, but it does not satisfy remote demand. + this.noteUpstreamLoadFailure(options, error) throw error } } } + /** + * The bookkeeping a failed upstream subset load owes, whichever path ran it. + * + * A cancelled demand must stop being ensured remotely; any other failure + * leaves the hydration readable but unsatisfied upstream, so it is queued + * for retry. The fast path forwards upstream without going through + * `loadSubset`, so it has to ask for this explicitly — otherwise an aborted + * acquisition is ensured for ever. + */ + noteUpstreamLoadFailure(options: LoadSubsetOptions, error: unknown): void { + if ( + options.signal?.aborted || + (typeof error === `object` && + error !== null && + `name` in error && + error.name === `AbortError`) + ) { + this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options)) + return + } + + console.warn(`Failed to trigger remote subset load:`, error) + this.queueRemoteSubsetEnsure(options) + } + + /** An upstream result with that bookkeeping attached, shape preserved. */ + settleUpstreamLoad( + options: LoadSubsetOptions, + result: true | Promise, + ): true | Promise { + if (result === true) return true + + return result.catch((error: unknown) => { + this.noteUpstreamLoadFailure(options, error) + throw error + }) + } + + private enterLoad(key: ReturnType): void { + this.loadsInFlight.set(key, (this.loadsInFlight.get(key) ?? 0) + 1) + } + + private leaveLoad(key: ReturnType): void { + const remaining = (this.loadsInFlight.get(key) ?? 1) - 1 + + if (remaining > 0) this.loadsInFlight.set(key, remaining) + else this.loadsInFlight.delete(key) + } + unloadSubset( options: LoadSubsetOptions, upstreamUnloadSubset?: (options: LoadSubsetOptions) => void, @@ -1187,12 +1250,14 @@ class PersistedCollectionRuntime< * already know the answer to costs a promise, and that promise is what makes * a live query `loading` when it should be `ready`. * - * Three conditions, and each rules out a way the rows could be absent: + * Four conditions, and each rules out a way the rows could be absent: * startup must have finished; no hydration may be in flight (there is a * window inside `truncateAndReloadUnsafe` where the collection is - * deliberately empty); and some acquisition must still hold this demand, - * because every in-generation invalidation path preserves the rows of - * demands that are still active and only those. + * deliberately empty); this demand must have no read of its own still + * running, because registration is synchronous while the read is not; and + * some acquisition must still hold it, because every in-generation + * invalidation path preserves the rows of demands that are still active and + * only those. */ tryAcquireHydratedSubset(options: LoadSubsetOptions): boolean { if (!this.startupSettled || this.isHydratingNow()) return false @@ -1200,6 +1265,7 @@ class PersistedCollectionRuntime< const key = getLoadSubsetDemandKey(options) if (!this.hydratedDemands.has(key)) return false + if (this.loadsInFlight.has(key)) return false if ((this.demandCoverage.get(key) ?? 0) <= 0) return false // The same registration the slow path performs, so this acquisition owns @@ -1375,12 +1441,12 @@ class PersistedCollectionRuntime< } private advanceLifecycle(): void { - this.lifecycleGeneration++ this.started = false this.startupSettled = false this.demandCoverage.clear() this.hydratedDemands.clear() + this.loadsInFlight.clear() this.startupMetadataPromise = null this.startPromise = null this.resumeBaselinePromise = null @@ -2844,8 +2910,16 @@ function createWrappedSyncConfig< // construction rather than `loading` for ever under Suspense. if (sourceResultSettled && runtime.tryAcquireHydratedSubset(options)) { try { - return forwardUpstream(sourceResult)(options) + // `settleUpstreamLoad` is what `runtime.loadSubset` would have + // applied: without it an aborted acquisition is never removed + // from the pending remote ensures and is retried for ever. + return runtime.settleUpstreamLoad( + options, + forwardUpstream(sourceResult)(options), + ) } catch (error) { + runtime.noteUpstreamLoadFailure(options, error) + // The async path below turns a synchronous upstream throw into a // rejection. Callers must not see a different failure shape // depending on whether the rows happened to be cached. diff --git a/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts index 13dc05b87c..98e6cae100 100644 --- a/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts +++ b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { createCollection } from '@tanstack/db' import { persistedCollectionOptions } from '../src' import type { @@ -32,6 +32,10 @@ function createLimitAdapter(initial: Array) { return Promise.resolve(limited) }, loadCollectionMetadata: () => Promise.resolve([]), + scanRows: () => + Promise.resolve( + Array.from(rows.values()).map((value) => ({ key: value.id, value })), + ), applyCommittedTx: (_id, tx) => { if (tx.truncate) rows.clear() for (const m of tx.mutations) { @@ -97,9 +101,14 @@ async function flush(times = 6): Promise { } /** - * The two ways a repeat acquisition could be answered from rows that are no - * longer in the collection. Both are about `hydratedDemands` outliving what it - * describes, and both produce a `true` where a promise is owed. + * The ways a repeat acquisition could be answered from rows that are not in + * the collection, and the bookkeeping the fast path still owes upstream. + * + * Four of these are about `hydratedDemands` outliving what it describes, each + * producing a `true` where a promise is owed. The fifth is the opposite + * direction: an acquisition that was cancelled must stop being ensured + * remotely, which the slow path has always done and the fast path has to be + * told to do. */ describe(`subset fast path`, () => { it(`does not leak coverage when one options object is acquired twice`, async () => { @@ -242,4 +251,192 @@ describe(`subset fast path`, () => { await collection.cleanup() }) + + it(`does not keep ensuring a fast-path acquisition whose upstream load aborted`, async () => { + vi.useFakeTimers() + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const abortError = Object.assign(new Error(`abort`), { name: `AbortError` }) + const ensured: Array = [] + const coordinator: PersistedCollectionCoordinator = { + getNodeId: () => `fast-path-abort`, + subscribe: () => () => {}, + publish: () => {}, + isLeader: () => true, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + requestEnsureRemoteSubset: (_id, options) => { + ensured.push(options) + return Promise.reject(new Error(`offline`)) + }, + } + const first: LoadSubsetOptions = { limit: 1 } + const second: LoadSubsetOptions = { limit: 1 } + const collection = createCollection( + persistedCollectionOptions({ + id: `fast-path-abort`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: (options: LoadSubsetOptions) => + options === second ? Promise.reject(abortError) : true, + unloadSubset: () => {}, + } + }, + }, + persistence: { adapter: createLimitAdapter([{ id: `1`, title: `a` }]), coordinator }, + }), + ) + try { + collection.startSyncImmediate() + await collection._sync.loadSubset(first) + + const result = await Promise.resolve( + collection._sync.loadSubset(second), + ).then( + () => `ready`, + (error: unknown) => error, + ) + expect(result).toBe(abortError) + + const secondEnsuresBefore = ensured.filter((o) => o === second).length + await vi.advanceTimersByTimeAsync(500) + const secondEnsuresAfter = ensured.filter((o) => o === second).length + // The aborted acquisition must not keep being ensured remotely. + expect(secondEnsuresAfter).toBe(secondEnsuresBefore) + } finally { + await collection.cleanup() + warning.mockRestore() + vi.useRealTimers() + } + }) + + it(`stops ensuring an aborted demand on the slow path too`, async () => { + vi.useFakeTimers() + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const abortError = Object.assign(new Error(`abort`), { name: `AbortError` }) + const ensured: Array = [] + const coordinator: PersistedCollectionCoordinator = { + getNodeId: () => `fast-path-abort`, + subscribe: () => () => {}, + publish: () => {}, + isLeader: () => true, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + requestEnsureRemoteSubset: (_id, options) => { + ensured.push(options) + return Promise.reject(new Error(`offline`)) + }, + } + const first: LoadSubsetOptions = { limit: 1 } + const second: LoadSubsetOptions = { limit: 2 } + const collection = createCollection( + persistedCollectionOptions({ + id: `fast-path-abort-control`, + getKey: (row) => row.id, + syncMode: `on-demand`, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: (options: LoadSubsetOptions) => + options === second ? Promise.reject(abortError) : true, + unloadSubset: () => {}, + } + }, + }, + persistence: { adapter: createLimitAdapter([{ id: `1`, title: `a` }]), coordinator }, + }), + ) + try { + collection.startSyncImmediate() + await collection._sync.loadSubset(first) + + const result = await Promise.resolve( + collection._sync.loadSubset(second), + ).then( + () => `ready`, + (error: unknown) => error, + ) + expect(result).toBe(abortError) + + const secondEnsuresBefore = ensured.filter((o) => o === second).length + await vi.advanceTimersByTimeAsync(500) + const secondEnsuresAfter = ensured.filter((o) => o === second).length + // The aborted acquisition must not keep being ensured remotely. + expect(secondEnsuresAfter).toBe(secondEnsuresBefore) + } finally { + await collection.cleanup() + warning.mockRestore() + vi.useRealTimers() + } + }) + + it(`does not mark a demand hydrated when its last holder left mid-read`, async () => { + const adapter = createLimitAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + { id: `3`, title: `three` }, + ]) + const read = adapter.loadSubset.bind(adapter) + let openGate!: () => void + let readEntered!: () => void + const gate = new Promise((resolve) => { + openGate = resolve + }) + const entered = new Promise((resolve) => { + readEntered = resolve + }) + let gated = true + adapter.loadSubset = async (...args) => { + if (gated) { + gated = false + readEntered() + await gate + } + return read(...args) + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `released-mid-read`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true, unloadSubset: () => {} } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + const only: LoadSubsetOptions = { limit: 3 } + const pending = collection._sync.loadSubset(only) + + // Released while its own read is still in flight, so by the time that + // read finishes nothing wants the rows any more. + await entered + collection._sync.unloadSubset(only) + openGate() + await pending + + // Recording the demand as hydrated here would leave an entry with no + // holder, which no release can ever remove. This asserts the invariant + // directly rather than the wrong answer it can lead to: reaching that + // needs the apply mutex held by non-hydrating work, which no public API + // on a persisted collection exposes. + const reacquired = collection._sync.loadSubset({ limit: 3 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(collection.get(`3`)).toBeDefined() + } finally { + await collection.cleanup() + } + }) }) From 13fc08fe49f9f21b0178fad182a711a79415fba2 Mon Sep 17 00:00:00 2001 From: Muhammad Amin Saffari Taheri Date: Sun, 20 Sep 2026 08:14:04 +0330 Subject: [PATCH 3/3] fix(db-sqlite-persistence-core): three more fast-path holes from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **A failed reload after a reset left demands marked hydrated.** `truncateAndReloadUnsafe` empties the collection and then refills it, and the refill can fail — it also does not go through the source's `truncate`, so `noteSourceTruncate` never ran for it. The demands are now forgotten before the truncate, so nothing is believed until it has been read again. **A repeat acquisition of the same options object stranded an upstream lease.** `unloadSubset` releases the exact acquisition created for `options`, but the wrapper replaced that record on every call. Two forwards, one release, one lease left held for ever. The record is now reused and a second forward for an object that already holds a lease is skipped. **An in-flight count could be decremented by a load from a previous lifecycle.** `advanceLifecycle` clears the map, but a load that started before it still runs its `finally`; if the current generation had re-entered the same demand, that late call decremented — or deleted — its count, and the fast path could answer while the current load was still settling. Entries carry the generation that created them and a mismatched release is ignored. The first two have failing-first tests; the third needs a lifecycle advance during an in-flight read, which no public API can orchestrate. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/persisted.ts | 57 +++++++-- .../tests/subset-fast-path.test.ts | 112 +++++++++++++++++- 2 files changed, 160 insertions(+), 9 deletions(-) diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 6c6f3d0c32..c4da0e01a5 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -818,7 +818,7 @@ class PersistedCollectionRuntime< */ private readonly loadsInFlight = new Map< ReturnType, - number + { generation: number; count: number } >() private startupSettled = false /** @@ -1106,7 +1106,7 @@ class PersistedCollectionRuntime< if (lifecycleGeneration !== this.lifecycleGeneration) return await this.waitForAppliedReceiptsAfter(appliedCursor) } finally { - this.leaveLoad(demandKey) + this.leaveLoad(demandKey, lifecycleGeneration) } // Only claim the rows are here if nothing invalidated them while we were @@ -1174,13 +1174,36 @@ class PersistedCollectionRuntime< } private enterLoad(key: ReturnType): void { - this.loadsInFlight.set(key, (this.loadsInFlight.get(key) ?? 0) + 1) + const entry = this.loadsInFlight.get(key) + + if (entry && entry.generation === this.lifecycleGeneration) { + entry.count++ + return + } + + this.loadsInFlight.set(key, { + generation: this.lifecycleGeneration, + count: 1, + }) } - private leaveLoad(key: ReturnType): void { - const remaining = (this.loadsInFlight.get(key) ?? 1) - 1 + /** + * `generation` is the one the load started in. + * + * `advanceLifecycle` clears this map, but a load from before it still runs + * its `finally`. Without the check that late call would decrement — or + * delete — a count belonging to the *current* generation, and the fast path + * would then answer while that load was still settling. + */ + private leaveLoad( + key: ReturnType, + generation: number, + ): void { + const entry = this.loadsInFlight.get(key) + + if (!entry || entry.generation !== generation) return - if (remaining > 0) this.loadsInFlight.set(key, remaining) + if (entry.count > 1) entry.count-- else this.loadsInFlight.delete(key) } @@ -1265,7 +1288,9 @@ class PersistedCollectionRuntime< const key = getLoadSubsetDemandKey(options) if (!this.hydratedDemands.has(key)) return false - if (this.loadsInFlight.has(key)) return false + if (this.loadsInFlight.get(key)?.generation === this.lifecycleGeneration) { + return false + } if ((this.demandCoverage.get(key) ?? 0) <= 0) return false // The same registration the slow path performs, so this acquisition owns @@ -2286,6 +2311,12 @@ class PersistedCollectionRuntime< } private async truncateAndReloadUnsafe(): Promise { + // The collection is about to be emptied, and the reload that refills it + // can fail. Nothing below may be believed until it is read again — and + // this path does not go through the source's `truncate`, so + // `noteSourceTruncate` never runs for it. + this.hydratedDemands.clear() + if (this.syncControls.begin && this.syncControls.commit) { this.withInternalApply(() => { this.syncControls.begin?.({ immediate: true }) @@ -2877,7 +2908,11 @@ function createWrappedSyncConfig< runtime.clearSyncControls() }, loadSubset: (options: LoadSubsetOptions): true | Promise => { - const acquisition = { forwarded: false } + // **One record per options object.** `unloadSubset` releases the + // exact acquisition created for `options`, so replacing the record + // on a repeat acquisition would strand the first upstream lease: one + // release would end only the latest. + const acquisition = acquisitions.get(options) ?? { forwarded: false } acquisitions.set(options, acquisition) // Hydration is another async boundary. A release before this point @@ -2892,6 +2927,12 @@ function createWrappedSyncConfig< return true } if (!resolved.loadSubset) return true + + // Already holding an upstream lease for this exact object: a + // second forward would take a lease the single release cannot + // give back. + if (acquisition.forwarded) return true + acquisition.forwarded = true try { // Returning a promise transfers its lease even if it rejects. diff --git a/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts index 98e6cae100..96b589d516 100644 --- a/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts +++ b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts @@ -50,7 +50,7 @@ function createLimitAdapter(initial: Array) { } type CoordinatorHarness = PersistedCollectionCoordinator & { - emit: (payload: TxCommitted, senderId?: string) => void + emit: (payload: TxCommitted | Record, senderId?: string) => void } function createCoordinatorHarness(collectionId: string): CoordinatorHarness { @@ -374,6 +374,116 @@ describe(`subset fast path`, () => { } }) + it(`does not answer from a demand whose reload failed after a reset`, async () => { + const adapter = createLimitAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + ]) + const coordinator = createCoordinatorHarness(`reset-reload-failed`) + const read = adapter.loadSubset.bind(adapter) + let failNextRead = false + adapter.loadSubset = (...args) => { + if (failNextRead) { + failNextRead = false + return Promise.reject(new Error(`store read failed`)) + } + return read(...args) + } + + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const collection = createCollection( + persistedCollectionOptions({ + id: `reset-reload-failed`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { loadSubset: () => true, unloadSubset: () => {} } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + collection.startSyncImmediate() + + try { + await collection._sync.loadSubset({ limit: 2 }) + expect(collection.size).toBe(2) + + // A reset from another node empties the collection and reloads it. The + // reload fails, so the rows are gone and nothing put them back. + failNextRead = true + coordinator.emit({ + type: `collection:reset`, + schemaVersion: 1, + resetEpoch: 1, + }) + await flush() + + expect(collection.size).toBe(0) + + // Answering `true` here would hand a live query rows that are not there. + const reacquired = collection._sync.loadSubset({ limit: 2 }) + expect(reacquired).not.toBe(true) + await reacquired + expect(collection.size).toBe(2) + } finally { + warning.mockRestore() + await collection.cleanup() + } + }) + + it(`keeps upstream loads and releases balanced for one options object`, async () => { + const adapter = createLimitAdapter([ + { id: `1`, title: `one` }, + { id: `2`, title: `two` }, + ]) + let loads = 0 + let releases = 0 + + const collection = createCollection( + persistedCollectionOptions({ + id: `balanced-leases`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: ({ markReady }) => { + markReady() + return { + loadSubset: () => { + loads++ + return true + }, + unloadSubset: () => { + releases++ + }, + } + }, + }, + persistence: { adapter }, + }), + ) + collection.startSyncImmediate() + + try { + // The *same* object twice. `unloadSubset` releases the exact acquisition + // created for it, so a second upstream lease taken here could never be + // given back. + const only: LoadSubsetOptions = { limit: 2 } + + await collection._sync.loadSubset(only) + expect(collection._sync.loadSubset(only)).toBe(true) + expect(loads).toBe(1) + + collection._sync.unloadSubset(only) + expect(releases).toBe(1) + expect(loads).toBe(releases) + } finally { + await collection.cleanup() + } + }) + it(`does not mark a demand hydrated when its last holder left mid-read`, async () => { const adapter = createLimitAdapter([ { id: `1`, title: `one` },