fix: preserve offline runtime correctness across replay and restart

.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, use React Native's subscribed connectivity snapshot as the initial authority, 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, 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.

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

@@ -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,24 @@ 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
+  }
+
+  private 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> {
@@ -126,3 +141,11 @@ export class KeyScheduler {
     )
   }
 }
+
+/** @internal Reconcile one replay snapshot without canceling issued work. */
+export function reconcilePendingTransactions(
+  scheduler: KeyScheduler,
+  transactionIds: Iterable<string>,
+): Array<string> {
+  return scheduler[`removePendingTransactions`](transactionIds)
+}

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

@@ -2,6 +2,7 @@ import { createTransaction } from '@tanstack/db'
 import { DefaultRetryPolicy } from '../retry/RetryPolicy'
 import { NonRetriableError } from '../types'
 import { withNestedSpan } from '../telemetry/tracer'
+import { reconcilePendingTransactions } from './KeyScheduler'
 import type { KeyScheduler } from './KeyScheduler'
 import type { OutboxManager } from '../outbox/OutboxManager'
 import type {
@@ -56,6 +57,7 @@ export class TransactionExecutor {
     } finally {
       this.isExecuting = false
       this.executionPromise = null
+      this.scheduleNextRetry()
     }
   }
 
@@ -73,9 +75,6 @@ export class TransactionExecutor {
 
       await this.executeTransaction(transaction)
     }
-
-    // Schedule next retry after execution completes
-    this.scheduleNextRetry()
   }
 
   private async executeTransaction(
@@ -180,8 +179,11 @@ export class TransactionExecutor {
         span.setAttribute(`shouldRetry`, shouldRetry)
 
         if (!shouldRetry) {
-          this.scheduler.markCompleted(transaction)
-          await this.outbox.remove(transaction.id)
+          try {
+            await this.outbox.remove(transaction.id)
+          } finally {
+            this.scheduler.markCompleted(transaction)
+          }
           console.warn(
             `Transaction ${transaction.id} failed permanently:`,
             error,
@@ -211,7 +213,6 @@ export class TransactionExecutor {
         span.setAttribute(`retryDelay`, delay)
         span.setAttribute(`nextRetryCount`, updatedTransaction.retryCount)
 
-        this.scheduler.markFailed(transaction)
         this.scheduler.updateTransaction(updatedTransaction)
 
         try {
@@ -221,10 +222,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()
       },
     )
   }
@@ -247,6 +247,14 @@ export class TransactionExecutor {
         this.scheduler.schedule(transaction),
       )
 
+      removedIds = transactions
+        .filter(
+          (tx) =>
+            !filteredTransactions.some((filtered) => filtered.id === tx.id),
+        )
+        .map(({ id }) => id)
+      removedIds = reconcilePendingTransactions(this.scheduler, removedIds)
+
       // Restore optimistic state for loaded transactions
       // This ensures the UI shows the optimistic data while transactions are pending
       this.restoreOptimisticState(newlyLoaded)
@@ -256,17 +264,17 @@ 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((id) =>
+          this.outbox
+            .remove(id)
+            .then(() => this.offlineExecutor.rejectTransaction(id, error)),
+        ),
+      )
     }
   }
 

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,