From b0077c397b1c0b797a08c8675e2eb6cb5c4dcc97 Mon Sep 17 00:00:00 2001
From: Isaac <isaac@example.com>
Date: Mon, 12 Oct 2026 14:28:33 -0600
Subject: [PATCH 1/5] test(db): a request after a retired replay updates the
 stale row

The publication oracle's model let no request change a row the
subscriber had already seen. After a replay retired because its last
owner was released, the source kept a newer value and the subscriber
kept showing the old one indefinitely. Production reconciles that row
on the next request, which subscription.ts documents as intended. The
model now expects the update, the law prose says why, and seed
-136205996 is a named history. The witness fails against the previous
model and against a production mutant that skips the reconciliation.

Co-authored-by: Isaac <no-reply@databricks.com>
---
 docs/contributing/oracle-coverage.md          |  4 +-
 ...ecycle-publication-oracle.property.test.ts | 70 ++++++++++++++++---
 2 files changed, 63 insertions(+), 11 deletions(-)

diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md
index 2c53ec87d6..feb0038fa8 100644
--- a/docs/contributing/oracle-coverage.md
+++ b/docs/contributing/oracle-coverage.md
@@ -469,7 +469,9 @@ invariant witness does not claim arbitrary manually constructed IR is valid.
 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
+  owns exact public rows and change batches. A replay that retires when its
+  last owner is released leaves its source writes installed; a later request
+  updates a row the subscriber still shows at an older value. 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.
 - The [pagination oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/query/pagination-oracle.property.test.ts)
