feat(db): ordered snapshot / layout-revision contract (RFC #1623 phase 4)

.changeset/live-query-observer.md

@@ -9,8 +9,8 @@
 
 Add an internal shared live-query observer and migrate all five framework adapters to it
 
-Introduces `createLiveQueryObserver` in `@tanstack/db`: given a resolved live-query collection (or `null` for a disabled query) it owns the lifecycle every adapter used to re-implement — sync activation on first subscribe, change and status subscriptions, a snapshot with stable identity per state revision for wholesale consumers, and delivery of the raw `ChangeMessage[]` for granular consumers. React, Vue, Svelte, Solid, and Angular's live-query hooks now materialize from the observer instead of their own hand-rolled subscription/status/snapshot machinery, keeping each adapter's native reactivity and each adapter's data-loading policy (wholesale adapters subscribe without initial state; granular adapters seed from it).
+Introduces `createLiveQueryObserver` in `@tanstack/db`: given a resolved live-query collection (or `null` for a disabled query) it owns the subscription lifecycle every adapter used to re-implement — change and status subscriptions, a snapshot with stable identity per state revision for wholesale consumers, and delivery of the raw `ChangeMessage[]` for granular consumers. React, Vue, Svelte, Solid, and Angular's live-query hooks now materialize from the observer instead of their own hand-rolled subscription/status/snapshot machinery, keeping each adapter's native reactivity and each adapter's data-loading policy (wholesale adapters subscribe without initial state; granular adapters seed from it).
 
 The observer is an **internal, unstable contract** for TanStack DB's official adapters — it is exported so the adapter packages can consume it, but it is not a public extension point yet and its API may change in any release.
 
-The migration also fixes several live-query lifecycle defects: status-only transitions (`error`, `cleaned-up`) now reach mounted consumers; snapshot identity is stable across unsubscribe/resubscribe and stays fresh while detached; dispatch is FIFO and non-reentrant with subscriptions identified by record rather than callback; disposing during the synchronous initial replay no longer leaks the collection subscription; subscribing after dispose throws instead of registering a dead listener; Solid guards its async resource continuations against superseded collections; and constructing an observer no longer activates sync (activation belongs to the first committed subscription).
+The migration also fixes several live-query lifecycle defects: status-only transitions (`error`, `cleaned-up`) now reach mounted consumers; snapshot identity is stable across unsubscribe/resubscribe and stays fresh while detached; dispatch is FIFO and non-reentrant with subscriptions identified by record rather than callback; disposing during the synchronous initial replay no longer leaks the collection subscription; subscribing after dispose throws instead of registering a dead listener; Solid guards its async resource continuations against superseded collections; and constructing an observer no longer activates sync. Observers activate on their first committed subscription unless an adapter has already started a pre-created collection supplied directly or returned from a callback.

.changeset/live-query-order-only-move.md

@@ -0,0 +1,13 @@
+---
+'@tanstack/db': patch
+---
+
+fix(db): republish ordered live queries on an order-only move
+
+An `orderBy` live query that reordered its rows without changing any projected
+row value (an "order-only move") previously emitted nothing, so `useLiveQuery`
+kept rendering the stale order. The live-query collection now publishes an
+explicit layout-change notification when this happens, and the shared live-query
+observer snapshot exposes a `layoutRevision` that increments on any visible
+membership, ordering, or order-only-move change. All five framework adapters
+pick this up via their existing wholesale re-read.

packages/db/src/collection/change-events.ts

@@ -253,7 +253,7 @@ export function createFilteredCallback<
 >(
   originalCallback: (changes: Array<ChangeMessage<T>>) => void,
   options: SubscribeChangesOptions<T, TKey>,
-): (changes: Array<ChangeMessage<T>>) => void {
+): (changes: Array<ChangeMessage<T>>) => boolean {
   const filterFn = createFilterFunctionFromExpression(options.whereExpression!)
 
   return (changes: Array<ChangeMessage<T>>) => {
@@ -303,7 +303,9 @@ export function createFilteredCallback<
     // if the original changes array was empty (which indicates a ready signal)
     if (filteredChanges.length > 0 || changes.length === 0) {
       originalCallback(filteredChanges)
+      return true
     }
+    return false
   }
 }
 

packages/db/src/collection/changes.ts

@@ -39,6 +39,13 @@ export class CollectionChangesManager<
    */
   public stateRevision = 0
 
+  /**
+   * 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.
+   */
+  public layoutRevision = 0
+
   /**
    * Creates a new CollectionChangesManager instance
    */
@@ -85,10 +92,12 @@ export class CollectionChangesManager<
   public emitEvents(
     changes: Array<ChangeMessage<TOutput, TKey>>,
     forceEmit = false,
+    layoutChanged = false,
   ): void {
     // The visible state was already committed by the caller, so the revision
     // advances even when the events below end up batched for later emission.
     if (changes.length > 0) this.stateRevision++
+    if (layoutChanged) this.layoutRevision++
 
     // Skip batching for user actions (forceEmit=true) to keep UI responsive
     if (this.shouldBatchEvents && !forceEmit) {
@@ -111,7 +120,7 @@ export class CollectionChangesManager<
       this.shouldBatchEvents = false
     }
 
-    if (rawEvents.length === 0) {
+    if (rawEvents.length === 0 && !layoutChanged) {
       return
     }
 
@@ -123,7 +132,7 @@ export class CollectionChangesManager<
 
     // Emit to all listeners
     for (const subscription of this.changeSubscriptions) {
-      subscription.emitEvents(enrichedEvents)
+      subscription.emitEvents(enrichedEvents, layoutChanged)
     }
   }
 

packages/db/src/collection/index.ts

@@ -429,6 +429,19 @@ export class CollectionImpl<
     return this._changes.stateRevision
   }
 
+  /**
+   * Monotonic revision of explicit layout-only publications.
+   * Internal — used to distinguish them from empty ready events.
+   */
+  public get _layoutRevision(): number {
+    return this._changes.layoutRevision
+  }
+
+  /** Mark the active sync transaction as layout-changing. Internal. */
+  public _markLayoutChange(): void {
+    this._sync.markLayoutChange()
+  }
+
   /**
    * Register a callback to be executed when the collection first becomes ready
    * Useful for preloading collections

packages/db/src/collection/state.ts

@@ -25,6 +25,7 @@ interface PendingSyncedTransaction<
   TKey extends string | number = string | number,
 > {
   committed: boolean
+  layoutChanged: boolean
   operations: Array<OptimisticChangeMessage<T>>
   truncate?: boolean
   deletedKeys: Set<string | number>
@@ -836,10 +837,12 @@ export class CollectionStateManager<
       uncommittedSyncedTransactions,
       hasTruncateSync,
       hasImmediateSync,
+      layoutChanged,
     } = this.pendingSyncedTransactions.reduce(
       (acc, t) => {
         if (t.committed) {
           acc.committedSyncedTransactions.push(t)
+          acc.layoutChanged ||= t.layoutChanged
           if (t.truncate) {
             acc.hasTruncateSync = true
           }
@@ -860,6 +863,7 @@ export class CollectionStateManager<
         >,
         hasTruncateSync: false,
         hasImmediateSync: false,
+        layoutChanged: false,
       },
     )
 
@@ -1331,7 +1335,7 @@ export class CollectionStateManager<
       }
 
       // End batching and emit all events (combines any batched events with sync events)
-      this.changes.emitEvents(events, true)
+      this.changes.emitEvents(events, true, layoutChanged)
 
       this.pendingSyncedTransactions = uncommittedSyncedTransactions
 

packages/db/src/collection/subscription.ts

@@ -83,7 +83,7 @@ export class CollectionSubscription
   // Track the last key sent via requestLimitedSnapshot for cursor-based pagination
   private lastSentKey: string | number | undefined
 
-  private filteredCallback: (changes: Array<ChangeMessage<any, any>>) => void
+  private filteredCallback: (changes: Array<ChangeMessage<any, any>>) => boolean
 
   private orderByIndex: IndexInterface<string | number> | undefined
 
@@ -99,6 +99,7 @@ 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 {
@@ -132,7 +133,10 @@ export class CollectionSubscription
     // Create a filtered callback if where clause is provided
     this.filteredCallback = options.whereExpression
       ? createFilteredCallback(this.callback, options)
-      : this.callback
+      : (changes) => {
+          this.callback(changes)
+          return true
+        }
 
     // Listen for truncate events to re-request data after must-refetch
     // When a truncate happens (e.g., from a 409 must-refetch), all collection data is cleared.
@@ -245,11 +249,14 @@ 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()
-    if (merged.length > 0) {
-      this.filteredCallback(merged)
+    const layoutChanged = this.truncateBufferHasLayoutChange
+    if (merged.length > 0 || layoutChanged) {
+      const delivered = this.filteredCallback(merged)
+      if (layoutChanged && !delivered) this.filteredCallback([])
     }
 
     this.truncateBuffer = []
+    this.truncateBufferHasLayoutChange = false
   }
 
   setOrderByIndex(index: IndexInterface<any>) {
@@ -318,7 +325,10 @@ export class CollectionSubscription
     return this.snapshotSent
   }
 
-  emitEvents(changes: Array<ChangeMessage<any, any>>) {
+  emitEvents(
+    changes: Array<ChangeMessage<any, any>>,
+    layoutChanged = false,
+  ): boolean {
     const newChanges = this.filterAndFlipChanges(changes)
 
     if (this.isBufferingForTruncate) {
@@ -327,8 +337,12 @@ export class CollectionSubscription
       if (newChanges.length > 0) {
         this.truncateBuffer.push(newChanges)
       }
+      if (layoutChanged) this.truncateBufferHasLayoutChange = true
+      return false
     } else {
-      this.filteredCallback(newChanges)
+      const delivered = this.filteredCallback(newChanges)
+      if (layoutChanged && !delivered) return this.filteredCallback([])
+      return delivered
     }
   }
 
@@ -728,6 +742,7 @@ 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

@@ -71,6 +71,11 @@ export class CollectionSyncManager<
     this._events = deps.events
   }
 
+  /** Mark the active sync transaction as changing collection layout. */
+  public markLayoutChange(): void {
+    this.getActivePendingSyncTransaction().layoutChanged = true
+  }
+
   /**
    * Start the sync process for this collection
    * This is called when the collection is first accessed or preloaded
@@ -92,6 +97,7 @@ export class CollectionSyncManager<
           begin: (options?: { immediate?: boolean }) => {
             this.state.pendingSyncedTransactions.push({
               committed: false,
+              layoutChanged: false,
               operations: [],
               deletedKeys: new Set(),
               rowMetadataWrites: new Map(),