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..c4da0e01a5 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,41 @@ class PersistedCollectionRuntime<
   > = []
   private readonly queuedTxCommitted: Array<TxCommitted> = []
   private readonly requestIds = new WeakMap<LoadSubsetOptions, string>()
+  /**
+   * 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<typeof getLoadSubsetDemandKey>,
+    number
+  >()
+  private readonly hydratedDemands = new Set<
+    ReturnType<typeof getLoadSubsetDemandKey>
+  >()
+  /**
+   * 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<typeof getLoadSubsetDemandKey>,
+    { generation: number; count: number }
+  >()
+  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<T, TKey, PersistedCollectionUtils> | null =
     null
@@ -928,8 +964,13 @@ class PersistedCollectionRuntime<
   private async hydrateBaseline(lifecycleGeneration: number): Promise<void> {
     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.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 +1012,10 @@ class PersistedCollectionRuntime<
     if (this.syncMode !== `on-demand`) {
       await this.hydrateBaseline(lifecycleGeneration)
     }
+
+    if (lifecycleGeneration === this.lifecycleGeneration) {
+      this.startupSettled = true
+    }
   }
 
   private async loadStartupMetadataInternal(
@@ -1045,49 +1090,226 @@ class PersistedCollectionRuntime<
     upstreamLoadSubset?: LoadSubsetFn,
   ): Promise<void> {
     const lifecycleGeneration = this.lifecycleGeneration
-    this.activeSubsets.set(this.getSubsetKey(options), options)
+    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, lifecycleGeneration)
+    }
+
+    // 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 &&
+      (this.demandCoverage.get(demandKey) ?? 0) > 0
+    ) {
+      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<void>,
+  ): true | Promise<void> {
+    if (result === true) return true
+
+    return result.catch((error: unknown) => {
+      this.noteUpstreamLoadFailure(options, error)
+      throw error
+    })
+  }
+
+  private enterLoad(key: ReturnType<typeof getLoadSubsetDemandKey>): void {
+    const entry = this.loadsInFlight.get(key)
+
+    if (entry && entry.generation === this.lifecycleGeneration) {
+      entry.count++
+      return
+    }
+
+    this.loadsInFlight.set(key, {
+      generation: this.lifecycleGeneration,
+      count: 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<typeof getLoadSubsetDemandKey>,
+    generation: number,
+  ): void {
+    const entry = this.loadsInFlight.get(key)
+
+    if (!entry || entry.generation !== generation) return
+
+    if (entry.count > 1) entry.count--
+    else this.loadsInFlight.delete(key)
+  }
+
   unloadSubset(
     options: LoadSubsetOptions,
     upstreamUnloadSubset?: (options: LoadSubsetOptions) => void,
   ): void {
-    this.activeSubsets.delete(this.getSubsetKey(options))
-    this.pendingRemoteSubsetEnsures.delete(this.getSubsetKey(options))
+    const subsetKey = this.getSubsetKey(options)
+    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<void>` 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`.
+   *
+   * 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); 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
+
+    const key = getLoadSubsetDemandKey(options)
+
+    if (!this.hydratedDemands.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
+    // 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<void> {
     const lifecycleGeneration = this.lifecycleGeneration
     // A one-shot refresh does not acquire an enduring subscription lease.
@@ -1232,6 +1454,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())
     }
@@ -1244,6 +1468,10 @@ 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
@@ -2083,6 +2311,12 @@ class PersistedCollectionRuntime<
   }
 
   private async truncateAndReloadUnsafe(): Promise<void> {
+    // 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 })
@@ -2181,6 +2415,7 @@ class PersistedCollectionRuntime<
 
   private async reloadActiveSubsetsUnsafe(): Promise<void> {
     const lifecycleGeneration = this.lifecycleGeneration
+    const sourceTruncateGeneration = this.sourceTruncateGeneration
     const activeSubsetOptions =
       this.activeSubsets.size > 0
         ? Array.from(this.activeSubsets.values())
@@ -2210,6 +2445,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 +2808,7 @@ function createWrappedSyncConfig<
           : undefined,
         truncate: () => {
           if (startupState.cleanedUp) return
+          runtime.noteSourceTruncate()
           const openTransaction = getOpenTransaction()
           if (!openTransaction) {
             params.truncate()
@@ -2629,17 +2882,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 +2907,81 @@ function createWrappedSyncConfig<
           runtime.cleanup()
           runtime.clearSyncControls()
         },
-        loadSubset: async (options: LoadSubsetOptions) => {
-          const acquisition = { forwarded: false }
+        loadSubset: (options: LoadSubsetOptions): true | Promise<void> => {
+          // **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)
-          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<void> => {
+              if (
+                startupState.cleanedUp ||
+                acquisitions.get(options) !== acquisition
+              ) {
+                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.
+                // 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 {
+              // `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.
+              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 +3028,10 @@ function createLoopbackSyncConfig<
           runtime.cleanup()
           runtime.clearSyncControls()
         },
-        loadSubset: (options: LoadSubsetOptions) => runtime.loadSubset(options),
+        loadSubset: (options: LoadSubsetOptions): true | Promise<void> =>
+          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<Todo, string>({
+        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<void>` 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<LoadSubsetOptions> = []
+    let publish!: (row: Todo) => void
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        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<Todo, string>({
+        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<Todo, string>({
+        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<Todo, string>({
+        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..96b589d516
--- /dev/null
+++ b/packages/db-sqlite-persistence-core/tests/subset-fast-path.test.ts
@@ -0,0 +1,552 @@
+import { describe, expect, it, vi } 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<Todo>) {
+  const rows = new Map(initial.map((r) => [r.id, r]))
+  const calls: Array<LoadSubsetOptions> = []
+  const adapter: PersistenceAdapter & {
+    rows: Map<string, Todo>
+    calls: Array<LoadSubsetOptions>
+  } = {
+    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([]),
+    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) {
+        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 | Record<string, unknown>, senderId?: string) => void
+}
+
+function createCoordinatorHarness(collectionId: string): CoordinatorHarness {
+  let subscriber: ((message: ProtocolEnvelope<unknown>) => 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<void> {
+  for (let i = 0; i < times; i++) {
+    await new Promise((resolve) => setTimeout(resolve, 0))
+  }
+}
+
+/**
+ * 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 () => {
+    const adapter = createLimitAdapter([
+      { id: `1`, title: `one` },
+      { id: `2`, title: `two` },
+      { id: `3`, title: `three` },
+    ])
+    const coordinator = createCoordinatorHarness(`sync-present`)
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        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<void>((r) => {
+      releaseReload = r
+    })
+    const entered = new Promise<void>((r) => {
+      reloadEntered = r
+    })
+    adapter.loadSubset = async (...args) => {
+      calls++
+      if (calls === 2) {
+        reloadEntered()
+        await reloadGate
+      }
+      return loadSubset(...args)
+    }
+
+    let truncateSource!: () => void
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        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()
+  })
+
+  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<LoadSubsetOptions> = []
+    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<Todo, string>({
+        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<LoadSubsetOptions> = []
+    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<Todo, string>({
+        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 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<Todo, string>({
+        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<Todo, string>({
+        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` },
+      { id: `2`, title: `two` },
+      { id: `3`, title: `three` },
+    ])
+    const read = adapter.loadSubset.bind(adapter)
+    let openGate!: () => void
+    let readEntered!: () => void
+    const gate = new Promise<void>((resolve) => {
+      openGate = resolve
+    })
+    const entered = new Promise<void>((resolve) => {
+      readEntered = resolve
+    })
+    let gated = true
+    adapter.loadSubset = async (...args) => {
+      if (gated) {
+        gated = false
+        readEntered()
+        await gate
+      }
+      return read(...args)
+    }
+
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        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()
+    }
+  })
+})