fix(db): recover a live query's demand when its source recovers from error

.changeset/source-error-demand-recovers.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/db': patch
+---
+
+A live query now recovers when a source Collection errors and then becomes ready again. Before, a subset load that rejected because its source entered `error` put a query with a join or include into a terminal error, and a direct query became ready on recovery without loading its subset again. The load is now requested again when the source recovers, and the query becomes ready when that load settles. A load that rejects while its source is healthy still puts the query in error.

docs/contributing/oracle-reviews/progressive-source-readiness.md

@@ -297,20 +297,33 @@ outcome of every async hold.
 
 ## Open cells
 
-- TrailBase on-demand: a change committed between a subset fetch and the
-  change subscription can be missed. The owner is
-  `packages/trailbase-db-collection/tests/ORACLE.md`. This predates the change.
-
+- **Fixed: TrailBase on-demand.** A load now waits for the change
+  subscription before it fetches, so a query is not ready before the
+  subscription and a change after the fetch is kept.
+  `packages/trailbase-db-collection/tests/ORACLE.md` owns the law and its
+  remaining limits.
 - Effect `skipInitial` over an ordered window or an include child has no lane.
 - Vue, Svelte, and Angular have no Suspense surface in the shared suite.
-- A persisted follower tab's load may settle before the leader's rows apply
-  in that tab. Only the two-tab OPFS e2e suite can show this; it has not run.
-- Electron IPC, two on-demand roots, and include children apart from lazy
-  joins are not exercised.
-- Whether a load that was waiting at a retried stream error is a real failure
-  is a product decision.
+- **Fixed: persisted follower tabs.** A follower's load settled at the
+  leader's reply, before the follower applied the leader's broadcast commit.
+  It now waits for that apply. The two-tab OPFS e2e suite fails 3 of 3 runs on
+  the original code and passes with the fix.
+- **Exercised: Electron IPC.** Five witnesses pass in-process and through a
+  real Electron main process. A follower renderer uses the same coordinator as
+  a follower tab and has no two-renderer witness.
+- Two on-demand roots and include children apart from lazy joins are not
+  exercised.
+- **Decided: a retried stream error that succeeds is a success.** Electric
+  keeps waiting loads through an `onError` retry. A load that rejects while
+  its source is in `error` now detaches and is acquired again when the source
+  recovers (ARCHITECTURE.md law 16), for root and lazy demand alike. Before,
+  lazy demand entered terminal error, and a root query became ready on
+  recovery without reacquiring its demand. Model D's source-error histories
+  fail on the original code (80 cells). Open: an Effect still disposes on a
+  source error; one-shot reads reject with it; the ordered window has no
+  source-error history.
 - The Electric e2e progressive suite has no run receipt (see the coverage map).
-- A rejected acquisition or a source error while a source lags is outside
-  Model D.
+- A real rejection while a source lags is a Model D `fail` history, and a
+  source error is a `source-error` history. Both cover live queries only.
 - Before the full sync, an Electric progressive query's rows do not receive
   live changes. The source contract owns that behavior, not this law.

packages/db/src/collection/subscription.ts

@@ -113,6 +113,8 @@ type SubsetDemand = {
   acquisition: SubsetAcquisitionRecord
   acquisitionState: `starting` | `active` | `detached`
   initialResult?: Deferred<void>
+  /** Detached by a source error; recovery acquires it again. */
+  awaitsErrorRecovery?: true
 }
 
 type TruncateReplayAttempt = {
@@ -145,6 +147,26 @@ function cancelAcquisition(acquisition: SubsetAcquisitionRecord): void {
   acquisition.removeRequestAbortListener?.()
 }
 
+/**
+ * Wait for detached demand to be acquired again: the restart replay settles
+ * the wait, and release or abort rejects it with `AbortError`.
+ */
+function awaitRecovery(demand: SubsetDemand): Promise<void> {
+  if (demand.initialResult) return demand.initialResult.promise
+  const initialResult = createDeferred<void>()
+  demand.initialResult = initialResult
+  const signal = demand.requestOptions.signal
+  const abort = () =>
+    initialResult.reject(new LoadSubsetOperationAbortedError())
+  signal?.addEventListener(`abort`, abort, { once: true })
+  const finish = () => {
+    signal?.removeEventListener(`abort`, abort)
+    demand.initialResult = undefined
+  }
+  void initialResult.promise.then(finish, finish)
+  return initialResult.promise
+}
+
 export class CollectionSubscription
   extends EventEmitter<SubscriptionEvents>
   implements Subscription
@@ -371,6 +393,8 @@ export class CollectionSubscription
     )
     const attempt = replayState.currentAttempt
     this.truncateReplayState = replayState
