fix(db): preserve restored rows across subscription restart

.changeset/fix-restarted-subscription-rows.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/db': patch
+---
+
+Preserve restored rows for direct `subscribeChanges()` consumers after Collection cleanup and restart, even when their values are unchanged. This prevents incorrect delete events when the new sync run restores the same rows.

docs/contributing/oracle-coverage.md

@@ -2349,6 +2349,37 @@ The following related oracles own distinct production paths or observation cuts:
   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 cleanup/restart equality crossing (`subscription-lifecycle.restart-publication`)
+  retains equal rows actually restored into a direct subscriber's private
+  replacement while deleting omitted retained rows. It runs the shared model,
+  driver, and exact per-command batch/source/subscriber comparisons over two
+  delivery paths (acquisition writes and independent source writes), six demand
+  lifetimes (pending, settled, two live demands, aborted, released, absent),
+  three prior-row relationships (absent, equal, changed), and resolve/reject
+  commands for live acquisitions. Acquisition histories without a live attempt
+  distinguish absent from retained rows and measure the settlement as a no-op;
+  they do not claim restored equality. The single scenario grammar has 24
+  acquisition and 27 source cases. Each case reaches its measured owner lifetime,
+  prior/delivered row relation, current outcome, and restarted acquisition count
+  12 times in each fixed/random campaign. Explicit generated row values and
+  zero to three prior retired acquisitions remove the fixture-attempt-ID premise.
+  Additional direct, filtered, ordered-snapshot and snapshot-before-source-events
+  drivers each measure absent/equal/changed rows 12 times per campaign and reuse
+  the owner's Map diff at held and settled publication cuts. The predicate
+  excludes a real nonmatching row; ordered selection covers the complete bounded
+  two-row replacement, not arbitrary windows. The exact issue #2310 history
+  remains a shared replay. Original suppression fails every restart campaign;
+  the earlier acquisition-only and retention-of-omitted-rows mutants remain
+  recorded. [Evidence and limits](oracle-reviews/2026-10-30-pr-2313-review.md).
+  This crossing concerns direct subscriptions under the controlled sync boundary.
+  Eager sources follow a separate accepted first-public-batch reconciliation law
+  in ARCHITECTURE's cleanup/restart section. The supplemental
+  `collection-restart-publication-boundaries.test.ts` observes its before-ready
+  deletion and later reinsertion, and checks actual compiled-query cleanup and
+  restart in eager/on-demand modes. Public cleanup refuses active dependents;
+  internal cleanup crashes them; restarting the query builds a fresh result.
+  This evidence does not claim source-only restart revives a query, nor does it
+  establish delegated includes, arbitrary ordered windows, or external providers.
   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.

packages/db/mangle-cache.json

@@ -106,7 +106,6 @@
   "retainedWindow": "b8",
   "setReadyIfIdle": "b9",
   "cachedCollectionRevision": "ca",
-  "collectionRestartCleanup": "cb",
   "pendingHydration": "cd",
   "removeFromBucket": "cf",
   "restorePublishedSnapshotTracking": "cg",
@@ -178,7 +177,6 @@
   "requireFullSourceRepair": "dy",
   "trackOrderedPublication": "dz",
   "canGarbageCollect": "dA",
-  "collectionCleanup": "dB",
   "extendPipelineWithChangeProcessing": "dC",
   "startSubsetDemand": "dD",
   "virtualPropsCache": "dE",
@@ -212,7 +210,6 @@
   "restoreInitialWindow": "d9",
   "settleTruncateReplay": "ea",
   "takeInternal": "eb",
-  "truncateCleanup": "ec",
   "_createRefForSource": "ee",
   "compileBasePipeline": "ef",
   "scheduleIdleCleanup": "eg",
@@ -265,7 +262,6 @@
   "publishEvents": "e5",
   "releaseDemand": "e6",
   "rollbackLease": "e7",
-  "trackSentKeys": "e9",
   "updateTimeout": "fa",
   "flushTruncateReplay": "fb",
   "accumulateActivity": "fc",
@@ -275,7 +271,6 @@
   "getOrCreateContext": "fg",
   "sentToD2Rows": "fh",
   "subscribeToChanges": "fi",
-  "trackPublishedRows": "fj",
   "_status": "fk",
   "clearSyncRunState": "fl",
   "retireEmptyReplay": "fm",
@@ -439,5 +434,6 @@
   "mergesOverRow": "b_",
   "awaitsSiblingReplays": "gN",
   "grownPrefixRows": "gZ",
-  "ownPrefixEnd": "he"
+  "ownPrefixEnd": "he",
+  "sourceListenerCleanups": "cb"
 }

packages/db/src/collection/subscription.ts

