fix(db): report incremental subset errors

.changeset/report-incremental-subset-errors.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/db': patch
+---
+
+Report incremental subset-load failures through subscriptions, live-query utilities, and effects while keeping cached source rows available.

docs/guides/error-handling.md

@@ -119,6 +119,42 @@ Error tracking methods:
 - **`errorCount`**: Returns the number of consecutive sync failures. This counter is incremented only when queries fail completely (not per retry attempt) and is reset on successful queries:
 - **`clearError()`**: Clears the error state and triggers a refetch of the query. This method resets both `lastError` and `errorCount`:
 
+## Incremental Subset Load Errors
+
+An incremental `loadSubset` failure does not discard rows that are already
+available or put the shared source collection into `error`. The failure belongs
+to the subscription that requested that subset:
+
+```ts
+const subscription = todoCollection.subscribeChanges(handleChanges, {
+  includeInitialState: false,
+})
+
+subscription.on('loadSubset:error', ({ error, options }) => {
+  console.error('Subset failed', options, error)
+})
+
+subscription.requestSnapshot()
+
+// The most recent failure remains available for diagnostics.
+console.log(subscription.lastError)
+```
+
+For ordered live queries, `utils.setWindow()` rejects with the same error. The
+last failure is also available as `utils.lastSubsetError`, while the last
+successful snapshot remains readable:
+
+```ts
+try {
+  await liveTodos.utils.setWindow({ offset: 0, limit: 100 })
+} catch (error) {
+  console.error(liveTodos.utils.lastSubsetError)
+}
+```
+
+Effects report subset failures through `onSourceError` and dispose because
+their incremental result can no longer be kept complete.
+
 ## Collection Status and Error States
 
 Collections track their status and transition between states:

packages/db/skills/db-core/custom-adapter/SKILL.md

@@ -198,6 +198,10 @@ return the fetched rows. `parseLoadSubsetOptions()` returns only `filters`,
 opaque backend cursor; translate or combine those expressions for your API.
 Return `unloadSubset` only when `loadSubset` creates an ongoing resource, such
 as a per-subset server subscription, that must be released.
+Ownership transfers to core only when `loadSubset` returns `true` or a promise.
+If it throws synchronously after partial setup, release that partial resource
+before throwing; core will not call `unloadSubset` for a request that never
+returned.
 
 ### Managing optimistic state duration
 

packages/db/src/collection/changes.ts

@@ -222,9 +222,6 @@ export class CollectionChangesManager<
     ) => void,
     options: SubscribeChangesOptions<TOutput, TKey> = {},
   ): CollectionSubscription {
-    // Start sync and track subscriber
-    this.addSubscriber()
-
     // Compile where callback to whereExpression if provided
     if (options.where && options.whereExpression) {
       throw new Error(
@@ -240,37 +237,56 @@ export class CollectionChangesManager<
       whereExpression = toExpression(result)
     }
 
-    const subscription = new CollectionSubscription(this.collection, callback, {
-      ...opts,
-      whereExpression,
-      onUnsubscribe: () => {
-        this.removeSubscriber()
-        this.changeSubscriptions.delete(subscription)
-      },
-    })
-
-    // Register status listener BEFORE requesting snapshot to avoid race condition.
-    // This ensures the listener catches all status transitions, even if the
-    // loadSubset promise resolves synchronously or very quickly.
-    if (options.onStatusChange) {
-      subscription.on(`status:change`, options.onStatusChange)
-    }
+    // Acquire ownership only after all fallible option validation and
+    // user-provided predicate compilation has completed.
+    this.addSubscriber()
 
-    if (options.includeInitialState) {
-      subscription.requestSnapshot({
-        trackLoadSubsetPromise: false,
-        orderBy: options.orderBy,
-        limit: options.limit,
-        onLoadSubsetResult: options.onLoadSubsetResult,
+    let subscription: CollectionSubscription | undefined
+    try {
+      subscription = new CollectionSubscription(this.collection, callback, {
+        ...opts,
+        whereExpression,
+        onUnsubscribe: () => {
+          this.removeSubscriber()
+          if (subscription) this.changeSubscriptions.delete(subscription)
+        },
       })
-    } else if (options.includeInitialState === false) {
-      // When explicitly set to false (not just undefined), mark all state as "seen"
-      // so that all future changes (including deletes) pass through unfiltered.
-      subscription.markAllStateAsSeen()
-    }
 
-    // Add to batched listeners
-    this.changeSubscriptions.add(subscription)
+      // Register status listener BEFORE requesting snapshot to avoid race condition.
+      // This ensures the listener catches all status transitions, even if the
+      // loadSubset promise resolves synchronously or very quickly.
+      if (options.onStatusChange) {
+        subscription.on(`status:change`, options.onStatusChange)
+      }
+
+      if (options.includeInitialState) {
+        subscription.requestSnapshot({
+          trackLoadSubsetPromise: false,
+          orderBy: options.orderBy,
+          limit: options.limit,
+          onLoadSubsetResult: options.onLoadSubsetResult,
+        })
+      } else if (options.includeInitialState === false) {
+        // When explicitly set to false (not just undefined), mark all state as "seen"
+        // so that all future changes (including deletes) pass through unfiltered.
+        subscription.markAllStateAsSeen()
+      }
+
+      // Add to batched listeners
+      this.changeSubscriptions.add(subscription)
+    } catch (error) {
+      if (subscription) {
+        try {
+          subscription.unsubscribe()
+        } catch {
+          // Preserve the setup error. Cleanup still releases subscriber
+          // ownership and attempts every subset unload before it throws.
+        }
+      } else {
+        this.removeSubscriber()
+      }
+      throw error
+    }
 
     return subscription
   }
@@ -283,12 +299,20 @@ export class CollectionChangesManager<
     this.activeSubscribersCount++
     this.lifecycle.cancelGCTimer()
 
-    // Start sync if collection was cleaned up
-    if (
-      this.lifecycle.status === `cleaned-up` ||
-      this.lifecycle.status === `idle`
-    ) {
-      this.sync.startSync()
+    try {
+      // Start sync if collection was cleaned up
+      if (
+        this.lifecycle.status === `cleaned-up` ||
+        this.lifecycle.status === `idle`
+      ) {
+        this.sync.startSync()
+      }
+    } catch (error) {
+      this.activeSubscribersCount = previousSubscriberCount
+      if (this.activeSubscribersCount === 0) {
+        this.lifecycle.startGCTimer()
+      }
+      throw error
     }
 
     this.events.emitSubscribersChange(