+    // From here the replay's own loading status holds dependent readiness.
+    for (const demand of demands) delete demand.awaitsErrorRecovery
     this.setStatus(`loadingSubset`)
     if (this.truncateReplayState !== replayState) return
     this.startTruncateReplayAttempt(replayState, attempt, demands)
@@ -895,6 +919,7 @@ export class CollectionSubscription
     if (!(syncResult instanceof Promise)) return
 
     const syncRunGeneration = this.collection._sync.getSyncRunGeneration()
+    const acquisition = demand.acquisition
     const participant = { demand, promise: syncResult }
 
     if (trackStatus) {
@@ -912,6 +937,10 @@ export class CollectionSubscription
     }
 
     void syncResult.then(finish, (error: unknown) => {
+      if (this.detachForErrorRecovery(demand, acquisition)) {
+        finish()
+        return
+      }
       if (
         this.isSyncRunGenerationCurrent(syncRunGeneration) &&
         shouldReportError()
@@ -925,6 +954,53 @@ export class CollectionSubscription
     })
   }
 
+  /**
+   * A load that fails while its source Collection is in error belongs to that
+   * source error, not to the request. Its demand detaches, and recovery
+   * acquires it again, as for demand requested during the error.
+   */
+  private detachForErrorRecovery(
+    demand: SubsetDemand,
+    acquisition: SubsetAcquisitionRecord,
+  ): boolean {
+    if (
+      this.collection.status !== `error` ||
+      demand.acquisition !== acquisition ||
+      demand.acquisitionState !== `active` ||
+      !this.isDemandActive(demand) ||
+      !this.isSyncRunGenerationCurrent(acquisition.syncRunGeneration) ||
+      demand.requestOptions.signal?.aborted
+    ) {
+      return false
+    }
+    cancelAcquisition(acquisition)
+    demand.acquisitionState = `detached`
+    demand.acquisition = {
+      options: demand.requestOptions,
+      syncRunGeneration: acquisition.syncRunGeneration,
+    }
+    demand.awaitsErrorRecovery = true
+    void awaitRecovery(demand)
+    return true
+  }
+
+  /**
+   * Whether demand detached by a source error waits for its restart. A
+   * dependent's readiness check can run on the source's `ready` before this
+   * subscription starts the restart replay.
+   */
+  public get awaitsErrorRecovery(): boolean {
+    return this.subsetDemands.some((demand) => demand.awaitsErrorRecovery)
+  }
+
+  /** Settles when every demand detached by a source error is acquired again. */
+  public get pendingErrorRecovery(): Promise<void> | undefined {
+    const waits = this.subsetDemands.flatMap((demand) =>
+      demand.awaitsErrorRecovery ? [awaitRecovery(demand)] : [],
+    )
+    return waits.length > 0 ? Promise.all(waits).then(() => {}) : undefined
+  }
+
   /** Give every logical observer of one transport rejection the same Error. */
   private normalizeLoadSubsetPromiseError(
     promise: Promise<unknown>,
@@ -1049,17 +1125,7 @@ export class CollectionSubscription
     ) {
       demand.acquisitionState = `detached`
       this.subsetDemands.push(demand)
-      const initialResult = createDeferred<void>()
-      demand.initialResult = initialResult
-      const abort = () =>
-        initialResult.reject(new LoadSubsetOperationAbortedError())
-      requestOptions.signal?.addEventListener(`abort`, abort, { once: true })
-      const finish = () => {
-        requestOptions.signal?.removeEventListener(`abort`, abort)
-        demand.initialResult = undefined
-      }
-      void initialResult.promise.then(finish, finish)
-      return { demand, result: initialResult.promise, started: false }
+      return { demand, result: awaitRecovery(demand), started: false }
     }
     const acquisition = this.createSubsetAcquisitionRecord(demand)
     demand.acquisition = acquisition

