Fix root cause of Electric infinite loop with ORDER BY/LIMIT queries

The infinite loop occurred because `loadMoreIfNeeded` kept trying to load
data even when the local index was exhausted. This happened when:
1. TopK had fewer items than limit (WHERE filtered out data)
2. loadMoreIfNeeded tried to load more → no local data found
3. Loop continued indefinitely since TopK still needed data

Root cause fix:
- Add `localIndexExhausted` flag to CollectionSubscriber
- Track when local index has no more data for current cursor
- Stop calling loadMoreIfNeeded when exhausted
- Reset flag when new data arrives from sync layer (inserts)
- requestLimitedSnapshot now returns boolean indicating if data was found

Error handling improvements (per review feedback):
- D2.run() now throws Error when iteration limit exceeded
- Caller catches and calls transitionToError() for proper error state
- requestLimitedSnapshot returns false when iteration limit hit
- Live query properly shows error state if safeguard limits are hit

This fixes the issue for both eager and progressive syncMode.

Author: claude

packages/db-ivm/src/d2.ts

@@ -64,14 +64,11 @@ export class D2 implements ID2 {
     let iterations = 0
 
     while (this.pendingWork()) {
-      iterations++
-      if (iterations > MAX_RUN_ITERATIONS) {
-        console.error(
-          `[D2 Graph] Execution exceeded ${MAX_RUN_ITERATIONS} iterations. ` +
-            `This may indicate an infinite loop in the dataflow graph. ` +
-            `Breaking out to prevent app freeze.`,
+      if (++iterations > MAX_RUN_ITERATIONS) {
+        throw new Error(
+          `D2 graph execution exceeded ${MAX_RUN_ITERATIONS} iterations. ` +
+            `This may indicate an infinite loop in the dataflow graph.`,
         )
-        break
       }
       this.step()
     }

packages/db/src/collection/subscription.ts

@@ -410,13 +410,15 @@ export class CollectionSubscription
    * Note 1: it may load more rows than the provided LIMIT because it loads all values equal to the first cursor value + limit values greater.
    *         This is needed to ensure that it does not accidentally skip duplicate values when the limit falls in the middle of some duplicated values.
    * Note 2: it does not send keys that have already been sent before.
+   *
+   * @returns true if local data was found and sent, false if the local index was exhausted
    */
   requestLimitedSnapshot({
     orderBy,
     limit,
     minValues,
     offset,
-  }: RequestLimitedSnapshotOptions) {
+  }: RequestLimitedSnapshotOptions): boolean {
     if (!limit) throw new Error(`limit is required`)
 
     if (!this.orderByIndex) {
@@ -508,15 +510,16 @@ export class CollectionSubscription
     // for any legitimate use case.
     const MAX_SNAPSHOT_ITERATIONS = 10000
     let snapshotIterations = 0
+    let hitIterationLimit = false
 
     while (valuesNeeded() > 0 && !collectionExhausted()) {
-      snapshotIterations++
-      if (snapshotIterations > MAX_SNAPSHOT_ITERATIONS) {
+      if (++snapshotIterations > MAX_SNAPSHOT_ITERATIONS) {
         console.error(
           `[TanStack DB] requestLimitedSnapshot exceeded ${MAX_SNAPSHOT_ITERATIONS} iterations. ` +
             `This may indicate an infinite loop in index iteration or filtering. ` +
             `Breaking out to prevent app freeze. Collection: ${this.collection.id}`,
         )
+        hitIterationLimit = true
         break
       }
 
@@ -611,6 +614,14 @@ export class CollectionSubscription
     // Track this loadSubset call
     this.loadedSubsets.push(loadOptions)
     this.trackLoadSubsetPromise(syncResult)
+
+    // Return whether local data was found and iteration completed normally.
+    // Return false if:
+    // - No local data was found (index exhausted)
+    // - Iteration limit was hit (abnormal exit)
+    // Either case signals that the caller should stop trying to load more.
+    // The async loadSubset may still return data later.
+    return changes.length > 0 && !hitIterationLimit
   }
 
   // TODO: also add similar test but that checks that it can also load it from the collection's loadSubset function

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

@@ -346,18 +346,25 @@ export class CollectionConfigBuilder<
         let iterations = 0
 
         while (syncState.graph.pendingWork()) {
-          iterations++
-          if (iterations > MAX_GRAPH_ITERATIONS) {
-            console.error(
-              `[TanStack DB] Graph execution exceeded ${MAX_GRAPH_ITERATIONS} iterations. ` +
-                `This may indicate an infinite loop caused by data loading triggering ` +
-                `continuous graph updates. Breaking out of the loop to prevent app freeze. ` +
-                `Query ID: ${this.id}`,
+          if (++iterations > MAX_GRAPH_ITERATIONS) {
+            this.transitionToError(
+              `Graph execution exceeded ${MAX_GRAPH_ITERATIONS} iterations. ` +
+                `This likely indicates an infinite loop caused by data loading ` +
+                `triggering continuous graph updates.`,
             )
-            break
+            return
           }
 
-          syncState.graph.run()
+          try {
+            syncState.graph.run()
+          } catch (error) {
+            // D2 graph throws when it exceeds its internal iteration limit
+            // Transition to error state so callers can detect incomplete data
+            this.transitionToError(
+              error instanceof Error ? error.message : String(error),
+            )
+            return
+          }
           // Flush accumulated changes after each graph step to commit them as one transaction.
           // This ensures intermediate join states (like null on one side) don't cause
           // duplicate key errors when the full join result arrives in the same step.

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

@@ -37,6 +37,12 @@ export class CollectionSubscriber<
   // can potentially send the same item to D2 multiple times.
   private sentToD2Keys = new Set<string | number>()
 
+  // Track when the local index has been exhausted for the current cursor position.
+  // When true, loadMoreIfNeeded will not try to load more data until new data arrives.
+  // This prevents infinite loops when the TopK can't be filled because WHERE filters
+  // out all available data.
+  private localIndexExhausted = false
+
   constructor(
     private alias: string,
     private collectionId: string,
@@ -301,11 +307,25 @@ export class CollectionSubscriber<
       return true
     }
 
+    // If we've already exhausted the local index, don't try to load more.
+    // This prevents infinite loops when the TopK can't be filled because
+    // the WHERE clause filters out all available local data.
+    // The flag is reset when new data arrives from the sync layer.
+    if (this.localIndexExhausted) {
+      return true
+    }
+
     // `dataNeeded` probes the orderBy operator to see if it needs more data
     // if it needs more data, it returns the number of items it needs
     const n = dataNeeded()
     if (n > 0) {
-      this.loadNextItems(n, subscription)
+      const foundLocalData = this.loadNextItems(n, subscription)
+      if (!foundLocalData) {
+        // No local data found - mark the index as exhausted so we don't
+        // keep trying in subsequent graph iterations. The sync layer's
+        // loadSubset has been called and may return data asynchronously.
+        this.localIndexExhausted = true
+      }
     }
     return true
   }
@@ -320,7 +340,19 @@ export class CollectionSubscriber<
       return
     }
 
-    const trackedChanges = this.trackSentValues(changes, orderByInfo.comparator)
+    // Reset localIndexExhausted when new data arrives from the sync layer.
+    // This allows loadMoreIfNeeded to try loading again since there's new data.
+    // We only reset on inserts since updates/deletes don't add new data to load.
+    const changesArray = Array.isArray(changes) ? changes : [...changes]
+    const hasInserts = changesArray.some((c) => c.type === `insert`)
+    if (hasInserts) {
+      this.localIndexExhausted = false
+    }
+
+    const trackedChanges = this.trackSentValues(
+      changesArray,
+      orderByInfo.comparator,
+    )
 
     // Cache the loadMoreIfNeeded callback on the subscription using a symbol property.
     // This ensures we pass the same function instance to the scheduler each time,
@@ -342,10 +374,14 @@ export class CollectionSubscriber<
 
   // Loads the next `n` items from the collection
   // starting from the biggest item it has sent
-  private loadNextItems(n: number, subscription: CollectionSubscription) {
+  // Returns true if local data was found, false if the local index is exhausted
+  private loadNextItems(
+    n: number,
+    subscription: CollectionSubscription,
+  ): boolean {
     const orderByInfo = this.getOrderByInfo()
     if (!orderByInfo) {
-      return
+      return false
     }
     const { orderBy, valueExtractorForRawRow, offset } = orderByInfo
     const biggestSentRow = this.biggest
@@ -369,7 +405,8 @@ export class CollectionSubscriber<
 
     // Take the `n` items after the biggest sent value
     // Pass the current window offset to ensure proper deduplication
-    subscription.requestLimitedSnapshot({
+    // Returns true if local data was found
+    return subscription.requestLimitedSnapshot({
       orderBy: normalizedOrderBy,
       limit: n,
       minValues,