fix(db): roll back a direct write whose subscriber throws

.changeset/fix-direct-write-subscriber-throw.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/db': patch
+---
+
+A direct write (`insert`, `update` or `delete` outside a transaction) whose change subscriber throws no longer leaves a pending transaction behind. Before, the write's transaction stayed pending with its optimistic row visible, its handler never ran, and the Collection kept tracking it. The write now rolls back its own transaction, without rolling back other transactions, and rethrows the subscriber's error. A local-only direct write that takes the fast path (the Collection has no handler for that operation and no other pending or persisting transaction) stores its rows when it publishes them, so it completes its transaction and rethrows. On the transaction path it rolls back and rethrows, as any other direct write does.

packages/db/src/collection/mutations.ts

@@ -3,6 +3,7 @@ import {
   withChangeTracking,
   withFlatChangeTracking,
 } from '../proxy'
+import { normalizeError } from '../utils/error.js'
 import { safeRandomUUID } from '../utils/uuid'
 import { createTransaction, getActiveTransaction } from '../transactions'
 import {
@@ -219,9 +220,14 @@ export class CollectionMutationsManager<
       mutationFn: () => Promise.resolve(),
     })
     transaction.applyMutations(mutations)
-    direct.write(mutations)
-    transaction.setState(`completed`)
-    transaction.isPersisted.resolve(transaction)
+    // The rows are stored when `write` publishes them, so a throwing
+    // subscriber cannot undo the write; the transaction still completes.
+    try {
+      direct.write(mutations)
+    } finally {
+      transaction.setState(`completed`)
+      transaction.isPersisted.resolve(transaction)
+    }
     return transaction
   }
 
@@ -268,7 +274,22 @@ export class CollectionMutationsManager<
           >,
         }),
     })
-    this.applyOwnedMutations(transaction, mutations)
+    try {
+      this.applyOwnedMutations(transaction, mutations)
+    } catch (error) {
+      // A subscriber threw before the handler could run. Roll back only this
+      // write, so no Collection keeps a transaction that nothing will settle.
+      // The admission error is the reported cause.
+      try {
+        transaction.rollback({
+          isSecondaryRollback: true,
+          error: normalizeError(error),
+        })
+      } catch {
+        // Settlement still ran every step.
+      }
+      throw error
+    }
     // Errors still reject `isPersisted.promise`. This catch only prevents an
     // unhandled rejection from the fire-and-forget commit.
     transaction.commit().catch(() => undefined)

packages/db/tests/change-event-history-oracle.test.ts