packages/db/src/query/live/ARCHITECTURE.md

@@ -830,6 +830,22 @@ demand; releasing it before recovery creates no physical acquisition or unload.
 Queued reacquisition must not retry a failed attempt merely because both
 loading and ready notifications scheduled it.
 
+A source Collection's `error` status is recoverable within its sync run, so a
+load that rejects while its source is in `error` belongs to that source error,
+not to the request. The subscription detaches that demand, and the source's
+next `ready` in the same sync run acquires it again through a fresh replay
+barrier. A source that reports a final stream error rejects its loads and then
+enters `error` in the same synchronous turn, so the rejection is classified
+after the status change. Root and lazy demand recover alike: neither puts the
+live query in terminal error, and initial-query readiness after recovery waits
+for the reacquired demand to settle. The live query's readiness check can run
+on the source's `ready` before the subscription starts that replay, so a
+subscription reports detached demand that awaits recovery until the replay
+starts. A preload that already rejected with the source error stays rejected.
+A load that rejects while its source is healthy is a query failure: it puts
+the live query in terminal error, as before. An Effect still disposes on a
+source error.
+
 Requests waiting for a loader report one pending promise synchronously through
 `onLoadSubsetResult`, including requests made during initial error or after
 cleanup. The callback is not delayed until acquisition: query callers capture
@@ -1555,6 +1571,12 @@ create recursive Collection machinery.
     counts as a preload. A deferring live query keeps its sources from
     garbage collection but is not counted in their `subscriberCount` or
     `subscribers:change`, which adapters read as a request for live data.
+16. **Source-error recovery:** a load that rejects while its source Collection
+    is in `error` cannot put a live query in terminal error. When the source
+    becomes ready again in the same sync run, its root and lazy demand are
+    acquired again, and initial-query readiness waits for that demand to
+    settle. A load that rejects while its source is healthy remains a terminal
+    query failure.
 
 ## Glossary
 

packages/db/src/query/live/collection-config-builder.ts

@@ -1178,7 +1178,12 @@ export class CollectionConfigBuilder<
     const allDemandsSettled = [...this.activeDemands.values()].every(
       (demand) => demand.settled,
     )
-    const isLoading = this.liveQueryCollection?.isLoadingSubset
+    const isLoading =
+      this.liveQueryCollection?.isLoadingSubset ||
+      // A recovered source acquires detached demand again after this check.
+      Object.values(this.subscriptions).some(
+        (subscription) => subscription.awaitsErrorRecovery,
+      )
     // Mark ready when:
     // 1. All subscriptions are set up (subscribedToAllCollections)
     // 2. All source collections are ready

packages/db/src/query/live/collection-subscriber.ts

@@ -216,9 +216,12 @@ export class CollectionSubscriber<
         void ready.then(
           () => this.collectionConfigBuilder.settleDemand(plan.id, generation),
           (error) => {
-            // A truncate replay aborts the acquisition it supersedes. Its
-            // replacement decides the demand; a real failure rejects it too.
-            const replacement = subscription.pendingTruncateReplacement
+            // A truncate replay aborts the acquisition it supersedes, and a
+            // source error detaches it until recovery. The replacement
+            // decides the demand; a real failure rejects it too.
+            const replacement =
+              subscription.pendingTruncateReplacement ??
+              subscription.pendingErrorRecovery
             if (replacement) settleWith(replacement)
             else
               this.collectionConfigBuilder.failDemand(