@@ -225,10 +225,8 @@ export class CollectionSubscription
     promise: Promise<unknown>
   }>()
 
-  // Cleanup function for truncate event listener
-  private truncateCleanup: (() => void) | undefined
-  private collectionCleanup: (() => void) | undefined
-  private collectionRestartCleanup: (() => void) | undefined
+  // Source listeners retire together on unsubscription.
+  private sourceListenerCleanups: Array<() => void>
 
   // One replay state owns the publication baseline, overlapping attempts,
   // and buffered changes until every attempt settles.
@@ -273,16 +271,20 @@ export class CollectionSubscription
       ensureIndexForExpression(options.whereExpression, this.collection)
     }
 
-    const callbackWithSentKeysTracking = (
-      changes: Array<ChangeMessage<any, any>>,
-    ) => {
-      this.trackPublishedRows(changes)
-      this.trackSentKeys(changes)
+    this.callback = (changes) => {
+      // Complete both publication records before a callback can reenter.
+      for (const change of changes) {
+        if (change.type === `delete`) {
+          this.publishedRows.delete(change.key)
+          this.sentKeys.delete(change.key)
+        } else {
+          this.publishedRows.set(change.key, change.value)
+          this.sentKeys.add(change.key)
+        }
+      }
       callback(changes)
     }
 
-    this.callback = callbackWithSentKeysTracking
-
     // Create a filtered callback if where clause is provided
     this.filteredCallback = options.whereExpression
       ? createFilteredCallback(this.callback, options)
@@ -295,15 +297,10 @@ export class CollectionSubscription
     // row, so every one re-requests the loaded subsets, also one that a
     // reload's new run sends after the reload's replay began or settled: the
     // replacement deleted what that replay loaded.
-    this.truncateCleanup = this.collection.on(`truncate`, () =>
-      this.handleTruncate(),
-    )
-    this.collectionCleanup = this.collection._onCleanupStart(() => {
-      this.handleCollectionCleanup()
-    })
-    this.collectionRestartCleanup = this.collection.on(
-      `status:change`,
-      ({ status }) => {
+    this.sourceListenerCleanups = [
+      this.collection.on(`truncate`, () => this.handleTruncate()),
+      this.collection._onCleanupStart(() => this.handleCollectionCleanup()),
+      this.collection.on(`status:change`, ({ status }) => {
         if (status !== `loading` && status !== `ready`) return
         const syncRunGeneration = this.collection._sync.getSyncRunGeneration()
         const replayState = this.truncateReplayState
@@ -319,8 +316,8 @@ export class CollectionSubscription
             this.restartDetachedDemands(syncRunGeneration)
           }
         })
-      },
-    )
+      }),
+    ]
   }
 
   /** Whether this subscription still defers provider work. */
@@ -1906,7 +1903,12 @@ export class CollectionSubscription
           value: previous,
           previousValue: undefined,
         })
