feat(db): preserve applied loadSubset outcomes

.changeset/add-load-subset-outcomes.md

@@ -0,0 +1,6 @@
+---
+'@tanstack/db': patch
+'@tanstack/db-sqlite-persistence-core': patch
+---
+
+Allow `loadSubset` adapters to report whether more rows exist and preserve applied, request-scoped outcomes through live-query demand, persistence, and window coordination.

packages/db-sqlite-persistence-core/src/persisted.ts

@@ -21,7 +21,9 @@ import type {
   CollectionIndexMetadata,
   DeleteMutationFnParams,
   InsertMutationFnParams,
+  LoadSubsetFn,
   LoadSubsetOptions,
+  LoadSubsetResult,
   PendingMutation,
   SyncAppliedReceipt,
   SyncConfig,
@@ -1015,8 +1017,8 @@ class PersistedCollectionRuntime<
 
   async loadSubset(
     options: LoadSubsetOptions,
-    upstreamLoadSubset?: (options: LoadSubsetOptions) => true | Promise<void>,
-  ): Promise<void> {
+    upstreamLoadSubset?: LoadSubsetFn,
+  ): Promise<void | LoadSubsetResult> {
     this.activeSubsets.set(this.getSubsetKey(options), options)
 
     const appliedCursor = this.appliedReceiptSequence
@@ -1031,12 +1033,13 @@ class PersistedCollectionRuntime<
       try {
         const maybePromise = upstreamLoadSubset(options)
         if (maybePromise instanceof Promise) {
-          await maybePromise.catch((error) => {
+          return await maybePromise.catch((error) => {
             console.warn(
               `Failed to load remote subset in persisted wrapper:`,
               error,
             )
             this.queueRemoteSubsetEnsure(options)
+            return undefined
           })
         }
       } catch (error) {
@@ -2602,7 +2605,7 @@ function createWrappedSyncConfig<
           if (startupState.cleanedUp || cancelledLoadKeys.has(loadKey)) {
             return
           }
-          await runtime.loadSubset(options, resolvedSourceResult.loadSubset)
+          return runtime.loadSubset(options, resolvedSourceResult.loadSubset)
         },
         unloadSubset: (options: LoadSubsetOptions) => {
           cancelledLoadKeys.add(getLoadKey(options))

packages/db-sqlite-persistence-core/tests/persisted.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'
 import {
   BasicIndex,
   DbClient,
+  DeduplicatedLoadSubset,
   IR,
   collectionOptions,
   createCollection,
@@ -27,13 +28,25 @@ import type {
   PullSinceResponse,
   TxCommitted,
 } from '../src'
-import type { LoadSubsetOptions, SyncConfig } from '@tanstack/db'
+import type {
+  AppliedLoadSubsetOutcome,
+  LoadSubsetOptions,
+  SyncConfig,
+} from '@tanstack/db'
 
 type Todo = {
   id: string
   title: string
 }
 
+type LoadSubsetTestCollection = {
+  _sync: {
+    loadSubset: (
+      options: LoadSubsetOptions,
+    ) => true | Promise<AppliedLoadSubsetOutcome>
+  }
+}
+
 type RecordingAdapter = PersistenceAdapter & {
   applyCommittedTxCalls: Array<{
     collectionId: string
@@ -1790,6 +1803,125 @@ describe(`persistedCollectionOptions`, () => {
     expect(ensureCalls).toBeGreaterThanOrEqual(2)
   })
 
+  it(`preserves authoritative source extent through persistence`, async () => {
+    const adapter = createRecordingAdapter()
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        id: `sync-present-source-extent`,
+        syncMode: `on-demand`,
+        getKey: (item) => item.id,
+        sync: {
+          sync: ({ markReady }) => {
+            markReady()
+            return {
+              loadSubset: () => Promise.resolve({ hasMore: false }),
+            }
+          },
+        },
+        persistence: {
+          adapter,
+          coordinator: createCoordinatorHarness(),
+        },
+      }),
+    )
+
+    collection.startSyncImmediate()
+    await flushAsyncWork()
+
+    const sync = (collection as unknown as LoadSubsetTestCollection)._sync
+    const outcome = await sync.loadSubset({ limit: 1 })
+
+    expect(outcome).not.toBe(true)
+    if (outcome !== true) {
+      expect(outcome.extent).toBe(`exhausted`)
+    }
+  })
+
+  it(`preserves exact physical-request provenance through persistence`, async () => {
+    const adapter = createRecordingAdapter()
+    let resolveLoad!: (result: { hasMore: boolean }) => void
+    const load = new Promise<{ hasMore: boolean }>((resolve) => {
+      resolveLoad = resolve
+    })
+    const physicalLimits: Array<number | undefined> = []
+    const deduplicated = new DeduplicatedLoadSubset({
+      loadSubset: (options) => {
+        physicalLimits.push(options.limit)
+        return load
+      },
+    })
+    let upstreamCalls = 0
+    let resolveFirstUpstream!: () => void
+    const firstUpstream = new Promise<void>((resolve) => {
+      resolveFirstUpstream = resolve
+    })
+    const upstreamHasMore = new Map<number | undefined, boolean | undefined>()
+    let resolveSecondUpstream!: () => void
+    const secondUpstream = new Promise<void>((resolve) => {
+      resolveSecondUpstream = resolve
+    })
+    const collection = createCollection(
+      persistedCollectionOptions<Todo, string>({
+        id: `sync-present-exact-source-extent`,
+        syncMode: `on-demand`,
+        getKey: (item) => item.id,
+        sync: {
+          sync: ({ markReady }) => {
+            markReady()
+            return {
+              loadSubset: async (options) => {
+                upstreamCalls++
+                const result = deduplicated.loadSubset(options)
+                if (upstreamCalls === 1) resolveFirstUpstream()
+                if (upstreamCalls === 2) resolveSecondUpstream()
+                if (result === true) return undefined
+                const sourceResult = await result
+                upstreamHasMore.set(options.limit, sourceResult?.hasMore)
+                return sourceResult === undefined
+                  ? undefined
+                  : { hasMore: sourceResult.hasMore }
+              },
+            }
+          },
+        },
+        persistence: {
+          adapter,
+          coordinator: createCoordinatorHarness(),
+        },
+      }),
+    )
+
+    try {
+      collection.startSyncImmediate()
+      await flushAsyncWork()
+
+      const sync = (collection as unknown as LoadSubsetTestCollection)._sync
+      const covering = sync.loadSubset({ limit: 10 })
+      await firstUpstream
+      const narrower = sync.loadSubset({ limit: 5 })
+
+      await secondUpstream
+      expect(physicalLimits).toEqual([10])
+      resolveLoad({ hasMore: false })
+
+      const [coveringOutcome, narrowerOutcome] = await Promise.all([
+        covering,
+        narrower,
+      ])
+      expect(upstreamHasMore).toEqual(
+        new Map([
+          [10, false],
+          [5, undefined],
+        ]),
+      )
+      expect(coveringOutcome).toMatchObject({ extent: `exhausted` })
+      expect(narrowerOutcome).toMatchObject({ extent: `unknown` })
+    } finally {
+      resolveLoad({ hasMore: false })
+      await collection.cleanup()
+    }
+  })
+
   it(`fails sync-absent persistence when follower ack omits mutation ids`, async () => {
     const adapter = createRecordingAdapter()
     const coordinator: PersistedCollectionCoordinator = {

packages/db/src/collection/subscription.ts

@@ -14,6 +14,7 @@ import type { IndexInterface } from '../indexes/base-index.js'
 import type {
   ChangeMessage,
   LoadSubsetOptions,
+  LoadSubsetRequestResult,
   Subscription,
   SubscriptionEvents,
   SubscriptionLoadSubsetErrorEvent,
@@ -31,8 +32,8 @@ type RequestSnapshotOptions = {
   orderBy?: OrderBy
   /** Optional limit to pass to loadSubset for backend optimization */
   limit?: number
-  /** Callback that receives the raw loadSubset result for external tracking */
-  onLoadSubsetResult?: (result: Promise<void> | true) => void
+  /** Callback that receives the normalized loadSubset result for internal tracking */
+  onLoadSubsetResult?: (result: LoadSubsetRequestResult) => void
   /** Called when the local snapshot must fall back from an index to a scan. */
   onUnoptimized?: () => void
 }
@@ -46,8 +47,8 @@ type RequestLimitedSnapshotOptions = {
   offset?: number
   /** Whether to track the loadSubset promise on this subscription (default: true) */
   trackLoadSubsetPromise?: boolean
-  /** Callback that receives the raw loadSubset result for external tracking */
-  onLoadSubsetResult?: (result: Promise<void> | true) => void
+  /** Callback that receives the normalized loadSubset result for internal tracking */
+  onLoadSubsetResult?: (result: LoadSubsetRequestResult) => void
 }
 
 type CollectionSubscriptionOptions = {
@@ -77,10 +78,11 @@ type SubsetAcquisition = {
 
 type SubsetDemand = SubsetAcquisition & {
   requestOptions: LoadSubsetOptions
+  releaseFailed: boolean
 }
 
 type TruncateReplayAttempt = {
-  pending: Set<{ promise: Promise<void> }>
+  pending: Set<{ promise: Promise<unknown> }>
   failed: boolean
   setupComplete: boolean
 }
@@ -135,7 +137,7 @@ export class CollectionSubscription
   // Status tracking
   private _status: SubscriptionStatus = `ready`
   private _lastError: unknown | undefined
-  private pendingLoadSubsetPromises: Set<Promise<void>> = new Set()
+  private pendingLoadSubsetPromises: Set<Promise<unknown>> = new Set()
 
   // Cleanup function for truncate event listener
   private truncateCleanup: (() => void) | undefined
@@ -276,7 +278,7 @@ export class CollectionSubscription
           this.truncateReplaySession === session &&
           session.currentAttempt === attempt
         const nextAcquisition = this.createSubsetAcquisition(demand)
-        let syncResult: Promise<void> | true
+        let syncResult: LoadSubsetRequestResult
         try {
           syncResult = this.loadSubset(
             nextAcquisition.options,
@@ -346,7 +348,7 @@ export class CollectionSubscription
   private settleTruncateReplay(
     session: TruncateReplaySession,
     attempt: TruncateReplayAttempt,
-    pending: { promise: Promise<void> },
+    pending: { promise: Promise<unknown> },
   ): void {
     if (this.truncateReplaySession !== session) return
     attempt.pending.delete(pending)
@@ -510,7 +512,7 @@ export class CollectionSubscription
 
   /** Observe an asynchronous subset load and restore status on settlement. */
   private observeLoadSubsetResult(
-    syncResult: Promise<void> | true,
+    syncResult: LoadSubsetRequestResult,
     options: LoadSubsetOptions,
     trackStatus: boolean,
     shouldReportError: () => boolean = () => true,
@@ -540,7 +542,7 @@ export class CollectionSubscription
   private loadSubset(
     options: LoadSubsetOptions,
     shouldReportError: () => boolean = () => true,
-  ): Promise<void> | true {
+  ): LoadSubsetRequestResult {
     try {
       return this.collection._sync.loadSubset(options)
     } catch (error) {
@@ -595,6 +597,10 @@ export class CollectionSubscription
     demand.abortController?.abort()
     try {
       this.collection._sync.unloadSubset(demand.options)
+      demand.releaseFailed = false
+    } catch (error) {
+      demand.releaseFailed = true
+      throw error
     } finally {
       demand.removeRequestAbortListener?.()
     }
@@ -603,23 +609,30 @@ export class CollectionSubscription
   /** Start and retain the first acquisition for one logical subset demand. */
   private startSubsetDemand(requestOptions: LoadSubsetOptions): {
     demand: SubsetDemand
-    result: Promise<void> | true
+    result: LoadSubsetRequestResult
   } {
     const demand: SubsetDemand = {
       requestOptions,
       options: requestOptions,
+      releaseFailed: false,
     }
     const acquisition = this.createSubsetAcquisition(demand)
+    demand.options = acquisition.options
+    demand.abortController = acquisition.abortController
+    demand.removeRequestAbortListener = acquisition.removeRequestAbortListener
+    // Reentrant release must see the exact acquisition before adapter work
+    // starts. A genuine load throw removes this tentative logical owner below.
+    this.subsetDemands.push(demand)
     try {
       const result = this.loadSubset(acquisition.options)
-      demand.options = acquisition.options
-      demand.abortController = acquisition.abortController
-      demand.removeRequestAbortListener = acquisition.removeRequestAbortListener
-      this.subsetDemands.push(demand)
       return { demand, result }
     } catch (error) {
-      acquisition.abortController.abort()
-      acquisition.removeRequestAbortListener?.()
+      const demandIndex = this.subsetDemands.indexOf(demand)
+      if (demandIndex !== -1 && !demand.releaseFailed) {
+        this.subsetDemands.splice(demandIndex, 1)
+        acquisition.abortController.abort()
+        acquisition.removeRequestAbortListener?.()
+      }
       throw error
     }
   }
@@ -776,8 +789,10 @@ export class CollectionSubscription
     )
     if (index === -1) return
 
-    const [demand] = this.subsetDemands.splice(index, 1)
-    if (demand) this.releaseSubsetDemand(demand)
+    const demand = this.subsetDemands[index]
+    if (!demand) return
+    this.releaseSubsetDemand(demand)
+    this.subsetDemands.splice(index, 1)
   }
 
   /**
@@ -1158,14 +1173,16 @@ export class CollectionSubscription
     this.stalePublishedRows.clear()
 
     // Release the current adapter acquisition for each logical subset demand.
+    const failedDemands: Array<SubsetDemand> = []
     for (const demand of this.subsetDemands) {
       try {
         this.releaseSubsetDemand(demand)
       } catch (error) {
         firstCleanupError ??= error
+        failedDemands.push(demand)
       }
     }
-    this.subsetDemands = []
+    this.subsetDemands = failedDemands
 
     try {
       this.emitInner(`unsubscribed`, {