diff --git a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
index e07af6d16a..81af06321b 100644
--- a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
+++ b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
@@ -32,6 +32,10 @@ import type {
  * failure keeps the replacement and later source changes private while the
  * last public snapshot remains visible. Retiring the final logical subset
  * owner can retire that replacement barrier without publishing its rows.
+ * Its source writes stay installed, so the subscriber can show an older value
+ * than the source. No barrier gates a later request, and that request
+ * delivers its installed row: an update when the subscriber shows an older
+ * value. Leaving the older value would keep a stale row visible.
  * The `Collection.subscribeChanges` API contract in `../src/collection/index.ts`
  * is the authority for callback batches and their per-key causal order. The
  * public Collection lifecycle and shared lifecycle grammar specify cleanup,
@@ -439,8 +443,9 @@ function projectPublication(
     publication.source.clear()
     publication.replacement = undefined
   } else if (command.type === `release`) {
-    // Release changes demand, not source retention. Only source writes or a
-    // successful replacement can change the subscriber's rows.
+    // Release changes demand, not source retention. Releasing alone never
+    // changes the subscriber's rows; a source write, a successful
+    // replacement, or a later request does.
     finishReplacement(publication, lifecycle)
   } else if (command.type === `settle` && effect.attemptId !== undefined) {
     const attempt = lifecycle.attempts[effect.attemptId]!
@@ -482,22 +487,39 @@ function projectPublication(
     index < lifecycle.publications;
     index++
   ) {
-    const row =
+    const requested =
       command.type === `request` &&
       lifecycle.active &&
       lifecycle.owners.filter(({ demand }) => demand === command.demand)
-        .length === 1 &&
-      !publication.sentKeys.has(command.demand)
+        .length === 1
         ? publication.source.get(command.demand)
         : undefined
+    // A request delivers its installed row. An unsent row is an insert. A
+    // sent row that the subscriber still shows at an older value, such as
+    // one kept after a replay retired without publishing, is an update to
+    // the installed value; leaving it would keep a stale row visible.
+    const shown = requested && publication.visible.get(requested.id)
+    const row =
+      requested &&
+      (!publication.sentKeys.has(requested.id) ||
+        (shown !== undefined && shown.value !== requested.value))
+        ? requested
+        : undefined
     publication.batches.push(
       row
         ? [
-            {
-              type: `insert`,
-              key: row.id,
-              value: cloneRow(row),
-            },
+            shown && publication.sentKeys.has(row.id)
+              ? {
+                  type: `update`,
+                  key: row.id,
+                  value: cloneRow(row),
+                  previousValue: cloneRow(shown),
+                }
+              : {
+                  type: `insert`,
+                  key: row.id,
+                  value: cloneRow(row),
+                },
           ]
         : [],
     )
@@ -1966,6 +1988,34 @@ describe(`CollectionSubscription lifecycle publication oracle`, () => {
     ])
   })
 
+  // Generated at seed -136205996: the replay retires when its last owner is
+  // released, so the source keeps a = 0 while the subscriber still shows
+  // a = 1. Requesting a again must update the stale row to the installed one.
+  it(`updates a row left stale by a retired replay when its demand is requested again`, async () => {
+    await runPublicationHistory([
+      { type: `source`, key: `a`, action: `upsert`, value: 1 },
+      { type: `request`, demand: `a` },
+      { type: `restart` },
+      { type: `abort`, demand: `b` },
+      { type: `truncate` },
+      {
+        type: `settle`,
+        demand: `a`,
+        scope: `obsolete`,
+        age: `oldest`,
+        outcome: `reject`,
+      },
+      { type: `release`, demand: `b` },
+      { type: `abort`, demand: `a` },
+      { type: `abort`, demand: `a` },
+      { type: `source`, key: `a`, action: `upsert`, value: 0 },
+      { type: `release`, demand: `a` },
+      { type: `request`, demand: `a` },
+      { type: `cleanup` },
+      { type: `restart` },
+    ])
+  })
+
   it(`retires failed private replacement rows when its final owner releases`, async () => {
     await runPublicationHistory(
       [

From cd5a65926335b1beffab6476122608cbf231d896 Mon Sep 17 00:00:00 2001
From: Isaac <isaac@example.com>
Date: Mon, 12 Oct 2026 15:20:32 -0600
Subject: [PATCH 2/5] fix(db): publish a retired replay's rows instead of
 keeping stale ones

When the last requested subset was released during a truncate replay,
the replay retired without publishing, and the subscriber kept values
the source had written, replaced or deleted during it. A later request
updated only its own row; other rows stayed stale until a write changed
them, and a delete was lost. A replay whose current attempt has not
failed now publishes the rows it holds when it retires. A failed replay
keeps the last public snapshot (law 9), now even after its failed owner
is released: releasing a demand deleted its failure, so that guard
needs a flag that outlives the release.

Both subscription oracles' models follow the new law. Named histories
cover a value written, rewritten or deleted during the replay, a row no
demand names, and a failed replay; they fail on main, and a mutant that
publishes failed replays fails the failed-replay witness. ARCHITECTURE.md
and the coverage map state the rule.

Co-authored-by: Isaac <no-reply@databricks.com>
---
 .changeset/fix-retired-replay-stale-rows.md   |   5 +
 docs/contributing/oracle-coverage.md          |  11 +-
 packages/db/src/collection/subscription.ts    |  35 ++++-
 packages/db/src/query/live/ARCHITECTURE.md    |   7 +-
 ...ecycle-publication-oracle.property.test.ts | 129 +++++++++++++-----
 ...ubscription-replay-oracle.property.test.ts |  35 ++++-
 .../tests/replay-publication-storage.test.ts  |  11 +-
 7 files changed, 187 insertions(+), 46 deletions(-)
 create mode 100644 .changeset/fix-retired-replay-stale-rows.md

diff --git a/.changeset/fix-retired-replay-stale-rows.md b/.changeset/fix-retired-replay-stale-rows.md
new file mode 100644
index 0000000000..058554c6f0
--- /dev/null
+++ b/.changeset/fix-retired-replay-stale-rows.md
@@ -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.
diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md
index feb0038fa8..620503468a 100644
--- a/docs/contributing/oracle-coverage.md
+++ b/docs/contributing/oracle-coverage.md
@@ -469,11 +469,14 @@ invariant witness does not claim arbitrary manually constructed IR is valid.
 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. A replay that retires when its
-  last owner is released leaves its source writes installed; a later request
-  updates a row the subscriber still shows at an older value. The
+  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, and a
+  later request that updates a row still shown at an older value.
 - The [pagination oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/query/pagination-oracle.property.test.ts)
   owns general ordered-window rows and publication. The
   [ordered-work oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/query/ordered-work-oracle.property.test.ts)
diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts
index 8f3a659fcb..d66760b9ad 100644
--- a/packages/db/src/collection/subscription.ts
+++ b/packages/db/src/collection/subscription.ts
@@ -131,6 +131,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>
 }
 
@@ -415,6 +417,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.
@@ -478,7 +481,9 @@ export class CollectionSubscription
       this.isSyncRunGenerationCurrent(replayState.syncRunGeneration) &&
       this.isDemandActive(demand)
     const fail = (error: unknown) => {
-      if (isCurrent()) replayState.failures.set(demand, normalizeError(error))
+      if (!isCurrent()) return
+      replayState.failures.set(demand, normalizeError(error))
+      replayState.failed = true
     }
     if (demand.initialResult) {
       void replayState.completion.promise.then(
@@ -594,6 +599,7 @@ export class CollectionSubscription
         ) {
           const normalized = this.normalizeLoadSubsetPromiseError(result, error)
           replayState.failures.set(demand, normalized)
+          replayState.failed = true
         }
         this.settleTruncateReplay(replayState, pending)
       },
@@ -799,6 +805,7 @@ export class CollectionSubscription
       pendingSetups: 1,
       currentAttempt: { pendingCount: 0, setupComplete: false },
       failures: new Map(),
+      failed: false,
       completion: createReplayCompletion(),
     }
   }
@@ -1095,6 +1102,7 @@ export class CollectionSubscription
           replayState.currentAttempt === replayAttempt
         ) {
           replayState.failures.set(demand, normalizeError(error))
+          replayState.failed = true
         }
         this.subsetDemands.splice(demandIndex, 1)
       }
@@ -1376,12 +1384,33 @@ 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
     }
     this.discardTruncateReplay()
