fix(db): settle subset loads after publication

.changeset/canonical-demand-identity.md

@@ -0,0 +1,6 @@
+---
+'@tanstack/db': patch
+'@tanstack/query-db-collection': patch
+---
+
+Canonicalize equivalent loadSubset queries to one demand identity while preserving observable output aliases, exact projected values, and distinct ordered windows. Query DB now reuses the same canonical identity for its on-demand cache keys.

.changeset/settle-subset-after-publication.md

@@ -0,0 +1,15 @@
+---
+'@tanstack/db': patch
+'@tanstack/db-sqlite-persistence-core': patch
+'@tanstack/electric-db-collection': patch
+'@tanstack/powersync-db-collection': patch
+'@tanstack/query-db-collection': patch
+'@tanstack/rxdb-db-collection': patch
+'@tanstack/trailbase-db-collection': patch
+---
+
+Settle subset loads only after their committed rows and events are visible. A
+commit receipt now rejects with `AbortError` when cancellation wins before
+application and ignores later aborts. Preserve causal publication,
+cancellation, persistence, and error handling across the affected sync
+adapters.

AGENTS.md

@@ -379,6 +379,16 @@ test('ignores snapshot that resolves after up-to-date message', async () => {
 })
 ```
 
+### Treat Every Review Bug as a Test Gap
+
+When a reviewer agent confirms a bug, it must also ask why the existing tests
+did not catch it. The finding should name the missing test law, state
+transition, generator dimension, adapter boundary, or assertion. If a test or
+oracle should already have caught the bug, identify the false-green model,
+classifier, fixture, or assertion that let it pass. Use that analysis to suggest
+the smallest test or oracle improvement that would catch the same class of bug,
+not only the reported example.
+
 ### Name Tests After Behavior
 
 Test names should state the behavior they prove. Do not put issue or pull

docs/guides/mutations.md

@@ -432,6 +432,12 @@ const todoCollection = createCollection({
 
 > [!IMPORTANT]
 > Operation handlers must not resolve until the server changes have synced back to the collection. Different collection types provide different patterns to ensure this happens correctly.
+>
+> Do not call or await `collection.preload()`, live-query `preload()`, or a
+> direct `loadSubset()` inside a mutation handler. The optimistic mutation is
+> already applied when the handler starts. A preload may need a sync commit
+> that is queued behind that same handler, which creates a deadlock. Use the
+> collection adapter's documented mutation acknowledgement pattern instead.
 
 ### Collection-Specific Handler Patterns
 

packages/db-sqlite-persistence-core/src/persisted.ts

@@ -1,4 +1,5 @@
 import {
+  SyncTransactionAbortedError,
   compileSingleRowExpression,
   safeRandomUUID,
   toBooleanPredicate,
@@ -22,6 +23,7 @@ import type {
   InsertMutationFnParams,
   LoadSubsetOptions,
   PendingMutation,
+  SyncAppliedReceipt,
   SyncConfig,
   SyncConfigRes,
   SyncMetadataApi,
@@ -433,7 +435,7 @@ type SyncControlFns<T extends object, TKey extends string | number> = {
           | { type: `delete`; key: TKey },
       ) => void)
     | null
-  commit: (() => void) | null
+  commit: ((signal?: AbortSignal) => SyncAppliedReceipt) | null
   truncate: (() => void) | null
   metadata: SyncMetadataApi<TKey> | null
 }
@@ -586,6 +588,9 @@ type BufferedSyncTransaction<T extends object, TKey extends string | number> = {
   >
   truncate: boolean
   internal: boolean
+  signal?: AbortSignal
+  resolveApplied?: () => void
+  rejectApplied?: (error: unknown) => void
 }
 
 type OpenSyncTransaction<
@@ -811,6 +816,8 @@ class PersistedCollectionRuntime<
   private startupMetadataPromise: Promise<void> | null = null
   private startPromise: Promise<void> | null = null
   private internalApplyDepth = 0
+  private appliedReceiptSequence = 0
+  private readonly pendingAppliedReceipts = new Map<number, Promise<void>>()
   private isHydrating = false
   private coordinatorUnsubscribe: (() => void) | null = null
   private indexAddedUnsubscribe: (() => void) | null = null
@@ -834,7 +841,32 @@ class PersistedCollectionRuntime<
   ) {}
 
   setSyncControls(syncControls: SyncControlFns<T, TKey>): void {
-    this.syncControls = syncControls
+    const commit = syncControls.commit
+    this.syncControls = {
+      ...syncControls,
+      commit: commit
+        ? (signal) => this.trackAppliedReceipt(commit(signal))
+        : null,
+    }
+  }
+
+  private trackAppliedReceipt(receipt: SyncAppliedReceipt): SyncAppliedReceipt {
+    const sequence = ++this.appliedReceiptSequence
+    if (receipt === true) {
+      return true
+    }
+    this.pendingAppliedReceipts.set(sequence, receipt)
+    const removeReceipt = () => this.pendingAppliedReceipts.delete(sequence)
+    void receipt.then(removeReceipt, removeReceipt)
+    return receipt
+  }
+
+  private async waitForAppliedReceiptsAfter(cursor: number): Promise<void> {
+    await Promise.all(
+      Array.from(this.pendingAppliedReceipts, ([sequence, receipt]) =>
+        sequence > cursor ? receipt : undefined,
+      ),
+    )
   }
 
   clearSyncControls(): void {
@@ -906,9 +938,11 @@ class PersistedCollectionRuntime<
 
     if (this.syncMode !== `on-demand`) {
       this.activeSubsets.set(this.getSubsetKey({}), {})
+      const appliedCursor = this.appliedReceiptSequence
       await this.applyMutex.run(() =>
         this.hydrateSubsetUnsafe({}, { requestRemoteEnsure: false }),
       )
+      await this.waitForAppliedReceiptsAfter(appliedCursor)
     }
   }
 
@@ -985,17 +1019,19 @@ class PersistedCollectionRuntime<
   ): Promise<void> {
     this.activeSubsets.set(this.getSubsetKey(options), options)
 
+    const appliedCursor = this.appliedReceiptSequence
     await this.applyMutex.run(() =>
       this.hydrateSubsetUnsafe(options, {
         requestRemoteEnsure: this.mode === `sync-present`,
       }),
     )
+    await this.waitForAppliedReceiptsAfter(appliedCursor)
 
     if (upstreamLoadSubset) {
       try {
         const maybePromise = upstreamLoadSubset(options)
         if (maybePromise instanceof Promise) {
-          maybePromise.catch((error) => {
+          await maybePromise.catch((error) => {
             console.warn(
               `Failed to load remote subset in persisted wrapper:`,
               error,
@@ -1156,15 +1192,18 @@ class PersistedCollectionRuntime<
 
     this.pendingRemoteSubsetEnsures.clear()
     this.activeSubsets.clear()
+    for (const transaction of this.queuedHydrationTransactions) {
+      transaction.rejectApplied?.(new SyncTransactionAbortedError())
+    }
     this.queuedHydrationTransactions.length = 0
     this.queuedTxCommitted.length = 0
     this.clearSyncControls()
   }
 
-  private withInternalApply(task: () => void): void {
+  private withInternalApply<TResult>(task: () => TResult): TResult {
     this.internalApplyDepth++
     try {
-      task()
+      return task()
     } finally {
       this.internalApplyDepth--
     }
@@ -1311,36 +1350,48 @@ class PersistedCollectionRuntime<
       if (!transaction) {
         continue
       }
-      await this.applyBufferedSyncTransactionUnsafe(transaction)
+      try {
+        await this.applyBufferedSyncTransactionUnsafe(transaction)
+      } catch (error) {
+        transaction.rejectApplied?.(error)
+        for (const abandoned of this.queuedHydrationTransactions) {
+          abandoned.rejectApplied?.(error)
+        }
+        this.queuedHydrationTransactions.length = 0
+        throw error
+      }
     }
   }
 
   private async applyBufferedSyncTransactionUnsafe(
     transaction: BufferedSyncTransaction<T, TKey>,
   ): Promise<void> {
-    if (
-      !this.syncControls.begin ||
-      !this.syncControls.write ||
-      !this.syncControls.commit
-    ) {
+    if (transaction.signal?.aborted) {
+      transaction.rejectApplied?.(new SyncTransactionAbortedError())
       return
     }
 
-    const applyToCollection = () => {
-      this.syncControls.begin?.()
+    const { begin, write, commit, truncate, metadata } = this.syncControls
+    if (!begin || !write || !commit) {
+      transaction.rejectApplied?.(new SyncTransactionAbortedError())
+      return
+    }
+
+    const applyToCollection = (): SyncAppliedReceipt => {
+      begin()
 
       if (transaction.truncate) {
-        this.syncControls.truncate?.()
+        truncate?.()
       }
 
       for (const operation of transaction.operations) {
         if (operation.type === `delete`) {
-          this.syncControls.write?.({
+          write({
             type: `delete`,
             key: operation.key,
           })
         } else {
-          this.syncControls.write?.({
+          write({
             type: `update`,
             value: operation.value,
             metadata: operation.metadata,
@@ -1350,30 +1401,39 @@ class PersistedCollectionRuntime<
 
       for (const [key, metadataWrite] of transaction.rowMetadataWrites) {
         if (metadataWrite.type === `delete`) {
-          this.syncControls.metadata?.row.delete(key)
+          metadata?.row.delete(key)
         } else {
-          this.syncControls.metadata?.row.set(key, metadataWrite.value)
+          metadata?.row.set(key, metadataWrite.value)
         }
       }
 
       for (const [key, metadataWrite] of transaction.collectionMetadataWrites) {
         if (metadataWrite.type === `delete`) {
-          this.syncControls.metadata?.collection.delete(key)
+          metadata?.collection.delete(key)
         } else {
-          this.syncControls.metadata?.collection.set(key, metadataWrite.value)
+          metadata?.collection.set(key, metadataWrite.value)
         }
       }
 
-      this.syncControls.commit?.()
+      return commit(transaction.signal)
     }
 
-    if (transaction.internal) {
-      this.withInternalApply(applyToCollection)
-      return
-    }
+    try {
+      const applied = transaction.internal
+        ? this.withInternalApply(applyToCollection)
+        : applyToCollection()
+      if (applied !== true) {
+        await applied
+      }
 
-    applyToCollection()
-    await this.persistAndBroadcastExternalSyncTransactionUnsafe(transaction)
+      if (!transaction.internal) {
+        await this.persistAndBroadcastExternalSyncTransactionUnsafe(transaction)
+      }
+      transaction.resolveApplied?.()
+    } catch (error) {
+      transaction.rejectApplied?.(error)
+      throw error
+    }
   }
 
   private async persistAndBroadcastExternalSyncTransactionUnsafe(
@@ -2457,43 +2517,57 @@ function createWrappedSyncConfig<
             params.truncate()
           }
         },
-        commit: () => {
+        commit: (signal?: AbortSignal) => {
           const openTransaction = transactionStack.pop()
           if (!openTransaction) {
-            params.commit()
-            return
+            return params.commit(signal)
           }
 
           if (openTransaction.queuedBecauseHydrating) {
+            if (signal?.aborted) {
+              const aborted = Promise.reject(new SyncTransactionAbortedError())
+              void aborted.catch(() => undefined)
+              return aborted
+            }
+            let resolveApplied!: () => void
+            let rejectApplied!: (error: unknown) => void
+            const applied = new Promise<void>((resolve, reject) => {
+              resolveApplied = resolve
+              rejectApplied = reject
+            })
+            void applied.catch(() => undefined)
             runtime.queueHydrationBufferedTransaction({
               operations: openTransaction.operations,
               rowMetadataWrites: openTransaction.rowMetadataWrites,
               collectionMetadataWrites:
                 openTransaction.collectionMetadataWrites,
               truncate: openTransaction.truncate,
               internal: openTransaction.internal,
+              signal,
+              resolveApplied,
+              rejectApplied,
             })
-            return
+            return applied
           }
 
-          params.commit()
+          const applied = params.commit(signal)
           if (!openTransaction.internal) {
-            void runtime
-              .persistAndBroadcastExternalSyncTransaction({
+            const persistAfterApplication = async () => {
+              if (applied !== true) await applied
+              await runtime.persistAndBroadcastExternalSyncTransaction({
                 operations: openTransaction.operations,
                 rowMetadataWrites: openTransaction.rowMetadataWrites,
                 collectionMetadataWrites:
                   openTransaction.collectionMetadataWrites,
                 truncate: openTransaction.truncate,
                 internal: false,
               })
-              .catch((error) => {
-                console.warn(
-                  `Failed to persist wrapped sync transaction:`,
-                  error,
-                )
-              })
+            }
+            const persisted = persistAfterApplication()
+            void persisted.catch(() => undefined)
+            return persisted
           }
+          return applied
         },
       }