feat(db): shared live-query window controller (RFC #1623 phase 5)

.changeset/live-query-window-controller.md

@@ -0,0 +1,10 @@
+---
+'@tanstack/db': patch
+'@tanstack/react-db': patch
+---
+
+Add the unstable, internal `createLiveQueryWindowController` primitive for
+forward pagination. It coordinates collection-scoped window leases, commits
+pages only after subset loads succeed, restores windows after failures and
+cleanup, and lets React's `useLiveInfiniteQuery` become a thin binding without
+changing its public API or resetting pages for structurally equal dependencies.

packages/db/src/collection/changes.ts

@@ -29,6 +29,7 @@ export class CollectionChangesManager<
   public changeSubscriptions = new Set<CollectionSubscription>()
   public batchedEvents: Array<ChangeMessage<TOutput, TKey>> = []
   public shouldBatchEvents = false
+  private layoutChangeListeners = new Set<() => void>()
 
   /**
    * Monotonic revision of the collection's visible state, advanced once per
@@ -41,8 +42,7 @@ export class CollectionChangesManager<
 
   /**
    * Monotonic revision advanced only for explicit layout-only publications.
-   * This distinguishes them from the legacy empty ready event, since both use
-   * an empty change batch at the public subscription boundary.
+   * Observers use it to detect reordered rows whose values did not change.
    */
   public layoutRevision = 0
 
@@ -124,6 +124,13 @@ export class CollectionChangesManager<
       return
     }
 
+    // Notify both internal layout consumers and the public subscription API.
+    // Public subscribers historically receive an empty batch for order-only
+    // moves because there is no row-value ChangeMessage to publish.
+    if (rawEvents.length === 0) {
+      for (const listener of this.layoutChangeListeners) listener()
+    }
+
     // Enrich all change messages with virtual properties
     // This uses the "add-if-missing" pattern to preserve pass-through semantics
     const enrichedEvents: Array<
@@ -132,10 +139,16 @@ export class CollectionChangesManager<
 
     // Emit to all listeners
     for (const subscription of this.changeSubscriptions) {
-      subscription.emitEvents(enrichedEvents, layoutChanged)
+      subscription.emitEvents(enrichedEvents)
     }
   }
 
+  /** Subscribe to layout-only publications. Internal observer channel. */
+  public subscribeLayoutChanges(listener: () => void): () => void {
+    this.layoutChangeListeners.add(listener)
+    return () => this.layoutChangeListeners.delete(listener)
+  }
+
   /**
    * Subscribe to changes in the collection
    */

packages/db/src/collection/index.ts

@@ -437,6 +437,11 @@ export class CollectionImpl<
     return this._changes.layoutRevision
   }
 
+  /** Subscribe to layout-only publications. Internal observer channel. */
+  public _subscribeLayoutChanges(listener: () => void): () => void {
+    return this._changes.subscribeLayoutChanges(listener)
+  }
+
   /** Mark the active sync transaction as layout-changing. Internal. */
   public _markLayoutChange(): void {
     this._sync.markLayoutChange()

packages/db/src/collection/subscription.ts

@@ -99,7 +99,6 @@ export class CollectionSubscription
   // This prevents a flash of missing content between deletes and new inserts
   private isBufferingForTruncate = false
   private truncateBuffer: Array<Array<ChangeMessage<any, any>>> = []
-  private truncateBufferHasLayoutChange = false
   private pendingTruncateRefetches: Set<Promise<void>> = new Set()
 
   public get status(): SubscriptionStatus {
@@ -249,14 +248,9 @@ export class CollectionSubscription
     // Flatten all buffered changes into a single array for atomic emission
     // This ensures consumers see all truncate changes (deletes + inserts) in one callback
     const merged = this.truncateBuffer.flat()
-    const layoutChanged = this.truncateBufferHasLayoutChange
-    if (merged.length > 0 || layoutChanged) {
-      const delivered = this.filteredCallback(merged)
-      if (layoutChanged && !delivered) this.filteredCallback([])
-    }
+    if (merged.length > 0) this.filteredCallback(merged)
 
     this.truncateBuffer = []
-    this.truncateBufferHasLayoutChange = false
   }
 
   setOrderByIndex(index: IndexInterface<any>) {
@@ -308,12 +302,13 @@ export class CollectionSubscription
       this.pendingLoadSubsetPromises.add(syncResult)
       this.setStatus(`loadingSubset`)
 
-      syncResult.finally(() => {
+      const finish = () => {
         this.pendingLoadSubsetPromises.delete(syncResult)
         if (this.pendingLoadSubsetPromises.size === 0) {
           this.setStatus(`ready`)
         }
-      })
+      }
+      void syncResult.then(finish, finish)
     }
   }
 
@@ -325,10 +320,7 @@ export class CollectionSubscription
     return this.snapshotSent
   }
 
