fix(db): batch D2 output callbacks to prevent duplicate key errors in joins (#1114)

KyleAMathews · claude · autofix-ci[bot] · web-flow · commit 29033b8f55b0 · 2026-01-12T11:59:43.000-07:00
* fix(db): batch D2 output callbacks to prevent duplicate key errors in joins

When D2's incremental join processes data, it can produce multiple outputs
during a single graph.run() iteration - first a partial join result, then
the full result plus a delete of the partial. Previously, each output
callback had its own begin()/commit() cycle, causing the second insert
for the same key to fail with DuplicateKeySyncError.

This fix:
- Accumulates all changes from output callbacks into a pendingChanges Map
- Flushes accumulated changes in a single transaction after each graph.run()
- Adds subscribedToAllCollections check to updateLiveQueryStatus() to ensure
  markReady() is only called after the graph has processed data
- Properly types flushPendingChanges on SyncState to avoid type assertions

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

* ci: apply automated fixes

* chore: add changeset for live query join fix

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

* chore: update changeset to mention isReady fix

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

* refactor: restore accumulateChanges helper function

Restore the accumulateChanges helper function instead of inlining its logic.
This improves code readability and maintainability by keeping the accumulation
logic in a separate, testable function.

Co-authored-by: Kevin &lt;kevin-dp@users.noreply.github.com&gt;

---------

Co-authored-by: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;
Co-authored-by: autofix-ci[bot] &lt;114827586+autofix-ci[bot]@users.noreply.github.com&gt;
Co-authored-by: claude[bot] &lt;41898282+claude[bot]@users.noreply.github.com&gt;
Co-authored-by: Kevin &lt;kevin-dp@users.noreply.github.com&gt;
diff --git a/.changeset/fix-live-query-join-duplicates.md b/.changeset/fix-live-query-join-duplicates.md
@@ -0,0 +1,7 @@
+---
+'@tanstack/db': patch
+---
+
+Fix `isReady()` returning `true` while `toArray()` returns empty results. The status now correctly waits until data has been processed through the graph before marking ready.
+
+Also fix duplicate key errors when live queries use joins with custom `getKey` functions. D2's incremental join can produce multiple outputs for the same key during a single graph run; this change batches all outputs into a single transaction to prevent conflicts.
diff --git a/packages/db/src/collection/sync.ts b/packages/db/src/collection/sync.ts
@@ -134,10 +134,10 @@ export class CollectionSyncManager<
                 !isTruncateTransaction
               ) {
                 const existingValue = this.state.syncedData.get(key)
-                if (
+                const valuesEqual =
                   existingValue !== undefined &&
                   deepEquals(existingValue, messageWithOptionalKey.value)
-                ) {
+                if (valuesEqual) {
                   // The "insert" is an echo of a value we already have locally.
                   // Treat it as an update so we preserve optimistic intent without
                   // throwing a duplicate-key error during reconciliation.
diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts
@@ -337,6 +337,10 @@ export class CollectionConfigBuilder<
       if (syncState.subscribedToAllCollections) {
         while (syncState.graph.pendingWork()) {
           syncState.graph.run()
+          // 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.
+          syncState.flushPendingChanges?.()
           callback?.()
         }
 
@@ -345,10 +349,14 @@ export class CollectionConfigBuilder<
         if (syncState.messagesCount === 0) {
           begin()
           commit()
-          // After initial commit, check if we should mark ready
-          // (in case all sources were already ready before we subscribed)
-          this.updateLiveQueryStatus(this.currentSyncConfig)
         }
+
+        // After graph processing completes, check if we should mark ready.
+        // This is the canonical place to transition to ready state because:
+        // 1. All data has been processed through the graph
+        // 2. All source collections have had a chance to send their initial data
+        // This prevents marking ready before data is processed (fixes isReady=true with empty data)
+        this.updateLiveQueryStatus(this.currentSyncConfig)
       }
     } finally {
       this.isGraphRunning = false
@@ -687,22 +695,35 @@ export class CollectionConfigBuilder<
     const { begin, commit } = config
     const { graph, inputs, pipeline } = this.maybeCompileBasePipeline()
 
+    // Accumulator for changes across all output callbacks within a single graph run.
+    // This allows us to batch all changes from intermediate join states into a single
+    // transaction, avoiding duplicate key errors when joins produce multiple outputs
+    // for the same key (e.g., first output with null, then output with joined data).
+    let pendingChanges: Map<unknown, Changes<TResult>> = new Map()
+
     pipeline.pipe(
       output((data) => {
         const messages = data.getInner()
         syncState.messagesCount += messages.length
 
-        begin()
-        messages
-          .reduce(
-            accumulateChanges<TResult>,
-            new Map<unknown, Changes<TResult>>(),
-          )
-          .forEach(this.applyChanges.bind(this, config))
-        commit()
+        // Accumulate changes from this output callback into the pending changes map.
+        // Changes for the same key are merged (inserts/deletes are added together).
+        messages.reduce(accumulateChanges<TResult>, pendingChanges)
       }),
     )
 
+    // Flush pending changes and reset the accumulator.
+    // Called at the end of each graph run to commit all accumulated changes.
+    syncState.flushPendingChanges = () => {
+      if (pendingChanges.size === 0) {
+        return
+      }
+      begin()
+      pendingChanges.forEach(this.applyChanges.bind(this, config))
+      commit()
+      pendingChanges = new Map()
+    }
+
     graph.finalize()
 
     // Extend the sync state with the graph, inputs, and pipeline
@@ -808,11 +829,14 @@ export class CollectionConfigBuilder<
       return
     }
 
-    // Mark ready when all source collections are ready AND
-    // the live query collection is not loading subset data.
-    // This prevents marking the live query ready before its data is loaded
+    // Mark ready when:
+    // 1. All subscriptions are set up (subscribedToAllCollections)
+    // 2. All source collections are ready
+    // 3. The live query collection is not loading subset data
+    // This prevents marking the live query ready before its data is processed
     // (fixes issue where useLiveQuery returns isReady=true with empty data)
     if (
+      this.currentSyncState?.subscribedToAllCollections &&
       this.allCollectionsReady() &&
       !this.liveQueryCollection?.isLoadingSubset
     ) {
@@ -913,8 +937,10 @@ export class CollectionConfigBuilder<
     // (graph only runs when all collections are subscribed)
     syncState.subscribedToAllCollections = true
 
-    // Initial status check after all subscriptions are set up
-    this.updateLiveQueryStatus(config)
+    // Note: We intentionally don't call updateLiveQueryStatus() here.
+    // The graph hasn't run yet, so marking ready would be premature.
+    // The canonical place to mark ready is after the graph processes data
+    // in maybeRunGraph(), which ensures data has been processed first.
 
     return loadSubsetDataCallbacks
   }
@@ -1096,8 +1122,11 @@ function accumulateChanges<T>(
     changes.deletes += Math.abs(multiplicity)
   } else if (multiplicity > 0) {
     changes.inserts += multiplicity
+    // Update value to the latest version for this key
     changes.value = value
-    changes.orderByIndex = orderByIndex
+    if (orderByIndex !== undefined) {
+      changes.orderByIndex = orderByIndex
+    }
   }
   acc.set(key, changes)
   return acc
diff --git a/packages/db/src/query/live/types.ts b/packages/db/src/query/live/types.ts
@@ -22,9 +22,11 @@ export type SyncState = {
   graph?: D2
   inputs?: Record<string, RootStreamBuilder<unknown>>
   pipeline?: ResultStream
+  flushPendingChanges?: () => void
 }
 
-export type FullSyncState = Required<SyncState>
+export type FullSyncState = Required<Omit<SyncState, `flushPendingChanges`>> &
+  Pick<SyncState, `flushPendingChanges`>
 
 /**
  * Configuration interface for live query collection options
diff --git a/packages/db/tests/query/live-query-collection.test.ts b/packages/db/tests/query/live-query-collection.test.ts
@@ -2293,10 +2293,11 @@ describe(`createLiveQueryCollection`, () => {
           .limit(10),
       )
 
-      // Trigger sync which will call loadSubset
-      await liveQueryCollection.preload()
+      // Start preload (don't await yet - it won't resolve until loadSubset completes)
+      const preloadPromise = liveQueryCollection.preload()
       await flushPromises()
 
+      // Verify loadSubset was called with the correct options
       expect(capturedOptions.length).toBeGreaterThan(0)
 
       // Find the call that has orderBy (the limited snapshot request)
@@ -2308,8 +2309,10 @@ describe(`createLiveQueryCollection`, () => {
       expect(callWithOrderBy?.orderBy?.[0]?.expression.type).toBe(`ref`)
       expect(callWithOrderBy?.limit).toBe(10)
 
+      // Resolve the loadSubset promise so preload can complete
       resolveLoadSubset!()
       await flushPromises()
+      await preloadPromise
     })
 
     it(`passes multiple orderBy columns to loadSubset when using limit`, async () => {
@@ -2350,10 +2353,11 @@ describe(`createLiveQueryCollection`, () => {
           .limit(10),
       )
 
-      // Trigger sync which will call loadSubset
-      await liveQueryCollection.preload()
+      // Start preload (don't await yet - it won't resolve until loadSubset completes)
+      const preloadPromise = liveQueryCollection.preload()
       await flushPromises()
 
+      // Verify loadSubset was called with the correct options
       expect(capturedOptions.length).toBeGreaterThan(0)
 
       // Find the call that has orderBy with multiple columns
@@ -2369,8 +2373,10 @@ describe(`createLiveQueryCollection`, () => {
       expect(callWithMultiOrderBy?.orderBy?.[1]?.expression.type).toBe(`ref`)
       expect(callWithMultiOrderBy?.limit).toBe(10)
 
+      // Resolve the loadSubset promise so preload can complete
       resolveLoadSubset!()
       await flushPromises()
+      await preloadPromise
     })
   })
 })
