fix: preserve accepted snapshots and fence stale replay reads

.changeset/fix-retention-and-stale-replay.md

@@ -0,0 +1,7 @@
+---
+'@tanstack/db': patch
+'@tanstack/db-ivm': patch
+'@tanstack/offline-transactions': patch
+---
+
+Preserve only captured accepted local inserts across a truncate. Preserve sparse-array length and RegExp state through ordered-query hashing, including hosts without a global File constructor. Prevent delayed replay reads from rerunning any transaction removed while the read was in flight, without rescanning the outbox.

docs/contributing/oracle-coverage.md

@@ -44,6 +44,22 @@ comment and the current API/architecture contract before extending its model.
 
 ## Acceptance map
 
+The post-merge review added three missing domains to existing owners:
+
+- [Top-K batch contracts](../../packages/db-ivm/tests/operators/topk-batch-contract.test.ts)
+  cross sparse-array length/holes and RegExp source/flags/position with equal
+  controls, replacement order, hash consolidation, and actual retained graph
+  output. Ordinary replacements also run without the global `File` constructor.
+- [Leadership replay](../../packages/offline-transactions/tests/leadership-replay.property.test.ts)
+  holds real storage-read delivery across successful and permanently rejected
+  durable removals, with bounded scans, concurrent loads, and unfinished peers.
+  This is distinct from exactly-once execution across independent owners.
+- [Accepted-snapshot retention](../../packages/db/tests/collection-state-retention-oracle.property.test.ts)
+  varies truncate before/during/after an optimistic delete, rejection versus
+  rollback, post-capture direct insertion, and later ordinary sync/key reuse. A
+  hidden accepted insert returns after rollback; an uncaptured insert retires,
+  and neither snapshot is rebased onto synced fields.
+
 | Issue obligation | Implemented evidence | Limit |
 | --- | --- | --- |
 | Metamorphic laws | Includes cross-formulation/partition, D2 independent-key commutation, DBSP incremental/full recomputation, pagination provider/UI boundaries, optimistic snapshot stability | Equivalence premises are explicit; not arbitrary query rewrites. |

packages/db-ivm/src/hashing/hash.ts

@@ -13,6 +13,7 @@ const UNDEFINED = randomHash()
 const KEY = randomHash()
 const FUNCTIONS = randomHash()
 const DATE_MARKER = randomHash()