+    const { privateRows } = replayState
+    if (privateRows && !replayState.failed) {
+      this.stalePublishedRows.clear()
+      try {
+        const replacement = this.createStateDiff(
+          this.publishedRows,
+          privateRows,
+        )
+        if (replacement.length > 0) this.filteredCallback(replacement)
+      } finally {
+        this.restorePublishedSnapshotTracking()
+        this.options.truncateReplayPublication?.succeed()
+      }
+      return
+    }
     this.stalePublishedRows = new Map(this.publishedRows)
     this.restorePublishedSnapshotTracking()
     this.options.truncateReplayPublication?.succeed()
diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md
index 7561e33172..7c05e51b92 100644
--- a/packages/db/src/query/live/ARCHITECTURE.md
+++ b/packages/db/src/query/live/ARCHITECTURE.md
@@ -1245,7 +1245,12 @@ 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 publishes the rows it
+holds at that retirement, 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
diff --git a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
index 81af06321b..e28d9b4bca 100644
--- a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
+++ b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
@@ -30,12 +30,14 @@ import type {
  * file adds the row-publication node. A truncate opens a private replacement;
  * source writes accumulate there until every gating attempt succeeds. Replay
  * failure keeps the replacement and later source changes private while the
- * last public snapshot remains visible. Retiring the final logical subset
- * owner can retire that replacement barrier without publishing its rows.
- * Its source writes stay installed, so the subscriber can show an older value
- * than the source. No barrier gates a later request, and that request
- * delivers its installed row: an update when the subscriber shows an older
- * value. Leaving the older value would keep a stale row visible.
+ * last public snapshot remains visible. Releasing the final logical subset
+ * owner retires the replacement, since nothing can complete it. A
+ * replacement whose current attempt has not failed then publishes the rows
+ * it holds, so the subscriber does not keep values the source replaced or
+ * deleted during it. One that failed stays private, as a failed replay does
+ * until an authoritative replay succeeds, even after its failed owner was
+ * released. A later request delivers its installed row, so a row that the
+ * subscriber still shows at an older value is updated then.
  * The `Collection.subscribeChanges` API contract in `../src/collection/index.ts`
  * is the authority for callback batches and their per-key causal order. The
  * public Collection lifecycle and shared lifecycle grammar specify cleanup,
@@ -114,6 +116,10 @@ type Replacement = {
   replayGeneration: number
   rows: Map<RowKey, Row>
   failed: boolean
+  // A current attempt failed, even if its owner has been released since.
+  // Only full retirement reads it: with owners left, a released owner's
+  // failure cannot stop a successful replacement.
+  everFailed: boolean
 }
 type PublicationModel = {
   source: Map<RowKey, Row>
@@ -311,6 +317,7 @@ function finishReplacement(
   replacement.failed = currentAttempts.some(
     ({ outcome }) => outcome === `reject`,
   )
+  replacement.everFailed ||= replacement.failed
   if (lifecycle.publicationBarrierOpen) return
   if (replacement.failed) {
     replacement.failed = true
@@ -326,6 +333,13 @@ function finishReplacement(
     publishIfChanged(publication, new Map(replacement.rows))
     publication.retainedKeys.clear()
     publication.replacement = undefined
+  } else if (lifecycle.owners.length === 0 && !replacement.everFailed) {
+    // Releasing the final owner retires a replay that has not failed. Nothing
+    // can complete it, so it publishes the rows it holds rather than keep
+    // values the source has replaced or deleted.
+    publishIfChanged(publication, new Map(replacement.rows))
+    publication.retainedKeys.clear()
+    publication.replacement = undefined
   } else {
     // Retire private publication work, not the source's independently applied state.
     publication.replacement = undefined
@@ -360,6 +374,7 @@ function projectPublication(
         replayGeneration: lifecycle.replayGeneration,
         rows: new Map(publication.source),
         failed: false,
+        everFailed: false,
       }
     } else {
       publication.replacement = undefined
@@ -401,6 +416,7 @@ function projectPublication(
       replayGeneration: lifecycle.replayGeneration,
       rows: new Map(),
       failed: false,
+      everFailed: false,
     }
   } else if (
     command.type === `source` &&
@@ -494,21 +510,17 @@ function projectPublication(
         .length === 1
         ? publication.source.get(command.demand)
         : undefined
-    // A request delivers its installed row. An unsent row is an insert. A
-    // sent row that the subscriber still shows at an older value, such as
-    // one kept after a replay retired without publishing, is an update to
-    // the installed value; leaving it would keep a stale row visible.
+    // A request delivers its installed row: an insert when the subscriber
+    // does not show it, an update when it shows an older value, such as a
+    // row a failed replay kept, and nothing when it shows the same value.
+    // Either way the row is no longer retained.
     const shown = requested && publication.visible.get(requested.id)
     const row =
-      requested &&
-      (!publication.sentKeys.has(requested.id) ||
-        (shown !== undefined && shown.value !== requested.value))
-        ? requested
-        : undefined
+      requested && requested.value !== shown?.value ? requested : undefined
     publication.batches.push(
       row
         ? [
-            shown && publication.sentKeys.has(row.id)
+            shown
               ? {
                   type: `update`,
                   key: row.id,
@@ -523,9 +535,9 @@ function projectPublication(
           ]
         : [],
     )
+    if (requested) publication.retainedKeys.delete(requested.id)
     if (row) {
       publication.visible.set(row.id, cloneRow(row))
-      publication.retainedKeys.delete(row.id)
       publication.sentKeys.add(row.id)
     }
   }
@@ -1988,31 +2000,86 @@ describe(`CollectionSubscription lifecycle publication oracle`, () => {
     ])
   })
 
-  // Generated at seed -136205996: the replay retires when its last owner is
-  // released, so the source keeps a = 0 while the subscriber still shows
-  // a = 1. Requesting a again must update the stale row to the installed one.
-  it(`updates a row left stale by a retired replay when its demand is requested again`, async () => {
+  // Releasing the final owner retires a replay that has not failed; it
+  // publishes the rows it holds, so none of these rows stays stale.
+  it.each([
+    [
+      `a value written during the replay`,
+      [
+        { type: `source`, key: `a`, action: `upsert`, value: 1 },
+        { type: `request`, demand: `a` },
+        { type: `truncate` },
+        { type: `abort`, demand: `a` },
+        { type: `source`, key: `a`, action: `upsert`, value: 0 },
+        { type: `release`, demand: `a` },
+        { type: `request`, demand: `a` },
+      ],
+    ],
+    [
+      `a value written again after the replay retired`,
+      [
+        { type: `source`, key: `a`, action: `upsert`, value: 1 },
+        { type: `request`, demand: `a` },
+        { type: `truncate` },
+        { type: `abort`, demand: `a` },
+        { type: `source`, key: `a`, action: `upsert`, value: 2 },
+        { type: `release`, demand: `a` },
+        { type: `source`, key: `a`, action: `upsert`, value: 2 },
+      ],
+    ],
+    [
+      `a row no demand names`,
+      [
+        { type: `source`, key: `c`, action: `upsert`, value: 1 },
+        { type: `request`, demand: `a` },
+        { type: `truncate` },
+        { type: `abort`, demand: `a` },
+        { type: `source`, key: `c`, action: `upsert`, value: 2 },
+        { type: `release`, demand: `a` },
+      ],
+    ],
+    [
+      `a row deleted during the replay`,
+      [
+        { type: `source`, key: `a`, action: `upsert`, value: 1 },
+        { type: `request`, demand: `a` },
+        { type: `truncate` },
+        { type: `abort`, demand: `a` },
+        { type: `source`, key: `a`, action: `delete`, value: 0 },
+        { type: `release`, demand: `a` },
+        { type: `request`, demand: `a` },
+      ],
+    ],
+  ] as Array<[string, Array<PublicationCommand>]>)(
+    `leaves no stale row after a retired replay: %s`,
+    async (_, history) => {
+      await runPublicationHistory(history)
+    },
+  )
+
+  // Law 9: a failed replay stays private until an authoritative replay
+  // succeeds, even after its failed owner is released and nothing remains.
+  it(`keeps a failed replay private when its final owner releases`, async () => {
     await runPublicationHistory([
       { type: `source`, key: `a`, action: `upsert`, value: 1 },
       { type: `request`, demand: `a` },
-      { type: `restart` },
-      { type: `abort`, demand: `b` },
+      {
+        type: `settle`,
+        demand: `a`,
+        scope: `current`,
+        age: `oldest`,
+        outcome: `resolve`,
+      },
       { type: `truncate` },
+      { type: `source`, key: `a`, action: `upsert`, value: 5 },
       {
         type: `settle`,
         demand: `a`,
-        scope: `obsolete`,
+        scope: `current`,
         age: `oldest`,
         outcome: `reject`,
       },
-      { type: `release`, demand: `b` },
-      { type: `abort`, demand: `a` },
-      { type: `abort`, demand: `a` },
-      { type: `source`, key: `a`, action: `upsert`, value: 0 },
       { type: `release`, demand: `a` },
-      { type: `request`, demand: `a` },
-      { type: `cleanup` },
-      { type: `restart` },
     ])
   })
 
diff --git a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
index cc60ca7ac6..dc8b8d03e1 100644
--- a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
+++ b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
@@ -604,6 +604,8 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
           pending: Set<number>
           currentAttemptIndex: number
           publicationCount: number
+          // The current attempt failed, even if its demand was released since.
+          failed: boolean
         }
       | undefined
 
@@ -651,6 +653,7 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
       } else {
         if (isCurrent) {
           lastReportedError = pending.error
+          replayState.failed = true
         } else {
           expect(pending.signal?.aborted).toBe(true)
         }
@@ -721,8 +724,14 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
         pending: new Set(),
         currentAttemptIndex: attemptIndex,
         publicationCount: expectedPublicationCount,
+        failed: false,
       }
-      modelReplayState.currentAttemptIndex = attemptIndex
+      // settleReplay records a failure from a callback, so reset without
+      // narrowing the field for this scope.
+      Object.assign(modelReplayState, {
+        currentAttemptIndex: attemptIndex,
+        failed: false,
+      })
 
       for (const load of attempt.loads) {
         queuedLoads.push({ attemptIndex, load })
@@ -757,8 +766,24 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
           }
         }
         // A released request does not retract rows already applied by the
-        // source, nor change the retained baseline of an unfinished replay.
+        // source. Releasing the final demand retires the replay: one that has
+        // not failed publishes the rows the source applied, since nothing can
+        // complete it; a failed one keeps the retained baseline (law 9).
         if (modelReplayState.pending.size === 0 && activeDemandIds.size === 0) {
+          if (!modelReplayState.failed) {
+            const previousPublication = new Map(expectedPublished)
+            expectedPublished.clear()
+            for (const [id, row] of sourceRows) {
+              expectedPublished.set(id, { ...row })
+            }
+            const expectedBatch = publicationDiff(
+              previousPublication,
+              expectedPublished,
+            )
+            expect(publicationCount - modelReplayState.publicationCount).toBe(
+              Number(expectedBatch.length > 0),
+            )
+          }
           expectedPublicationCount = publicationCount
           modelReplayState = undefined
         }
@@ -4253,7 +4278,11 @@ describe(`CollectionSubscription replay oracle`, () => {
         expect(sortedChanges(batches[0]!)).toEqual([
           { type: `insert`, key: `one`, value: { id: `one`, value: 1 } },
         ])
-        expect(sortedChanges(batches.at(-1)!)).toEqual([
+        // Retiring the replay publishes its applied row at the release; the
+        // reacquired demand then has nothing new to deliver.
+        expect(
+          sortedChanges(batches.filter((batch) => batch.length > 0).at(-1)!),
+        ).toEqual([
           {
             type: `update`,
             key: `one`,
diff --git a/packages/db/tests/replay-publication-storage.test.ts b/packages/db/tests/replay-publication-storage.test.ts
index ca7392f365..b85ee5f7de 100644
--- a/packages/db/tests/replay-publication-storage.test.ts
+++ b/packages/db/tests/replay-publication-storage.test.ts
@@ -298,7 +298,7 @@ describe(`Replay publication storage`, () => {
     })
   })
 
-  it(`direct: releasing the last demand during replay retires it; re-acquisition reconciles`, async () => {
+  it(`direct: releasing the last demand during replay retires it and publishes its rows; re-acquisition reconciles`, async () => {
     return withReplayResources(async (resources) => {
       const s = makeSource(`probe-release`)
       resources.sources.push(() => s.source.cleanup())
@@ -337,16 +337,19 @@ describe(`Replay publication storage`, () => {
       expect((outcome as PromiseRejectedResult).reason.name).toBe(`AbortError`)
       expect(sub.status).toBe(`ready`)
       expect(sub.hasPendingTruncateReplacement).toBe(false)
-      expect(visible.get(1)).toBe(1)
+      // Nothing can complete the retired replay, and its attempt did not
+      // fail, so it publishes the row the source replaced during it.
+      expect(visible.get(1)).toBe(2)
+      expect(batches.at(-1)).toEqual([[`update`, 1, 2]])
       expect(replayStates(s.source)).toEqual([])
 
       // Late settlement of the released transport changes nothing.
       hold.resolve()
       await flushPromises()
-      expect(visible.get(1)).toBe(1)
+      expect(visible.get(1)).toBe(2)
       expect(sub.status).toBe(`ready`)
 
-      // Re-acquiring reconciles the retained row against the source.
+      // Re-acquiring reconciles the row against the source.
       s.setHold(undefined)
       s.setVersion(3)
       sub.requestSnapshot({ where, optimizedOnly: false })

From e04b0f36e24d41dc64fc4ade12eec6634b734a86 Mon Sep 17 00:00:00 2001
From: Isaac <isaac@example.com>
Date: Mon, 12 Oct 2026 15:42:36 -0600
Subject: [PATCH 3/5] fix(db): drop stale rows when a retired replay publishes

A retired replay seeded its private rows from the public snapshot, so a row
retained by an earlier failed replay and deleted by the source stayed visible
and lost its stale marker. Share flushTruncateReplay's publish step, which
removes stale keys first, with retireEmptyReplay.

Add the two ghost-row histories to the publication oracle, and witnesses for
each synchronous failure site at retirement to the replay oracle. Compare the
retire batch's contents in the replay oracle, not only its count.

Co-authored-by: Isaac <no-reply@databricks.com>
---
 docs/contributing/oracle-coverage.md          |  11 +-
 packages/db/src/collection/subscription.ts    |  37 +++---
 ...ecycle-publication-oracle.property.test.ts |  68 ++++++++++
 ...ubscription-replay-oracle.property.test.ts | 118 +++++++++++++++++-
 4 files changed, 207 insertions(+), 27 deletions(-)

diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md
index 620503468a..dff1f3784c 100644
--- a/docs/contributing/oracle-coverage.md
+++ b/docs/contributing/oracle-coverage.md
@@ -475,8 +475,15 @@ The following related oracles own distinct production paths or observation cuts:
   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, and a
-  later request that updates a row still shown at an older value.
+  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.
 - The [pagination oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/query/pagination-oracle.property.test.ts)
   owns general ordered-window rows and publication. The
   [ordered-work oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/query/ordered-work-oracle.property.test.ts)
diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts
index d66760b9ad..3e74889282 100644
--- a/packages/db/src/collection/subscription.ts
+++ b/packages/db/src/collection/subscription.ts
@@ -481,9 +481,10 @@ export class CollectionSubscription
       this.isSyncRunGenerationCurrent(replayState.syncRunGeneration) &&
       this.isDemandActive(demand)
     const fail = (error: unknown) => {
-      if (!isCurrent()) return
-      replayState.failures.set(demand, normalizeError(error))
-      replayState.failed = true
+      if (isCurrent()) {
+        replayState.failures.set(demand, normalizeError(error))
+        replayState.failed = true
+      }
     }
     if (demand.initialResult) {
       void replayState.completion.promise.then(
@@ -664,9 +665,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 {
@@ -681,8 +692,6 @@ export class CollectionSubscription
     } finally {
       // Restore tracking even when a subscriber rejects the replacement.
       this.restorePublishedSnapshotTracking()
-      replayState.completion.resolve()
-      this.options.truncateReplayPublication?.succeed()
     }
   }
 
@@ -1396,19 +1405,9 @@ export class CollectionSubscription
       return
     }
     this.discardTruncateReplay()
-    const { privateRows } = replayState
-    if (privateRows && !replayState.failed) {
-      this.stalePublishedRows.clear()
-      try {
-        const replacement = this.createStateDiff(
-          this.publishedRows,
-          privateRows,
-        )
-        if (replacement.length > 0) this.filteredCallback(replacement)
-      } finally {
-        this.restorePublishedSnapshotTracking()
-        this.options.truncateReplayPublication?.succeed()
-      }
+    // Only a direct subscriber holds private rows; it has no delegate to release.
+    if (replayState.privateRows && !replayState.failed) {
+      this.publishReplayRows(replayState.privateRows)
       return
     }
     this.stalePublishedRows = new Map(this.publishedRows)
diff --git a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
index e28d9b4bca..5cad8f6ef4 100644
--- a/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
+++ b/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts
@@ -2050,6 +2050,74 @@ describe(`CollectionSubscription lifecycle publication oracle`, () => {
         { type: `request`, demand: `a` },
       ],
     ],
+    [
+      `a row retained by a failed replay and deleted before the next one`,
+      [
+        { type: `source`, key: `a`, action: `upsert`, value: 1 },
+        { type: `request`, demand: `a` },
+        {
+          type: `settle`,
+          demand: `a`,
+          scope: `current`,
+          age: `oldest`,
+          outcome: `resolve`,
+        },
+        { type: `truncate` },
+        { type: `source`, key: `a`, action: `delete`, value: 0 },
+        {
+          type: `settle`,
+          demand: `a`,
+          scope: `current`,
+          age: `oldest`,
+          outcome: `reject`,
+        },
+        { type: `release`, demand: `a` },
+        { type: `request`, demand: `b` },
+        {
+          type: `settle`,
+          demand: `b`,
+          scope: `current`,
+          age: `oldest`,
+          outcome: `resolve`,
+        },
+        { type: `truncate` },
+        { type: `release`, demand: `b` },
+      ],
+    ],
+    [
+      `an undemanded row retained by a failed replay and deleted before the next one`,
+      [
+        { type: `source`, key: `c`, action: `upsert`, value: 1 },
+        { type: `request`, demand: `a` },
+        {
+          type: `settle`,
+          demand: `a`,
+          scope: `current`,
+          age: `oldest`,
+          outcome: `resolve`,
+        },
+        { type: `truncate` },
+        { type: `source`, key: `c`, action: `delete`, value: 0 },
+        {
+          type: `settle`,
+          demand: `a`,
+          scope: `current`,
+          age: `oldest`,
+          outcome: `reject`,
+        },
+        { type: `release`, demand: `a` },
+        { type: `request`, demand: `b` },
+        {
+          type: `settle`,
+          demand: `b`,
+          scope: `current`,
+          age: `oldest`,
+          outcome: `resolve`,
+        },
+        { type: `truncate` },
+        { type: `release`, demand: `b` },
+      ],
+    ],
   ] as Array<[string, Array<PublicationCommand>]>)(
     `leaves no stale row after a retired replay: %s`,
     async (_, history) => {
diff --git a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
index dc8b8d03e1..42457823af 100644
--- a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
+++ b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
@@ -609,6 +609,15 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
         }
       | undefined
 
+    // A new attempt supersedes the previous one, and its failure with it.
+    const startModelAttempt = (
+      state: NonNullable<typeof modelReplayState>,
+      attemptIndex: number,
+    ): void => {
+      state.currentAttemptIndex = attemptIndex
+      state.failed = false
+    }
+
     const writeReplayRows = (
       pending: PendingReplay,
       isCurrent: boolean,
@@ -726,12 +735,7 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
         publicationCount: expectedPublicationCount,
         failed: false,
       }
-      // settleReplay records a failure from a callback, so reset without
-      // narrowing the field for this scope.
-      Object.assign(modelReplayState, {
-        currentAttemptIndex: attemptIndex,
-        failed: false,
-      })
+      startModelAttempt(modelReplayState, attemptIndex)
 
       for (const load of attempt.loads) {
         queuedLoads.push({ attemptIndex, load })
@@ -783,6 +787,11 @@ async function runReplayScenario(scenario: ReplayScenario): Promise<void> {
             expect(publicationCount - modelReplayState.publicationCount).toBe(
               Number(expectedBatch.length > 0),
             )
+            if (expectedBatch.length > 0) {
+              expect(sortedChanges(publicationBatches.at(-1)!)).toEqual(
+                sortedChanges(expectedBatch),
+              )
+            }
           }
           expectedPublicationCount = publicationCount
           modelReplayState = undefined
@@ -2412,6 +2421,103 @@ describe(`CollectionSubscription replay oracle`, () => {
     })
   })
 
+  // Law 9 at retirement: a failure in the current attempt keeps the replay
+  // private even after the last owner is released. Each site writes a row
+  // before it fails, so publishing the private rows would show that row.
+  it.each([
+    `replay load throws`,
+    `old lease unload throws`,
+    `new request throws`,
+  ])(
+    `keeps a replay private at retirement after a synchronous failure: %s`,
+    async (site) => {
+      let begin!: () => void
+      let write!: (message: {
+        type: `insert` | `update`
+        value: ReplayRow
+      }) => void
+      let commit!: () => void
+      let truncate!: () => void
+      let loads = 0
+      let unloads = 0
+      const writeRow = (row: ReplayRow) => {
+        begin()
+        write({ type: `insert`, value: row })
+        commit()
+      }
+      const collection = createCollection<ReplayRow>({
+        id: `retired-replay-synchronous-failure-${site}`,
+        getKey: ({ id }) => id,
+        syncMode: `on-demand`,
+        sync: {
+          sync: (operations) => {
+            begin = operations.begin
+            write = operations.write
+            commit = operations.commit
+            truncate = operations.truncate
+            operations.markReady()
+            return {
+              loadSubset: () => {
+                loads++
+                if (loads === 1) {
+                  writeRow({ id: `one`, value: 1 })
+                  return true
+                }
+                if (site === `replay load throws` && loads === 2) {
+                  writeRow({ id: `one`, value: 5 })
+                  throw new Error(`replay load failed`)
+                }
+                if (site === `new request throws` && loads === 3) {
+                  writeRow({ id: `two`, value: 2 })
+                  throw new Error(`request failed`)
+                }
+                return createDeferred<void>().promise
+              },
+              unloadSubset: () => {
+                unloads++
+                if (site === `old lease unload throws` && unloads === 1) {
+                  writeRow({ id: `two`, value: 2 })
+                  throw new Error(`unload failed`)
+                }
+              },
+            }
+          },
+        },
+      })
+      const visible = new Map<string | number, ReplayRow>()
+      const subscription = collection.subscribeChanges((changes) => {
+        recordPublishedChanges(visible, changes)
+      })
+      const where = new Func(`eq`, [new PropRef([`id`]), new Value(`one`)])
+      try {
+        subscription.requestSnapshot({ where, optimizedOnly: false })
+        await flushPromises()
+        expect(sortedRows(visible)).toEqual([{ id: `one`, value: 1 }])
+
+        begin()
+        truncate()
+        commit()
+        await flushPromises()
+        if (site === `new request throws`) {
+          expect(() =>
+            subscription.requestSnapshot({
+              where: new Func(`eq`, [new PropRef([`id`]), new Value(`two`)]),
+              optimizedOnly: false,
+            }),
+          ).toThrow(`request failed`)
+        }
+        subscription.releaseSnapshot(where)
+        await flushPromises()
+
+        expect(subscription.pendingTruncateReplacement).toBeUndefined()
+        expect(sortedRows(visible)).toEqual([{ id: `one`, value: 1 }])
+      } finally {
+        subscription.unsubscribe()
+        await collection.cleanup()
+      }
+    },
+  )
+
   it(`replaces a retained snapshot with a later empty replay`, async () => {
     await runReplayScenario({
       initialRows: [{ id: `one`, value: 1 }],

From 85c0f697430e8c13f3e25035f05558066f1a91de Mon Sep 17 00:00:00 2001
From: Isaac <isaac@example.com>
Date: Mon, 12 Oct 2026 17:01:17 -0600
Subject: [PATCH 4/5] fix(db): resolve a retired replay's completion when it
 publishes

A retired replay that has not failed now finishes through
flushTruncateReplay, so its completion resolves with the published rows
instead of rejecting with AbortError. No production consumer reads that
completion on this path; the tests that pinned the abort now expect it to
resolve.

Co-authored-by: Isaac <no-reply@databricks.com>
---
 packages/db/src/collection/subscription.ts           |  5 ++---
 packages/db/src/query/live/ARCHITECTURE.md           |  7 ++++---
 ...ction-subscription-replay-oracle.property.test.ts | 12 ++++--------
 packages/db/tests/replay-publication-storage.test.ts |  4 ++--
 4 files changed, 12 insertions(+), 16 deletions(-)

diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts
index 3e74889282..ddeb576bfe 100644
--- a/packages/db/src/collection/subscription.ts
+++ b/packages/db/src/collection/subscription.ts
@@ -1404,12 +1404,11 @@ export class CollectionSubscription
     if (this.subsetDemands.length !== 0 || !replayState) {
       return
     }
-    this.discardTruncateReplay()
-    // Only a direct subscriber holds private rows; it has no delegate to release.
     if (replayState.privateRows && !replayState.failed) {
-      this.publishReplayRows(replayState.privateRows)
+      this.flushTruncateReplay(replayState)
       return
     }
+    this.discardTruncateReplay()
     this.stalePublishedRows = new Map(this.publishedRows)
     this.restorePublishedSnapshotTracking()
     this.options.truncateReplayPublication?.succeed()
diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md
index 7c05e51b92..1706b5ae3f 100644
--- a/packages/db/src/query/live/ARCHITECTURE.md
+++ b/packages/db/src/query/live/ARCHITECTURE.md
@@ -1246,9 +1246,10 @@ 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. A direct
-subscriber's replay whose current attempt has not failed publishes the rows it
-holds at that retirement, because nothing can complete it and keeping the old
-snapshot would leave values the source replaced or deleted. A replay whose
+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
diff --git a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
index 42457823af..71c7130b30 100644
--- a/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
+++ b/packages/db/tests/collection-subscription-replay-oracle.property.test.ts
@@ -4345,11 +4345,10 @@ describe(`CollectionSubscription replay oracle`, () => {
         reacquireOnReady = reacquireTiming === `during-ready`
         reacquireInUnload = reacquireTiming === `during-unload`
         subscription.releaseSnapshot(where)
+        // Releasing the final demand retires the replay, which publishes the
+        // row it holds, so the replacement resolves.
         if (reacquireTiming === `after-release`) {
-          await expect(settlement).resolves.toMatchObject({
-            status: `rejected`,
-            error: { name: `AbortError` },
-          })
+          await expect(settlement).resolves.toEqual({ status: `resolved` })
           subscription.requestSnapshot({ where })
         } else if (reacquireTiming === `during-unload`) {
           let settled = false
@@ -4373,10 +4372,7 @@ describe(`CollectionSubscription replay oracle`, () => {
           await expect(settlement).resolves.toEqual({ status: `resolved` })
         } else {
           expect(subscription.pendingTruncateReplacement).toBeUndefined()
-          await expect(settlement).resolves.toMatchObject({
-            status: `rejected`,
-            error: { name: `AbortError` },
-          })
+          await expect(settlement).resolves.toEqual({ status: `resolved` })
         }
 
         expect(subscription.pendingTruncateReplacement).toBeUndefined()
diff --git a/packages/db/tests/replay-publication-storage.test.ts b/packages/db/tests/replay-publication-storage.test.ts
index b85ee5f7de..c5215773c5 100644
--- a/packages/db/tests/replay-publication-storage.test.ts
+++ b/packages/db/tests/replay-publication-storage.test.ts
@@ -333,8 +333,8 @@ describe(`Replay publication storage`, () => {
 
       sub.releaseSnapshot(where)
       const [outcome] = await settled
-      expect(outcome.status).toBe(`rejected`)
-      expect((outcome as PromiseRejectedResult).reason.name).toBe(`AbortError`)
+      // The retired replay published its rows, so its replacement resolves.
+      expect(outcome.status).toBe(`fulfilled`)
       expect(sub.status).toBe(`ready`)
       expect(sub.hasPendingTruncateReplacement).toBe(false)
       // Nothing can complete the retired replay, and its attempt did not

From df7eb161b298e1d73ab308b60ea74c1fce65c730 Mon Sep 17 00:00:00 2001
From: Isaac <isaac@example.com>
Date: Mon, 12 Oct 2026 23:26:58 -0600
Subject: [PATCH 5/5] fix: restore trailbase type assertions a merge commit's
 lint fix removed

Co-authored-by: Isaac <no-reply@databricks.com>
---
 packages/trailbase-db-collection/src/trailbase.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/trailbase-db-collection/src/trailbase.ts b/packages/trailbase-db-collection/src/trailbase.ts
index 53ea350800..193bb71d89 100644
--- a/packages/trailbase-db-collection/src/trailbase.ts
+++ b/packages/trailbase-db-collection/src/trailbase.ts
@@ -264,7 +264,7 @@ export function trailBaseCollectionOptions<
           for (let i = 0; i < Math.min(length, remaining); ++i) {
             write({
               type: `insert`,
-              value: parse(response.records[i]),
+              value: parse(response.records[i]!),
             })
           }
 
@@ -287,7 +287,7 @@ export function trailBaseCollectionOptions<
               (opts.subscription === undefined || lastKey !== undefined)
             ) {
               cursors.set(
-                getKey(parse(response.records.at(-1))),
+                getKey(parse(response.records.at(-1)!)),
                 response.cursor,
               )
             }