Fix: deleted items not disappearing from live queries with .limit() (#1044)

* fix: emit delete events when subscribing with includeInitialState: false

When subscribing to a collection with includeInitialState: false,
delete events were being filtered out because the sentKeys set was
empty. This affected live queries with limit/offset where users
would subscribe to get future changes after already loading
initial data via preload() or values().

Changes:
- Add skipFiltering flag separate from loadedInitialState to allow
  filtering to be skipped while still allowing requestSnapshot to work
- Call markAllStateAsSeen() when includeInitialState is explicitly false
- Change internal subscriptions to not pass includeInitialState: false
  explicitly, so they can be distinguished from user subscriptions
- Add tests for optimistic delete behavior with limit

Fixes the issue where deleted items would not disappear from live
queries when using .limit() and subscribing with includeInitialState: false.

* debug: add extensive logging to track delete event flow

This is a DEBUG BUILD with [TanStack-DB-DEBUG] logs to help track down
why delete events may not be reaching subscribers when using limit/offset.

The debug logs cover:
- subscribeChanges: when subscriptions are created
- emitEvents: when events are emitted to subscriptions
- Subscription.emitEvents: when individual subscriptions receive events
- filterAndFlipChanges: when events are filtered or passed through
- recomputeOptimisticState: when optimistic state is recomputed and events emitted
- sendChangesToPipeline: when changes flow through the D2 pipeline
- applyChanges: when D2 pipeline outputs to the live query collection

To use: Filter browser console for "[TanStack-DB-DEBUG]"

Also includes the fix for includeInitialState: false not emitting deletes.

* ci: apply automated fixes

* debug: add more logging to track delete event flow in live queries

Add comprehensive debug logging to:
- createFilteredCallback in change-events.ts for whereExpression filtering
- sendChangesToInput for D2 pipeline input
- subscribeToOrderedChanges for orderBy/limit path
- splitUpdates for update event handling
- recomputeOptimisticState for pending sync key filtering

This additional logging helps track where delete events may be filtered
out when using live queries with limit/offset and where clauses.

* ci: apply automated fixes

* debug: add logging to graph scheduling and execution

Add debug logging to track:
- scheduleGraphRun: when graph run is scheduled
- executeGraphRun: when graph run executes or returns early
- maybeRunGraph: when graph actually runs, pending work status

This helps diagnose issues where deletes are sent to D2 pipeline
but never appear in the output (applyChanges not called).

* ci: apply automated fixes

* debug: add detailed logging to D2 reduce and topK operators

Add debug logging to track:
- ReduceOperator: input processing, key handling, and result output
- topK: consolidation, sorting, slicing, and result details

Also add two new test cases:
1. Test delete from different page (page 1 delete while viewing page 2)
   - Verifies items shift correctly when delete occurs on earlier page
2. Test delete beyond TopK window (no-op case)
   - Verifies deleting item outside window doesn't affect results

These tests and debug logs will help diagnose issues where deleted
items don't disappear from live queries when using limit/offset.

* ci: apply automated fixes

* debug: add more detailed logging to D2 graph and subscription

Add additional debug logging to help diagnose delete issues:

D2 graph (d2.ts):
- Log when run() starts and completes with step count
- Log pendingWork() results with operator IDs
- Log when operators have pending work in step()

Output operator (output.ts):
- Log when run is called with message count
- Log items in each message being processed

Subscription (subscription.ts):
- Log trackSentKeys with keys being added
- Show total sentKeys count

This should help diagnose scenarios where delete events are sent
to D2 but no applyChanges output is produced.

* ci: apply automated fixes

* debug: add operator type logging to trace D2 pipeline

Add operatorType property to Operator base class and log it when
operators run. This will help identify which operators are processing
the delete and where the data is being lost.

Also add detailed logging to LinearUnaryOperator.run() to show:
- Input message count
- Input/output item counts
- Sample of input and output items

This should reveal exactly which operator is dropping the delete.

* debug: add logging to TopKWithFractionalIndexOperator

This is the key operator for orderBy+limit queries. Add detailed logging to:
- run(): Show message count and index size
- processElement(): Show key, multiplicity changes, and action (INSERT/DELETE/NO_CHANGE)
- processElement result: Show moveIn/moveOut keys

This should reveal exactly why deletes aren't producing output changes
when the item exists in the TopK index.

* ci: apply automated fixes

* fix: filter duplicate inserts in subscription to prevent D2 multiplicity issues

When an item is inserted multiple times without a delete in between,
D2 multiplicity goes above 1. Then when a single delete arrives,
multiplicity goes from 2 to 1 (not 0), so TopK doesn't emit a DELETE event.

This fix:
1. Filters out duplicate inserts in filterAndFlipChanges when key already in sentKeys
2. Removes keys from sentKeys on delete in both filterAndFlipChanges and trackSentKeys
3. Updates test expectation to reflect correct behavior (2 events instead of 3)

Root cause: Multiple subscriptions or sync mechanisms could send duplicate
insert events for the same key, causing D2 to track multiplicity > 1.

* fix: add D2 input level deduplication to prevent multiplicity > 1

The previous fix in CollectionSubscription.filterAndFlipChanges was only
catching duplicates at the subscription level. But each live query has its
own CollectionSubscriber with its own D2 pipeline.

This fix adds a sentToD2Keys set in CollectionSubscriber to track which
keys have been sent to the D2 input, preventing duplicate inserts at the
D2 level regardless of which code path triggers them.

Also clears the tracking on truncate events.

* docs: add detailed changeset for delete fix

* ci: apply automated fixes

* chore: remove debug logging from D2 pipeline and subscription code

Remove all TanStack-DB-DEBUG console statements that were added during
investigation of the deleted items not disappearing from live queries bug.

The fix for duplicate D2 inserts is preserved - just removing the verbose
debug output now that the issue is resolved.

* debug: add logging to trace source of duplicate D2 inserts

Add targeted debug logging to understand where duplicate inserts originate:

1. recomputeOptimisticState: Track what events are generated and when
2. CollectionSubscription.filterAndFlipChanges: Trace filtering decisions
3. CollectionSubscriber.sendChangesToPipeline: Track D2-level deduplication

This will help determine if duplicates come from:
- Multiple calls to recomputeOptimisticState for the same key
- Overlap between initial snapshot and change events
- Multiple code paths feeding the D2 pipeline

* ci: apply automated fixes

* fix: prevent race condition in snapshot loading by adding keys to sentKeys before callback

The race condition occurred because snapshot methods (requestSnapshot, requestLimitedSnapshot)
added keys to sentKeys AFTER calling the callback, while filterAndFlipChanges added keys
BEFORE. If a change event arrived during callback execution, it would not see the keys
in sentKeys yet, allowing duplicate inserts.

Changes:
- Add keys to sentKeys BEFORE calling callback in requestSnapshot and requestLimitedSnapshot
- Remove redundant D2-level deduplication (sentToD2Keys) - subscription-level filtering is sufficient
- Remove debug logging added during investigation

* docs: update changeset to reflect race condition fix

* cleanup

* simplify changeset

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Sam Willis <sam.willis@gmail.com>

Author: 4 people

.changeset/fix-deleted-items-not-disappearing-from-live-queries.md

@@ -0,0 +1,11 @@
+---
+'@tanstack/db': patch
+---
+
+fix: deleted items not disappearing from live queries with `.limit()`
+
+Fixed a bug where deleting an item from a live query with `.orderBy()` and `.limit()` would not remove it from the query results. The `subscribeChanges` callback would never fire with a delete event.
+
+The issue was caused by duplicate inserts reaching the D2 pipeline, which corrupted the multiplicity tracking used by `TopKWithFractionalIndexOperator`. A delete would decrement multiplicity from 2 to 1 instead of 1 to 0, so the item remained visible.
+
+Fixed by ensuring `sentKeys` is updated before callbacks execute (preventing race conditions) and filtering duplicate inserts in `filterAndFlipChanges`.

packages/db/src/collection/changes.ts

@@ -109,6 +109,10 @@ export class CollectionChangesManager<
 
     if (options.includeInitialState) {
       subscription.requestSnapshot({ trackLoadSubsetPromise: false })
+    } else if (options.includeInitialState === false) {
+      // When explicitly set to false (not just undefined), mark all state as "seen"
+      // so that all future changes (including deletes) pass through unfiltered.
+      subscription.markAllStateAsSeen()
     }
 
     // Add to batched listeners

packages/db/src/collection/subscription.ts

@@ -52,6 +52,11 @@ export class CollectionSubscription
 {
   private loadedInitialState = false
 
+  // Flag to skip filtering in filterAndFlipChanges.
+  // This is separate from loadedInitialState because we want to allow
+  // requestSnapshot to still work even when filtering is skipped.
+  private skipFiltering = false
+
   // Flag to indicate that we have sent at least 1 snapshot.
   // While `snapshotSent` is false we filter out all changes from subscription to the collection.
   private snapshotSent = false
@@ -244,6 +249,13 @@ export class CollectionSubscription
       (change) => !this.sentKeys.has(change.key),
     )
 
+    // Add keys to sentKeys BEFORE calling callback to prevent race condition.
+    // If a change event arrives while the callback is executing, it will see
+    // the keys already in sentKeys and filter out duplicates correctly.
+    for (const change of filteredSnapshot) {
+      this.sentKeys.add(change.key)
+    }
+
     this.snapshotSent = true
     this.callback(filteredSnapshot)
     return true
@@ -367,6 +379,13 @@ export class CollectionSubscription
     // Use the current count as the offset for this load
     const currentOffset = this.limitedSnapshotRowCount
 
+    // Add keys to sentKeys BEFORE calling callback to prevent race condition.
+    // If a change event arrives while the callback is executing, it will see
+    // the keys already in sentKeys and filter out duplicates correctly.
+    for (const change of changes) {
+      this.sentKeys.add(change.key)
+    }
+
     this.callback(changes)
 
     // Update the row count and last key after sending (for next call's offset/cursor)
@@ -441,43 +460,74 @@ export class CollectionSubscription
    * Filters and flips changes for keys that have not been sent yet.
    * Deletes are filtered out for keys that have not been sent yet.
    * Updates are flipped into inserts for keys that have not been sent yet.
+   * Duplicate inserts are filtered out to prevent D2 multiplicity > 1.
    */
   private filterAndFlipChanges(changes: Array<ChangeMessage<any, any>>) {
-    if (this.loadedInitialState) {
-      // We loaded the entire initial state
+    if (this.loadedInitialState || this.skipFiltering) {
+      // We loaded the entire initial state or filtering is explicitly skipped
       // so no need to filter or flip changes
       return changes
     }
 
     const newChanges = []
     for (const change of changes) {
       let newChange = change
-      if (!this.sentKeys.has(change.key)) {
+      const keyInSentKeys = this.sentKeys.has(change.key)
+
+      if (!keyInSentKeys) {
         if (change.type === `update`) {
           newChange = { ...change, type: `insert`, previousValue: undefined }
         } else if (change.type === `delete`) {
           // filter out deletes for keys that have not been sent
           continue
         }
         this.sentKeys.add(change.key)
+      } else {
+        // Key was already sent - handle based on change type
+        if (change.type === `insert`) {
+          // Filter out duplicate inserts - the key was already inserted.
+          // This prevents D2 multiplicity from going above 1, which would
+          // cause deletes to not properly remove items (multiplicity would
+          // go from 2 to 1 instead of 1 to 0).
+          continue
+        } else if (change.type === `delete`) {
+          // Remove from sentKeys so future inserts for this key are allowed
+          // (e.g., after truncate + reinsert)
+          this.sentKeys.delete(change.key)
+        }
       }
       newChanges.push(newChange)
     }
     return newChanges
   }
 
   private trackSentKeys(changes: Array<ChangeMessage<any, string | number>>) {
-    if (this.loadedInitialState) {
-      // No need to track sent keys if we loaded the entire state.
-      // Since we sent everything, all keys must have been observed.
+    if (this.loadedInitialState || this.skipFiltering) {
+      // No need to track sent keys if we loaded the entire state or filtering is skipped.
+      // Since filtering won't be applied, all keys are effectively "observed".
       return
     }
 
     for (const change of changes) {
-      this.sentKeys.add(change.key)
+      if (change.type === `delete`) {
+        // Remove deleted keys from sentKeys so future re-inserts are allowed
+        this.sentKeys.delete(change.key)
+      } else {
+        // For inserts and updates, track the key as sent
+        this.sentKeys.add(change.key)
+      }
     }
   }
 
+  /**
+   * Mark that the subscription should not filter any changes.
+   * This is used when includeInitialState is explicitly set to false,
+   * meaning the caller doesn't want initial state but does want ALL future changes.
+   */
+  markAllStateAsSeen() {
+    this.skipFiltering = true
+  }
+
   unsubscribe() {
     // Unload all subsets that this subscription loaded
     // We pass the exact same LoadSubsetOptions we used for loadSubset

packages/db/tests/collection-subscribe-changes.test.ts

@@ -1193,8 +1193,10 @@ describe(`Collection.subscribeChanges`, () => {
     f.write({ type: `insert`, value: { id: 1, value: `server-after` } })
     f.commit()
 
-    // Expect delete, insert with optimistic value, and an empty event from markReady
-    expect(changeEvents.length).toBe(3)
+    // Expect delete and insert with optimistic value
+    // Note: Previously there was a duplicate insert event that was incorrectly
+    // being sent, causing 3 events. Now duplicates are filtered correctly.
+    expect(changeEvents.length).toBe(2)
     expect(changeEvents[0]).toEqual({
       type: `delete`,
       key: 1,