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,11 @@
+---
+'@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. Preserve causal publication, cancellation, 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

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

@@ -22,6 +22,7 @@ import type {
   InsertMutationFnParams,
   LoadSubsetOptions,
   PendingMutation,
+  SyncAppliedReceipt,
   SyncConfig,
   SyncConfigRes,
   SyncMetadataApi,
@@ -433,7 +434,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 +587,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 +815,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 +840,31 @@ 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)
+    void receipt.then(() => this.pendingAppliedReceipts.delete(sequence))
+    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 +936,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 +1017,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 +1190,18 @@ class PersistedCollectionRuntime<
 
     this.pendingRemoteSubsetEnsures.clear()
     this.activeSubsets.clear()
+    for (const transaction of this.queuedHydrationTransactions) {
+      transaction.resolveApplied?.()
+    }
     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 +1348,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.resolveApplied?.()
+      return
+    }
+
+    const { begin, write, commit, truncate, metadata } = this.syncControls
+    if (!begin || !write || !commit) {
+      transaction.resolveApplied?.()
       return
     }
 
-    const applyToCollection = () => {
-      this.syncControls.begin?.()
+    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 +1399,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 && !transaction.signal?.aborted) {
+        await this.persistAndBroadcastExternalSyncTransactionUnsafe(transaction)
+      }
+      transaction.resolveApplied?.()
+    } catch (error) {
+      transaction.rejectApplied?.(error)
+      throw error
+    }
   }
 
   private async persistAndBroadcastExternalSyncTransactionUnsafe(
@@ -2457,43 +2515,51 @@ 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) return true
+            let resolveApplied!: () => void
+            let rejectApplied!: (error: unknown) => void
+            const applied = new Promise<void>((resolve, reject) => {
+              resolveApplied = resolve
+              rejectApplied = reject
+            })
             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
+              if (signal?.aborted) return
+              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,
-                )
-              })
+            }
+            return persistAfterApplication()
           }
+          return applied
         },
       }