+const REGEXP_MARKER = randomHash()
 const STRUCTURAL_MARKERS = {
   object: randomHash(),
   array: randomHash(),
@@ -89,12 +90,19 @@ function hashObject(input: object, context: HashContext): number {
       valueHash = hashUint8Array(input)
     } else if (isTemporal(input)) {
       valueHash = hashTemporal(input)
+    } else if (input instanceof RegExp) {
+      valueHash = hashPlainObject(input, REGEXP_MARKER, context, [
+        input.source,
+        input.flags,
+        input.lastIndex,
+      ])
     } else {
       const [kind, plainObjectInput] = structuralShape(input)
       valueHash = hashPlainObject(
         plainObjectInput,
         STRUCTURAL_MARKERS[kind],
         context,
+        kind === `array` ? [input instanceof Array ? input.length : 0] : [],
       )
     }
   } finally {
@@ -136,11 +144,13 @@ function hashPlainObject(
   input: object,
   marker: number,
   context: HashContext,
+  headerValues: ReadonlyArray<unknown> = [],
 ): number {
   const hasher = new MurmurHashStream()
 
   // Mark the type of the input
   hasher.update(marker)
+  for (const value of headerValues) updateHasher(hasher, value, context)
   const keys = Object.keys(input)
   keys.sort(keySort)
   for (const key of keys) {
@@ -239,7 +249,7 @@ function getCachedHash(input: object, context?: HashContext): number {
 
 function isReferenceHashedObject(input: object): boolean {
   return (
-    input instanceof File ||
+    (typeof File !== `undefined` && input instanceof File) ||
     (isBinaryValue(input) &&
       input.byteLength > UINT8ARRAY_CONTENT_HASH_THRESHOLD)
   )
@@ -302,6 +312,17 @@ export function equalHashValues(left: unknown, right: unknown): boolean {
         a[Symbol.toStringTag] === b[Symbol.toStringTag] &&
         a.toString() === b.toString()
       )
+    if (a instanceof RegExp || b instanceof RegExp) {
+      if (
+        !(a instanceof RegExp && b instanceof RegExp) ||
+        a.source !== b.source ||
+        a.flags !== b.flags ||
+        a.lastIndex !== b.lastIndex
+      )
+        return false
+    }
+    if (Array.isArray(a) && Array.isArray(b) && a.length !== b.length)
+      return false
 
     // Revisited pairs close cycles and avoid expanding shared subtrees.
     const peers = compared.get(a)

packages/db-ivm/tests/operators/topk-batch-contract.test.ts

@@ -84,6 +84,36 @@ const valueRelations: Array<{
     pair: (n) => [new Set([n, n + 1]), new Set([n + 1, n]), false],
   },
   { name: `array versus object`, pair: (n) => [[n], { 0: n }, false] },
+  { name: `sparse array length`, pair: (n) => [[], Array(n + 1), false] },
+  { name: `equal sparse arrays`, pair: (n) => [Array(n), Array(n), true] },
+  { name: `hole versus undefined`, pair: () => [Array(1), [undefined], false] },
+  {
+    name: `regexp source`,
+    pair: (n) => [
+      new RegExp(`old${n}`, `g`),
+      new RegExp(`new${n}`, `g`),
+      false,
+    ],
+  },
+  { name: `regexp flags`, pair: () => [/same/g, /same/i, false] },
+  {
+    name: `regexp position`,
+    pair: (n) => {
+      const left = /same/g
+      const right = /same/g
+      right.lastIndex = n + 1
+      return [left, right, false]
+    },
+  },
+  {
+    name: `equal regexp state`,
+    pair: (n) => {
+      const left = /same/g
+      const right = /same/g
+      left.lastIndex = right.lastIndex = n
+      return [left, right, true]
+    },
+  },
   {
     name: `shared symbol property`,
     pair: (n) => {
@@ -145,13 +175,18 @@ it.each(
   ({ pair, largeGroup, collision }) => {
     if (collision) vi.spyOn(hashing, `hash`).mockReturnValue(7)
     fc.assert(
-      fc.property(fc.integer({ min: 0, max: 254 }), (n) => {
-        const [left, right, equivalent] = pair(n)
-        const before: [number, { value: unknown }] = [1, { value: left }]
-        const after: [number, { value: unknown }] = [1, { value: right }]
-        const transient: [number, { value: unknown }] = [1, { value: Symbol() }]
-        const actual = [
-          ...topKBatch([
+      fc.property(
+        fc.integer({ min: 0, max: 254 }),
+        fc.boolean(),
+        (n, retractFirst) => {
+          const [left, right, equivalent] = pair(n)
+          const before: [number, { value: unknown }] = [1, { value: left }]
+          const after: [number, { value: unknown }] = [1, { value: right }]
+          const transient: [number, { value: unknown }] = [
+            1,
+            { value: Symbol() },
+          ]
+          const messages = [
             new MultiSet([[after, 1]]),
             ...(largeGroup
               ? [
@@ -162,23 +197,119 @@ it.each(
                 ]
               : []),
             new MultiSet([[before, -1]]),
-          ]),
-        ]
-        if (equivalent) expect(actual).toEqual([])
-        else {
-          expect(actual).toHaveLength(2)
-          expect(actual[0]![0]).toBe(before)
-          expect(actual[0]![1]).toBe(-1)
-          expect(actual[1]![0]).toBe(after)
-          expect(actual[1]![1]).toBe(1)
-        }
-      }),
+          ]
+          if (retractFirst) messages.reverse()
+          const actual = [...topKBatch(messages)]
+          if (equivalent) expect(actual).toEqual([])
+          else {
+            expect(actual).toHaveLength(2)
+            expect(actual[0]![0]).toBe(before)
+            expect(actual[0]![1]).toBe(-1)
+            expect(actual[1]![0]).toBe(after)
+            expect(actual[1]![1]).toBe(1)
+          }
+          // A correct helper alone does not prove the ordered graph retains the
+          // replacement. Keep raw signed output; never erase a missed retraction.
+          const graph = new D2()
+          const input = graph.newInput<typeof before>()
+          const retained = new Map<(typeof before)[1], number>()
+          input.pipe(
+            topKWithFractionalIndex(() => 0, { limit: 1 }),
+            output((message) => {
+              for (const [[, [row]], weight] of message.getInner())
+                retained.set(row, (retained.get(row) ?? 0) + weight)
+            }),
+          )
+          graph.finalize()
+          input.sendData(new MultiSet([[before, 1]]))
+          graph.run()
+          for (const message of messages) input.sendData(message)
+          graph.run()
+          const live = [...retained].filter(([, weight]) => weight !== 0)
+          expect(live).toHaveLength(1)
+          expect(live[0]![0]).toBe(equivalent ? before[1] : after[1])
+          expect(live[0]![1]).toBe(1)
+        },
+      ),
       { seed: 409033, numRuns: 25 },
     )
     if (collision) expect(hashing.hash).not.toHaveBeenCalled()
   },
 )
 
+it.each([
+  { name: `RegExp source`, before: /old/g, after: /new/g },
+  { name: `RegExp flags`, before: /same/g, after: /same/i },
+  {
+    name: `RegExp position`,
+    before: Object.assign(/same/g, { lastIndex: 0 }),
+    after: Object.assign(/same/g, { lastIndex: 1 }),
+  },
+  { name: `sparse-array length`, before: [], after: Array(2) },
+])(
+  `keeps a $name replacement through hash consolidation`,
+  ({ before, after }) => {
+    // Relation: a retraction and a distinct addition must remain observable to
+    // an ordered operator even when an earlier stage consolidates the batch.
+    const previous = { id: 1, value: before }
+    const next = { id: 1, value: after }
+    expect(
+      new MultiSet([
+        [previous, -1],
+        [next, 1],
+      ])
+        .consolidate()
+        .getInner(),
+    ).toEqual([
+      [previous, -1],
+      [next, 1],
+    ])
+  },
+)
+
+it.each([true, false])(
+  `replaces ordinary rows with File available=%s`,
+  (available) => {
+    const descriptor = Object.getOwnPropertyDescriptor(globalThis, `File`)
+    if (!available) Reflect.deleteProperty(globalThis, `File`)
+    try {
+      fc.assert(
+        fc.property(fc.integer(), fc.boolean(), (rank, retractFirst) => {
+          const graph = new D2()
+          const input = graph.newInput<[number, { rank: number }]>()
+          const rows = new Map<object, number>()
+          input.pipe(
+            topKWithFractionalIndex((a, b) => a.rank - b.rank, { limit: 1 }),
+            output((message) => {
+              for (const [[, [row]], weight] of message.getInner())
+                rows.set(row, (rows.get(row) ?? 0) + weight)
+            }),
+          )
+          graph.finalize()
+          const before = { rank }
+          const after = { rank: rank + 1 }
+          input.sendData(new MultiSet([[[1, before], 1]]))
+          graph.run()
+          const changes: Array<[[number, typeof before], number]> = [
+            [[1, after], 1],
+            [[1, before], -1],
+          ]
+          if (retractFirst) changes.reverse()
+          input.sendData(new MultiSet(changes))
+          graph.run()
+          const live = [...rows].filter(([, weight]) => weight !== 0)
+          expect(live).toHaveLength(1)
+          expect(live[0]![0]).toBe(after)
+          expect(live[0]![1]).toBe(1)
+        }),
+        { seed: 409039, numRuns: 25 },
+      )
+    } finally {
+      if (descriptor) Object.defineProperty(globalThis, `File`, descriptor)
+    }
+  },
+)
+
 it(`keeps distinct row keys when structural hashes collide`, () => {
   // A hash is an accelerator, never proof that two keyed rows are equal.
   vi.spyOn(hashing, `hash`).mockReturnValue(7)

packages/db-ivm/tests/utils.test.ts

@@ -599,15 +599,24 @@ describe(`hash`, () => {
       const regex1 = /test/g
       const regex2 = /test/g
       const regex3 = /different/i
+      const regex4 = /test/g
+      regex4.lastIndex = 1
 
       const hash1 = hash(regex1)
       const hash2 = hash(regex2)
       const hash3 = hash(regex3)
+      const hash4 = hash(regex4)
 
       expect(typeof hash1).toBe(hashType)
       expect(hash1).toBe(hash2) // Same regex should have same hash
-      // Note: RegExp don't have enumerable properties so they all produce the same hash
-      expect(hash1).toBe(hash3) // All RegExp objects have the same hash
+      expect(hash1).not.toBe(hash3)
+      expect(hash1).not.toBe(hash4)
+    })
+
+    it(`should include sparse array length in its hash`, () => {
+      expect(hash([])).not.toBe(hash(Array(1)))
+      expect(hash(Array(1))).not.toBe(hash(Array(2)))
+      expect(hash(Array(2))).toBe(hash(Array(2)))
     })
 
     it(`should hash nested objects`, () => {

packages/db/src/collection/state.ts

@@ -1226,17 +1226,22 @@ export class CollectionStateManager<
       // sync commit has been applied, stop retaining completed optimistic keys
       // that were not confirmed by this commit so the temporary row is removed.
       for (const key of this.pendingOptimisticDirectUpserts) {
-        // Truncate republishes this captured snapshot. Keep its existing
-        // retention marker so the next sync can also publish its removal.
+        // An active delete can hide an accepted snapshot. Retain it through
+        // truncate so rollback can restore it. A direct insert completed after
+        // snapshot capture has no support in the replacement and must retire.
         if (
           hasTruncateSync &&
-          truncateOptimisticSnapshot?.upserts.has(key) &&
+          this.pendingOptimisticUpserts.has(key) &&
+          (truncateOptimisticSnapshot?.upserts.has(key) === true ||
+            truncateOptimisticSnapshot?.deletes.has(key) === true) &&
           !changedKeys.has(key)
         )
           continue
         if (!changedKeys.has(key)) {
           changedKeys.add(key)
-          if (!currentVisibleState.has(key)) {
+          // Truncate already emitted the prior visible rows as its clear
+          // prefix. Reconstructing one here would publish a duplicate delete.
+          if (!hasTruncateSync && !currentVisibleState.has(key)) {
             const previousValue = previousOptimisticUpserts.get(key)
             if (previousValue !== undefined) {
               currentVisibleState.set(key, previousValue)