fix: invalidate stale predicate-scoped cache entries on manual-sync writes

.changeset/fix-cache-poisoning-on-manual-writes.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/query-db-collection': patch
+---
+
+Prevent on-demand manual-sync writes from replacing predicate, order, or pagination cache entries with the full synced snapshot. Refetch active enabled entries and remove inactive or disabled entries, while eager collections still patch their full-result cache.

docs/collections/query-collection.md

@@ -314,9 +314,9 @@ derived Query cache entries instead.
 
 If a stale initial response triggers a fetch, the initial rows remain available
 while it is in flight. A successful response reconciles them through the normal
-row ownership pipeline; an error retains the initial rows. Direct writes use the
-same Query cache-patching rules as fetched data, and a later successful server
-response may reconcile or replace those writes.
+row ownership pipeline; an error retains the initial rows. Direct writes patch
+the eager Query cache in place, and a later successful server response may
+reconcile or replace those writes.
 
 ### Selecting Rows from Wrapped Responses
 
@@ -356,7 +356,7 @@ preserving the envelope in the Query cache.
 
 This differs from TanStack Query's observer-level `select`: query-db-collection uses this option to bridge Query's response object into DB's normalized row store.
 
-Direct write utilities such as `writeInsert`, `writeUpdate`, and `writeDelete` make a best-effort attempt to update the matching row array inside wrapped Query cache entries while preserving wrapper metadata.
+In eager mode, direct write utilities such as `writeInsert`, `writeUpdate`, and `writeDelete` make a best-effort attempt to update the matching row array inside wrapped Query cache entries while preserving wrapper metadata. In on-demand mode, they revalidate active scoped queries and remove inactive or disabled cache entries instead of patching them with the full collection snapshot.
 
 This works automatically for simple wrappers such as:
 
@@ -566,7 +566,7 @@ The collection provides these utility methods via `collection.utils`:
 
 ## Direct Writes
 
-Direct writes are intended for scenarios where the normal query/mutation flow doesn't fit your needs. They allow you to write directly to the synced data store, bypassing the optimistic update system and query refetch mechanism.
+Direct writes are intended for scenarios where the normal query/mutation flow doesn't fit your needs. They write directly to the synced data store and bypass the optimistic update system. How they update the Query cache depends on the collection's sync mode.
 
 ### Understanding the Data Stores
 
@@ -621,9 +621,9 @@ These operations:
 
 - Write directly to the synced data store
 - Do NOT create optimistic mutations
-- Do NOT trigger automatic query refetches
-- Update the TanStack Query cache immediately
 - Are immediately visible in the UI
+- In eager mode, update the full-result Query cache in place without refetching
+- In on-demand mode, refetch active enabled queries and remove inactive or disabled cache entries
 
 ### Batch Operations
 
