diff --git a/.changeset/fix-refetch-query-ownership.md b/.changeset/fix-refetch-query-ownership.md new file mode 100644 index 0000000000..5a1321bd12 --- /dev/null +++ b/.changeset/fix-refetch-query-ownership.md @@ -0,0 +1,5 @@ +--- +'@tanstack/query-db-collection': patch +--- + +Fix query ownership cleanup across overlapping refetch and observer lifecycles so removed rows do not remain visible because of stale persisted metadata. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 00fec5a8c7..2d3753e74e 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -1055,6 +1055,29 @@ held-request checkpoint. These finite histories leave same-run external durable writers, larger overlapping-fetch schedules, and direct-patch provenance after Query cache dehydration outside their proven scope. +A fixed accepted-ownership history acquires three exact subset claims for one +row, then accepts two complete empty results while an unrelated optimistic +mutation delays publication. Its independent model stores the latest complete +result per active Query and exposes their union: the third result keeps the row +public until its own release, after which the union is empty. The ordered +refetch owner's held-delete replay is the killing witness for trusting stale +published owners over accepted ownership. A release/reacquisition history also +holds publication, reacquires the same cached Query, then proves the older +release cannot erase the new claim by releasing it and observing an empty +public row set. Larger overlap schedules, window ownership outside the ordered +replay, and durable-adapter ordering remain open under the owners named above. + +A controlled cold persisted scan gives one Query two rows in both scan orders, +then returns a current result that omits one row. Its independent replacement +model requires the survivor to remain public with its owner and the controlled +persistence boundary to receive a deletion for the omission. A row-by-row +reconstruction that treats the first persisted edge as a resolved Query fails +one order. A sibling DbClient hydration history supplies the same two rows in +both orders, then compares the public Collection with the same complete current +result and releases its demand. A row-by-row reconstruction fails both orders +at the public-row checkpoint. These finite histories do not cover larger cold +results or hydration interleaved with another active Query result. + The alias-scope owner also has a compiler invariant witness for a malformed include plan: removing a captured ancestor's parent projection after the builder creates it must fail before child rows are read. Without that check, diff --git a/packages/query-db-collection/src/query.ts b/packages/query-db-collection/src/query.ts index f0d45963f6..e9e9da6b82 100644 --- a/packages/query-db-collection/src/query.ts +++ b/packages/query-db-collection/src/query.ts @@ -1110,21 +1110,15 @@ export function queryCollectionOptions( const removeQueryOwnership = (hashedQueryKey: string) => { const nextOwnersByRow = new Map>() - const rowKeys = - queryToRows.get(hashedQueryKey) ?? new Set() - - rowKeys.forEach((rowKey) => { - const owners = rowToQueries.get(rowKey) - - if (!owners) { - return - } - - const nextOwners = new Set(owners) + queryToRows.get(hashedQueryKey)?.forEach((rowKey) => { + const nextOwners = new Set(rowToQueries.get(rowKey)) nextOwners.delete(hashedQueryKey) nextOwnersByRow.set(rowKey, nextOwners) + if (nextOwners.size === 0) rowToQueries.delete(rowKey) + else rowToQueries.set(rowKey, nextOwners) }) + queryToRows.set(hashedQueryKey, new Set()) return nextOwnersByRow } @@ -1504,26 +1498,34 @@ export function queryCollectionOptions( Record | undefined } - const getOwnersFromMetadata = (rowMetadata: unknown) => { - if (!rowMetadata || typeof rowMetadata !== `object`) { + const getOwnersFromMetadata = ( + rowMetadata: unknown, + rowKey?: string | number, + ) => { + if (!rowMetadata || typeof rowMetadata !== `object`) return new Set() - } const queryMetadata = (rowMetadata as Record) .queryCollection - if (!queryMetadata || typeof queryMetadata !== `object`) { + if (!queryMetadata || typeof queryMetadata !== `object`) return new Set() - } const owners = (queryMetadata as Record).owners - if (!owners || typeof owners !== `object`) { - return new Set() - } - - return new Set(Object.keys(owners as Record)) + if (!owners || typeof owners !== `object`) return new Set() + + const result = new Set(Object.keys(owners as Record)) + if (rowKey === undefined) return result + // Resolved Queries supersede stale persisted edges; accepted in-memory + // edges supplement owners whose metadata has not published yet. + result.forEach((owner) => { + const rows = queryToRows.get(owner) + if (rows && !rows.has(rowKey)) result.delete(owner) + }) + rowToQueries.get(rowKey)?.forEach((owner) => result.add(owner)) + return result } const getPersistedOwners = (rowKey: string | number) => - getOwnersFromMetadata(getRowMetadata(rowKey)) + getOwnersFromMetadata(getRowMetadata(rowKey), rowKey) const setPersistedOwners = ( rowKey: string | number, @@ -1612,8 +1614,11 @@ export function queryCollectionOptions( } const ownedRows = new Set() - for (const [rowKey] of collection._state.syncedData.entries()) { - const owners = getPersistedOwners(rowKey) + const hydratedOwners = Array.from( + collection._state.syncedData.keys(), + (rowKey) => [rowKey, getPersistedOwners(rowKey)] as const, + ) + for (const [rowKey, owners] of hydratedOwners) { if (owners.size === 0) { continue } @@ -1695,19 +1700,13 @@ export function queryCollectionOptions( return baseline } - scannedRows.forEach((row) => { - const rowMetadata = row.metadata as Record | undefined - const queryMetadata = rowMetadata?.queryCollection - if (!queryMetadata || typeof queryMetadata !== `object`) { - return - } - - const owners = (queryMetadata as Record).owners - if (!owners || typeof owners !== `object`) { - return - } - - const ownerSet = new Set(Object.keys(owners as Record)) + // Project the whole scan before adding any reconstructed edges, so one + // row cannot make a persisted owner look resolved for the next row. + const scannedOwners = scannedRows.map((row) => + getOwnersFromMetadata(row.metadata, row.key), + ) + scannedRows.forEach((row, index) => { + const ownerSet = scannedOwners[index]! if (ownerSet.size === 0) { return } @@ -2652,10 +2651,7 @@ export function queryCollectionOptions( try { // From this point onward the result, including an empty result, is the // authoritative ownership baseline until this query is cleaned up. - queryToRows.set( - hashedQueryKey, - queryToRows.get(hashedQueryKey) ?? new Set(), - ) + queryToRows.set(hashedQueryKey, new Set(previouslyOwnedRows)) begin() transactionActive = true @@ -3432,6 +3428,11 @@ export function queryCollectionOptions( invalidatePendingResultApplication(hashedQueryKey) const nextOwnersByRow = removeQueryOwnership(hashedQueryKey) + const ownershipTombstone = queryToRows.get(hashedQueryKey) + const retireOwnership = () => { + if (queryToRows.get(hashedQueryKey) === ownershipTombstone) + queryToRows.delete(hashedQueryKey) + } const rowsToDelete: Array = [] nextOwnersByRow.forEach((nextOwners, rowKey) => { @@ -3453,12 +3454,6 @@ export function queryCollectionOptions( } nextOwnersByRow.forEach((owners, rowKey) => { - if (owners.size === 0) { - rowToQueries.delete(rowKey) - } else { - rowToQueries.set(rowKey, owners) - } - if (shouldWriteMetadata) { setPersistedOwners(rowKey, owners) } @@ -3474,24 +3469,23 @@ export function queryCollectionOptions( metadata.collection.delete(retentionKey) } - if (needsTransaction) { - const applied = commit() - const retireDeletedPositions = () => { - rowsToDelete.forEach((row) => { - const key = getKey(row) - if (collection._state.getAcceptedSyncedRow(key) === undefined) - acceptedRowPositions.delete(key) - }) - } - const accepted = whenSyncAccepted(applied) - if (accepted === true) retireDeletedPositions() - else void accepted.then(retireDeletedPositions, () => undefined) + const applied = needsTransaction ? commit() : true + const retireDeletedPositions = () => { + rowsToDelete.forEach((row) => { + const key = getKey(row) + if (collection._state.getAcceptedSyncedRow(key) === undefined) + acceptedRowPositions.delete(key) + }) } + const accepted = whenSyncAccepted(applied) + if (accepted === true) retireDeletedPositions() + else void accepted.then(retireDeletedPositions, () => undefined) + if (applied === true) retireOwnership() + else void applied.then(retireOwnership, () => undefined) state.observers.delete(hashedQueryKey) if (query?.state.fetchStatus !== `fetching`) fetchStartGenerations.delete(hashedQueryKey) - queryToRows.delete(hashedQueryKey) hashToQueryKey.delete(hashedQueryKey) queryRefCounts.delete(hashedQueryKey) effectivePersistedGcTimes.delete(hashedQueryKey) diff --git a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts index 18b33a2e1b..721b4db5c6 100644 --- a/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts +++ b/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts @@ -7,7 +7,9 @@ import { isCancelledError, } from '@tanstack/query-core' import { + DbClient, IR, + collectionOptions, createCollection, createLiveQueryCollection, createOptimisticAction, @@ -5852,6 +5854,119 @@ describe(`query collection ownership lifecycle`, () => { ).toEqual([]) }) + /** + * Releasing and then reacquiring one demand creates a new active claim. An + * older release may publish later, but it cannot erase that newer claim. The + * independent model removes the first claim, adds the second, and leaves the + * shared row until the second release. The unrelated blocker only delays + * publication. + */ + it(`keeps a reacquired claim across an older release publication`, async () => { + const id = `reacquired-claim-old-release` + const detail = categorySubset(`detail`) + const blocker = { id: `blocker`, category: `local`, name: `Blocker` } + const persistence = createDeferred() + const { collection, queryFn } = createOwnershipFixture({ + id, + results: [[shared]], + staleTime: Number.POSITIVE_INFINITY, + }) + const mutation = createTransaction({ + mutationFn: () => persistence.promise, + }) + + await collection._sync.loadSubset(detail) + mutation.mutate(() => collection.insert(blocker)) + cleanups.push(async () => { + persistence.resolve() + await mutation.isPersisted.promise.catch(() => undefined) + }) + + const pendingBefore = collection._state.pendingSyncedTransactions.length + collection._sync.unloadSubset(detail) + await vi.waitFor(() => + expect( + collection._state.pendingSyncedTransactions.length, + ).toBeGreaterThan(pendingBefore), + ) + const pendingAfterRelease = + collection._state.pendingSyncedTransactions.length + const reacquired = Promise.resolve(collection._sync.loadSubset(detail)) + void reacquired.catch(() => undefined) + await vi.waitFor(() => + expect( + collection._state.pendingSyncedTransactions.length, + ).toBeGreaterThan(pendingAfterRelease), + ) + + persistence.resolve() + await mutation.isPersisted.promise + await reacquired + await vi.waitFor(() => expect(rows(collection)).toEqual([shared.id])) + + collection._sync.unloadSubset(detail) + await vi.waitFor(() => expect(rows(collection)).toEqual([])) + expect(queryFn).toHaveBeenCalledOnce() + }) + + /** + * Accepted Query results update ownership in acceptance order even while + * publication is held. The independent model stores each active Query's + * latest complete result and exposes their union: after the detail and list + * results become empty, the peer still exposes the row; after the peer is + * released, the union is empty. Published metadata remains stale until the + * hold ends, so it cannot be the model's source of truth. + */ + it(`composes accepted owner retirements ahead of stale metadata`, async () => { + const id = `accepted-owner-retirements` + const detail = categorySubset(`detail`) + const list = categorySubset(`list`) + const peer: LoadSubsetOptions = { + where: new IR.Func(`in`, [ + new IR.PropRef([`category`]), + new IR.Value([`shared`, `peer`]), + ]), + } + const blocker = { id: `blocker`, category: `local`, name: `Blocker` } + const persistence = createDeferred() + const { collection, queryFn } = createOwnershipFixture({ + id, + results: [[shared], [shared], [shared], [], [], [shared]], + }) + const mutation = createTransaction({ + mutationFn: () => persistence.promise, + }) + + await collection._sync.loadSubset(detail) + await collection._sync.loadSubset(list) + await collection._sync.loadSubset(peer) + + mutation.mutate(() => collection.insert(blocker)) + cleanups.push(async () => { + persistence.resolve() + await mutation.isPersisted.promise.catch(() => undefined) + }) + expect(mutation.state).toBe(`persisting`) + + const pendingBefore = collection._state.pendingSyncedTransactions.length + const refetch = collection.utils.refetch() + await vi.waitFor(() => expect(queryFn).toHaveBeenCalledTimes(6)) + await vi.waitFor(() => + expect( + collection._state.pendingSyncedTransactions.length, + ).toBeGreaterThanOrEqual(pendingBefore + 2), + ) + + persistence.resolve() + await mutation.isPersisted.promise + await refetch + await vi.waitFor(() => expect(rows(collection)).toEqual([shared.id])) + + collection._sync.unloadSubset(peer) + await vi.waitFor(() => expect(rows(collection)).toEqual([])) + expect(queryFn).toHaveBeenCalledTimes(6) + }) + // Law: once a result's sync transaction is accepted, the rows it owns stay // owned when a later result supersedes it. The later result diffs against // that ownership, so a row only the superseded result added is deleted, and @@ -11869,6 +11984,132 @@ describe(`query collection ownership lifecycle`, () => { ) }) + /** + * A current complete Query result replaces its cold persisted result. The + * independent model starts with `{kept, omitted}` and replaces it with + * `{kept}`. Persisted row order cannot change that result, so both scan + * orders must expose the survivor and delete the omission. + */ + it.each([ + { scanOrder: [`kept`, `omitted`] }, + { scanOrder: [`omitted`, `kept`] }, + ] as const)( + `replaces a cold two-row result in scan order $scanOrder`, + async ({ scanOrder }) => { + const id = `cold-multi-row-owner-${scanOrder.join(`-`)}` + const queryHash = hashKey([id]) + const kept: Item = { id: `kept`, category: `shared`, name: `Kept` } + const omitted: Item = { + id: `omitted`, + category: `shared`, + name: `Omitted`, + } + const byId = new Map([ + [kept.id, kept], + [omitted.id, omitted], + ]) + const ownerMetadata = { + queryCollection: { owners: { [queryHash]: true } }, + } + const metadata: MetadataRecorder = { rows: new Map(), writes: [] } + const scanPersistedRows = vi.fn(() => + Promise.resolve( + scanOrder.map((rowId) => ({ + key: rowId, + value: byId.get(rowId)!, + metadata: ownerMetadata, + })), + ), + ) + const { collection } = createOwnershipFixture({ + id, + results: [[kept]], + metadataRecorder: metadata, + scanPersistedRows, + setupMetadata: (syncMetadata) => + syncMetadata.collection.set(`queryCollection:gc:${queryHash}`, { + queryHash, + mode: `until-revalidated`, + }), + }) + + await collection._sync.loadSubset({}) + + expect(rows(collection)).toEqual([kept.id]) + expect(metadata.writes).toContainEqual({ + type: `delete`, + key: omitted.id, + }) + }, + ) + + /** + * Hydrated rows obey the same complete-result replacement law. The public + * Collection begins with `{kept, omitted}` and the Query returns `{kept}`. + * Hydration order cannot change the public result or keep the survivor alive + * after its demand is released. + */ + it.each([ + { hydrationOrder: [`kept`, `omitted`] }, + { hydrationOrder: [`omitted`, `kept`] }, + ] as const)( + `replaces a hydrated two-row result in order $hydrationOrder`, + async ({ hydrationOrder }) => { + const id = `hydrated-multi-row-owner-${hydrationOrder.join(`-`)}` + const queryHash = hashKey([id]) + const kept: Item = { id: `kept`, category: `shared`, name: `Kept` } + const omitted: Item = { + id: `omitted`, + category: `shared`, + name: `Omitted`, + } + const byId = new Map([ + [kept.id, kept], + [omitted.id, omitted], + ]) + const queryClient = createQueryClient() + const queryFn = vi.fn(() => Promise.resolve([kept])) + const descriptor = collectionOptions( + queryCollectionOptions({ + id, + queryClient, + queryKey: [id], + queryFn, + getKey: (item) => item.id, + syncMode: `on-demand`, + startSync: true, + }), + ) + const client = new DbClient({ queryClient }) + client.hydrate({ + collections: [ + { + collectionId: id, + rows: hydrationOrder.map((rowId) => ({ + key: rowId, + value: byId.get(rowId)!, + metadata: { + queryCollection: { owners: { [queryHash]: true } }, + }, + })), + }, + ], + }) + const collection = client.collection(descriptor) + cleanups.push(async () => { + await client.cleanup() + queryClient.clear() + }) + + await collection._sync.loadSubset({}) + expect(rows(collection)).toEqual([kept.id]) + + collection._sync.unloadSubset({}) + await vi.waitFor(() => expect(rows(collection)).toEqual([])) + expect(queryFn).toHaveBeenCalledOnce() + }, + ) + /** * A retained subset owner cannot be settled by acquiring a different Query * key. The independent source model has one `detail` row in durable storage; diff --git a/packages/query-db-collection/tests/refetch-ordered-prefix-oracle.test.ts b/packages/query-db-collection/tests/refetch-ordered-prefix-oracle.test.ts index 9e2591408d..f4d4ad4e90 100644 --- a/packages/query-db-collection/tests/refetch-ordered-prefix-oracle.test.ts +++ b/packages/query-db-collection/tests/refetch-ordered-prefix-oracle.test.ts @@ -1928,6 +1928,43 @@ describe(`query-db on-demand refetch keeps served ordered prefixes whole`, () => }) }) + // Found by the random campaign. A Query invalidation lets another consumer + // observe row 1 before the backend deletes it, and the final refetch applies + // while an optimistic update holds source publication back. The public law + // is unchanged: after release, the ordered window equals the pure backend + // window, regardless of which consumer previously observed a deleted row. + it(`a held refetch removes a deleted row owned by an observed query`, async () => { + await checkCase({ + scenario: { + encoding: `cursor`, + functionKey: false, + terms: 1, + orderKey: `rank`, + pageCap: undefined, + limit: 1, + offset: 1, + }, + initial: [ + { id: 1, rank: 0, marked: true }, + { id: 0, rank: 0, marked: false }, + ], + history: [ + { type: `hold`, kind: `update`, id: 0 }, + { type: `upsert`, id: 2, rank: 0 }, + { type: `refetch` }, + { type: `release` }, + { type: `flip`, id: 0 }, + { type: `upsert`, id: 0, rank: 1 }, + { type: `observe`, id: 1 }, + { type: `invalidate` }, + { type: `delete`, id: 1 }, + { type: `hold`, kind: `update`, id: 0 }, + { type: `refetch` }, + { type: `release` }, + ], + }) + }) + // The premise-directed campaign for the skip under a hold. Its reach // witness requires refetches that report while a hold is running. it.runIf(replay.replayPath === undefined)(