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-retired-replay-stale-rows.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@tanstack/db': patch
---

A subscriber no longer keeps stale rows after a truncate replay that loses its last demand. When the last requested subset is released while the replay is still loading, nothing can complete the replay. If its current attempt has not failed, the subscriber now receives the rows the source wrote, replaced or deleted during it, instead of keeping the values from before the truncate. A replay whose attempt failed still keeps the last complete snapshot, as before, and a later request updates any row it still shows at an older value.
14 changes: 13 additions & 1 deletion docs/contributing/oracle-coverage.md
Original file line number Diff line number Diff line change
Expand Up @@ -681,7 +681,19 @@ The following related oracles own distinct production paths or observation cuts:
- The [Collection lifecycle publication oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts)
owns exact public rows and change batches. The
[replay oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-subscription-replay-oracle.property.test.ts)
applies that law to competing replay schedules and same-key writes.
applies that law to competing replay schedules and same-key writes. Both
models publish a replay's rows when its last owner is released and its
current attempt has not failed, and keep a failed one private; the
publication oracle's named histories cover a value written, rewritten or
deleted during the replay, a row no demand names, a failed replay, a
later request that updates a row still shown at an older value, and a row
retained by a failed replay that the source deletes before a later replay
retires. The replay oracle's named witnesses fail each synchronous failure
site (a replay load, an old lease's unload, a new request) and then release
the last owner; removing that site's sticky failure fails its witness.
Limit: the generated grammars reach a failed retirement followed by another
retired replay only rarely (the replay grammar releases one demand, on the
last attempt), so the named histories own that sequence.
Bounded three-row histories now compare a subscription's selected keys with
callback key/type batches across full, explicit, filtered, and empty snapshots,
two later ordered selections, both initial-state modes, and callback reentry.
Expand Down
39 changes: 33 additions & 6 deletions packages/db/src/collection/subscription.ts
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,8 @@ type TruncateReplayState = {
pendingSetups: number
currentAttempt: TruncateReplayAttempt
failures: Map<SubsetDemand, Error>
/** The current attempt failed, even if its failed demand was released since. */
failed: boolean
completion: Deferred<void>
}

Expand Down Expand Up @@ -408,6 +410,7 @@ export class CollectionSubscription
// before a request returns its promise and joins the pending set.
replayState.pendingSetups++
replayState.failures.clear()
replayState.failed = false
replayState.currentAttempt = { pendingCount: 0, setupComplete: false }
} else {
// Every overlapping attempt shares one publication baseline and buffer.
Expand Down Expand Up @@ -471,7 +474,10 @@ export class CollectionSubscription
this.isSyncRunGenerationCurrent(replayState.syncRunGeneration) &&
this.isDemandActive(demand)
const fail = (error: unknown) => {
if (isCurrent()) replayState.failures.set(demand, normalizeError(error))
if (isCurrent()) {
replayState.failures.set(demand, normalizeError(error))
replayState.failed = true
}
}
if (demand.initialResult) {
void replayState.completion.promise.then(
Expand Down Expand Up @@ -587,6 +593,7 @@ export class CollectionSubscription
) {
const normalized = this.normalizeLoadSubsetPromiseError(result, error)
replayState.failures.set(demand, normalized)
replayState.failed = true
}
this.settleTruncateReplay(replayState, pending)
},
Expand Down Expand Up @@ -649,9 +656,19 @@ export class CollectionSubscription
if (this.truncateReplayState !== replayState) return
this.truncateReplayState = undefined
this.truncateReplacementPending = false
try {
this.publishReplayRows(replayState.privateRows)
} finally {
replayState.completion.resolve()
this.options.truncateReplayPublication?.succeed()
}
}

/** Replace the public snapshot with a replay's private rows in one batch. */
private publishReplayRows(
privateRows: Map<string | number, object> | undefined,
): void {
// Retained rows the source never re-delivered leave the replacement.
const { privateRows } = replayState
for (const key of this.stalePublishedRows.keys()) privateRows?.delete(key)
this.stalePublishedRows.clear()
try {
Expand All @@ -666,8 +683,6 @@ export class CollectionSubscription
} finally {
// Restore tracking even when a subscriber rejects the replacement.
this.restorePublishedSnapshotTracking()
replayState.completion.resolve()
this.options.truncateReplayPublication?.succeed()
}
}

Expand Down Expand Up @@ -778,6 +793,7 @@ export class CollectionSubscription
pendingSetups: 1,
currentAttempt: { pendingCount: 0, setupComplete: false },
failures: new Map(),
failed: false,
completion: createReplayCompletion(),
}
}
Expand Down Expand Up @@ -1074,6 +1090,7 @@ export class CollectionSubscription
replayState.currentAttempt === replayAttempt
) {
replayState.failures.set(demand, normalizeError(error))
replayState.failed = true
}
this.subsetDemands.splice(demandIndex, 1)
}
Expand Down Expand Up @@ -1358,9 +1375,19 @@ export class CollectionSubscription
runAllCallbacks(releaseCallbacks)
}

/** A replay with no remaining logical demand cannot establish more rows. */
/**
* A replay with no remaining logical demand cannot establish more rows. One
* that has not failed publishes the rows it holds, so the subscriber does
* not keep values the source has replaced or deleted. A failed replay keeps
* the last public snapshot until an authoritative replay succeeds.
*/
private retireEmptyReplay(): void {
if (this.subsetDemands.length !== 0 || !this.truncateReplayState) {
const replayState = this.truncateReplayState
if (this.subsetDemands.length !== 0 || !replayState) {
return
}
if (replayState.privateRows && !replayState.failed) {
this.flushTruncateReplay(replayState)
return
}
this.discardTruncateReplay()
Expand Down
8 changes: 7 additions & 1 deletion packages/db/src/query/live/ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -1263,7 +1263,13 @@ replacement. Replay failure is scoped to the logical demand that failed. If
that demand retires, its failure cannot poison a successful replacement for the
remaining demand. If the last logical demand retires, the now-unreachable source
replay rejects its completion with `AbortError` and stops gating the shared
graph; unrelated parent or sibling changes may then publish. If release
graph; unrelated parent or sibling changes may then publish. A direct
subscriber's replay whose current attempt has not failed instead publishes the
rows it holds at that retirement and resolves its completion, because nothing
can complete it and keeping the old snapshot would leave values the source
replaced or deleted. A replay whose
current attempt failed keeps the last public snapshot, even after its failed
owner was released; a later request reconciles a row it still shows. If release
publication or adapter unload synchronously acquires new demand, core checks
completion after that callback: the new demand joins the private replacement,
while the retired transport can no longer gate it. A genuine replay failure is
Expand Down
Loading
Loading