fix: preserve offline runtime correctness across replay and restart (#1837)

* fix(offline-transactions): preserve runtime correctness

Co-authored-by: Colton Demetriou <cdemetriou@valinor.co>

* fix(offline-transactions): reject unrestorable scalars

* fix(offline-transactions): address runtime review gaps

* fix(offline-transactions): preserve metadata JSON semantics

* test(offline-transactions): validate Temporal tags once

* fix(offline-transactions): settle successful replay removals

* docs: clarify offline retry liveness scope

* fix(offline-transactions): isolate cleanup failures

* fix(offline-transactions): await initial RN connectivity

---------

Co-authored-by: Colton Demetriou <cdemetriou@valinor.co>

Author: KyleAMathews

.changeset/fix-offline-runtime-correctness.md

@@ -0,0 +1,7 @@
+---
+'@tanstack/offline-transactions': patch
+---
+
+Fix offline replay filtering so it preserves concurrently admitted and issued transactions, wait for React Native's subscribed connectivity snapshot before reporting online, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. Filtered replay work now settles and rolls back after each successful durable removal even when a sibling removal fails, while a failed filtered-work cleanup no longer aborts unrelated startup replay. Successful and permanently failed provider work settles its own caller even when durable cleanup also fails, retry scheduling remains live when a retry update or permanent-failure removal fails, and metadata keeps standard `toJSON(key)` replacement semantics. Recognized native scalars now fail before storage when the matching global constructor is unavailable.
+
+Offline storage compatibility: new records use `valueEncoding: 3`. Older clients cannot read these records, so do not run old and new clients against the same pending outbox or downgrade while new records remain pending.

docs/contributing/oracle-coverage.md

@@ -82,6 +82,7 @@ does not award oracle credit for a filename alone.
 | [#1833](https://github.com/TanStack/db/pull/1833) | Explicit oracle | `packages/db/tests/query/pagination-oracle.property.test.ts` owns inherited collection collation, actual `item2`/`item10` order, exact request options, hostile lexical/numeric controls, and both scan and auto-index paths. It runs in `@tanstack/db`'s `test:oracles` campaign. |
 | [#1834](https://github.com/TanStack/db/pull/1834) | Explicit oracle | `packages/db/tests/query/cold-join-reconciliation-oracle.test.ts` owns join/predicate equality equivalence across the established value domains, binary/string and nullish controls, replacement histories, raw lazy demand, and scan/auto-index paths. It runs in `@tanstack/db`'s `test:oracles` campaign. |
 | [#1835](https://github.com/TanStack/db/pull/1835) | Explicit oracle | The existing `packages/db/tests/collection-state-retention-oracle.property.test.ts` and `packages/db/tests/optimistic-transaction-oracle.property.test.ts` owners cover separate collection-state and transaction-history laws. Both were already registered in `@tanstack/db`'s `test:oracles` campaign; focused storage/local-only tests remain collateral. |
+| [#1837](https://github.com/TanStack/db/pull/1837) | Explicit oracle | The offline scheduler, leadership replay, and serializer owners cover selective replay retirement, durable per-ID settlement, stale-read fencing, lifecycle recovery, native scalar encoding, and prior wire compatibility. They run in the package test campaign; generated owners expose `OFFLINE_ORACLE_{SEED,PATH,RUNS}` or the scheduler's `TANSTACK_DB_OFFLINE_ORACLE_*` replay interface. |
 | [#1842](https://github.com/TanStack/db/pull/1842) | No shipped-law case | The PR changed only focused observer tests and introduced no production behavior. `packages/db/tests/live-query-observer.test.ts` remains the correct evidence; no synthetic oracle or campaign claim is added. |
 
 [PR #1816](https://github.com/TanStack/db/pull/1816) preserves existing witnesses,

packages/offline-transactions/src/OfflineExecutor.ts

@@ -101,7 +101,8 @@ export class OfflineExecutor {
       this.initResolve = resolve
       this.initReject = reject
     })
-
+    // Handle constructor-started rejection; waitForInit still observes it.
+    void this.initPromise.catch(() => {})
     this.initialize()
   }
 

packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts

@@ -14,7 +14,7 @@ export class ReactNativeOnlineDetector implements OnlineDetector {
   private netInfoUnsubscribe: (() => void) | null = null
   private appStateSubscription: NativeEventSubscription | null = null
   private isListening = false
-  private wasConnected = true
+  private wasConnected = false
 
   constructor() {
     this.startListening()
@@ -27,16 +27,6 @@ export class ReactNativeOnlineDetector implements OnlineDetector {
 
     this.isListening = true
 
-    if (typeof NetInfo.fetch === `function`) {
-      void NetInfo.fetch()
-        .then((state) => {
-          this.wasConnected = this.toConnectivityState(state)
-        })
-        .catch(() => {
-          // Ignore initial fetch failures and rely on subscription updates.
-        })
-    }
-
     // Subscribe to network state changes
     this.netInfoUnsubscribe = NetInfo.addEventListener((state) => {
       const isConnected = this.toConnectivityState(state)

packages/offline-transactions/src/executor/KeyScheduler.ts

@@ -3,7 +3,7 @@ import type { OfflineTransaction } from '../types'
 
 export class KeyScheduler {
   private pendingTransactions: Array<OfflineTransaction> = []
-  private isRunning = false
+  private activeTransactionId: string | undefined
 
   schedule(transaction: OfflineTransaction): boolean {
     return withSyncSpan(
@@ -35,7 +35,10 @@ export class KeyScheduler {
       `scheduler.getNext`,
       { pendingCount: this.pendingTransactions.length },
       (span) => {
-        if (this.isRunning || this.pendingTransactions.length === 0) {
+        if (
+          this.activeTransactionId !== undefined ||
+          this.pendingTransactions.length === 0
+        ) {
           span.setAttribute(`result`, `empty`)
           return undefined
         }
@@ -59,17 +62,17 @@ export class KeyScheduler {
     return Date.now() >= transaction.nextAttemptAt
   }
 
-  markStarted(_transaction: OfflineTransaction): void {
-    this.isRunning = true
+  markStarted(transaction: OfflineTransaction): void {
+    this.activeTransactionId = transaction.id
   }
 
   markCompleted(transaction: OfflineTransaction): void {
     this.removeTransaction(transaction)
-    this.isRunning = false
+    this.activeTransactionId = undefined
   }
 
   markFailed(_transaction: OfflineTransaction): void {
-    this.isRunning = false
+    this.activeTransactionId = undefined
   }
 
   private removeTransaction(transaction: OfflineTransaction): void {
@@ -99,12 +102,23 @@ export class KeyScheduler {
   }
 
   getRunningCount(): number {
-    return this.isRunning ? 1 : 0
+    return this.activeTransactionId === undefined ? 0 : 1
   }
 
   clear(): void {
     this.pendingTransactions = []
-    this.isRunning = false
+    this.activeTransactionId = undefined
+  }
+
+  /** @internal Reconcile one replay snapshot without canceling issued work. */
+  removePendingTransactions(transactionIds: Iterable<string>): Array<string> {
+    const ids = new Set(transactionIds)
+    if (this.activeTransactionId !== undefined)
+      ids.delete(this.activeTransactionId)
+    this.pendingTransactions = this.pendingTransactions.filter(
+      ({ id }) => !ids.has(id),
+    )
+    return [...ids]
   }
 
   getAllPendingTransactions(): Array<OfflineTransaction> {

packages/offline-transactions/src/executor/TransactionExecutor.ts

@@ -56,6 +56,7 @@ export class TransactionExecutor {
     } finally {
       this.isExecuting = false
       this.executionPromise = null
+      this.scheduleNextRetry()
     }
   }
 
@@ -73,9 +74,6 @@ export class TransactionExecutor {
 
       await this.executeTransaction(transaction)
     }
-
-    // Schedule next retry after execution completes
-    this.scheduleNextRetry()
   }
 
   private async executeTransaction(
@@ -97,18 +95,9 @@ export class TransactionExecutor {
             span.setAttribute(`retry.attempt`, transaction.retryCount)
           }
 
+          let result: void
           try {
-            const result = await this.runMutationFn(transaction)
-
-            try {
-              // Replay can still see this ID until durable deletion settles.
-              await this.outbox.remove(transaction.id)
-            } finally {
-              this.scheduler.markCompleted(transaction)
-            }
-
-            span.setAttribute(`result`, `success`)
-            this.offlineExecutor.resolveTransaction(transaction.id, result)
+            result = await this.runMutationFn(transaction)
           } catch (error) {
             const err =
               error instanceof Error ? error : new Error(String(error))
@@ -119,6 +108,20 @@ export class TransactionExecutor {
             ;(err as any)[HANDLED_EXECUTION_ERROR] = true
             throw err
           }
+
+          let removalError: unknown
+          try {
+            // Replay can still see this ID until durable deletion settles.
+            await this.outbox.remove(transaction.id)
+          } catch (error) {
+            removalError = error
+          } finally {
+            this.scheduler.markCompleted(transaction)
+          }
+
+          span.setAttribute(`result`, `success`)
+          this.offlineExecutor.resolveTransaction(transaction.id, result)
+          if (removalError !== undefined) throw removalError
         },
       )
     } catch (error) {
@@ -180,8 +183,14 @@ export class TransactionExecutor {
         span.setAttribute(`shouldRetry`, shouldRetry)
 
         if (!shouldRetry) {
-          this.scheduler.markCompleted(transaction)
-          await this.outbox.remove(transaction.id)
+          let removalError: unknown
+          try {
+            await this.outbox.remove(transaction.id)
+          } catch (cleanupError) {
+            removalError = cleanupError
+          } finally {
+            this.scheduler.markCompleted(transaction)
+          }
           console.warn(
             `Transaction ${transaction.id} failed permanently:`,
             error,
@@ -190,6 +199,7 @@ export class TransactionExecutor {
           span.setAttribute(`result`, `permanent_failure`)
           // Signal permanent failure to the waiting transaction
           this.offlineExecutor.rejectTransaction(transaction.id, error)
+          if (removalError !== undefined) throw removalError
           return
         }
 
@@ -211,7 +221,6 @@ export class TransactionExecutor {
         span.setAttribute(`retryDelay`, delay)
         span.setAttribute(`nextRetryCount`, updatedTransaction.retryCount)
 
-        this.scheduler.markFailed(transaction)
         this.scheduler.updateTransaction(updatedTransaction)
 
         try {
@@ -221,10 +230,9 @@ export class TransactionExecutor {
           span.recordException(persistError as Error)
           span.setAttribute(`result`, `persist_failed`)
           throw persistError
+        } finally {
+          this.scheduler.markFailed(transaction)
         }
-
-        // Schedule retry timer
-        this.scheduleNextRetry()
       },
     )
   }
@@ -240,13 +248,21 @@ export class TransactionExecutor {
         filteredTransactions = this.config.beforeRetry(transactions)
       }
 
-      // The outbox read or retry hook may outlive this owner's right to replay.
+      // The retry hook is user code and may synchronously revoke replay rights.
       if (!this.offlineExecutor.isOfflineEnabled) return
 
       const newlyLoaded = filteredTransactions.filter((transaction) =>
         this.scheduler.schedule(transaction),
       )
 
+      removedIds = transactions
+        .filter(
+          (tx) =>
+            !filteredTransactions.some((filtered) => filtered.id === tx.id),
+        )
+        .map(({ id }) => id)
+      removedIds = this.scheduler.removePendingTransactions(removedIds)
+
       // Restore optimistic state for loaded transactions
       // This ensures the UI shows the optimistic data while transactions are pending
       this.restoreOptimisticState(newlyLoaded)
@@ -256,17 +272,24 @@ export class TransactionExecutor {
 
       // Schedule retry timer for loaded transactions
       this.scheduleNextRetry()
-
-      removedIds = transactions
-        .filter(
-          (tx) =>
-            !filteredTransactions.some((filtered) => filtered.id === tx.id),
-        )
-        .map(({ id }) => id)
     })
 
     if (removedIds.length > 0) {
-      await this.outbox.removeMany(removedIds)
+      const error = new NonRetriableError(`Transaction excluded by beforeRetry`)
+      await Promise.all(
+        removedIds.map(async (id) => {
+          try {
+            await this.outbox.remove(id)
+            this.offlineExecutor.rejectTransaction(id, error)
+          } catch (cleanupError) {
+            console.warn(
+              `Failed to remove transaction excluded by beforeRetry:`,
+              id,
+              cleanupError,
+            )
+          }
+        }),
+      )
     }
   }
 

packages/offline-transactions/src/outbox/OutboxManager.ts

@@ -1,5 +1,8 @@
 import { withSpan } from '../telemetry/tracer'
-import { TransactionSerializer } from './TransactionSerializer'
+import {
+  MissingTemporalConstructorError,
+  TransactionSerializer,
+} from './TransactionSerializer'
 import type { OfflineTransaction, StorageAdapter } from '../types'
 import type { Collection } from '@tanstack/db'
 
@@ -74,6 +77,10 @@ export class OutboxManager {
         span.setAttribute(`result`, `found`)
         return transaction
       } catch (error) {
+        if (error instanceof MissingTemporalConstructorError) {
+          error.message = `transaction ${id}: ${error.message}`
+          throw error
+        }
         console.warn(`Failed to deserialize transaction ${id}:`, error)
         span.setAttribute(`result`, `deserialize_error`)
         return null
@@ -108,6 +115,10 @@ export class OutboxManager {
               const transaction = this.serializer.deserialize(data)
               transactions.push(transaction)
             } catch (error) {
+              if (error instanceof MissingTemporalConstructorError) {
+                error.message = `transaction ${key.slice(this.keyPrefix.length)}: ${error.message}`
+                throw error
+              }
               console.warn(
                 `Failed to deserialize transaction from key ${key}:`,
                 error,