@@ -10,6 +10,7 @@ import {
 import { localOnlyCollectionOptions } from '../src/local-only'
 import { PropRef } from '../src/query/ir'
 import { oraclePropertyOptions, oracleRuns } from './oracle-config'
+import { captureCreatedTransactions } from './utils'
 import type { LocalOnlyCollectionUtils } from '../src/local-only'
 import type { Collection } from '../src/index'
 import type { BaseIndex } from '../src/indexes/base-index'
@@ -47,6 +48,13 @@ import type { ChangeMessage, SyncConfig } from '../src/types'
  * repeated changes and complete row values, while leaving cross-key
  * interleaving unconstrained.
  *
+ * About one generated history in four meets a second subscriber, registered
+ * after the mirror, that throws while one operation publishes. A local-only
+ * direct write stores its rows as it publishes, so the call rethrows, its
+ * transaction is already completed, and the rows and mirror still follow the
+ * model. Whether a throwing subscriber may stop a later subscriber of the same
+ * publication is not decided here; the throwing subscriber is always last.
+ *
  * This partial oracle does not cover failed persistence, sync-transaction
  * cancellation, general callback batch shape, callback-time index agreement, or
  * publications without change messages.
@@ -84,6 +92,8 @@ type GeneratedHistory = {
   initialKeys: ReadonlyArray<Key>
   sequence: ReadonlyArray<Op>
   mode: `batched` | `sequential`
+  /** The operation whose publication meets a throwing subscriber. */
+  throwOn?: number
 }
 
 type GeneratedStep = { key: Key; deleteWhenPresent: boolean }
@@ -359,6 +369,7 @@ async function runHistory(
   mode: `batched` | `sequential`,
   id: string,
   indexType?: IndexType,
+  throwOn?: number,
 ): Promise<void> {
   const initialRows = [...expectedRowsAfter(initialKeys, [])].map(
     ([, row]) => row,
@@ -436,9 +447,46 @@ async function runHistory(
         ? collection.subscribeChanges(onChanges, { includeInitialState: false })
         : collection.subscribeChanges(onChanges)
 
+    // A second subscriber, after the mirror, throws while one operation
+    // publishes. A local-only direct write stores its rows as it publishes
+    // them, so the throw cannot undo the write: the call rethrows, and its
+    // transaction still settles. The rows and the mirror follow the model.
+    const injected = new Error(`subscriber failed`)
+    let armed = false
+    const throwing =
+      throwOn === undefined
+        ? undefined
+        : collection.subscribeChanges(
+            () => {
+              if (armed) throw injected
+            },
+            { includeInitialState: true },
+          )
+    const { created, restore: restoreCreate } =
+      captureCreatedTransactions(collection)
+    const applyAt = (op: Op, operationIndex: number) => {
+      if (operationIndex !== throwOn) return applyOp(collection, op)
+      armed = true
+      const before = created.length
+      try {
+        applyOp(collection, op)
+      } catch (error) {
+        expect(error, `operation ${operationIndex} rethrows`).toBe(injected)
+        expect(created.length, `one transaction`).toBe(before + 1)
+        expect(
+          created[before]!.state,
+          `operation ${operationIndex} settles`,
+        ).toBe(`completed`)
+        return created[before]!
+      } finally {
+        armed = false
+      }
+      throw new Error(`operation ${operationIndex} did not meet the throw`)
+    }
+
     if (mode === `batched`) {
       const transactions = sequence.map((op, operationIndex) => {
-        const transaction = applyOp(collection, op)
+        const transaction = applyAt(op, operationIndex)
         if (index) {
           const expected = expectedRowsAfter(
             initialKeys,
@@ -455,8 +503,8 @@ async function runHistory(
       if (index) assertIndexAgreement(index, expected)
     } else {
       const settled: Array<Op> = []
-      for (const op of sequence) {
-        await settle(applyOp(collection, op))
+      for (const [operationIndex, op] of sequence.entries()) {
+        await settle(applyAt(op, operationIndex))
         settled.push(op)
         const expected = expectedRowsAfter(initialKeys, settled)
         assertFinalAgreement(collection, mirror, expected)
@@ -473,6 +521,8 @@ async function runHistory(
     for (const publication of publications) {
       expect(publication.mirrorRows).toEqual(publication.publicRows)
     }
+    throwing?.unsubscribe()
+    restoreCreate()
     expect(protocolViolations, `change-message protocol`).toEqual([])
   } catch (error) {
     primaryFailure = error
@@ -509,12 +559,15 @@ async function verifyGeneratedHistory({
   initialKeys,
   sequence,
   mode,
+  throwOn,
 }: GeneratedHistory): Promise<void> {
   await runHistory(
     sequence,
     initialKeys,
     mode,
     `history-oracle-generated-${initialKeys.join(``)}-${mode}-${describeSequence(sequence)}`,
+    undefined,
+    throwOn === undefined ? undefined : throwOn % sequence.length,
   )
 }
 
@@ -660,10 +713,13 @@ describe(`change-event history oracle`, () => {
         { minLength: 1, maxLength: 20 },
       ),
       mode: fc.constantFrom(`batched` as const, `sequential` as const),
+      // About two histories in three arm a throwing subscriber; about one in
+      // four reaches the operation it throws on.
+      throwOn: fc.option(fc.nat(19), { nil: undefined, freq: 2 }),
     })
-    .map(({ initialKeys, steps, mode }) => {
+    .map(({ initialKeys, steps, mode, throwOn }) => {
       const sequence = opsFromPresence(initialKeys, steps, 1)
-      return { initialKeys, sequence, mode }
+      return { initialKeys, sequence, mode, throwOn }
     })
 
   // The fixed prefix forces reused keys through an already-built index.

packages/db/tests/local-only-direct-write-oracle.test.ts

@@ -4,6 +4,7 @@ import { createCollection } from '../src/index'
 import { localOnlyCollectionOptions } from '../src/local-only'
 import { createTransaction } from '../src/transactions'
 import { createDeferred } from '../src/deferred'
+import { captureCreatedTransactions } from './utils'
 
 /**
  * # When does a local-only direct write skip the optimistic stage?
@@ -138,6 +139,88 @@ describe(`local-only direct writes`, () => {
     expect(orders.get(1)?.value).toBe(`a`)
   })
 
+  // The direct write stores its rows when it publishes them, so a throwing
+  // subscriber cannot undo the write. The call rethrows, and the transaction
+  // still completes instead of staying pending.
+  it(`completes a direct write whose subscriber throws`, async () => {
+    const orders = createOrders()
+    const { created, restore: restoreCreate } =
+      captureCreatedTransactions(orders)
+    const failure = new Error(`subscriber failed`)
+    const subscription = orders.subscribeChanges(() => {
+      throw failure
+    })
+    expect(() => orders.insert({ id: 2, value: `b` })).toThrow(failure)
+    subscription.unsubscribe()
+    restoreCreate()
+
+    expect(orders.get(2)?.value).toBe(`b`)
+    expect(created.map((transaction) => transaction.state)).toEqual([
+      `completed`,
+    ])
+    await expect(created[0]!.isPersisted.promise).resolves.toBe(created[0])
+    // The next direct write still takes the direct path.
+    expect(orders.insert({ id: 3, value: `c` }).state).toBe(`completed`)
+  })
+
+  // A subscriber may write the key it was told about, as a normalizing or
+  // rejecting subscriber does. The first write stored its rows before that,
+  // so it completes and does not throw.
+  for (const kind of [`update`, `delete`] as const) {
+    it(`completes a direct insert whose subscriber ${kind}s the same key`, async () => {
+      const orders = createOrders()
+      const { created, restore: restoreCreate } =
+        captureCreatedTransactions(orders)
+      let reacted = false
+      const subscription = orders.subscribeChanges((changes) => {
+        if (reacted || changes.every((change) => change.key !== 2)) return
+        reacted = true
+        if (kind === `update`)
+          orders.update(2, (draft) => {
+            draft.value = `normalized`
+          })
+        else orders.delete(2)
+      })
+      try {
+        expect(() => orders.insert({ id: 2, value: `b` })).not.toThrow()
+      } finally {
+        subscription.unsubscribe()
+        restoreCreate()
+      }
+      expect(orders.get(2)?.value).toBe(
+        kind === `update` ? `normalized` : undefined,
+      )
+      await Promise.all(
+        created.map((transaction) => transaction.isPersisted.promise),
+      )
+      expect(created.map((transaction) => transaction.state)).toEqual([
+        `completed`,
+        `completed`,
+      ])
+    })
+  }
+
+  // A subscriber may throw a value that is not an Error. The rolled-back
+  // transaction still records an Error, so its receipt names the failure.
+  it(`records an Error when a direct write's subscriber throws a non-Error`, async () => {
+    const orders = createOrders({ onInsert: async () => {} })
+    const { created, restore: restoreCreate } =
+      captureCreatedTransactions(orders)
+    const subscription = orders.subscribeChanges(() => {
+      throw `bad`
+    })
+    try {
+      expect(() => orders.insert({ id: 2, value: `b` })).toThrow(`bad`)
+    } finally {
+      subscription.unsubscribe()
+      restoreCreate()
+    }
+    expect(created.map((transaction) => transaction.state)).toEqual([`failed`])
+    expect(created[0]!.error?.error).toBeInstanceOf(Error)
+    expect(created[0]!.error?.message).toBe(`bad`)
+    await expect(created[0]!.isPersisted.promise).rejects.toBeInstanceOf(Error)
+  })
+
   it(`runs the user's handler for that operation type`, async () => {
     let calls = 0
     const orders = createOrders({