-  emitEvents(
-    changes: Array<ChangeMessage<any, any>>,
-    layoutChanged = false,
-  ): boolean {
+  emitEvents(changes: Array<ChangeMessage<any, any>>): boolean {
     const newChanges = this.filterAndFlipChanges(changes)
 
     if (this.isBufferingForTruncate) {
@@ -337,12 +329,9 @@ export class CollectionSubscription
       if (newChanges.length > 0) {
         this.truncateBuffer.push(newChanges)
       }
-      if (layoutChanged) this.truncateBufferHasLayoutChange = true
       return false
     } else {
-      const delivered = this.filteredCallback(newChanges)
-      if (layoutChanged && !delivered) return this.filteredCallback([])
-      return delivered
+      return this.filteredCallback(newChanges)
     }
   }
 
@@ -742,7 +731,6 @@ export class CollectionSubscription
     // Clean up truncate buffer state
     this.isBufferingForTruncate = false
     this.truncateBuffer = []
-    this.truncateBufferHasLayoutChange = false
     this.pendingTruncateRefetches.clear()
 
     // Unload all subsets that this subscription loaded

packages/db/src/collection/sync.ts

@@ -444,6 +444,18 @@ export class CollectionSyncManager<
     return this.pendingLoadSubsetPromises.size > 0
   }
 
+  /** Wait for the subset loads that are active during the current operation. */
+  public waitForCurrentLoadSubset(): true | Promise<void> {
+    if (this.pendingLoadSubsetPromises.size === 0) return true
+    return this.waitForPendingLoadSubset()
+  }
+
+  private async waitForPendingLoadSubset(): Promise<void> {
+    do {
+      await Promise.all([...this.pendingLoadSubsetPromises])
+    } while (this.pendingLoadSubsetPromises.size > 0)
+  }
+
   /**
    * Tracks a load promise for isLoadingSubset state.
    * @internal This is for internal coordination (e.g., live-query glue code), not for general use.
@@ -462,7 +474,7 @@ export class CollectionSyncManager<
       })
     }
 
-    promise.finally(() => {
+    const finish = () => {
       const loadingEnding =
         this.pendingLoadSubsetPromises.size === 1 &&
         this.pendingLoadSubsetPromises.has(promise)
@@ -477,7 +489,8 @@ export class CollectionSyncManager<
           loadingSubsetTransition: `end`,
         })
       }
-    })
+    }
+    void promise.then(finish, finish)
   }
 
   /**

packages/db/src/errors.ts

@@ -141,6 +141,12 @@ export class LiveQueryObserverDisposedError extends CollectionStateError {
   }
 }
 
+export class LiveQueryWindowControllerDisposedError extends CollectionStateError {
+  constructor() {
+    super(`Cannot subscribe to a disposed LiveQueryWindowController`)
+  }
+}
+
 // Collection Operation Errors
 export class CollectionOperationError extends TanStackDBError {
   constructor(message: string) {

packages/db/src/index.ts

@@ -12,6 +12,8 @@ export * from './query/index.js'
 export * from './optimistic-action'
 export * from './live-query-adapter'
 export * from './live-query-observer'
+/** @internal Unstable adapter primitive for RFC #1623. */
+export * from './live-query-window-controller'
 export * from './local-only'
 export * from './local-storage'
 export * from './errors'

packages/db/src/live-query-observer.ts

@@ -44,7 +44,10 @@ export interface LiveQuerySnapshot<
   isEnabled: boolean
 }
 
-/** Listener payload: the change set, or `undefined` for the synthetic ready notify. */
+/**
+ * Listener payload: changes, `[]` for an internal layout-only publication, or
+ * `undefined` for a synthetic status/ready notification.
+ */
 export type LiveQueryObserverListener<
   T extends object,
   TKey extends string | number,
@@ -357,11 +360,16 @@ class LiveQueryObserverImpl<
     const notify = (
       changes: Array<ChangeMessage<T, TKey>> | undefined,
       status: CollectionStatus = collection.status,
+      explicitLayoutChange = false,
     ) => {
       if (this.disposed || this.subscriptions.size === 0) return
       const layoutRevision = this.getCollectionLayoutRevision(collection)
-      let layoutChanged = false
-      if (changes !== undefined && changes.length === 0) {
+      let layoutChanged = explicitLayoutChange
+      if (
+        !explicitLayoutChange &&
+        changes !== undefined &&
+        changes.length === 0
+      ) {
         // Empty ready events predate the explicit layout signal and share its
         // empty-array payload. Only forward an empty batch when the collection
         // confirms that a new layout-only publication occurred.
@@ -401,6 +409,17 @@ class LiveQueryObserverImpl<
     const statusUnsub = collection.on(`status:change`, ({ status }) =>
       notify(undefined, status),
     )
+    const subscribeLayoutChanges = (
+      collection as Collection<T, TKey, any> & {
+        _subscribeLayoutChanges?: (listener: () => void) => () => void
+      }
+    )._subscribeLayoutChanges
+    const layoutUnsub =
+      typeof subscribeLayoutChanges === `function`
+        ? subscribeLayoutChanges.call(collection, () =>
+            notify([], collection.status, true),
+          )
+        : () => {}
 
     // `subscribeChanges` delivers the initial state synchronously, so a
     // listener can dispose the observer while the collection subscription is
@@ -410,6 +429,7 @@ class LiveQueryObserverImpl<
     let subscription: { unsubscribe: () => void } | null = null
     const release = () => {
       statusUnsub()
+      layoutUnsub()
       subscription?.unsubscribe()
     }
     this.collectionUnsub = release