From b71e4a96c507b2c38b59ae88ff555caec89b06e3 Mon Sep 17 00:00:00 2001 From: Marc MacLeod Date: Sun, 26 Apr 2026 17:52:02 -0500 Subject: [PATCH 1/3] fix: invalidate stale predicate-scoped cache entries on manual-sync writes --- .../fix-cache-poisoning-on-manual-writes.md | 9 + packages/query-db-collection/src/query.ts | 37 ++-- .../query-db-collection/tests/query.test.ts | 169 ++++++++++++++++++ 3 files changed, 201 insertions(+), 14 deletions(-) create mode 100644 .changeset/fix-cache-poisoning-on-manual-writes.md diff --git a/.changeset/fix-cache-poisoning-on-manual-writes.md b/.changeset/fix-cache-poisoning-on-manual-writes.md new file mode 100644 index 0000000000..38f1addc2e --- /dev/null +++ b/.changeset/fix-cache-poisoning-on-manual-writes.md @@ -0,0 +1,9 @@ +--- +'@tanstack/query-db-collection': patch +--- + +fix: invalidate stale predicate-scoped cache entries on manual-sync writes + +In `syncMode: 'on-demand'`, manual-sync writes (`writeInsert`/`writeUpdate`/`writeDelete`/`writeUpsert`/`writeBatch`) used to overwrite every cache entry under the collection's `queryKey` with the full post-write `syncedData` snapshot — regardless of the predicate that originally produced each entry. For stale entries (no live observer) within `gcTime`, this stamped them with rows that didn't satisfy their original `where`. A subsequent cache-hit re-subscribe re-applied those wrong rows via `applySuccessfulResult`, the per-subscription `where` filter discarded them, and the subscriber received `[]` for predicates whose matching rows still existed in the source. + +Stale cache entries are now invalidated (`removeQueries`) instead of overwritten, forcing the next subscribe to re-run `queryFn` against the source of truth. Active entries continue to receive the full snapshot (predicate scoping for the consumer is enforced downstream by `subscribeChanges`'s `where`). The original ghost-row protection is preserved because deleted rows no longer reappear from a stale snapshot. diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index b29aac8730..d84bdd74ed 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -1813,29 +1813,38 @@ 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). + * Active entries (observer still subscribed): write the full syncedData + * snapshot back. Predicate scoping for the consumer is enforced downstream + * by the per-subscription `where` filter on `subscribeChanges`. * - * By updating all cache entries (active and stale), we ensure the cache always reflects - * the current syncedData state. + * Stale entries (no live observer): invalidate by removing the entry. Each + * entry was originally produced by `queryFn` for a specific predicate, so + * stamping it with the full snapshot would poison it — rows that don't + * satisfy the predicate end up in the cache, then a cache-hit re-subscribe + * within `gcTime` re-applies those wrong rows via `applySuccessfulResult`, + * the subscription's `where` filter discards them, and the subscriber sees + * `[]`. The next subscribe will re-run `queryFn` against the source of + * truth, which also preserves ghost-row protection (deleted rows won't + * reappear from a stale snapshot). */ const updateCacheData = (items: Array): void => { const allCached = queryClient.getQueryCache().findAll({ queryKey: baseKey }) - if (allCached.length > 0) { - for (const query of allCached) { - updateCacheDataForKey(query.queryKey, items) - } - } else { + 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) { + if (!state.observers.has(hashKey(query.queryKey))) { + queryClient.removeQueries({ queryKey: query.queryKey, exact: true }) + continue + } + updateCacheDataForKey(query.queryKey, items) } } diff --git a/packages/query-db-collection/tests/query.test.ts b/packages/query-db-collection/tests/query.test.ts index 113fc0e1ea..4e9cb9a23e 100644 --- a/packages/query-db-collection/tests/query.test.ts +++ b/packages/query-db-collection/tests/query.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { QueryClient, hashKey } from '@tanstack/query-core' import { BTreeIndex, + compileSingleRowExpression, createCollection, createLiveQueryCollection, eq, @@ -6342,4 +6343,172 @@ describe(`QueryCollection`, () => { customQueryClient.clear() }) }) + + describe(`Predicate-scoped cache invalidation on manual writes`, () => { + // Each cache entry under an on-demand collection's queryKey was originally + // produced by queryFn with a specific where predicate. A manual-sync write + // (writeInsert/writeUpdate/writeDelete/writeUpsert) used to overwrite every + // such entry with the full post-write syncedData snapshot, including rows + // that didn't satisfy the entry's predicate. After unsubscribe + re-subscribe + // within gcTime, the cache-hit re-applied those wrong rows, the per- + // subscription where filter discarded them, and the subscriber saw []. + // + // The fix invalidates stale (no-live-observer) cache entries on each write, + // forcing the next subscribe to re-run queryFn against the source of truth. + + interface ScopedRow { + id: string + category: string + } + + function makeQueryFn(getServerRows: () => Array) { + return async (ctx: QueryFunctionContext) => { + const where = (ctx.meta as { loadSubsetOptions?: { where?: any } }) + .loadSubsetOptions?.where + if (!where) return getServerRows() + const evaluator = compileSingleRowExpression(where) + return getServerRows().filter((r) => + evaluator(r as unknown as Record), + ) + } + } + + function makeCollection(serverRows: Array) { + const customQueryClient = new QueryClient({ + defaultOptions: { + queries: { + gcTime: 5 * 60 * 1000, + staleTime: Infinity, + retry: false, + }, + }, + }) + + const collection = createCollection( + queryCollectionOptions({ + id: `predicate-cache-${Math.random()}`, + queryClient: customQueryClient, + queryKey: [`predicate-cache-test`], + syncMode: `on-demand`, + getKey: (r) => r.id, + queryFn: makeQueryFn(() => serverRows), + onInsert: async () => ({ refetch: false }), + onUpdate: async () => ({ refetch: false }), + onDelete: async () => ({ refetch: false }), + }), + ) + + return { collection, customQueryClient } + } + + it(`should return correct rows on re-subscribe after an unrelated writeUpsert`, async () => { + const serverRows: Array = [ + { id: `1`, category: `A` }, + { id: `2`, category: `B` }, + ] + const { collection, customQueryClient } = makeCollection(serverRows) + + // Subscribe to category=A, then unsubscribe. Leaves a stale cache entry + // whose contents should be [{id:1}]. + const wave1 = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, `A`)), + }) + await wave1.preload() + expect(wave1.toArray.map((r) => r.id)).toEqual([`1`]) + await wave1.cleanup() + + // Manual write of a row outside the predicate. Before the fix, this + // overwrote every cache entry with the full snapshot, poisoning + // category=A's stale entry with [{id:1},{id:2},{id:99}]. + collection.utils.writeUpsert({ id: `99`, category: `unrelated` }) + + // Re-subscribe to the same predicate. With the fix, the stale entry was + // invalidated and queryFn re-runs, returning only the matching row. + const wave2 = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, `A`)), + }) + await wave2.preload() + expect(wave2.toArray.map((r) => r.id)).toEqual([`1`]) + + await wave2.cleanup() + customQueryClient.clear() + }) + + it(`should pick up newly inserted rows that match the predicate on re-subscribe`, async () => { + const serverRows: Array = [{ id: `1`, category: `A` }] + const { collection, customQueryClient } = makeCollection(serverRows) + + const wave1 = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, `A`)), + }) + await wave1.preload() + expect(wave1.toArray.map((r) => r.id)).toEqual([`1`]) + await wave1.cleanup() + + // Insert a new matching row both server-side and via writeInsert. + serverRows.push({ id: `2`, category: `A` }) + collection.utils.writeInsert({ id: `2`, category: `A` }) + + const wave2 = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, `A`)), + }) + await wave2.preload() + expect(wave2.toArray.map((r) => r.id).sort()).toEqual([`1`, `2`]) + + await wave2.cleanup() + customQueryClient.clear() + }) + + it(`should not poison a stale narrow predicate when a write lands in a broader active predicate`, async () => { + const serverRows: Array = [{ id: `1`, category: `A` }] + const { collection, customQueryClient } = makeCollection(serverRows) + + // Narrow predicate (id=1) subscribes then unsubscribes — leaves a stale + // cache entry containing [{id:1}]. + const narrow = createLiveQueryCollection({ + query: (q) => + q.from({ item: collection }).where(({ item }) => eq(item.id, `1`)), + }) + await narrow.preload() + await narrow.cleanup() + + // Broader predicate (category=A) becomes active. + const broad = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, `A`)), + }) + await broad.preload() + + // Insert a row in the broader scope but outside the narrow predicate. + serverRows.push({ id: `2`, category: `A` }) + collection.utils.writeInsert({ id: `2`, category: `A` }) + + // Re-subscribing to the narrow predicate must still return only id=1, not + // be poisoned by id=2 from the active broader entry. + const narrow2 = createLiveQueryCollection({ + query: (q) => + q.from({ item: collection }).where(({ item }) => eq(item.id, `1`)), + }) + await narrow2.preload() + expect(narrow2.toArray.map((r) => r.id)).toEqual([`1`]) + + await narrow2.cleanup() + await broad.cleanup() + customQueryClient.clear() + }) + }) }) From 4fa5b8ea184eb6b13297082aebca6704221d2792 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Tue, 21 Jul 2026 13:49:16 -0600 Subject: [PATCH 2/3] fix(query-db): revalidate on-demand cache after direct writes --- .../fix-cache-poisoning-on-manual-writes.md | 6 +- docs/collections/query-collection.md | 26 +- .../interfaces/QueryCollectionUtils.md | 13 +- packages/query-db-collection/src/query.ts | 88 +++- .../query-db-collection/tests/query.test.ts | 436 ++++++++++++++---- 5 files changed, 429 insertions(+), 140 deletions(-) diff --git a/.changeset/fix-cache-poisoning-on-manual-writes.md b/.changeset/fix-cache-poisoning-on-manual-writes.md index 38f1addc2e..5b7b957184 100644 --- a/.changeset/fix-cache-poisoning-on-manual-writes.md +++ b/.changeset/fix-cache-poisoning-on-manual-writes.md @@ -2,8 +2,8 @@ '@tanstack/query-db-collection': patch --- -fix: invalidate stale predicate-scoped cache entries on manual-sync writes +fix: invalidate predicate-scoped cache entries on manual-sync writes -In `syncMode: 'on-demand'`, manual-sync writes (`writeInsert`/`writeUpdate`/`writeDelete`/`writeUpsert`/`writeBatch`) used to overwrite every cache entry under the collection's `queryKey` with the full post-write `syncedData` snapshot — regardless of the predicate that originally produced each entry. For stale entries (no live observer) within `gcTime`, this stamped them with rows that didn't satisfy their original `where`. A subsequent cache-hit re-subscribe re-applied those wrong rows via `applySuccessfulResult`, the per-subscription `where` filter discarded them, and the subscriber received `[]` for predicates whose matching rows still existed in the source. +In `syncMode: 'on-demand'`, manual-sync writes (`writeInsert`, `writeUpdate`, `writeDelete`, `writeUpsert`, and `writeBatch`) no longer retain the full post-write `syncedData` snapshot in predicate-scoped Query cache entries. Those entries may also encode ordering and pagination, so a normalized collection snapshot cannot preserve their shape. -Stale cache entries are now invalidated (`removeQueries`) instead of overwritten, forcing the next subscribe to re-run `queryFn` against the source of truth. Active entries continue to receive the full snapshot (predicate scoping for the consumer is enforced downstream by `subscribeChanges`'s `where`). The original ghost-row protection is preserved because deleted rows no longer reappear from a stale snapshot. +Inactive and disabled entries are now removed so a later subscription reruns `queryFn`. Fetchable active entries are refetched without replacing their scoped data with the normalized snapshot; stale notifications are ignored until a newer server result arrives. Eager collections still update their full-result cache in place. diff --git a/docs/collections/query-collection.md b/docs/collections/query-collection.md index e55fb94138..95253e1bd6 100644 --- a/docs/collections/query-collection.md +++ b/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 diff --git a/docs/reference/query-db-collection/interfaces/QueryCollectionUtils.md b/docs/reference/query-db-collection/interfaces/QueryCollectionUtils.md index b2ab5a7c36..68d477e03b 100644 --- a/docs/reference/query-db-collection/interfaces/QueryCollectionUtils.md +++ b/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 diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index 8e6ff9701c..7fb5004429 100644 --- a/packages/query-db-collection/src/query.ts +++ b/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) => 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 | Array>) => 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) => 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 | Array>) => 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 void>() const pendingReadyUnsubscribes = new Map void>>() + const manualWriteDataUpdateCounts = new Map() // 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 = [] @@ -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) @@ -2018,21 +2037,48 @@ export function queryCollectionOptions( /** * Updates the query cache after a manual-sync write. * - * Active entries (observer still subscribed): write the full syncedData - * snapshot back. Predicate scoping for the consumer is enforced downstream - * by the per-subscription `where` filter on `subscribeChanges`. - * - * Stale entries (no live observer): invalidate by removing the entry. Each - * entry was originally produced by `queryFn` for a specific predicate, so - * stamping it with the full snapshot would poison it — rows that don't - * satisfy the predicate end up in the cache, then a cache-hit re-subscribe - * within `gcTime` re-applies those wrong rows via `applySuccessfulResult`, - * the subscription's `where` filter discards them, and the subscriber sees - * `[]`. The next subscribe will re-run `queryFn` against the source of - * truth, which also preserves ghost-row protection (deleted rows won't - * reappear from a stale snapshot). + * 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): void => { + if (syncMode === `on-demand`) { + const refetchingQueries = new Set() + + 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. + }) + } + + const deferredRefresh = writeContext?.collection.deferDataRefresh + if (deferredRefresh) { + void deferredRefresh.then(refetchObserver, refetchObserver) + } else { + refetchObserver() + } + } + + queryClient.removeQueries({ + queryKey: baseKey, + predicate: (query) => !refetchingQueries.has(query.queryHash), + }) + return + } + const allCached = queryClient.getQueryCache().findAll({ queryKey: baseKey }) if (allCached.length === 0) { @@ -2043,10 +2089,6 @@ export function queryCollectionOptions( } for (const query of allCached) { - if (!state.observers.has(hashKey(query.queryKey))) { - queryClient.removeQueries({ queryKey: query.queryKey, exact: true }) - continue - } updateCacheDataForKey(query.queryKey, items) } } diff --git a/packages/query-db-collection/tests/query.test.ts b/packages/query-db-collection/tests/query.test.ts index e69e7ff917..a6c5d6c2c9 100644 --- a/packages/query-db-collection/tests/query.test.ts +++ b/packages/query-db-collection/tests/query.test.ts @@ -7699,17 +7699,14 @@ describe(`QueryCollection`, () => { }) }) - describe(`On-demand collection directWrite cache update`, () => { - it(`should update query cache for all active query keys when using writeUpdate with computed queryKey`, async () => { - // Ensures writeUpdate on on-demand collections with computed query keys - // updates all active cache keys to prevent data loss on remount - - const items: Array = [ + describe(`On-demand collection directWrite cache invalidation`, () => { + it(`should refetch a computed queryKey after writeUpdate`, async () => { + let items: Array = [ { id: `1`, name: `Item 1`, category: `A` }, { id: `2`, name: `Item 2`, category: `A` }, ] - const queryFn = vi.fn().mockResolvedValue(items) + const queryFn = vi.fn().mockImplementation(() => Promise.resolve(items)) // Use a custom queryClient with longer gcTime to prevent cache from being removed const customQueryClient = new QueryClient({ @@ -7757,14 +7754,14 @@ describe(`QueryCollection`, () => { expect(collection.size).toBe(2) }) - // Perform a direct write update + items = items.map((item) => + item.id === `1` ? { ...item, name: `Updated Item 1` } : item, + ) collection.utils.writeUpdate({ id: `1`, name: `Updated Item 1` }) // Verify the collection reflects the update expect(collection.get(`1`)?.name).toBe(`Updated Item 1`) - // IMPORTANT: Simulate remount by cleaning up and recreating the live query - // This is where the bug manifests - the updated data should persist await query1.cleanup() await flushPromises() @@ -7784,9 +7781,7 @@ describe(`QueryCollection`, () => { expect(collection.size).toBe(2) }) - // BUG ASSERTION: After remount, the updated data should persist - // With the bug, this will fail because writeUpdate updated the wrong cache key - // and on remount, the stale cached data is loaded instead + expect(queryFn).toHaveBeenCalledTimes(2) expect(collection.get(`1`)?.name).toBe(`Updated Item 1`) // Cleanup @@ -7794,16 +7789,13 @@ describe(`QueryCollection`, () => { customQueryClient.clear() }) - it(`should update query cache for static queryKey with where clause in on-demand mode`, async () => { - // Scenario: static queryKey + on-demand mode + where clause - // The where clause causes a computed query key to be generated - - const items: Array = [ + it(`should refetch a static queryKey with a where clause after writeUpdate`, async () => { + let items: Array = [ { id: `1`, name: `Item 1`, category: `A` }, { id: `2`, name: `Item 2`, category: `A` }, ] - const queryFn = vi.fn().mockResolvedValue(items) + const queryFn = vi.fn().mockImplementation(() => Promise.resolve(items)) const customQueryClient = new QueryClient({ defaultOptions: { @@ -7843,7 +7835,9 @@ describe(`QueryCollection`, () => { expect(collection.size).toBe(2) }) - // Perform a direct write update + items = items.map((item) => + item.id === `1` ? { ...item, name: `Updated Item 1` } : item, + ) collection.utils.writeUpdate({ id: `1`, name: `Updated Item 1` }) expect(collection.get(`1`)?.name).toBe(`Updated Item 1`) @@ -7866,23 +7860,20 @@ describe(`QueryCollection`, () => { expect(collection.size).toBe(2) }) - // After remount, the updated data should persist + expect(queryFn).toHaveBeenCalledTimes(2) expect(collection.get(`1`)?.name).toBe(`Updated Item 1`) await query2.cleanup() customQueryClient.clear() }) - it(`should update query cache for function queryKey that returns constant value in on-demand mode`, async () => { - // Scenario: function queryKey that returns same value - // This creates an undefined entry in the cache - - const items: Array = [ + it(`should refetch a constant function queryKey after writeUpdate`, async () => { + let items: Array = [ { id: `1`, name: `Item 1` }, { id: `2`, name: `Item 2` }, ] - const queryFn = vi.fn().mockResolvedValue(items) + const queryFn = vi.fn().mockImplementation(() => Promise.resolve(items)) const customQueryClient = new QueryClient({ defaultOptions: { @@ -7917,7 +7908,9 @@ describe(`QueryCollection`, () => { expect(collection.size).toBe(2) }) - // Perform a direct write update + items = items.map((item) => + item.id === `1` ? { ...item, name: `Updated Item 1` } : item, + ) collection.utils.writeUpdate({ id: `1`, name: `Updated Item 1` }) expect(collection.get(`1`)?.name).toBe(`Updated Item 1`) @@ -7936,7 +7929,7 @@ describe(`QueryCollection`, () => { expect(collection.size).toBe(2) }) - // After remount, the updated data should persist + expect(queryFn).toHaveBeenCalledTimes(2) expect(collection.get(`1`)?.name).toBe(`Updated Item 1`) await query2.cleanup() @@ -8265,8 +8258,7 @@ describe(`QueryCollection`, () => { // Only one assignment with resource_id=4 const allItems = Array.from(collection.values()) const carolAssignments = allItems.filter((a) => a.resource_id === 4) - expect(carolAssignments).toHaveLength(1) - expect(carolAssignments[0]?.id).toBe(101) + expect(carolAssignments.map((assignment) => assignment.id)).toEqual([101]) expect(collection.size).toBe(4) @@ -8278,30 +8270,24 @@ describe(`QueryCollection`, () => { }) describe(`Predicate-scoped cache invalidation on manual writes`, () => { - // Each cache entry under an on-demand collection's queryKey was originally - // produced by queryFn with a specific where predicate. A manual-sync write - // (writeInsert/writeUpdate/writeDelete/writeUpsert) used to overwrite every - // such entry with the full post-write syncedData snapshot, including rows - // that didn't satisfy the entry's predicate. After unsubscribe + re-subscribe - // within gcTime, the cache-hit re-applied those wrong rows, the per- - // subscription where filter discarded them, and the subscriber saw []. - // - // The fix invalidates stale (no-live-observer) cache entries on each write, - // forcing the next subscribe to re-run queryFn against the source of truth. - interface ScopedRow { id: string category: string } function makeQueryFn(getServerRows: () => Array) { - return async (ctx: QueryFunctionContext) => { - const where = (ctx.meta as { loadSubsetOptions?: { where?: any } }) - .loadSubsetOptions?.where - if (!where) return getServerRows() + return (ctx: QueryFunctionContext) => { + const where = ( + ctx.meta as { + loadSubsetOptions?: Pick + } + ).loadSubsetOptions?.where + if (!where) return Promise.resolve(getServerRows()) const evaluator = compileSingleRowExpression(where) - return getServerRows().filter((r) => - evaluator(r as unknown as Record), + return Promise.resolve( + getServerRows().filter((row) => + evaluator(row as unknown as Record), + ), ) } } @@ -8316,6 +8302,7 @@ describe(`QueryCollection`, () => { }, }, }) + const queryFn = vi.fn(makeQueryFn(() => serverRows)) const collection = createCollection( queryCollectionOptions({ @@ -8323,49 +8310,44 @@ describe(`QueryCollection`, () => { queryClient: customQueryClient, queryKey: [`predicate-cache-test`], syncMode: `on-demand`, - getKey: (r) => r.id, - queryFn: makeQueryFn(() => serverRows), + getKey: (row) => row.id, + queryFn, onInsert: async () => ({ refetch: false }), onUpdate: async () => ({ refetch: false }), onDelete: async () => ({ refetch: false }), }), ) - return { collection, customQueryClient } + return { collection, customQueryClient, queryFn } } - it(`should return correct rows on re-subscribe after an unrelated writeUpsert`, async () => { + function createCategoryQuery( + collection: ReturnType[`collection`], + category: string, + ) { + return createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, category)), + }) + } + + it(`should refetch a stale predicate after an unrelated writeUpsert`, async () => { const serverRows: Array = [ { id: `1`, category: `A` }, { id: `2`, category: `B` }, ] const { collection, customQueryClient } = makeCollection(serverRows) - // Subscribe to category=A, then unsubscribe. Leaves a stale cache entry - // whose contents should be [{id:1}]. - const wave1 = createLiveQueryCollection({ - query: (q) => - q - .from({ item: collection }) - .where(({ item }) => eq(item.category, `A`)), - }) + const wave1 = createCategoryQuery(collection, `A`) await wave1.preload() expect(wave1.toArray.map((r) => r.id)).toEqual([`1`]) await wave1.cleanup() - // Manual write of a row outside the predicate. Before the fix, this - // overwrote every cache entry with the full snapshot, poisoning - // category=A's stale entry with [{id:1},{id:2},{id:99}]. collection.utils.writeUpsert({ id: `99`, category: `unrelated` }) - // Re-subscribe to the same predicate. With the fix, the stale entry was - // invalidated and queryFn re-runs, returning only the matching row. - const wave2 = createLiveQueryCollection({ - query: (q) => - q - .from({ item: collection }) - .where(({ item }) => eq(item.category, `A`)), - }) + const wave2 = createCategoryQuery(collection, `A`) await wave2.preload() expect(wave2.toArray.map((r) => r.id)).toEqual([`1`]) @@ -8377,26 +8359,15 @@ describe(`QueryCollection`, () => { const serverRows: Array = [{ id: `1`, category: `A` }] const { collection, customQueryClient } = makeCollection(serverRows) - const wave1 = createLiveQueryCollection({ - query: (q) => - q - .from({ item: collection }) - .where(({ item }) => eq(item.category, `A`)), - }) + const wave1 = createCategoryQuery(collection, `A`) await wave1.preload() expect(wave1.toArray.map((r) => r.id)).toEqual([`1`]) await wave1.cleanup() - // Insert a new matching row both server-side and via writeInsert. serverRows.push({ id: `2`, category: `A` }) collection.utils.writeInsert({ id: `2`, category: `A` }) - const wave2 = createLiveQueryCollection({ - query: (q) => - q - .from({ item: collection }) - .where(({ item }) => eq(item.category, `A`)), - }) + const wave2 = createCategoryQuery(collection, `A`) await wave2.preload() expect(wave2.toArray.map((r) => r.id).sort()).toEqual([`1`, `2`]) @@ -8408,8 +8379,6 @@ describe(`QueryCollection`, () => { const serverRows: Array = [{ id: `1`, category: `A` }] const { collection, customQueryClient } = makeCollection(serverRows) - // Narrow predicate (id=1) subscribes then unsubscribes — leaves a stale - // cache entry containing [{id:1}]. const narrow = createLiveQueryCollection({ query: (q) => q.from({ item: collection }).where(({ item }) => eq(item.id, `1`)), @@ -8417,21 +8386,12 @@ describe(`QueryCollection`, () => { await narrow.preload() await narrow.cleanup() - // Broader predicate (category=A) becomes active. - const broad = createLiveQueryCollection({ - query: (q) => - q - .from({ item: collection }) - .where(({ item }) => eq(item.category, `A`)), - }) + const broad = createCategoryQuery(collection, `A`) await broad.preload() - // Insert a row in the broader scope but outside the narrow predicate. serverRows.push({ id: `2`, category: `A` }) collection.utils.writeInsert({ id: `2`, category: `A` }) - // Re-subscribing to the narrow predicate must still return only id=1, not - // be poisoned by id=2 from the active broader entry. const narrow2 = createLiveQueryCollection({ query: (q) => q.from({ item: collection }).where(({ item }) => eq(item.id, `1`)), @@ -8443,5 +8403,289 @@ describe(`QueryCollection`, () => { await broad.cleanup() customQueryClient.clear() }) + + it(`should keep an active predicate cache scoped while revalidating it`, async () => { + const serverRows: Array = [{ id: `1`, category: `A` }] + const { collection, customQueryClient, queryFn } = + makeCollection(serverRows) + const active = createCategoryQuery(collection, `A`) + + await active.preload() + const cacheSnapshots: Array> = [] + const unsubscribe = customQueryClient + .getQueryCache() + .subscribe((event) => { + if ( + event.type === `updated` && + Array.isArray(event.query.state.data) + ) { + cacheSnapshots.push( + event.query.state.data.map((row: ScopedRow) => row.id), + ) + } + }) + collection.utils.writeInsert({ id: `outside`, category: `B` }) + + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(2)) + expect(cacheSnapshots).not.toContainEqual([`1`, `outside`]) + expect(active.toArray.map((row) => row.id)).toEqual([`1`]) + expect( + customQueryClient.getQueriesData>({ + queryKey: [`predicate-cache-test`], + }), + ).toEqual([[expect.any(Array), [{ id: `1`, category: `A` }]]]) + + unsubscribe() + await active.cleanup() + customQueryClient.clear() + }) + + it(`should not resolve an initial subset from a manual write`, async () => { + const deferred = createDeferred>() + const customQueryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const queryFn = vi.fn(() => deferred.promise) + const collection = createCollection( + queryCollectionOptions({ + id: `predicate-cache-in-flight`, + queryClient: customQueryClient, + queryKey: [`predicate-cache-in-flight`], + syncMode: `on-demand`, + getKey: (row) => row.id, + queryFn, + }), + ) + const active = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .where(({ item }) => eq(item.category, `A`)), + }) + let preloadSettled = false + const preload = active.preload().then(() => { + preloadSettled = true + }) + + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(1)) + collection.utils.writeInsert({ id: `outside`, category: `B` }) + await flushPromises() + + expect(preloadSettled).toBe(false) + expect( + customQueryClient.getQueryCache().findAll({ + queryKey: [`predicate-cache-in-flight`], + })[0]?.state.data, + ).toBeUndefined() + + deferred.resolve([{ id: `1`, category: `A` }]) + await preload + expect(active.toArray.map((row) => row.id)).toEqual([`1`]) + + await active.cleanup() + customQueryClient.clear() + }) + + it(`should replenish an active limited query after a manual delete`, async () => { + let serverRows: Array = [ + { id: `1`, category: `A` }, + { id: `2`, category: `A` }, + { id: `3`, category: `A` }, + ] + const customQueryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const queryFn = vi.fn((ctx: QueryFunctionContext) => { + const options = ( + ctx.meta as { loadSubsetOptions?: Pick } + ).loadSubsetOptions + return Promise.resolve(serverRows.slice(0, options?.limit)) + }) + const collection = createCollection( + queryCollectionOptions({ + id: `predicate-cache-limit`, + queryClient: customQueryClient, + queryKey: [`predicate-cache-limit`], + syncMode: `on-demand`, + getKey: (row) => row.id, + queryFn, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + }), + ) + const active = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .orderBy(({ item }) => item.id, `asc`) + .limit(2), + }) + + await active.preload() + expect(active.toArray.map((row) => row.id)).toEqual([`1`, `2`]) + + serverRows = serverRows.filter((row) => row.id !== `1`) + collection.utils.writeDelete(`1`) + + await vi.waitFor(() => { + expect(queryFn.mock.calls.length).toBeGreaterThan(1) + expect(active.toArray.map((row) => row.id)).toEqual([`2`, `3`]) + }) + + await active.cleanup() + customQueryClient.clear() + }) + + it(`should replenish an active static query after a manual delete`, async () => { + let serverRows: Array = [ + { id: `1`, category: `A` }, + { id: `2`, category: `A` }, + { id: `3`, category: `A` }, + ] + const customQueryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const queryFn = vi.fn((ctx: QueryFunctionContext) => { + const options = ( + ctx.meta as { loadSubsetOptions?: Pick } + ).loadSubsetOptions + return Promise.resolve(serverRows.slice(0, options?.limit)) + }) + const collection = createCollection( + queryCollectionOptions({ + id: `predicate-cache-static`, + queryClient: customQueryClient, + queryKey: [`predicate-cache-static`], + syncMode: `on-demand`, + staleTime: `static`, + getKey: (row) => row.id, + queryFn, + autoIndex: `eager`, + defaultIndexType: BTreeIndex, + }), + ) + const active = createLiveQueryCollection({ + query: (q) => + q + .from({ item: collection }) + .orderBy(({ item }) => item.id, `asc`) + .limit(2), + }) + + await active.preload() + expect(active.toArray.map((row) => row.id)).toEqual([`1`, `2`]) + + serverRows = serverRows.filter((row) => row.id !== `1`) + collection.utils.writeDelete(`1`) + + await vi.waitFor(() => { + expect(queryFn.mock.calls.length).toBeGreaterThan(1) + expect(active.toArray.map((row) => row.id)).toEqual([`2`, `3`]) + }) + + await active.cleanup() + customQueryClient.clear() + }) + + it(`should revalidate after a deferred refresh rejects`, async () => { + const serverRows: Array = [{ id: `1`, category: `A` }] + const { collection, customQueryClient, queryFn } = + makeCollection(serverRows) + const active = createCategoryQuery(collection, `A`) + const deferredRefresh = createDeferred() + + await active.preload() + collection.deferDataRefresh = deferredRefresh.promise + void deferredRefresh.promise.catch(() => { + collection.deferDataRefresh = null + }) + + serverRows.push({ id: `2`, category: `A` }) + collection.utils.writeInsert({ id: `outside`, category: `B` }) + await flushPromises() + expect(queryFn).toHaveBeenCalledTimes(1) + + deferredRefresh.reject(new Error(`refresh failed`)) + + await vi.waitFor(() => { + expect(queryFn).toHaveBeenCalledTimes(2) + expect(active.toArray.map((row) => row.id)).toEqual([`1`, `2`]) + }) + + await active.cleanup() + customQueryClient.clear() + }) + + it(`should remove a disabled cache entry after a manual write`, async () => { + const customQueryClient = new QueryClient() + const queryKey = [`predicate-cache-disabled`] + customQueryClient.setQueryData(queryKey, [{ id: `1`, category: `A` }]) + const collection = createCollection( + queryCollectionOptions({ + id: `predicate-cache-disabled`, + queryClient: customQueryClient, + queryKey: () => queryKey, + queryFn: () => Promise.resolve([{ id: `1`, category: `A` }]), + syncMode: `on-demand`, + enabled: false, + getKey: (row) => row.id, + }), + ) + const active = createCategoryQuery(collection, `A`) + + await active.preload() + expect(active.toArray.map((row) => row.id)).toEqual([`1`]) + + collection.utils.writeInsert({ id: `outside`, category: `B` }) + + expect( + customQueryClient.getQueryCache().findAll({ queryKey }), + ).toHaveLength(0) + + await active.cleanup() + customQueryClient.clear() + }) + + it(`should accept a revalidation result after the Query state resets`, async () => { + const interruptedRefetch = createDeferred>() + const queryKey = [`predicate-cache-reset`] + const customQueryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const queryFn = vi + .fn() + .mockResolvedValueOnce([{ id: `1`, category: `A` }]) + .mockImplementationOnce(() => interruptedRefetch.promise) + .mockResolvedValue([ + { id: `1`, category: `A` }, + { id: `2`, category: `A` }, + ]) + const collection = createCollection( + queryCollectionOptions({ + id: `predicate-cache-reset`, + queryClient: customQueryClient, + queryKey: () => queryKey, + queryFn, + syncMode: `on-demand`, + getKey: (row) => row.id, + }), + ) + const active = createCategoryQuery(collection, `A`) + + await active.preload() + customQueryClient.setQueryData(queryKey, (rows) => rows) + customQueryClient.setQueryData(queryKey, (rows) => rows) + + collection.utils.writeInsert({ id: `outside`, category: `B` }) + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(2)) + + await customQueryClient.resetQueries({ queryKey, exact: true }) + + expect(queryFn).toHaveBeenCalledTimes(3) + expect(active.toArray.map((row) => row.id)).toEqual([`1`, `2`]) + + await active.cleanup() + customQueryClient.clear() + }) }) }) From f55784a0b0954fa73a33e2cbcfcc820ba1e2f823 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Tue, 21 Jul 2026 13:50:15 -0600 Subject: [PATCH 3/3] docs: tighten query cache changeset --- .changeset/fix-cache-poisoning-on-manual-writes.md | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/.changeset/fix-cache-poisoning-on-manual-writes.md b/.changeset/fix-cache-poisoning-on-manual-writes.md index 5b7b957184..1d1fd7fe69 100644 --- a/.changeset/fix-cache-poisoning-on-manual-writes.md +++ b/.changeset/fix-cache-poisoning-on-manual-writes.md @@ -2,8 +2,4 @@ '@tanstack/query-db-collection': patch --- -fix: invalidate predicate-scoped cache entries on manual-sync writes - -In `syncMode: 'on-demand'`, manual-sync writes (`writeInsert`, `writeUpdate`, `writeDelete`, `writeUpsert`, and `writeBatch`) no longer retain the full post-write `syncedData` snapshot in predicate-scoped Query cache entries. Those entries may also encode ordering and pagination, so a normalized collection snapshot cannot preserve their shape. - -Inactive and disabled entries are now removed so a later subscription reruns `queryFn`. Fetchable active entries are refetched without replacing their scoped data with the normalized snapshot; stale notifications are ignored until a newer server result arrives. Eager collections still update their full-result cache in place. +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.