@@ -641,7 +641,7 @@ todosCollection.utils.writeBatch(() => {
 ### Real-World Example: WebSocket Integration
 
 ```typescript
-// Handle real-time updates from WebSocket without triggering full refetches
+// Apply real-time updates from a WebSocket to the synced store
 ws.on("todos:update", (changes) => {
   todosCollection.utils.writeBatch(() => {
     changes.forEach((change) => {
@@ -823,13 +823,15 @@ This pattern allows you to:
 
 ### Direct Writes and Query Sync
 
-Direct writes update the collection immediately and also update the TanStack Query cache. However, they do not prevent the normal query sync behavior. If your `queryFn` returns data that conflicts with your direct writes, the query data will take precedence.
+Direct writes update the collection immediately. In eager mode, they also patch the full-result TanStack Query cache in place.
+
+In on-demand mode, each Query cache entry may represent a different predicate, order, limit, or offset. A full collection snapshot cannot safely replace those scoped results. Direct writes therefore refetch active enabled queries and remove inactive or disabled entries so their next subscription fetches fresh data. A successful `queryFn` result remains authoritative and may reconcile or replace a direct write.
 
 To handle this properly:
 
-1. Use `{ refetch: false }` in your persistence handlers when using direct writes
-2. Set appropriate `staleTime` to prevent unnecessary refetches
-3. Design your `queryFn` to be aware of incremental updates (e.g., only fetch new data)
+1. Use `{ refetch: false }` in persistence handlers to avoid the handler's additional refetch after a direct write. On-demand cache revalidation still runs.
+2. Make sure an on-demand `queryFn` returns the current server result for its pushed-down predicate, order, limit, and offset.
+3. Use eager mode when direct writes must update one complete cached result without a network request.
 
 ## Complete Direct Write API Reference
 

docs/reference/query-db-collection/interfaces/QueryCollectionUtils.md

@@ -8,7 +8,8 @@ title: QueryCollectionUtils
 Defined in: [packages/query-db-collection/src/query.ts:235](https://github.com/TanStack/db/blob/main/packages/query-db-collection/src/query.ts#L235)
 
 Utility methods available on Query Collections for direct writes and manual operations.
-Direct writes bypass the normal query/mutation flow and write directly to the synced data store.
+Direct writes bypass optimistic mutations and write to the synced data store.
+Eager collections patch Query cache; on-demand collections revalidate scoped entries.
 
 ## Extends
 
@@ -187,7 +188,7 @@ writeBatch: (callback) => void;
 
 Defined in: [packages/query-db-collection/src/query.ts:252](https://github.com/TanStack/db/blob/main/packages/query-db-collection/src/query.ts#L252)
 
-Execute multiple write operations as a single atomic batch to the synced data store
+Execute direct writes as one atomic batch, then update or revalidate the Query cache
 
 #### Parameters
 
@@ -209,7 +210,7 @@ writeDelete: (keys) => void;
 
 Defined in: [packages/query-db-collection/src/query.ts:248](https://github.com/TanStack/db/blob/main/packages/query-db-collection/src/query.ts#L248)
 
-Delete one or more items directly from the synced data store without triggering a query refetch or optimistic update
+Delete items without an optimistic update. On-demand queries revalidate their scoped cache entries.
 
 #### Parameters
 
@@ -231,7 +232,7 @@ writeInsert: (data) => void;
 
 Defined in: [packages/query-db-collection/src/query.ts:244](https://github.com/TanStack/db/blob/main/packages/query-db-collection/src/query.ts#L244)
 
-Insert one or more items directly into the synced data store without triggering a query refetch or optimistic update
+Insert items without an optimistic update. On-demand queries revalidate their scoped cache entries.
 
 #### Parameters
 
@@ -253,7 +254,7 @@ writeUpdate: (updates) => void;
 
 Defined in: [packages/query-db-collection/src/query.ts:246](https://github.com/TanStack/db/blob/main/packages/query-db-collection/src/query.ts#L246)
 
-Update one or more items directly in the synced data store without triggering a query refetch or optimistic update
+Update items without an optimistic update. On-demand queries revalidate their scoped cache entries.
 
 #### Parameters
 
@@ -275,7 +276,7 @@ writeUpsert: (data) => void;
 
 Defined in: [packages/query-db-collection/src/query.ts:250](https://github.com/TanStack/db/blob/main/packages/query-db-collection/src/query.ts#L250)
 
-Insert or update one or more items directly in the synced data store without triggering a query refetch or optimistic update
+Insert or update items without an optimistic update. On-demand queries revalidate their scoped cache entries.
 
 #### Parameters
 

packages/query-db-collection/src/query.ts

@@ -247,7 +247,8 @@ export type RefetchFn = (opts?: {
 
 /**
  * Utility methods available on Query Collections for direct writes and manual operations.
- * Direct writes bypass the normal query/mutation flow and write directly to the synced data store.
+ * Direct writes bypass optimistic mutations and write to the synced data store.
+ * Eager collections patch Query cache; on-demand collections revalidate scoped entries.
  * @template TItem - The type of items stored in the collection
  * @template TKey - The type of the item keys
  * @template TInsertInput - The type accepted for insert operations
@@ -261,15 +262,15 @@ export interface QueryCollectionUtils<
 > extends UtilsRecord {
   /** Manually trigger a refetch of the query */
   refetch: RefetchFn
-  /** Insert one or more items directly into the synced data store without triggering a query refetch or optimistic update */
+  /** Insert items without an optimistic update. On-demand queries revalidate their scoped cache entries. */
   writeInsert: (data: TInsertInput | Array<TInsertInput>) => void
-  /** Update one or more items directly in the synced data store without triggering a query refetch or optimistic update */
+  /** Update items without an optimistic update. On-demand queries revalidate their scoped cache entries. */
   writeUpdate: (updates: Partial<TItem> | Array<Partial<TItem>>) => void
-  /** Delete one or more items directly from the synced data store without triggering a query refetch or optimistic update */
+  /** Delete items without an optimistic update. On-demand queries revalidate their scoped cache entries. */
   writeDelete: (keys: TKey | Array<TKey>) => void
-  /** Insert or update one or more items directly in the synced data store without triggering a query refetch or optimistic update */
+  /** Insert or update items without an optimistic update. On-demand queries revalidate their scoped cache entries. */
   writeUpsert: (data: Partial<TItem> | Array<Partial<TItem>>) => void
-  /** Execute multiple write operations as a single atomic batch to the synced data store */
+  /** Execute direct writes as one atomic batch, then update or revalidate the Query cache */
   writeBatch: (callback: () => void) => void
 
   // Query Observer State (getters)
@@ -799,6 +800,7 @@ export function queryCollectionOptions(
   // queryKey → QueryObserver's unsubscribe function
   const unsubscribes = new Map<string, () => void>()
   const pendingReadyUnsubscribes = new Map<string, Set<() => void>>()
+  const manualWriteDataUpdateCounts = new Map<string, number>()
 
   // queryKey → reference count (how many loadSubset calls are active)
   // Reference counting for QueryObserver lifecycle management
@@ -1545,6 +1547,21 @@ export function queryCollectionOptions(
             return
           }
 
+          const dataUpdateCountBeforeWrite =
+            manualWriteDataUpdateCounts.get(hashedQueryKey)
+          if (dataUpdateCountBeforeWrite !== undefined) {
+            const dataUpdateCount = state.observers
+              .get(hashedQueryKey)
+              ?.getCurrentQuery().state.dataUpdateCount
+            if (
+              dataUpdateCount === undefined ||
+              dataUpdateCount === dataUpdateCountBeforeWrite
+            ) {
+              return
+            }
+            manualWriteDataUpdateCounts.delete(hashedQueryKey)
+          }
+
           if (retainedQueriesPendingRevalidation.has(hashedQueryKey)) {
             const query = queryClient.getQueryCache().find({
               queryKey,
@@ -1693,6 +1710,7 @@ export function queryCollectionOptions(
       unsubscribePendingReadyListeners(hashedQueryKey)
       cancelPersistedRetentionExpiry(hashedQueryKey)
       retainedQueriesPendingRevalidation.delete(hashedQueryKey)
+      manualWriteDataUpdateCounts.delete(hashedQueryKey)
 
       const nextOwnersByRow = removeQueryOwnership(hashedQueryKey)
       const rowsToDelete: Array<any> = []
@@ -1804,6 +1822,7 @@ export function queryCollectionOptions(
         }
         unsubscribes.get(hashedQueryKey)?.()
         unsubscribes.delete(hashedQueryKey)
+        manualWriteDataUpdateCounts.delete(hashedQueryKey)
         state.observers.delete(hashedQueryKey)
         hashToQueryKey.delete(hashedQueryKey)
         queryRefCounts.set(hashedQueryKey, 0)
@@ -2016,29 +2035,61 @@ export function queryCollectionOptions(
   }
 
   /**
-   * Updates the query cache with new items for ALL query keys matching this collection,
-   * including stale/inactive cache entries from destroyed observers.
+   * Updates the query cache after a manual-sync write.
    *
-   * This prevents ghost items: when an observer is destroyed but gcTime > 0, TanStack Query
-   * keeps the cached data. If syncedData changes (via writeDelete/writeInsert/writeUpdate)
-   * after the observer is destroyed, the stale cache becomes inconsistent. When a new observer
-   * later picks up this stale cache, makeQueryResultHandler would create spurious sync
-   * operations (re-inserting deleted items, reverting updated values, etc).
-   *
-   * By updating all cache entries (active and stale), we ensure the cache always reflects
-   * the current syncedData state.
+   * An on-demand cache entry belongs to one queryFn result, which may include
+   * predicates, ordering, and pagination. The full syncedData snapshot cannot
+   * preserve that shape. Revalidate active entries against the source of truth
+   * and remove inactive entries so a later subscription fetches them again.
    */
   const updateCacheData = (items: Array<any>): void => {
-    const allCached = queryClient.getQueryCache().findAll({ queryKey: baseKey })
+    if (syncMode === `on-demand`) {
+      const refetchingQueries = new Set<string>()
+
+      for (const [queryHash, observer] of state.observers) {
+        const query = observer.getCurrentQuery()
+        if (query.isDisabled()) {
+          continue
+        }
+
+        refetchingQueries.add(queryHash)
+        manualWriteDataUpdateCounts.set(queryHash, query.state.dataUpdateCount)
+
+        const refetchObserver = () => {
+          if (state.observers.get(queryHash) !== observer) {
+            return
+          }
+          observer.refetch().catch(() => {
+            // Errors are handled by the query result handler.
+          })
+        }
 
-    if (allCached.length > 0) {
-      for (const query of allCached) {
-        updateCacheDataForKey(query.queryKey, items)
+        const deferredRefresh = writeContext?.collection.deferDataRefresh
+        if (deferredRefresh) {
+          void deferredRefresh.then(refetchObserver, refetchObserver)
+        } else {
+          refetchObserver()
+        }
       }
-    } else {
+
+      queryClient.removeQueries({
+        queryKey: baseKey,
+        predicate: (query) => !refetchingQueries.has(query.queryHash),
+      })
+      return
+    }
+
+    const allCached = queryClient.getQueryCache().findAll({ queryKey: baseKey })
+
+    if (allCached.length === 0) {
       // Fallback: no queries in cache yet, seed the base query key.
       // This handles the case where updateCacheData is called before any queries are created.
       updateCacheDataForKey(baseKey, items)
+      return
+    }
+
+    for (const query of allCached) {
+      updateCacheDataForKey(query.queryKey, items)
     }
   }