Fix a bug that could result in a duplicate delete event for a row (#621)

* Add tests that repo the bug

* ensure that redundent sync delete events are skipped

* changeset

* typo

* remove console logs

Author: samwillis

.changeset/olive-boxes-lie.md

@@ -0,0 +1,5 @@
+---
+"@tanstack/db": patch
+---
+
+Fixed a bug that could result in a duplicate delete event for a row

packages/db/src/collection/state.ts

@@ -708,10 +708,23 @@ export class CollectionStateManager<
 
         // Check if this sync operation is redundant with a completed optimistic operation
         const completedOp = completedOptimisticOps.get(key)
-        const isRedundantSync =
-          completedOp &&
-          newVisibleValue !== undefined &&
-          deepEquals(completedOp.value, newVisibleValue)
+        let isRedundantSync = false
+
+        if (completedOp) {
+          if (
+            completedOp.type === `delete` &&
+            previousVisibleValue !== undefined &&
+            newVisibleValue === undefined &&
+            deepEquals(completedOp.value, previousVisibleValue)
+          ) {
+            isRedundantSync = true
+          } else if (
+            newVisibleValue !== undefined &&
+            deepEquals(completedOp.value, newVisibleValue)
+          ) {
+            isRedundantSync = true
+          }
+        }
 
         if (!isRedundantSync) {
           if (

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

@@ -1379,4 +1379,141 @@ describe(`Collection.subscribeChanges`, () => {
     expect(changeEvents.length).toBe(0)
     expect(collection.state.has(1)).toBe(false)
   })
+
+  it(`only emit a single event when a sync mutation is triggered from inside a mutation handler callback`, async () => {
+    const callback = vi.fn()
+
+    interface TestItem extends Record<string, unknown> {
+      id: number
+      number: number
+    }
+
+    let callBegin!: () => void
+    let callWrite!: (message: Omit<ChangeMessage<TestItem>, `key`>) => void
+    let callCommit!: () => void
+
+    // Create collection with pre-populated data
+    const collection = createCollection<TestItem>({
+      id: `test`,
+      getKey: (item) => item.id,
+      sync: {
+        sync: ({ begin, write, commit, markReady }) => {
+          callBegin = begin
+          callWrite = write
+          callCommit = commit
+          // Immediately populate with initial data
+          begin()
+          write({
+            type: `insert`,
+            value: { id: 0, number: 15 },
+          })
+          commit()
+          markReady()
+        },
+      },
+      onDelete: ({ transaction }) => {
+        const { original } = transaction.mutations[0]
+
+        // IMMEDIATELY synchronously trigger the sync inside the onDelete callback promise
+        callBegin()
+        callWrite({ type: `delete`, value: original })
+        callCommit()
+
+        return Promise.resolve()
+      },
+    })
+
+    // Subscribe to changes
+    const subscription = collection.subscribeChanges(callback, {
+      includeInitialState: true,
+    })
+
+    callback.mockReset()
+
+    // Delete item 0
+    collection.delete(0)
+
+    await new Promise((resolve) => setTimeout(resolve, 10))
+
+    expect(callback.mock.calls.length).toBe(1)
+    expect(callback.mock.calls[0]![0]).toEqual([
+      {
+        type: `delete`,
+        key: 0,
+        value: { id: 0, number: 15 },
+      },
+    ])
+
+    subscription.unsubscribe()
+  })
+
+  it(`only emit a single event when a sync mutation is triggered from inside a mutation handler callback after a short delay`, async () => {
+    const callback = vi.fn()
+
+    interface TestItem extends Record<string, unknown> {
+      id: number
+      number: number
+    }
+
+    let callBegin!: () => void
+    let callWrite!: (message: Omit<ChangeMessage<TestItem>, `key`>) => void
+    let callCommit!: () => void
+
+    // Create collection with pre-populated data
+    const collection = createCollection<TestItem>({
+      id: `test`,
+      getKey: (item) => item.id,
+      sync: {
+        sync: ({ begin, write, commit, markReady }) => {
+          callBegin = begin
+          callWrite = write
+          callCommit = commit
+          // Immediately populate with initial data
+          begin()
+          write({
+            type: `insert`,
+            value: { id: 0, number: 15 },
+          })
+          commit()
+          markReady()
+        },
+      },
+      onDelete: async ({ transaction }) => {
+        const { original } = transaction.mutations[0]
+
+        // Simulate waiting for some async operation
+        await new Promise((resolve) => setTimeout(resolve, 0))
+
+        // Synchronously trigger the sync inside the onDelete callback promise,
+        // but after a short delay.
+        // Ordering here is important to test for a race condition!
+        callBegin()
+        callWrite({ type: `delete`, value: original })
+        callCommit()
+      },
+    })
+
+    // Subscribe to changes
+    const subscription = collection.subscribeChanges(callback, {
+      includeInitialState: true,
+    })
+
+    callback.mockReset()
+
+    // Delete item 0
+    collection.delete(0)
+
+    await new Promise((resolve) => setTimeout(resolve, 10))
+
+    expect(callback.mock.calls.length).toBe(1)
+    expect(callback.mock.calls[0]![0]).toEqual([
+      {
+        type: `delete`,
+        key: 0,
+        value: { id: 0, number: 15 },
+      },
+    ])
+
+    subscription.unsubscribe()
+  })
 })

packages/db/tests/local-only.test.ts

@@ -1,5 +1,6 @@
 import { beforeEach, describe, expect, it, vi } from "vitest"
-import { createCollection } from "../src/index"
+import { createCollection, liveQueryCollectionOptions } from "../src/index"
+import { sum } from "../src/query/builder/functions"
 import { localOnlyCollectionOptions } from "../src/local-only"
 import type { LocalOnlyCollectionUtils } from "../src/local-only"
 import type { Collection } from "../src/index"
@@ -8,6 +9,7 @@ interface TestItem extends Record<string, unknown> {
   id: number
   name: string
   completed?: boolean
+  number?: number
 }
 
 describe(`LocalOnly Collection`, () => {
@@ -435,4 +437,53 @@ describe(`LocalOnly Collection`, () => {
       expect(testCollection.get(200)).toEqual({ id: 200, name: `Added Item` })
     })
   })
+
+  describe(`Live Query integration`, () => {
+    it(`aggregation should work when there is a onDelete callback`, async () => {
+      // This is a reproduction of this issue: https://github.com/TanStack/db/issues/609
+      // The underlying bug is covered by the "only emit a single event when a sync
+      // mutation is triggered from inside an mutation handler callback after a short
+      // delay" test in collection-subscribe-changes.test.ts
+
+      const testCollection = createCollection<TestItem, number>(
+        localOnlyCollectionOptions({
+          id: `numbers`,
+          getKey: (item) => item.id,
+          initialData: [
+            { id: 0, number: 15 },
+            { id: 1, number: 15 },
+            { id: 2, number: 15 },
+          ] as Array<TestItem>,
+          onDelete: () => {
+            return Promise.resolve()
+          },
+          autoIndex: `off`,
+        })
+      )
+
+      testCollection.subscribeChanges((changes) => {
+        console.log({ testCollectionChanges: changes })
+      })
+
+      const query = createCollection(
+        liveQueryCollectionOptions({
+          startSync: true,
+          query: (q) =>
+            q.from({ numbers: testCollection }).select(({ numbers }) => ({
+              totalNumber: sum(numbers.number),
+            })),
+        })
+      )
+
+      query.subscribeChanges((changes) => {
+        console.log({ queryChanges: changes })
+      })
+
+      testCollection.delete(0)
+
+      await new Promise((resolve) => setTimeout(resolve, 10))
+
+      expect(query.toArray).toEqual([{ totalNumber: 30 }])
+    })
+  })
 })