Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fix-refetch-query-ownership.md
Original file line number Diff line number Diff line change
@@ -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.
23 changes: 23 additions & 0 deletions docs/contributing/oracle-coverage.md
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
114 changes: 54 additions & 60 deletions packages/query-db-collection/src/query.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1110,21 +1110,15 @@ export function queryCollectionOptions(
const removeQueryOwnership = (hashedQueryKey: string) => {
const nextOwnersByRow = new Map<string | number, Set<string>>()

const rowKeys =
queryToRows.get(hashedQueryKey) ?? new Set<string | number>()

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
}

Expand Down Expand Up @@ -1504,26 +1498,34 @@ export function queryCollectionOptions(
Record<string, unknown> | 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<string>()
}
const queryMetadata = (rowMetadata as Record<string, unknown>)
.queryCollection
if (!queryMetadata || typeof queryMetadata !== `object`) {
if (!queryMetadata || typeof queryMetadata !== `object`)
return new Set<string>()
}

const owners = (queryMetadata as Record<string, unknown>).owners
if (!owners || typeof owners !== `object`) {
return new Set<string>()
}

return new Set(Object.keys(owners as Record<string, true>))
if (!owners || typeof owners !== `object`) return new Set<string>()

const result = new Set(Object.keys(owners as Record<string, true>))
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,
Expand Down Expand Up @@ -1612,8 +1614,11 @@ export function queryCollectionOptions(
}

const ownedRows = new Set<string | number>()
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
}
Expand Down Expand Up @@ -1695,19 +1700,13 @@ export function queryCollectionOptions(
return baseline
}

scannedRows.forEach((row) => {
const rowMetadata = row.metadata as Record<string, unknown> | undefined
const queryMetadata = rowMetadata?.queryCollection
if (!queryMetadata || typeof queryMetadata !== `object`) {
return
}

const owners = (queryMetadata as Record<string, unknown>).owners
if (!owners || typeof owners !== `object`) {
return
}

const ownerSet = new Set(Object.keys(owners as Record<string, true>))
// 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
}
Expand Down Expand Up @@ -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<string | number>(),
)
queryToRows.set(hashedQueryKey, new Set(previouslyOwnedRows))

begin()
transactionActive = true
Expand Down Expand Up @@ -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<any> = []

nextOwnersByRow.forEach((nextOwners, rowKey) => {
Expand All @@ -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)
}
Expand All @@ -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)
Expand Down
Loading
Loading