-      } else if (!deepEquals(previous, change.value)) {
+      } else if (
+        this.truncateReplayState?.privateRows ||
+        !deepEquals(previous, change.value)
+      ) {
+        // An equal retained row still belongs in the private replacement.
+        // Its final diff suppresses the redundant public update.
         reconciled.push({
           ...change,
           type: `update`,
@@ -1929,28 +1931,6 @@ export class CollectionSubscription
     return reconciled
   }
 
-  private trackPublishedRows(
-    changes: Array<ChangeMessage<any, string | number>>,
-  ): void {
-    for (const change of changes) {
-      if (change.type === `delete`) {
-        this.publishedRows.delete(change.key)
-      } else {
-        this.publishedRows.set(change.key, change.value)
-      }
-    }
-  }
-
-  private trackSentKeys(changes: Array<ChangeMessage<any, string | number>>) {
-    for (const change of changes) {
-      if (change.type === `delete`) {
-        this.sentKeys.delete(change.key)
-      } else {
-        this.sentKeys.add(change.key)
-      }
-    }
-  }
-
   /**
    * Mark that the subscription should not filter any changes.
    * This is used when includeInitialState is explicitly set to false,
@@ -1966,17 +1946,11 @@ export class CollectionSubscription
     // Stop any status listener set already being iterated. Clearing the
     // emitter's map cannot invalidate that captured Set by itself.
     this.statusRevision++
-    const sourceListenerCleanups = [
-      this.truncateCleanup,
-      this.collectionCleanup,
-      this.collectionRestartCleanup,
-    ]
-    this.truncateCleanup = undefined
-    this.collectionCleanup = undefined
-    this.collectionRestartCleanup = undefined
+    const sourceListenerCleanups = this.sourceListenerCleanups
+    this.sourceListenerCleanups = []
 
     runAllCallbacks([
-      ...sourceListenerCleanups.map((cleanup) => () => cleanup?.()),
+      ...sourceListenerCleanups,
       () => {
         // Stop any buffered replay from publishing after unsubscription.
         this.discardTruncateReplay()

packages/db/tests/collection-restart-publication-boundaries.test.ts

@@ -0,0 +1,122 @@
+import { expect, it, vi } from 'vitest'
+import {
+  CollectionHasSubscribersError,
+  createCollection,
+  createLiveQueryCollection,
+} from '../src/index.js'
+import { flushPromises } from './utils.js'
+import type { SyncConfig } from '../src/types.js'
+
+type Row = { id: string; value: number }
+type Ops = Parameters<SyncConfig<Row, string>[`sync`]>[0]
+
+// ARCHITECTURE's cleanup/restart contract gives an eager source's first batch
+// replacement authority. Readiness is a separate checkpoint.
+it(`reconciles retained eager rows on the first restarted public batch`, async () => {
+  let ops!: Ops
+  const source = createCollection<Row, string>({
+    getKey: (row) => row.id,
+    sync: {
+      sync: (next) => {
+        ops = next
+      },
+    },
+  })
+  const visible = new Map<string, number>()
+  const batches: Array<Array<string>> = []
+  const sub = source.subscribeChanges(
+    (changes) => {
+      batches.push(changes.map((change) => `${change.type}:${change.key}`))
+      for (const change of changes) {
+        if (change.type === `delete`) visible.delete(change.key)
+        else visible.set(change.key, change.value.value)
+      }
+    },
+    { includeInitialState: false },
+  )
+  const write = (keys: Array<string>) => {
+    ops.begin()
+    keys.forEach((id) => ops.write({ type: `insert`, value: { id, value: 1 } }))
+    ops.commit()
+  }
+  try {
+    write([`a`, `b`, `c`])
+    ops.markReady()
+    await source.cleanup()
+    source.startSyncImmediate()
+    batches.length = 0
+    write([`a`])
+    expect(source.status).toBe(`loading`)
+    expect([...visible.keys()]).toEqual([`a`])
+    expect(batches).toEqual([[`delete:b`, `delete:c`]])
+    write([`b`, `c`])
+    ops.markReady()
+    expect([...visible.keys()]).toEqual([`a`, `b`, `c`])
+  } finally {
+    sub.unsubscribe()
+    await source.cleanup()
+  }
+})
+
+// Cleanup ends a dependent graph. Source restart alone cannot revive it.
+// These exact b/d observations supplement the cleanup admission oracle.
+for (const syncMode of [`eager`, `on-demand`] as const) {
+  for (const cleanup of [`public`, `dependents`, `internal`] as const) {
+    it(`${syncMode} compiled queries obey ${cleanup} cleanup before restart`, async () => {
+      let ops!: Ops
+      let run = 0
+      const reports = vi.spyOn(console, `error`).mockImplementation(() => {})
+      const source = createCollection<Row, string>({
+        getKey: (row) => row.id,
+        syncMode,
+        sync: {
+          sync: (next) => {
+            ops = next
+            run++
+            ops.begin()
+            for (const id of run === 1 ? [`b`, `d`] : [`b`])
+              ops.write({ type: `insert`, value: { id, value: 1 } })
+            ops.commit()
+            ops.markReady()
+            return { loadSubset: () => true }
+          },
+        },
+      })
+      const live = createLiveQueryCollection((q) =>
+        q.from({ row: source }).select(({ row }) => row),
+      )
+      const sub = live.subscribeChanges(() => {})
+      try {
+        await live.preload()
+        expect([...live.keys()].sort()).toEqual([`b`, `d`])
+        if (cleanup === `public`) {
+          await expect(source.cleanup()).rejects.toBeInstanceOf(
+            CollectionHasSubscribersError,
+          )
+          expect(run).toBe(1)
+          expect(live.status).toBe(`ready`)
+        } else {
+          if (cleanup === `dependents`)
+            await source.cleanup({ dependents: true })
+          else await source._lifecycle.cleanup()
+          if (cleanup === `internal`) {
+            expect(live.status).toBe(`crashed`)
+            source.startSyncImmediate()
+            await flushPromises()
+            expect(live.status).toBe(`crashed`)
+            expect([...live.keys()].sort()).toEqual([`b`, `d`])
+            await live.cleanup()
+          }
+          await live.preload()
+          expect(live.status).toBe(`ready`)
+          expect([...live.keys()]).toEqual([`b`])
+        }
+      } finally {
+        sub.unsubscribe()
+        await live.cleanup()
+        await source.cleanup()
+        reports.mockRestore()
+      }
+    })
+  }
+}