From e3d9e30cfc4b7ae6dec6dff441b23d05b8dcaa04 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 14:22:22 -0600 Subject: [PATCH 1/7] fix(offline-transactions): preserve runtime correctness Co-authored-by: Colton Demetriou --- .changeset/fix-offline-runtime-correctness.md | 5 + .../connectivity/ReactNativeOnlineDetector.ts | 6 +- .../src/executor/KeyScheduler.ts | 39 +- .../src/executor/TransactionExecutor.ts | 26 +- .../src/outbox/OutboxManager.ts | 7 +- .../src/outbox/TransactionSerializer.ts | 94 +++- packages/offline-transactions/src/types.ts | 2 +- .../tests/KeyScheduler.property.test.ts | 95 +++- .../tests/OfflineExecutor.test.ts | 65 +++ .../tests/ReactNativeOnlineDetector.test.ts | 160 +++++-- .../tests/leadership-replay.property.test.ts | 292 ++++++++++++ .../transaction-serializer.property.test.ts | 415 +++++++++++++++++- 12 files changed, 1144 insertions(+), 62 deletions(-) create mode 100644 .changeset/fix-offline-runtime-correctness.md diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md new file mode 100644 index 0000000000..07fabca8b5 --- /dev/null +++ b/.changeset/fix-offline-runtime-correctness.md @@ -0,0 +1,5 @@ +--- +'@tanstack/offline-transactions': patch +--- + +Fix offline replay filtering so it preserves concurrently admitted and issued transactions, keep React Native live connectivity events authoritative over stale startup snapshots, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. diff --git a/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts b/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts index 5c38c06911..745d796ac7 100644 --- a/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts +++ b/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts @@ -26,11 +26,14 @@ export class ReactNativeOnlineDetector implements OnlineDetector { } this.isListening = true + let receivedLiveState = false if (typeof NetInfo.fetch === `function`) { void NetInfo.fetch() .then((state) => { - this.wasConnected = this.toConnectivityState(state) + if (this.isListening && !receivedLiveState) { + this.wasConnected = this.toConnectivityState(state) + } }) .catch(() => { // Ignore initial fetch failures and rely on subscription updates. @@ -39,6 +42,7 @@ export class ReactNativeOnlineDetector implements OnlineDetector { // Subscribe to network state changes this.netInfoUnsubscribe = NetInfo.addEventListener((state) => { + receivedLiveState = true const isConnected = this.toConnectivityState(state) // Only notify when transitioning to online diff --git a/packages/offline-transactions/src/executor/KeyScheduler.ts b/packages/offline-transactions/src/executor/KeyScheduler.ts index 20fdd1dbf3..4eebe8dcd5 100644 --- a/packages/offline-transactions/src/executor/KeyScheduler.ts +++ b/packages/offline-transactions/src/executor/KeyScheduler.ts @@ -3,7 +3,7 @@ import type { OfflineTransaction } from '../types' export class KeyScheduler { private pendingTransactions: Array = [] - 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, + ): Array { + 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 { @@ -126,3 +141,11 @@ export class KeyScheduler { ) } } + +/** @internal Reconcile one replay snapshot without canceling issued work. */ +export function reconcilePendingTransactions( + scheduler: KeyScheduler, + transactionIds: Iterable, +): Array { + return scheduler[`removePendingTransactions`](transactionIds) +} diff --git a/packages/offline-transactions/src/executor/TransactionExecutor.ts b/packages/offline-transactions/src/executor/TransactionExecutor.ts index b6bfbd964a..97fb1733f1 100644 --- a/packages/offline-transactions/src/executor/TransactionExecutor.ts +++ b/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 { @@ -180,8 +181,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 +215,6 @@ export class TransactionExecutor { span.setAttribute(`retryDelay`, delay) span.setAttribute(`nextRetryCount`, updatedTransaction.retryCount) - this.scheduler.markFailed(transaction) this.scheduler.updateTransaction(updatedTransaction) try { @@ -221,6 +224,8 @@ export class TransactionExecutor { span.recordException(persistError as Error) span.setAttribute(`result`, `persist_failed`) throw persistError + } finally { + this.scheduler.markFailed(transaction) } // Schedule retry timer @@ -247,6 +252,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,13 +269,6 @@ 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) { diff --git a/packages/offline-transactions/src/outbox/OutboxManager.ts b/packages/offline-transactions/src/outbox/OutboxManager.ts index 82cccd3780..588a8244a7 100644 --- a/packages/offline-transactions/src/outbox/OutboxManager.ts +++ b/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,7 @@ export class OutboxManager { span.setAttribute(`result`, `found`) return transaction } catch (error) { + if (error instanceof MissingTemporalConstructorError) throw error console.warn(`Failed to deserialize transaction ${id}:`, error) span.setAttribute(`result`, `deserialize_error`) return null @@ -108,6 +112,7 @@ export class OutboxManager { const transaction = this.serializer.deserialize(data) transactions.push(transaction) } catch (error) { + if (error instanceof MissingTemporalConstructorError) throw error console.warn( `Failed to deserialize transaction from key ${key}:`, error, diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index 0eeb9f65df..02af63dd01 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -6,6 +6,33 @@ import type { } from '../types' import type { Collection, PendingMutation } from '@tanstack/db' +const temporalConstructorNames = [ + `Duration`, + `Instant`, + `PlainDate`, + `PlainDateTime`, + `PlainMonthDay`, + `PlainTime`, + `PlainYearMonth`, + `ZonedDateTime`, +] as const + +type TemporalConstructorName = (typeof temporalConstructorNames)[number] + +function getTemporalConstructorName( + type: unknown, +): TemporalConstructorName | undefined { + if (typeof type !== `string` || !type.startsWith(`Temporal.`)) return + const constructorName = type.slice( + `Temporal.`.length, + ) as TemporalConstructorName + return temporalConstructorNames.includes(constructorName) + ? constructorName + : undefined +} + +export class MissingTemporalConstructorError extends Error {} + function setDataProperty( object: Record, key: string, @@ -39,8 +66,9 @@ export class TransactionSerializer { serialize(transaction: OfflineTransaction): string { const serialized: SerializedOfflineTransaction = { ...transaction, - valueEncoding: 2, + valueEncoding: 3, createdAt: transaction.createdAt.toISOString(), + metadata: this.serializeValue(transaction.metadata), mutations: transaction.mutations.map((mutation) => this.serializeMutation(mutation), ), @@ -57,7 +85,11 @@ export class TransactionSerializer { }: Omit & { valueEncoding?: unknown } = JSON.parse(data) - if (valueEncoding !== undefined && valueEncoding !== 2) { + if ( + valueEncoding !== undefined && + valueEncoding !== 2 && + valueEncoding !== 3 + ) { throw new Error( `Unsupported transaction value encoding: ${valueEncoding}`, ) @@ -73,8 +105,12 @@ export class TransactionSerializer { return { ...parsed, createdAt, + metadata: + valueEncoding === 3 + ? this.deserializeValue(parsed.metadata, valueEncoding) + : parsed.metadata, mutations: parsed.mutations.map((mutationData) => - this.deserializeMutation(mutationData, valueEncoding === 2), + this.deserializeMutation(mutationData, valueEncoding), ), } } @@ -99,14 +135,14 @@ export class TransactionSerializer { private deserializeMutation( data: SerializedMutation, - escapedObjects: boolean, + valueEncoding: 2 | 3 | undefined, ): PendingMutation { const collection = this.collections[data.collectionId] if (!collection) { throw new Error(`Collection with id ${data.collectionId} not found`) } - const modified = this.deserializeValue(data.modified, escapedObjects) + const modified = this.deserializeValue(data.modified, valueEncoding) // Extract the key from the modified data using the collection's getKey function // This is needed for optimistic state restoration to work correctly @@ -118,8 +154,8 @@ export class TransactionSerializer { globalKey: data.globalKey, type: data.type as any, modified, - original: this.deserializeValue(data.original, escapedObjects), - changes: this.deserializeValue(data.changes, escapedObjects) ?? {}, + original: this.deserializeValue(data.original, valueEncoding), + changes: this.deserializeValue(data.changes, valueEncoding) ?? {}, collection, // These fields would need to be reconstructed by the executor mutationId: ``, // Will be regenerated @@ -141,6 +177,17 @@ export class TransactionSerializer { return { __type: `Date`, value: value.toISOString() } } + if ( + typeof value === `object` && + getTemporalConstructorName(value[Symbol.toStringTag]) + ) { + return { + __type: `Temporal`, + type: value[Symbol.toStringTag], + value: value.toString(), + } + } + if (typeof value === `object`) { const result: any = Array.isArray(value) ? [] : {} for (const key in value) { @@ -157,7 +204,7 @@ export class TransactionSerializer { return value } - private deserializeValue(value: any, escapedObjects: boolean): any { + private deserializeValue(value: any, valueEncoding: 2 | 3 | undefined): any { if (value === null || value === undefined) { return value } @@ -175,9 +222,36 @@ export class TransactionSerializer { return date } + if ( + valueEncoding === 3 && + typeof value === `object` && + value.__type === `Temporal` + ) { + const constructorName = getTemporalConstructorName(value.type) + if (!constructorName) + throw new Error(`Corrupted Temporal marker: invalid type field`) + if (typeof value.value !== `string`) + throw new Error(`Corrupted Temporal marker: missing value field`) + const constructor = ( + globalThis as { + Temporal?: Partial< + Record< + TemporalConstructorName, + { from: (value: string) => unknown } + > + > + } + ).Temporal?.[constructorName] + if (typeof constructor?.from !== `function`) + throw new MissingTemporalConstructorError( + `Cannot restore Temporal.${constructorName}: missing global constructor`, + ) + return constructor.from(value.value) + } + if (typeof value === `object`) { // Unwrap once, then decode only the fields: the object's own __type is data. - if (escapedObjects && value.__type === `Object`) { + if (valueEncoding !== undefined && value.__type === `Object`) { if ( value.value === null || typeof value.value !== `object` || @@ -193,7 +267,7 @@ export class TransactionSerializer { setDataProperty( result, key, - this.deserializeValue(value[key], escapedObjects), + this.deserializeValue(value[key], valueEncoding), ) } } diff --git a/packages/offline-transactions/src/types.ts b/packages/offline-transactions/src/types.ts index df09189dd4..209d3d8b82 100644 --- a/packages/offline-transactions/src/types.ts +++ b/packages/offline-transactions/src/types.ts @@ -58,7 +58,7 @@ export interface OfflineTransaction { // Serialized representation for storage export interface SerializedOfflineTransaction { /** Absent for the original Date-marker format. */ - valueEncoding?: 2 + valueEncoding?: 2 | 3 id: string mutationFnName: string mutations: Array diff --git a/packages/offline-transactions/tests/KeyScheduler.property.test.ts b/packages/offline-transactions/tests/KeyScheduler.property.test.ts index 43831d0cb7..9bacd9bed6 100644 --- a/packages/offline-transactions/tests/KeyScheduler.property.test.ts +++ b/packages/offline-transactions/tests/KeyScheduler.property.test.ts @@ -1,6 +1,9 @@ import { fc } from '@fast-check/vitest' import { afterEach, describe, expect, it, vi } from 'vitest' -import { KeyScheduler } from '../src/executor/KeyScheduler' +import { + KeyScheduler, + reconcilePendingTransactions, +} from '../src/executor/KeyScheduler' import type { OfflineTransaction } from '../src/types' type Command = @@ -17,6 +20,7 @@ type Command = | { type: `fail` } | { type: `retry`; delay: number; payload: number } | { type: `bulkUpdate`; payload: number } + | { type: `remove`; ids: Array } | { type: `advance`; duration: number } | { type: `clear` } @@ -160,6 +164,11 @@ function applyPlanningCommand( state.retryableId = undefined } else if (nextCommand.type === `advance`) { state.now += nextCommand.duration * 1000 + } else if (nextCommand.type === `remove`) { + const ids = new Set(nextCommand.ids) + state.pending = state.pending.filter( + ({ id }) => id === state.activeId || !ids.has(id), + ) } else if (nextCommand.type === `clear`) { state.pending = [] state.activeId = undefined @@ -182,6 +191,22 @@ function buildLegalHistory(tokens: Array): Array { { type: `advance`, duration: token.duration }, { type: `clear` }, ] + const removable = state.pending.filter(({ id }) => id !== state.activeId) + if (removable.length > 0) { + const mode = Math.abs(token.payload) % 4 + const ids = + mode === 0 + ? [] + : mode === 1 + ? [removable[token.slot % removable.length]!.id] + : mode === 2 + ? removable.map(({ id }) => id) + : [removable[0]!.id, `missing`] + choices.push({ + type: `remove`, + ids, + }) + } if (state.pending.length < 5) { choices.push({ @@ -419,6 +444,18 @@ function runHistory( const duration = nextCommand.duration * 1000 vi.advanceTimersByTime(duration) model.now += duration + } else if (nextCommand.type === `remove`) { + const expectedRemoved = [ + ...new Set(nextCommand.ids.filter((id) => id !== model.activeId)), + ] + expect(reconcilePendingTransactions(scheduler, nextCommand.ids)).toEqual( + expectedRemoved, + ) + const ids = new Set(nextCommand.ids) + model.pending = model.pending.filter( + ({ transaction }) => + transaction.id === model.activeId || !ids.has(transaction.id), + ) } else { scheduler.clear() model.pending = [] @@ -458,6 +495,7 @@ describe(`KeyScheduler generated lifecycle`, () => { { type: `schedule`, slot: 1, createdAt: 0, delay: 0, payload: 2 }, { type: `getNext` }, { type: `start` }, + { type: `remove`, ids: [`tx-0`, `tx-1`, `missing`] }, { type: `fail` }, { type: `retry`, delay: 2, payload: 3 }, { type: `getNext` }, @@ -476,6 +514,7 @@ describe(`KeyScheduler generated lifecycle`, () => { `fail`, `retry`, `bulkUpdate`, + `remove`, `advance`, `complete`, `clear`, @@ -594,4 +633,58 @@ describe(`KeyScheduler generated lifecycle`, () => { }), ).toThrow() }) + + it(`rejects retaining selectively revoked work on its scheduler path`, () => { + const history: Array = [ + { type: `schedule`, slot: 0, createdAt: 0, delay: 0, payload: 1 }, + { type: `schedule`, slot: 1, createdAt: 1, delay: 0, payload: 2 }, + { type: `remove`, ids: [`tx-0`] }, + ] + + expect(() => + runHistory(history, { + commandIndex: 2, + apply: (actual) => ({ + ...actual, + pending: [ + { + id: `tx-0`, + createdAt: BASE_TIME, + nextAttemptAt: BASE_TIME, + retryCount: 0, + payload: 1, + }, + ...actual.pending, + ], + pendingCount: actual.pendingCount + 1, + }), + }), + ).toThrow() + }) + + it(`selectively removes only unissued work`, () => { + const scheduler = new KeyScheduler() + const active = createTransaction(``, BASE_TIME, BASE_TIME, 1) + const removed = createTransaction(`removed`, BASE_TIME + 1, BASE_TIME, 2) + const retained = createTransaction(`retained`, BASE_TIME + 2, BASE_TIME, 3) + scheduler.schedule(active) + scheduler.schedule(removed) + scheduler.schedule(retained) + scheduler.markStarted(active) + + expect( + reconcilePendingTransactions(scheduler, [ + active.id, + removed.id, + `missing`, + ]), + ).toEqual([removed.id, `missing`]) + + expect({ + pending: scheduler.getAllPendingTransactions().map(({ id }) => id), + running: scheduler.getRunningCount(), + }).toEqual({ pending: [active.id, retained.id], running: 1 }) + scheduler.markCompleted(active) + expect(scheduler.getNext()?.id).toBe(retained.id) + }) }) diff --git a/packages/offline-transactions/tests/OfflineExecutor.test.ts b/packages/offline-transactions/tests/OfflineExecutor.test.ts index 19e48eb6cd..bcca7cd7c1 100644 --- a/packages/offline-transactions/tests/OfflineExecutor.test.ts +++ b/packages/offline-transactions/tests/OfflineExecutor.test.ts @@ -1,5 +1,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { LocalStorageAdapter, startOfflineExecutor } from '../src/index' +import { MissingTemporalConstructorError } from '../src/outbox/TransactionSerializer' +import { FakeStorageAdapter } from './harness' import type { OfflineConfig } from '../src/types' describe(`OfflineExecutor`, () => { @@ -73,4 +75,67 @@ describe(`OfflineExecutor`, () => { expect(() => executor.dispose()).not.toThrow() }) + + it(`rejects startup when persisted native scalars cannot be restored`, async () => { + const storage = new FakeStorageAdapter() + await storage.set( + `tx:native-scalar`, + JSON.stringify({ + valueEncoding: 3, + id: `native-scalar`, + mutationFnName: `syncData`, + mutations: [ + { + globalKey: `test-collection:one`, + type: `insert`, + modified: { + id: `one`, + due: { + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `2026-09-16`, + }, + }, + original: {}, + changes: {}, + collectionId: `test-collection`, + }, + ], + keys: [`test-collection:one`], + idempotencyKey: `once`, + createdAt: new Date(0).toISOString(), + retryCount: 0, + nextAttemptAt: 0, + version: 1, + }), + ) + const temporalGlobal = globalThis as { + Temporal?: Record + } + const previousTemporal = temporalGlobal.Temporal + temporalGlobal.Temporal = {} + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const executor = startOfflineExecutor({ + ...config, + storage, + leaderElection: { + requestLeadership: async () => true, + releaseLeadership: () => {}, + isLeader: () => true, + onLeadershipChange: () => () => {}, + }, + }) + + try { + await expect(executor.waitForInit()).rejects.toBeInstanceOf( + MissingTemporalConstructorError, + ) + expect(storage.snapshot()).toHaveProperty(`tx:native-scalar`) + } finally { + executor.dispose() + warning.mockRestore() + if (previousTemporal === undefined) delete temporalGlobal.Temporal + else temporalGlobal.Temporal = previousTemporal + } + }) }) diff --git a/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts b/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts index f1bd98218c..c69e585e30 100644 --- a/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts +++ b/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts @@ -38,36 +38,46 @@ vi.mock(`react-native`, () => { // Mock the @react-native-community/netinfo module vi.mock(`@react-native-community/netinfo`, () => { - const listeners: Array< - (state: { - isConnected: boolean - isInternetReachable: boolean | null - }) => void - > = [] + type NetworkState = { + isConnected: boolean + isInternetReachable: boolean | null + } + const listeners: Array<(state: NetworkState) => void> = [] + let holdNextFetch = false + let resolveFetch: ((state: NetworkState) => void) | undefined return { default: { - addEventListener: vi.fn( - ( - callback: (state: { - isConnected: boolean - isInternetReachable: boolean | null - }) => void, - ) => { - listeners.push(callback) - return () => { - const index = listeners.indexOf(callback) - if (index > -1) { - listeners.splice(index, 1) - } + fetch: vi.fn(() => { + if (!holdNextFetch) return Promise.reject(new Error(`unavailable`)) + holdNextFetch = false + return new Promise((resolve) => { + resolveFetch = resolve + }) + }), + addEventListener: vi.fn((callback: (state: NetworkState) => void) => { + listeners.push(callback) + return () => { + const index = listeners.indexOf(callback) + if (index > -1) { + listeners.splice(index, 1) } - }, - ), + } + }), + __holdNextFetch: () => { + holdNextFetch = true + }, + __resolveFetch: (state: NetworkState) => { + if (!resolveFetch) throw new Error(`No startup fetch is pending`) + resolveFetch(state) + resolveFetch = undefined + }, + __resetFetch: () => { + holdNextFetch = false + resolveFetch = undefined + }, // Expose for testing __listeners: listeners, - __triggerState: (state: { - isConnected: boolean - isInternetReachable: boolean | null - }) => { + __triggerState: (state: NetworkState) => { for (const listener of listeners) { listener(state) } @@ -82,6 +92,7 @@ describe(`ReactNativeOnlineDetector`, () => { // Clear internal listener arrays ;(AppState as any).__listeners.length = 0 ;(NetInfo as any).__listeners.length = 0 + ;(NetInfo as any).__resetFetch() }) describe(`initialization`, () => { @@ -110,6 +121,92 @@ describe(`ReactNativeOnlineDetector`, () => { }) describe(`network connectivity changes`, () => { + it(`gives live events authority over a late startup snapshot`, async () => { + ;(NetInfo as any).__holdNextFetch() + const detector = new ReactNativeOnlineDetector() + const callback = vi.fn() + detector.subscribe(callback) + try { + expect(NetInfo.fetch).toHaveBeenCalledOnce() + ;(NetInfo as any).__triggerState({ + isConnected: false, + isInternetReachable: false, + }) + expect(detector.isOnline()).toBe(false) + ;(NetInfo as any).__resolveFetch({ + isConnected: true, + isInternetReachable: true, + }) + await Promise.resolve() + ;(NetInfo as any).__triggerState({ + isConnected: true, + isInternetReachable: true, + }) + + expect({ + notifications: callback.mock.calls.length, + isOnline: detector.isOnline(), + }).toEqual({ notifications: 1, isOnline: true }) + } finally { + detector.dispose() + } + }) + + it(`does not let a late startup snapshot invent an offline edge`, async () => { + ;(NetInfo as any).__holdNextFetch() + const detector = new ReactNativeOnlineDetector() + const callback = vi.fn() + detector.subscribe(callback) + try { + expect(NetInfo.fetch).toHaveBeenCalledOnce() + ;(NetInfo as any).__triggerState({ + isConnected: true, + isInternetReachable: true, + }) + ;(NetInfo as any).__resolveFetch({ + isConnected: false, + isInternetReachable: false, + }) + await Promise.resolve() + ;(NetInfo as any).__triggerState({ + isConnected: true, + isInternetReachable: true, + }) + + expect({ + notifications: callback.mock.calls.length, + isOnline: detector.isOnline(), + }).toEqual({ notifications: 0, isOnline: true }) + } finally { + detector.dispose() + } + }) + + it(`uses the startup snapshot until a live event arrives`, async () => { + ;(NetInfo as any).__holdNextFetch() + const detector = new ReactNativeOnlineDetector() + const callback = vi.fn() + detector.subscribe(callback) + try { + ;(NetInfo as any).__resolveFetch({ + isConnected: false, + isInternetReachable: false, + }) + await Promise.resolve() + expect(detector.isOnline()).toBe(false) + ;(NetInfo as any).__triggerState({ + isConnected: true, + isInternetReachable: true, + }) + expect({ + notifications: callback.mock.calls.length, + isOnline: detector.isOnline(), + }).toEqual({ notifications: 1, isOnline: true }) + } finally { + detector.dispose() + } + }) + it(`should notify subscribers when transitioning from offline to online`, () => { const detector = new ReactNativeOnlineDetector() const callback = vi.fn() @@ -268,6 +365,19 @@ describe(`ReactNativeOnlineDetector`, () => { }) describe(`disposal`, () => { + it(`ignores startup snapshots that settle after disposal`, async () => { + ;(NetInfo as any).__holdNextFetch() + const detector = new ReactNativeOnlineDetector() + detector.dispose() + ;(NetInfo as any).__resolveFetch({ + isConnected: false, + isInternetReachable: false, + }) + await Promise.resolve() + + expect(detector.isOnline()).toBe(true) + }) + it(`should unsubscribe from all native events on dispose`, () => { const detector = new ReactNativeOnlineDetector() const callback = vi.fn() diff --git a/packages/offline-transactions/tests/leadership-replay.property.test.ts b/packages/offline-transactions/tests/leadership-replay.property.test.ts index 6168e59c15..104b1ab719 100644 --- a/packages/offline-transactions/tests/leadership-replay.property.test.ts +++ b/packages/offline-transactions/tests/leadership-replay.property.test.ts @@ -32,6 +32,298 @@ const storedTransaction = (id: string): OfflineTransaction => ({ version: 1, }) +it(`revokes only replay work excluded by the retry hook`, async () => { + // The hook classifies one captured replay snapshot. Reconciliation may + // revoke IDs from that snapshot, but must preserve work admitted later. + const captured = gate() + const delivery = gate() + let hold = false + let scans = 0 + class Storage extends FakeStorageAdapter { + override async keys() { + scans++ + return super.keys() + } + + override async get(key: string) { + const value = await super.get(key) + if (hold && key === `tx:filtered`) { + captured.resolve() + await delivery.promise + } + return value + } + } + + const filtered = storedTransaction(`filtered`) + const retained = { + ...storedTransaction(`retained`), + createdAt: new Date(1), + } + const admitted = { + ...storedTransaction(`admitted`), + createdAt: new Date(2), + } + const storage = new Storage() + const outbox = new OutboxManager(storage, {}) + await outbox.add(filtered) + await outbox.add(retained) + + const hookInputs: Array> = [] + const calls: Array = [] + let filterReplay = false + let online = false + const scheduler = new KeyScheduler() + const executor = new TransactionExecutor( + scheduler, + outbox, + { + collections: {}, + mutationFns: { + syncData: async ({ transaction }) => { + calls.push(transaction.id) + }, + }, + beforeRetry: (transactions) => { + hookInputs.push(transactions.map(({ id }) => id)) + return filterReplay + ? transactions.filter(({ id }) => id !== filtered.id) + : transactions + }, + jitter: false, + }, + { + isOfflineEnabled: true, + isOnline: () => online, + resolveTransaction: () => {}, + rejectTransaction: () => {}, + registerRestorationTransaction: () => {}, + }, + ) + + try { + await executor.loadPendingTransactions() + expect(scheduler.getAllPendingTransactions().map(({ id }) => id)).toEqual([ + filtered.id, + retained.id, + ]) + + filterReplay = true + hold = true + const loading = executor.loadPendingTransactions() + await atOracleCheckpoint(captured.promise, `retry scan captured filtered`) + await outbox.add(admitted) + await executor.execute(admitted) + hold = false + delivery.resolve() + await atOracleCheckpoint(loading, `filtered retry scan delivered`) + + const queued = scheduler.getAllPendingTransactions().map(({ id }) => id) + const durable = (await outbox.getAll()).map(({ id }) => id) + online = true + await atOracleCheckpoint(executor.executeAll(), `retained work drained`) + + expect({ queued, durable, calls, hookInputs, scans }).toEqual({ + queued: [retained.id, admitted.id], + durable: [retained.id, admitted.id], + calls: [retained.id, admitted.id], + hookInputs: [ + [filtered.id, retained.id], + [filtered.id, retained.id], + ], + scans: 3, + }) + } finally { + hold = false + delivery.resolve() + executor.clear() + } +}) + +it(`keeps issued work durable when a replay hook excludes it`, async () => { + const entered = gate() + const release = gate() + const retryRead = gate() + const retryWrite = gate() + let holdRetryUpdate = false + class Storage extends FakeStorageAdapter { + override async get(key: string) { + const value = await super.get(key) + if (holdRetryUpdate && key === `tx:active`) { + holdRetryUpdate = false + retryRead.resolve() + await retryWrite.promise + } + return value + } + } + const active = storedTransaction(`active`) + const filtered = { + ...storedTransaction(`filtered`), + createdAt: new Date(1), + } + const retained = { + ...storedTransaction(`retained`), + createdAt: new Date(2), + } + const storage = new Storage() + const outbox = new OutboxManager(storage, {}) + await Promise.all( + [active, filtered, retained].map((transaction) => outbox.add(transaction)), + ) + const scheduler = new KeyScheduler() + let online = true + for (const transaction of [active, filtered, retained]) + scheduler.schedule(transaction) + const executor = new TransactionExecutor( + scheduler, + outbox, + { + collections: {}, + mutationFns: { + syncData: async () => { + entered.resolve() + await release.promise + holdRetryUpdate = true + throw new Error(`retry`) + }, + }, + beforeRetry: (transactions) => + transactions.filter(({ id }) => id === retained.id), + jitter: false, + }, + { + isOfflineEnabled: true, + isOnline: () => online, + resolveTransaction: () => {}, + rejectTransaction: () => {}, + registerRestorationTransaction: () => {}, + }, + ) + + let executing: Promise | undefined + try { + executing = executor.executeAll() + await atOracleCheckpoint(entered.promise, `issued work entered provider`) + await executor.loadPendingTransactions() + + expect({ + queued: scheduler.getAllPendingTransactions().map(({ id }) => id), + durable: (await outbox.getAll()).map(({ id }) => id), + running: scheduler.getRunningCount(), + }).toEqual({ + queued: [active.id, retained.id], + durable: [active.id, retained.id], + running: 1, + }) + + release.resolve() + await atOracleCheckpoint(retryRead.promise, `retry persistence read issued`) + await executor.loadPendingTransactions() + expect({ + queued: scheduler.getAllPendingTransactions().map(({ id }) => id), + durable: (await outbox.getAll()).map(({ id }) => id), + running: scheduler.getRunningCount(), + }).toEqual({ + queued: [active.id, retained.id], + durable: [active.id, retained.id], + running: 1, + }) + + online = false + retryWrite.resolve() + await atOracleCheckpoint(executing, `issued work scheduled its retry`) + expect({ + active: await outbox.get(active.id), + filtered: await outbox.get(filtered.id), + queued: scheduler.getAllPendingTransactions().map(({ id }) => id), + running: scheduler.getRunningCount(), + }).toMatchObject({ + active: { id: active.id, retryCount: 1 }, + filtered: null, + queued: [active.id, retained.id], + running: 0, + }) + } finally { + online = false + release.resolve() + retryWrite.resolve() + await executing?.catch(() => undefined) + executor.clear() + } +}) + +it(`keeps permanently failed work owned until durable deletion settles`, async () => { + const deleting = gate() + const deleteRelease = gate() + class Storage extends FakeStorageAdapter { + override async delete(key: string) { + if (key === `tx:active`) { + deleting.resolve() + await deleteRelease.promise + } + return super.delete(key) + } + } + const active = storedTransaction(`active`) + const storage = new Storage() + const outbox = new OutboxManager(storage, {}) + await outbox.add(active) + const scheduler = new KeyScheduler() + scheduler.schedule(active) + const calls: Array = [] + let online = true + const executor = new TransactionExecutor( + scheduler, + outbox, + { + collections: {}, + mutationFns: { + syncData: async ({ transaction }) => { + calls.push(transaction.id) + throw new NonRetriableError(`permanent`) + }, + }, + jitter: false, + }, + { + isOfflineEnabled: true, + isOnline: () => online, + resolveTransaction: () => {}, + rejectTransaction: () => {}, + registerRestorationTransaction: () => {}, + }, + ) + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + let executing: Promise | undefined + + try { + executing = executor.executeAll() + await atOracleCheckpoint(deleting.promise, `durable rejection started`) + await executor.loadPendingTransactions() + expect({ + queued: scheduler.getAllPendingTransactions().map(({ id }) => id), + durable: (await outbox.getAll()).map(({ id }) => id), + running: scheduler.getRunningCount(), + }).toEqual({ queued: [active.id], durable: [active.id], running: 1 }) + + online = false + deleteRelease.resolve() + await atOracleCheckpoint(executing, `durable rejection settled`) + expect({ + queued: scheduler.getAllPendingTransactions().map(({ id }) => id), + durable: (await outbox.getAll()).map(({ id }) => id), + calls, + }).toEqual({ queued: [], durable: [], calls: [active.id] }) + } finally { + online = false + deleteRelease.resolve() + await executing?.catch(() => undefined) + executor.clear() + warning.mockRestore() + } +}) + it.each([`construction`, `leadership`, `outbox read`, `retry hook`] as const)( `does not revive a disposed executor after %s`, async (boundary) => { diff --git a/packages/offline-transactions/tests/transaction-serializer.property.test.ts b/packages/offline-transactions/tests/transaction-serializer.property.test.ts index 3273001aec..e6d27fac59 100644 --- a/packages/offline-transactions/tests/transaction-serializer.property.test.ts +++ b/packages/offline-transactions/tests/transaction-serializer.property.test.ts @@ -1,9 +1,15 @@ import { createCollection, createTransaction } from '@tanstack/db' import fc from 'fast-check' import { expect, it, vi } from 'vitest' -import { TransactionSerializer } from '../src/outbox/TransactionSerializer' +import { OutboxManager } from '../src/outbox/OutboxManager' +import { + MissingTemporalConstructorError, + TransactionSerializer, +} from '../src/outbox/TransactionSerializer' import { cleanupOfflineOracle } from './oracle-lifecycle' +import { FakeStorageAdapter } from './harness' import type { OfflineTransaction } from '../src/types' +import type { PendingMutation } from '@tanstack/db' type Value = | null @@ -108,6 +114,7 @@ async function checkRoundtrip( fault: Fault = `none`, boundary: `encoder` | `decoder` = `encoder`, legacy = false, + versionTwo = false, ) { const row = (index: number, revision: number, payload: Value): Row => ({ id: `row:${index}`, @@ -206,7 +213,7 @@ async function checkRoundtrip( } const expectedWire = { ...envelope, - valueEncoding: 2, + valueEncoding: 3, createdAt: new Date(time).toISOString(), mutations: edits.map((edit, index) => ({ globalKey: transaction.mutations[index]!.globalKey, @@ -237,7 +244,7 @@ async function checkRoundtrip( if (fault === `omit-changes`) encoded = encoded.replaceAll(`"changes":`, `"lostChanges":`) if (fault === `unknown-encoding`) - encoded = encoded.replace(`"valueEncoding":2`, `"valueEncoding":3`) + encoded = encoded.replace(`"valueEncoding":3`, `"valueEncoding":4`) return encoded } const serialized = serializer.serialize(offline) @@ -257,6 +264,8 @@ async function checkRoundtrip( const { valueEncoding: _encoding, ...oldWire } = expectedWire wires.push(JSON.stringify(oldWire)) } + if (versionTwo) + wires.push(JSON.stringify({ ...expectedWire, valueEncoding: 2 })) for (const wire of wires) { const decoded = fresh.deserialize(wire) const { mutations, ...rest } = decoded @@ -299,6 +308,373 @@ const twin: Pair = { wire: `2024-01-01T00:00:00.000Z`, } +const temporalCases = [ + [`Duration`, `PT1H30M`], + [`Instant`, `2026-09-16T12:34:56Z`], + [`PlainDate`, `2026-09-16`], + [`PlainDateTime`, `2026-09-16T12:34:56`], + [`PlainMonthDay`, `09-16`], + [`PlainTime`, `12:34:56`], + [`PlainYearMonth`, `2026-09`], + [`ZonedDateTime`, `2026-09-16T12:34:56-06:00[America/Denver]`], +] as const + +type TemporalName = (typeof temporalCases)[number][0] + +class TemporalStub { + readonly #name: TemporalName + readonly #value: string + + constructor(name: TemporalName, value: string) { + this.#name = name + this.#value = value + } + + get [Symbol.toStringTag](): `Temporal.${TemporalName}` { + return `Temporal.${this.#name}` + } + + toString(): string { + return this.#value + } +} + +it(`preserves native scalar identity across storage restart`, async () => { + type NativeRow = { + id: string + values: Record + } + const writer = createCollection({ + id: `native-scalar-writer`, + getKey: (row) => row.id, + sync: { sync: ({ markReady }) => markReady() }, + }) + const reader = createCollection({ + id: `native-scalar-reader`, + getKey: (row) => row.id, + sync: { sync: ({ markReady }) => markReady() }, + }) + const previousTemporal = ( + globalThis as { Temporal?: Record } + ).Temporal + ;(globalThis as { Temporal?: Record }).Temporal = + Object.fromEntries( + temporalCases.map(([name]) => [ + name, + { from: (value: string) => new TemporalStub(name, value) }, + ]), + ) + + const values = Object.fromEntries( + temporalCases.map(([name, value]) => [name, new TemporalStub(name, value)]), + ) as NativeRow[`values`] + const mutation = { + globalKey: `native-scalar-writer:one`, + type: `update`, + modified: { id: `one`, values }, + original: { id: `one`, values }, + changes: { values }, + collection: writer, + } as unknown as PendingMutation + const transaction: OfflineTransaction = { + id: `native-scalars`, + mutationFnName: `persist`, + mutations: [mutation], + keys: [mutation.globalKey], + idempotencyKey: `once`, + createdAt: new Date(0), + retryCount: 0, + nextAttemptAt: 0, + metadata: { nested: { values } }, + version: 1, + } + + try { + const encoded = new TransactionSerializer({ rows: writer }).serialize( + transaction, + ) + const wire = JSON.parse(encoded) + expect(wire.valueEncoding).toBe(3) + const encodedLocations = [ + wire.mutations[0].modified.values, + wire.mutations[0].original.values, + wire.mutations[0].changes.values, + wire.metadata.nested.values, + ] as Array> + for (const location of encodedLocations) + for (const [name, value] of temporalCases) + expect(location[name]).toEqual({ + __type: `Temporal`, + type: `Temporal.${name}`, + value, + }) + + const restarted = new TransactionSerializer({ rows: reader }) + const decoded = restarted.deserialize(encoded) + const decodedMutation = decoded.mutations[0]! + const restored = [ + (decodedMutation.modified as NativeRow).values, + (decodedMutation.original as NativeRow).values, + (decodedMutation.changes as { values: NativeRow[`values`] }).values, + ( + decoded.metadata as { + nested: { values: NativeRow[`values`] } + } + ).nested.values, + ] as Array> + + for (const location of restored) { + for (const [name, value] of temporalCases) { + expect(location[name]).toBeInstanceOf(TemporalStub) + expect(Object.prototype.toString.call(location[name])).toBe( + `[object Temporal.${name}]`, + ) + expect(String(location[name])).toBe(value) + } + } + } finally { + if (previousTemporal === undefined) + delete (globalThis as { Temporal?: Record }).Temporal + else + (globalThis as { Temporal?: Record }).Temporal = + previousTemporal + await writer.cleanup() + await reader.cleanup() + } +}) + +it(`preserves marker-shaped user data through current wire encoding`, async () => { + const runtime = { + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `2026-09-16`, + } + await checkRoundtrip( + [ + { + kind: `insert`, + slot: 0, + before: twin, + after: { runtime, wire: objectWire(runtime) }, + }, + ], + 0, + ) +}) + +it(`preserves prior wire meanings when reading native scalar markers`, async () => { + const collection = createCollection<{ + id: string + due: unknown + createdAt: unknown + }>({ + id: `native-scalar-compatibility`, + getKey: (row) => row.id, + sync: { sync: ({ markReady }) => markReady() }, + }) + const serializer = new TransactionSerializer({ rows: collection }) + const temporalData = { + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `2026-09-16`, + } + const dateMarker = { + __type: `Date`, + value: `2026-09-16T12:34:56.000Z`, + } + const baseWire = { + id: `compatibility`, + mutationFnName: `persist`, + mutations: [ + { + globalKey: `rows:one`, + type: `insert`, + modified: { id: `one`, due: temporalData, createdAt: dateMarker }, + original: {}, + changes: {}, + collectionId: `rows`, + }, + ], + keys: [`rows:one`], + idempotencyKey: `once`, + createdAt: new Date(0).toISOString(), + retryCount: 0, + nextAttemptAt: 0, + metadata: { due: temporalData, createdAt: dateMarker }, + version: 1, + } + + try { + const unversioned = serializer.deserialize(JSON.stringify(baseWire)) + expect(unversioned.mutations[0]!.modified).toEqual({ + id: `one`, + due: temporalData, + createdAt: new Date(dateMarker.value), + }) + expect(unversioned.metadata).toEqual(baseWire.metadata) + + const versionTwo = serializer.deserialize( + JSON.stringify({ + ...baseWire, + valueEncoding: 2, + mutations: [ + { + ...baseWire.mutations[0], + modified: { + id: `one`, + due: { __type: `Object`, value: temporalData }, + createdAt: dateMarker, + }, + }, + ], + }), + ) + expect(versionTwo.mutations[0]!.modified).toEqual({ + id: `one`, + due: temporalData, + createdAt: new Date(dateMarker.value), + }) + expect(versionTwo.metadata).toEqual(baseWire.metadata) + } finally { + await collection.cleanup() + } +}) + +it(`fails visibly when a stored native scalar cannot be restored`, async () => { + const collection = createCollection<{ id: string; due: unknown }>({ + id: `native-scalar-missing-runtime`, + getKey: (row) => row.id, + sync: { sync: ({ markReady }) => markReady() }, + }) + const marker = { + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `2026-09-16`, + } + const wire = JSON.stringify({ + valueEncoding: 3, + id: `missing-runtime`, + mutationFnName: `persist`, + mutations: [ + { + globalKey: `rows:one`, + type: `insert`, + modified: { id: `one`, due: marker }, + original: {}, + changes: {}, + collectionId: `rows`, + }, + ], + keys: [`rows:one`], + idempotencyKey: `once`, + createdAt: new Date(0).toISOString(), + retryCount: 0, + nextAttemptAt: 0, + metadata: { due: marker }, + version: 1, + }) + const temporalGlobal = globalThis as { Temporal?: Record } + const previousTemporal = temporalGlobal.Temporal + temporalGlobal.Temporal = {} + const storage = new FakeStorageAdapter() + await storage.set(`tx:missing-runtime`, wire) + const outbox = new OutboxManager(storage, { rows: collection }) + + try { + expect(() => + new TransactionSerializer({ rows: collection }).deserialize(wire), + ).toThrow(MissingTemporalConstructorError) + await expect(outbox.get(`missing-runtime`)).rejects.toThrow( + MissingTemporalConstructorError, + ) + await expect(outbox.getAll()).rejects.toThrow( + MissingTemporalConstructorError, + ) + expect(storage.snapshot()).toHaveProperty(`tx:missing-runtime`, wire) + } finally { + if (previousTemporal === undefined) delete temporalGlobal.Temporal + else temporalGlobal.Temporal = previousTemporal + await collection.cleanup() + } +}) + +it(`rejects malformed native scalar markers and constructor failures`, async () => { + const collection = createCollection<{ id: string; due: unknown }>({ + id: `native-scalar-invalid`, + getKey: (row) => row.id, + sync: { sync: ({ markReady }) => markReady() }, + }) + const serializer = new TransactionSerializer({ rows: collection }) + const wire = (marker: unknown) => + JSON.stringify({ + valueEncoding: 3, + id: `invalid-native-scalar`, + mutationFnName: `persist`, + mutations: [ + { + globalKey: `rows:one`, + type: `insert`, + modified: { id: `one`, due: marker }, + original: {}, + changes: {}, + collectionId: `rows`, + }, + ], + keys: [`rows:one`], + idempotencyKey: `once`, + createdAt: new Date(0).toISOString(), + retryCount: 0, + nextAttemptAt: 0, + version: 1, + }) + + try { + expect(() => + serializer.deserialize( + wire({ + __type: `Temporal`, + type: `Temporal.Calendar`, + value: `iso8601`, + }), + ), + ).toThrow(`Corrupted Temporal marker: invalid type field`) + expect(() => + serializer.deserialize( + wire({ __type: `Temporal`, type: `Temporal.PlainDate` }), + ), + ).toThrow(`Corrupted Temporal marker: missing value field`) + + const temporalGlobal = globalThis as { + Temporal?: Record + } + const previousTemporal = temporalGlobal.Temporal + const constructorFailure = new Error(`constructor rejected value`) + temporalGlobal.Temporal = { + PlainDate: { + from: () => { + throw constructorFailure + }, + }, + } + try { + expect(() => + serializer.deserialize( + wire({ + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `not-a-date`, + }), + ), + ).toThrow(constructorFailure) + } finally { + if (previousTemporal === undefined) delete temporalGlobal.Temporal + else temporalGlobal.Temporal = previousTemporal + } + } finally { + await collection.cleanup() + } +}) + // Roundtrips generate valid envelopes. Corrupted wire must be rejected before // it can replace any mutation field with an invented empty object. it.each([`modified`, `original`, `changes`] as const)( @@ -335,7 +711,7 @@ it.each([`modified`, `original`, `changes`] as const)( JSON.stringify({ id: `bad`, createdAt: new Date(0).toISOString(), - valueEncoding: 2, + valueEncoding: 3, mutations: [mutation], }), ), @@ -471,6 +847,35 @@ it.each([20260915, undefined])( }, ) +it.each([20260916, undefined])( + `reads version-two escaped values across restart (seed %s)`, + async (seed) => { + await fc.assert( + fc.asyncProperty( + fc.array( + fc.record({ + kind: fc.constantFrom(`insert`, `update`, `delete`), + slot: fc.integer({ min: 0, max: 1 }), + before: tree(2), + after: tree(2), + }), + { minLength: 1, maxLength: 6 }, + ), + async (edits) => + checkRoundtrip(edits, 0, `none`, `encoder`, false, true), + ), + { + seed: seed ?? replaySeed, + numRuns, + ...(seed === undefined && replayPath !== undefined + ? { path: replayPath } + : {}), + examples: [[pinned]], + }, + ) + }, +) + it.each( ( [ @@ -499,7 +904,7 @@ it.each( message: fault === `wrong-registry` ? `Collection with id writer:0 not found` - : `Unsupported transaction value encoding: 3`, + : `Unsupported transaction value encoding: 4`, } await expect( checkRoundtrip(pinned, 0, fault, boundary), From 88bf6a07e49d607fb77367c0811872ec4e34afd6 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 14:44:41 -0600 Subject: [PATCH 2/7] fix(offline-transactions): reject unrestorable scalars --- .changeset/fix-offline-runtime-correctness.md | 2 +- .../src/outbox/TransactionSerializer.ts | 42 ++++++++++--------- .../transaction-serializer.property.test.ts | 32 ++++++++++++++ 3 files changed, 56 insertions(+), 20 deletions(-) diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md index 07fabca8b5..2be63e3069 100644 --- a/.changeset/fix-offline-runtime-correctness.md +++ b/.changeset/fix-offline-runtime-correctness.md @@ -2,4 +2,4 @@ '@tanstack/offline-transactions': patch --- -Fix offline replay filtering so it preserves concurrently admitted and issued transactions, keep React Native live connectivity events authoritative over stale startup snapshots, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. +Fix offline replay filtering so it preserves concurrently admitted and issued transactions, keep React Native live connectivity events authoritative over stale startup snapshots, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. Recognized native scalars now fail before storage when the matching global constructor is unavailable. diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index 02af63dd01..61ba91589b 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -18,6 +18,7 @@ const temporalConstructorNames = [ ] as const type TemporalConstructorName = (typeof temporalConstructorNames)[number] +type TemporalConstructor = { from: (value: string) => unknown } function getTemporalConstructorName( type: unknown, @@ -31,6 +32,21 @@ function getTemporalConstructorName( : undefined } +function requireTemporalConstructor( + name: TemporalConstructorName, +): TemporalConstructor { + const constructor = ( + globalThis as { + Temporal?: Partial> + } + ).Temporal?.[name] + if (typeof constructor?.from !== `function`) + throw new MissingTemporalConstructorError( + `Missing global Temporal.${name} constructor`, + ) + return constructor +} + export class MissingTemporalConstructorError extends Error {} function setDataProperty( @@ -177,10 +193,12 @@ export class TransactionSerializer { return { __type: `Date`, value: value.toISOString() } } - if ( - typeof value === `object` && - getTemporalConstructorName(value[Symbol.toStringTag]) - ) { + const temporalConstructorName = + typeof value === `object` + ? getTemporalConstructorName(value[Symbol.toStringTag]) + : undefined + if (temporalConstructorName) { + requireTemporalConstructor(temporalConstructorName) return { __type: `Temporal`, type: value[Symbol.toStringTag], @@ -232,21 +250,7 @@ export class TransactionSerializer { throw new Error(`Corrupted Temporal marker: invalid type field`) if (typeof value.value !== `string`) throw new Error(`Corrupted Temporal marker: missing value field`) - const constructor = ( - globalThis as { - Temporal?: Partial< - Record< - TemporalConstructorName, - { from: (value: string) => unknown } - > - > - } - ).Temporal?.[constructorName] - if (typeof constructor?.from !== `function`) - throw new MissingTemporalConstructorError( - `Cannot restore Temporal.${constructorName}: missing global constructor`, - ) - return constructor.from(value.value) + return requireTemporalConstructor(constructorName).from(value.value) } if (typeof value === `object`) { diff --git a/packages/offline-transactions/tests/transaction-serializer.property.test.ts b/packages/offline-transactions/tests/transaction-serializer.property.test.ts index e6d27fac59..d2ce8fde4b 100644 --- a/packages/offline-transactions/tests/transaction-serializer.property.test.ts +++ b/packages/offline-transactions/tests/transaction-serializer.property.test.ts @@ -339,6 +339,38 @@ class TemporalStub { } } +it(`rejects native scalars before storage when global restoration is unavailable`, async () => { + const temporalGlobal = globalThis as { Temporal?: Record } + const previousTemporal = temporalGlobal.Temporal + temporalGlobal.Temporal = {} + const storage = new FakeStorageAdapter() + const outbox = new OutboxManager(storage, {}) + const transaction: OfflineTransaction = { + id: `unrestorable-native-scalar`, + mutationFnName: `persist`, + mutations: [], + keys: [], + idempotencyKey: `once`, + createdAt: new Date(0), + retryCount: 0, + nextAttemptAt: 0, + metadata: { + due: new TemporalStub(`PlainDate`, `2026-09-16`), + }, + version: 1, + } + + try { + await expect(outbox.add(transaction)).rejects.toThrow( + MissingTemporalConstructorError, + ) + expect(storage.snapshot()).toEqual({}) + } finally { + if (previousTemporal === undefined) delete temporalGlobal.Temporal + else temporalGlobal.Temporal = previousTemporal + } +}) + it(`preserves native scalar identity across storage restart`, async () => { type NativeRow = { id: string From 4320478dd1846fdd2b53b80df3075ea0b8fa17b0 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 15:41:30 -0600 Subject: [PATCH 3/7] fix(offline-transactions): address runtime review gaps --- .changeset/fix-offline-runtime-correctness.md | 4 +- .../src/OfflineExecutor.ts | 3 +- .../connectivity/ReactNativeOnlineDetector.ts | 14 - .../src/executor/TransactionExecutor.ts | 10 +- .../src/outbox/OutboxManager.ts | 10 +- .../src/outbox/TransactionSerializer.ts | 25 +- .../tests/OfflineExecutor.test.ts | 152 ++++++++--- .../tests/ReactNativeOnlineDetector.test.ts | 132 ++++------ .../tests/leadership-replay.property.test.ts | 241 +++++++++++++++++- .../transaction-serializer.property.test.ts | 72 +++++- 10 files changed, 508 insertions(+), 155 deletions(-) diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md index 2be63e3069..71bec55660 100644 --- a/.changeset/fix-offline-runtime-correctness.md +++ b/.changeset/fix-offline-runtime-correctness.md @@ -2,4 +2,6 @@ '@tanstack/offline-transactions': patch --- -Fix offline replay filtering so it preserves concurrently admitted and issued transactions, keep React Native live connectivity events authoritative over stale startup snapshots, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. Recognized native scalars now fail before storage when the matching global constructor is unavailable. +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, and retry scheduling remains live when an outbox update or removal fails. 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. diff --git a/packages/offline-transactions/src/OfflineExecutor.ts b/packages/offline-transactions/src/OfflineExecutor.ts index ccda53af59..223086c1b1 100644 --- a/packages/offline-transactions/src/OfflineExecutor.ts +++ b/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() } diff --git a/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts b/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts index 745d796ac7..627f733c36 100644 --- a/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts +++ b/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts @@ -26,23 +26,9 @@ export class ReactNativeOnlineDetector implements OnlineDetector { } this.isListening = true - let receivedLiveState = false - - if (typeof NetInfo.fetch === `function`) { - void NetInfo.fetch() - .then((state) => { - if (this.isListening && !receivedLiveState) { - 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) => { - receivedLiveState = true const isConnected = this.toConnectivityState(state) // Only notify when transitioning to online diff --git a/packages/offline-transactions/src/executor/TransactionExecutor.ts b/packages/offline-transactions/src/executor/TransactionExecutor.ts index 97fb1733f1..662d1bd178 100644 --- a/packages/offline-transactions/src/executor/TransactionExecutor.ts +++ b/packages/offline-transactions/src/executor/TransactionExecutor.ts @@ -57,6 +57,7 @@ export class TransactionExecutor { } finally { this.isExecuting = false this.executionPromise = null + this.scheduleNextRetry() } } @@ -74,9 +75,6 @@ export class TransactionExecutor { await this.executeTransaction(transaction) } - - // Schedule next retry after execution completes - this.scheduleNextRetry() } private async executeTransaction( @@ -227,9 +225,6 @@ export class TransactionExecutor { } finally { this.scheduler.markFailed(transaction) } - - // Schedule retry timer - this.scheduleNextRetry() }, ) } @@ -273,6 +268,9 @@ export class TransactionExecutor { if (removedIds.length > 0) { await this.outbox.removeMany(removedIds) + const error = new NonRetriableError(`Transaction excluded by beforeRetry`) + for (const id of removedIds) + this.offlineExecutor.rejectTransaction(id, error) } } diff --git a/packages/offline-transactions/src/outbox/OutboxManager.ts b/packages/offline-transactions/src/outbox/OutboxManager.ts index 588a8244a7..3b9dc366ee 100644 --- a/packages/offline-transactions/src/outbox/OutboxManager.ts +++ b/packages/offline-transactions/src/outbox/OutboxManager.ts @@ -77,7 +77,10 @@ export class OutboxManager { span.setAttribute(`result`, `found`) return transaction } catch (error) { - if (error instanceof MissingTemporalConstructorError) throw 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 @@ -112,7 +115,10 @@ export class OutboxManager { const transaction = this.serializer.deserialize(data) transactions.push(transaction) } catch (error) { - if (error instanceof MissingTemporalConstructorError) throw 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, diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index 61ba91589b..3f856272b7 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -84,7 +84,7 @@ export class TransactionSerializer { ...transaction, valueEncoding: 3, createdAt: transaction.createdAt.toISOString(), - metadata: this.serializeValue(transaction.metadata), + metadata: this.serializeValue(transaction.metadata, `metadata`), mutations: transaction.mutations.map((mutation) => this.serializeMutation(mutation), ), @@ -184,7 +184,7 @@ export class TransactionSerializer { } as PendingMutation } - private serializeValue(value: any): any { + private serializeValue(value: any, jsonKey?: string): any { if (value === null || value === undefined) { return value } @@ -207,10 +207,29 @@ export class TransactionSerializer { } if (typeof value === `object`) { + if (jsonKey !== undefined && typeof value.toJSON === `function`) { + const jsonValue = value.toJSON(jsonKey) + if (jsonValue !== value) return this.serializeValue(jsonValue, jsonKey) + } + if ( + jsonKey !== undefined && + (value instanceof Boolean || + value instanceof Number || + value instanceof String) + ) { + return value.valueOf() + } const result: any = Array.isArray(value) ? [] : {} for (const key in value) { if (Object.prototype.hasOwnProperty.call(value, key)) { - setDataProperty(result, key, this.serializeValue(value[key])) + setDataProperty( + result, + key, + this.serializeValue( + value[key], + jsonKey === undefined ? undefined : key, + ), + ) } } return !Array.isArray(value) && diff --git a/packages/offline-transactions/tests/OfflineExecutor.test.ts b/packages/offline-transactions/tests/OfflineExecutor.test.ts index bcca7cd7c1..21506beb4e 100644 --- a/packages/offline-transactions/tests/OfflineExecutor.test.ts +++ b/packages/offline-transactions/tests/OfflineExecutor.test.ts @@ -1,6 +1,5 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { LocalStorageAdapter, startOfflineExecutor } from '../src/index' -import { MissingTemporalConstructorError } from '../src/outbox/TransactionSerializer' import { FakeStorageAdapter } from './harness' import type { OfflineConfig } from '../src/types' @@ -76,63 +75,142 @@ describe(`OfflineExecutor`, () => { expect(() => executor.dispose()).not.toThrow() }) - it(`rejects startup when persisted native scalars cannot be restored`, async () => { + it(`keeps constructor-started initialization failures observable`, async () => { + const storageError = new Error(`storage unavailable`) + class Storage extends FakeStorageAdapter { + override async keys(): Promise> { + throw storageError + } + } + const unhandled: Array = [] + const onUnhandled = (error: unknown) => unhandled.push(error) + process.on(`unhandledRejection`, onUnhandled) + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const executor = startOfflineExecutor({ + ...config, + storage: new Storage(), + leaderElection: { + requestLeadership: async () => true, + releaseLeadership: () => {}, + isLeader: () => true, + onLeadershipChange: () => () => {}, + }, + }) + + try { + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(unhandled).toEqual([]) + await expect(executor.waitForInit()).rejects.toBe(storageError) + } finally { + executor.dispose() + warning.mockRestore() + process.off(`unhandledRejection`, onUnhandled) + } + }) + + it(`identifies unreadable rows for targeted removal before restart`, async () => { const storage = new FakeStorageAdapter() - await storage.set( - `tx:native-scalar`, + const record = (id: string, metadata: Record) => JSON.stringify({ valueEncoding: 3, - id: `native-scalar`, + id, mutationFnName: `syncData`, - mutations: [ - { - globalKey: `test-collection:one`, - type: `insert`, - modified: { - id: `one`, - due: { - __type: `Temporal`, - type: `Temporal.PlainDate`, - value: `2026-09-16`, - }, - }, - original: {}, - changes: {}, - collectionId: `test-collection`, - }, - ], - keys: [`test-collection:one`], - idempotencyKey: `once`, + mutations: [], + keys: [], + idempotencyKey: `${id}/once`, createdAt: new Date(0).toISOString(), retryCount: 0, nextAttemptAt: 0, + metadata, version: 1, + }) + await storage.set( + `tx:native-scalar`, + record(`native-scalar`, { + due: { + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `2026-09-16`, + }, }), ) + await storage.set(`tx:readable`, record(`readable`, { note: `safe` })) const temporalGlobal = globalThis as { Temporal?: Record } const previousTemporal = temporalGlobal.Temporal temporalGlobal.Temporal = {} const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) - const executor = startOfflineExecutor({ - ...config, - storage, - leaderElection: { - requestLeadership: async () => true, - releaseLeadership: () => {}, - isLeader: () => true, - onLeadershipChange: () => () => {}, - }, - }) + const unhandled: Array = [] + const onUnhandled = (error: unknown) => unhandled.push(error) + process.on(`unhandledRejection`, onUnhandled) + const calls: Array<{ id: string; metadata: unknown }> = [] + let firstExecutor: ReturnType | undefined + let secondExecutor: ReturnType | undefined + const leaderElection = { + requestLeadership: async () => true, + releaseLeadership: () => {}, + isLeader: () => true, + onLeadershipChange: () => () => {}, + } try { - await expect(executor.waitForInit()).rejects.toBeInstanceOf( - MissingTemporalConstructorError, + mockMutationFn.mockImplementation( + ({ transaction }: { transaction: { id: string; metadata: unknown } }) => + calls.push({ id: transaction.id, metadata: transaction.metadata }), ) + firstExecutor = startOfflineExecutor({ + ...config, + storage, + leaderElection, + }) + await expect(firstExecutor.waitForInit()).rejects.toThrow( + /transaction native-scalar/, + ) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(calls).toEqual([]) expect(storage.snapshot()).toHaveProperty(`tx:native-scalar`) + expect(storage.snapshot()).toHaveProperty(`tx:readable`) + expect(unhandled).toEqual([]) + + await firstExecutor.removeFromOutbox(`native-scalar`) + firstExecutor.dispose() + const recoveredReplay = new Promise((resolve) => { + mockMutationFn.mockImplementation( + ({ + transaction, + }: { + transaction: { id: string; metadata: unknown } + }) => { + calls.push({ id: transaction.id, metadata: transaction.metadata }) + if (transaction.id === `readable`) resolve() + }, + ) + }) + secondExecutor = startOfflineExecutor({ + ...config, + storage, + leaderElection, + }) + await expect(secondExecutor.waitForInit()).resolves.toBeUndefined() + await recoveredReplay + expect(calls).toEqual([ + { + id: `readable`, + metadata: { note: `safe` }, + }, + ]) + expect(storage.snapshot()).toEqual({}) + expect(warning).toHaveBeenCalledWith( + `Failed to initialize offline executor:`, + expect.objectContaining({ + message: expect.stringMatching(/transaction native-scalar/), + }), + ) } finally { - executor.dispose() + firstExecutor?.dispose() + secondExecutor?.dispose() + process.off(`unhandledRejection`, onUnhandled) warning.mockRestore() if (previousTemporal === undefined) delete temporalGlobal.Temporal else temporalGlobal.Temporal = previousTemporal diff --git a/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts b/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts index c69e585e30..43463ed428 100644 --- a/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts +++ b/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts @@ -43,19 +43,24 @@ vi.mock(`@react-native-community/netinfo`, () => { isInternetReachable: boolean | null } const listeners: Array<(state: NetworkState) => void> = [] - let holdNextFetch = false - let resolveFetch: ((state: NetworkState) => void) | undefined + let latestState: NetworkState = { + isConnected: true, + isInternetReachable: true, + } + let deliverNextSubscriptionAsync = false return { default: { - fetch: vi.fn(() => { - if (!holdNextFetch) return Promise.reject(new Error(`unavailable`)) - holdNextFetch = false - return new Promise((resolve) => { - resolveFetch = resolve - }) - }), + fetch: vi.fn(() => Promise.resolve(latestState)), addEventListener: vi.fn((callback: (state: NetworkState) => void) => { listeners.push(callback) + // NetInfo promises the latest information soon after subscription. + const state = latestState + if (deliverNextSubscriptionAsync) { + deliverNextSubscriptionAsync = false + void Promise.resolve().then(() => { + if (listeners.includes(callback)) callback(state) + }) + } else callback(state) return () => { const index = listeners.indexOf(callback) if (index > -1) { @@ -63,21 +68,19 @@ vi.mock(`@react-native-community/netinfo`, () => { } } }), - __holdNextFetch: () => { - holdNextFetch = true + __setLatestState: (state: NetworkState) => { + latestState = state }, - __resolveFetch: (state: NetworkState) => { - if (!resolveFetch) throw new Error(`No startup fetch is pending`) - resolveFetch(state) - resolveFetch = undefined + __deliverNextSubscriptionAsync: () => { + deliverNextSubscriptionAsync = true }, - __resetFetch: () => { - holdNextFetch = false - resolveFetch = undefined + __resetSubscriptionDelivery: () => { + deliverNextSubscriptionAsync = false }, // Expose for testing __listeners: listeners, __triggerState: (state: NetworkState) => { + latestState = state for (const listener of listeners) { listener(state) } @@ -92,7 +95,11 @@ describe(`ReactNativeOnlineDetector`, () => { // Clear internal listener arrays ;(AppState as any).__listeners.length = 0 ;(NetInfo as any).__listeners.length = 0 - ;(NetInfo as any).__resetFetch() + ;(NetInfo as any).__resetSubscriptionDelivery() + ;(NetInfo as any).__setLatestState({ + isConnected: true, + isInternetReachable: true, + }) }) describe(`initialization`, () => { @@ -121,83 +128,51 @@ describe(`ReactNativeOnlineDetector`, () => { }) describe(`network connectivity changes`, () => { - it(`gives live events authority over a late startup snapshot`, async () => { - ;(NetInfo as any).__holdNextFetch() + it(`uses the initial state delivered by the network subscription`, () => { + ;(NetInfo as any).__setLatestState({ + isConnected: false, + isInternetReachable: false, + }) const detector = new ReactNativeOnlineDetector() - const callback = vi.fn() - detector.subscribe(callback) try { - expect(NetInfo.fetch).toHaveBeenCalledOnce() - ;(NetInfo as any).__triggerState({ - isConnected: false, - isInternetReachable: false, - }) expect(detector.isOnline()).toBe(false) - ;(NetInfo as any).__resolveFetch({ - isConnected: true, - isInternetReachable: true, - }) - await Promise.resolve() - ;(NetInfo as any).__triggerState({ - isConnected: true, - isInternetReachable: true, - }) - - expect({ - notifications: callback.mock.calls.length, - isOnline: detector.isOnline(), - }).toEqual({ notifications: 1, isOnline: true }) + expect(NetInfo.fetch).not.toHaveBeenCalled() } finally { detector.dispose() } }) - it(`does not let a late startup snapshot invent an offline edge`, async () => { - ;(NetInfo as any).__holdNextFetch() + it(`accepts an asynchronously delivered initial subscription state`, async () => { + ;(NetInfo as any).__setLatestState({ + isConnected: false, + isInternetReachable: false, + }) + ;(NetInfo as any).__deliverNextSubscriptionAsync() const detector = new ReactNativeOnlineDetector() - const callback = vi.fn() - detector.subscribe(callback) try { - expect(NetInfo.fetch).toHaveBeenCalledOnce() - ;(NetInfo as any).__triggerState({ - isConnected: true, - isInternetReachable: true, - }) - ;(NetInfo as any).__resolveFetch({ - isConnected: false, - isInternetReachable: false, - }) + expect(detector.isOnline()).toBe(true) await Promise.resolve() - ;(NetInfo as any).__triggerState({ - isConnected: true, - isInternetReachable: true, - }) - - expect({ - notifications: callback.mock.calls.length, - isOnline: detector.isOnline(), - }).toEqual({ notifications: 0, isOnline: true }) + expect(detector.isOnline()).toBe(false) + expect(NetInfo.fetch).not.toHaveBeenCalled() } finally { detector.dispose() } }) - it(`uses the startup snapshot until a live event arrives`, async () => { - ;(NetInfo as any).__holdNextFetch() + it(`notifies for changes after the subscription's initial state`, () => { + ;(NetInfo as any).__setLatestState({ + isConnected: false, + isInternetReachable: false, + }) const detector = new ReactNativeOnlineDetector() const callback = vi.fn() detector.subscribe(callback) try { - ;(NetInfo as any).__resolveFetch({ - isConnected: false, - isInternetReachable: false, - }) - await Promise.resolve() - expect(detector.isOnline()).toBe(false) ;(NetInfo as any).__triggerState({ isConnected: true, isInternetReachable: true, }) + expect({ notifications: callback.mock.calls.length, isOnline: detector.isOnline(), @@ -365,19 +340,6 @@ describe(`ReactNativeOnlineDetector`, () => { }) describe(`disposal`, () => { - it(`ignores startup snapshots that settle after disposal`, async () => { - ;(NetInfo as any).__holdNextFetch() - const detector = new ReactNativeOnlineDetector() - detector.dispose() - ;(NetInfo as any).__resolveFetch({ - isConnected: false, - isInternetReachable: false, - }) - await Promise.resolve() - - expect(detector.isOnline()).toBe(true) - }) - it(`should unsubscribe from all native events on dispose`, () => { const detector = new ReactNativeOnlineDetector() const callback = vi.fn() diff --git a/packages/offline-transactions/tests/leadership-replay.property.test.ts b/packages/offline-transactions/tests/leadership-replay.property.test.ts index 104b1ab719..e769b9370e 100644 --- a/packages/offline-transactions/tests/leadership-replay.property.test.ts +++ b/packages/offline-transactions/tests/leadership-replay.property.test.ts @@ -8,7 +8,7 @@ import { KeyScheduler } from '../src/executor/KeyScheduler' import { NonRetriableError } from '../src/types' import { FakeStorageAdapter, createTestOfflineEnvironment } from './harness' import { atOracleCheckpoint, cleanupOfflineOracle } from './oracle-lifecycle' -import type { OfflineTransaction } from '../src/types' +import type { OfflineTransaction, OnlineDetector } from '../src/types' function gate() { let resolve!: () => void @@ -140,6 +140,245 @@ it(`revokes only replay work excluded by the retry hook`, async () => { } }) +it(`settles replay work discarded by the retry hook`, async () => { + const persisted = gate() + const removed = gate() + class Storage extends FakeStorageAdapter { + override async set(key: string, value: string) { + await super.set(key, value) + persisted.resolve() + } + + override async delete(key: string) { + await super.delete(key) + removed.resolve() + } + } + const onlineDetector: OnlineDetector = { + subscribe: () => () => {}, + notifyOnline: () => {}, + isOnline: () => false, + dispose: () => {}, + } + let discardReplay = false + const storage = new Storage() + const env = createTestOfflineEnvironment({ + storage, + config: { + onlineDetector, + beforeRetry: (transactions) => (discardReplay ? [] : transactions), + }, + }) + let commitStatus: unknown = `pending` + let waitStatus: unknown = `pending` + let commitObserved: Promise | undefined + let waitObserved: Promise | undefined + let transactionId = `` + let hasPrimaryFailure = false + try { + await env.waitForLeader() + const transaction = env.executor.createOfflineTransaction({ + mutationFnName: env.mutationFnName, + autoCommit: false, + }) + transactionId = transaction.id + waitObserved = env.executor + .waitForTransactionCompletion(transaction.id) + .then( + () => { + waitStatus = `fulfilled` + }, + (error: unknown) => { + waitStatus = error + }, + ) + transaction.mutate(() => { + env.collection.insert({ + id: `discarded`, + value: `optimistic`, + completed: false, + updatedAt: new Date(0), + }) + }) + commitObserved = transaction.commit().then( + () => { + commitStatus = `fulfilled` + }, + (error: unknown) => { + commitStatus = error + }, + ) + await atOracleCheckpoint(persisted.promise, `discarded work persisted`) + expect(env.collection.get(`discarded`)).toMatchObject({ + value: `optimistic`, + }) + + env.leader.setLeader(false) + discardReplay = true + env.leader.setLeader(true) + await atOracleCheckpoint(removed.promise, `discarded work removed`) + await turn() + + expect(commitStatus).toBeInstanceOf(NonRetriableError) + expect(waitStatus).toBe(commitStatus) + expect(env.collection.get(`discarded`)).toBeUndefined() + expect(storage.snapshot()).not.toHaveProperty(`tx:${transaction.id}`) + } catch (error) { + hasPrimaryFailure = true + throw error + } finally { + if (commitStatus === `pending` && transactionId) + env.executor.rejectTransaction( + transactionId, + new NonRetriableError(`oracle cleanup`), + ) + await cleanupOfflineOracle( + [ + () => Promise.all([commitObserved, waitObserved]), + () => env.executor.dispose(), + () => env.collection.cleanup(), + ], + hasPrimaryFailure, + ) + } +}) + +it(`keeps retry timers live when a retry record update fails`, async () => { + vi.useFakeTimers() + vi.setSystemTime(0) + const storageError = new Error(`retry update failed`) + class Storage extends FakeStorageAdapter { + private failed = false + + override async set(key: string, value: string) { + if (!this.failed && JSON.parse(value).retryCount > 0) { + this.failed = true + throw storageError + } + await super.set(key, value) + } + } + const transaction = storedTransaction(`retry-after-update-failure`) + const storage = new Storage() + const outbox = new OutboxManager(storage, {}) + await outbox.add(transaction) + const scheduler = new KeyScheduler() + const calls: Array = [] + const completed: Array = [] + const executor = new TransactionExecutor( + scheduler, + outbox, + { + collections: {}, + mutationFns: { + syncData: async ({ transaction: current }) => { + calls.push(current.id) + if (calls.length === 1) throw new Error(`provider unavailable`) + }, + }, + jitter: false, + }, + { + isOfflineEnabled: true, + isOnline: () => true, + resolveTransaction: (id) => completed.push(id), + rejectTransaction: () => {}, + registerRestorationTransaction: () => {}, + }, + ) + + try { + await expect(executor.execute(transaction)).rejects.toBe(storageError) + expect({ calls, completed, pending: executor.getPendingCount() }).toEqual({ + calls: [transaction.id], + completed: [], + pending: 1, + }) + + await vi.advanceTimersByTimeAsync(1000) + expect({ calls, completed, pending: executor.getPendingCount() }).toEqual({ + calls: [transaction.id, transaction.id], + completed: [transaction.id], + pending: 0, + }) + expect(await outbox.get(transaction.id)).toBeNull() + } finally { + executor.clear() + vi.useRealTimers() + } +}) + +it(`keeps later work live when a permanent record removal fails`, async () => { + vi.useFakeTimers() + vi.setSystemTime(0) + const storageError = new Error(`permanent removal failed`) + class Storage extends FakeStorageAdapter { + private failed = false + + override async delete(key: string) { + if (!this.failed && key === `tx:permanent`) { + this.failed = true + throw storageError + } + await super.delete(key) + } + } + const permanent = storedTransaction(`permanent`) + const later = { ...storedTransaction(`later`), createdAt: new Date(1) } + const storage = new Storage() + const outbox = new OutboxManager(storage, {}) + await outbox.add(permanent) + await outbox.add(later) + const scheduler = new KeyScheduler() + scheduler.schedule(permanent) + scheduler.schedule(later) + const calls: Array = [] + const completed: Array = [] + const executor = new TransactionExecutor( + scheduler, + outbox, + { + collections: {}, + mutationFns: { + syncData: async ({ transaction }) => { + calls.push(transaction.id) + if (transaction.id === permanent.id) + throw new NonRetriableError(`permanent`) + }, + }, + jitter: false, + }, + { + isOfflineEnabled: true, + isOnline: () => true, + resolveTransaction: (id) => completed.push(id), + rejectTransaction: () => {}, + registerRestorationTransaction: () => {}, + }, + ) + + try { + await expect(executor.executeAll()).rejects.toBe(storageError) + expect({ calls, completed, pending: executor.getPendingCount() }).toEqual({ + calls: [permanent.id], + completed: [], + pending: 1, + }) + + await vi.advanceTimersByTimeAsync(0) + expect({ calls, completed, pending: executor.getPendingCount() }).toEqual({ + calls: [permanent.id, later.id], + completed: [later.id], + pending: 0, + }) + expect(await outbox.get(permanent.id)).toEqual(permanent) + expect(await outbox.get(later.id)).toBeNull() + } finally { + executor.clear() + vi.useRealTimers() + } +}) + it(`keeps issued work durable when a replay hook excludes it`, async () => { const entered = gate() const release = gate() diff --git a/packages/offline-transactions/tests/transaction-serializer.property.test.ts b/packages/offline-transactions/tests/transaction-serializer.property.test.ts index d2ce8fde4b..9dece477f7 100644 --- a/packages/offline-transactions/tests/transaction-serializer.property.test.ts +++ b/packages/offline-transactions/tests/transaction-serializer.property.test.ts @@ -371,6 +371,70 @@ it(`rejects native scalars before storage when global restoration is unavailable } }) +it(`preserves metadata toJSON values without changing mutation value semantics`, () => { + class JsonValue { + toJSON(key: string) { + return `from-toJSON:${key}` + } + } + const collection = { + id: `metadata-json-writer`, + getKeyFromItem: (value: { id: string }) => value.id, + } as any + const serializer = new TransactionSerializer({ rows: collection }) + const url = new URL(`https://example.com/path`) + const jsonValue = new JsonValue() + const metadata = { + url, + jsonValue, + nested: [jsonValue], + boxed: [Object(7), Object(`value`), Object(false)], + } + Object.defineProperty(metadata, `toJSON`, { + value: (key: string) => ({ ...metadata, metadataKey: key }), + }) + const transaction: OfflineTransaction = { + id: `metadata-json`, + mutationFnName: `persist`, + mutations: [ + { + globalKey: `metadata-json-writer:one`, + type: `insert`, + modified: { id: `one`, url, jsonValue }, + original: {}, + changes: {}, + collection, + } as unknown as PendingMutation, + ], + keys: [`metadata-json-writer:one`], + idempotencyKey: `once`, + createdAt: new Date(0), + retryCount: 0, + nextAttemptAt: 0, + metadata, + version: 1, + } + + const encoded = serializer.serialize(transaction) + const wire = JSON.parse(encoded) + expect(wire.metadata).toEqual({ + url: `https://example.com/path`, + jsonValue: `from-toJSON:jsonValue`, + nested: [`from-toJSON:0`], + boxed: [7, `value`, false], + metadataKey: `metadata`, + }) + expect(wire.mutations[0].modified).toEqual({ + id: `one`, + url: {}, + jsonValue: {}, + }) + + const decoded = serializer.deserialize(encoded) + expect(decoded.metadata).toEqual(wire.metadata) + expect(decoded.mutations[0]!.modified).toEqual(wire.mutations[0].modified) +}) + it(`preserves native scalar identity across storage restart`, async () => { type NativeRow = { id: string @@ -572,7 +636,7 @@ it(`preserves prior wire meanings when reading native scalar markers`, async () } }) -it(`fails visibly when a stored native scalar cannot be restored`, async () => { +it(`fails visibly with the retained native scalar transaction id`, async () => { const collection = createCollection<{ id: string; due: unknown }>({ id: `native-scalar-missing-runtime`, getKey: (row) => row.id, @@ -617,11 +681,9 @@ it(`fails visibly when a stored native scalar cannot be restored`, async () => { new TransactionSerializer({ rows: collection }).deserialize(wire), ).toThrow(MissingTemporalConstructorError) await expect(outbox.get(`missing-runtime`)).rejects.toThrow( - MissingTemporalConstructorError, - ) - await expect(outbox.getAll()).rejects.toThrow( - MissingTemporalConstructorError, + /transaction missing-runtime/, ) + await expect(outbox.getAll()).rejects.toThrow(/transaction missing-runtime/) expect(storage.snapshot()).toHaveProperty(`tx:missing-runtime`, wire) } finally { if (previousTemporal === undefined) delete temporalGlobal.Temporal From 26d458a4c251f592a4ca4be5d253d03e85ab559a Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 16:19:03 -0600 Subject: [PATCH 4/7] fix(offline-transactions): preserve metadata JSON semantics --- .changeset/fix-offline-runtime-correctness.md | 2 +- .../src/outbox/TransactionSerializer.ts | 45 +++--- .../transaction-serializer.property.test.ts | 134 ++++++++++++++++++ 3 files changed, 159 insertions(+), 22 deletions(-) diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md index 71bec55660..568e4bcdd3 100644 --- a/.changeset/fix-offline-runtime-correctness.md +++ b/.changeset/fix-offline-runtime-correctness.md @@ -2,6 +2,6 @@ '@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, and retry scheduling remains live when an outbox update or removal fails. Recognized native scalars now fail before storage when the matching global constructor is unavailable. +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, retry scheduling remains live when an outbox update or 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. diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index 3f856272b7..9ce003268b 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -184,17 +184,17 @@ export class TransactionSerializer { } as PendingMutation } - private serializeValue(value: any, jsonKey?: string): any { + private serializeValue(value: any, jsonKey?: string | false): any { if (value === null || value === undefined) { return value } - if (value instanceof Date) { + if (jsonKey !== false && value instanceof Date) { return { __type: `Date`, value: value.toISOString() } } const temporalConstructorName = - typeof value === `object` + jsonKey !== false && typeof value === `object` ? getTemporalConstructorName(value[Symbol.toStringTag]) : undefined if (temporalConstructorName) { @@ -207,33 +207,36 @@ export class TransactionSerializer { } if (typeof value === `object`) { - if (jsonKey !== undefined && typeof value.toJSON === `function`) { - const jsonValue = value.toJSON(jsonKey) - if (jsonValue !== value) return this.serializeValue(jsonValue, jsonKey) - } + const toJSON = typeof jsonKey === `string` && value.toJSON + if (typeof toJSON === `function`) + return this.serializeValue(toJSON.call(value, jsonKey), false) if ( jsonKey !== undefined && (value instanceof Boolean || + value instanceof BigInt || value instanceof Number || value instanceof String) ) { return value.valueOf() } - const result: any = Array.isArray(value) ? [] : {} - for (const key in value) { - if (Object.prototype.hasOwnProperty.call(value, key)) { - setDataProperty( - result, - key, - this.serializeValue( - value[key], - jsonKey === undefined ? undefined : key, - ), - ) - } + const isArray = Array.isArray(value) + const result: any = isArray ? [] : {} + const keys = isArray + ? Array.from({ length: value.length }, (_, index) => String(index)) + : Object.keys(value) + for (const key of keys) { + setDataProperty( + result, + key, + this.serializeValue( + value[key], + jsonKey === undefined ? undefined : key, + ), + ) } - return !Array.isArray(value) && - Object.prototype.hasOwnProperty.call(value, `__type`) + if (jsonKey === false && typeof result.toJSON === `function`) + delete result.toJSON + return !isArray && Object.prototype.hasOwnProperty.call(value, `__type`) ? { __type: `Object`, value: result } : result } diff --git a/packages/offline-transactions/tests/transaction-serializer.property.test.ts b/packages/offline-transactions/tests/transaction-serializer.property.test.ts index 9dece477f7..d11ef8295e 100644 --- a/packages/offline-transactions/tests/transaction-serializer.property.test.ts +++ b/packages/offline-transactions/tests/transaction-serializer.property.test.ts @@ -339,6 +339,23 @@ class TemporalStub { } } +function metadataTransaction( + metadata: Record, +): OfflineTransaction { + return { + id: `metadata-json`, + mutationFnName: `persist`, + mutations: [], + keys: [], + idempotencyKey: `once`, + createdAt: new Date(0), + retryCount: 0, + nextAttemptAt: 0, + metadata, + version: 1, + } +} + it(`rejects native scalars before storage when global restoration is unavailable`, async () => { const temporalGlobal = globalThis as { Temporal?: Record } const previousTemporal = temporalGlobal.Temporal @@ -435,6 +452,123 @@ it(`preserves metadata toJSON values without changing mutation value semantics`, expect(decoded.mutations[0]!.modified).toEqual(wire.mutations[0].modified) }) +it(`does not invoke toJSON again on its immediate replacement`, () => { + const nested = { + toJSON(key: string) { + return `nested:${key}` + }, + } + const replacement = { + nested, + toJSON() { + return `incorrect second invocation` + }, + } + const transaction = metadataTransaction({ + value: { + toJSON(key: string) { + return key === `value` ? replacement : `incorrect key:${key}` + }, + }, + }) + + const wire = JSON.parse(new TransactionSerializer({}).serialize(transaction)) + + expect(wire.metadata).toEqual({ value: { nested: `nested:nested` } }) +}) + +it(`treats immediate native-scalar replacements as ordinary JSON values`, () => { + const temporalReplacement = { + [Symbol.toStringTag]: `Temporal.PlainDate`, + toString() { + return `2026-09-16` + }, + } + const transaction = metadataTransaction({ + date: { toJSON: () => new Date(0) }, + temporal: { toJSON: () => temporalReplacement }, + }) + + const wire = JSON.parse(new TransactionSerializer({}).serialize(transaction)) + + expect(wire.metadata).toEqual({ date: {}, temporal: {} }) +}) + +it(`preserves JSON array length and indexed-property semantics in metadata`, () => { + const values = Array(3) as Array + Object.defineProperty(values, 0, { enumerable: false, value: `hidden` }) + values[1] = `visible` + const shrinking = Array(3) as Array + Object.defineProperty(shrinking, 0, { + enumerable: true, + get() { + shrinking.length = 1 + return `first` + }, + }) + const transaction = metadataTransaction({ values, shrinking }) + + const wire = JSON.parse(new TransactionSerializer({}).serialize(transaction)) + + expect(wire.metadata).toEqual({ + values: [`hidden`, `visible`, null], + shrinking: [`first`, null, null], + }) +}) + +it(`retains JSON's error for boxed BigInt metadata`, () => { + const transaction = metadataTransaction({ value: Object(1n) }) + + expect(() => new TransactionSerializer({}).serialize(transaction)).toThrow( + TypeError, + ) +}) + +it(`does not recurse through fresh toJSON replacement objects`, () => { + const freshReplacement = (): Record => ({ + toJSON: freshReplacement, + }) + const transaction = metadataTransaction({ + value: { toJSON: freshReplacement }, + }) + + const wire = JSON.parse(new TransactionSerializer({}).serialize(transaction)) + + expect(wire.metadata).toEqual({ value: {} }) +}) + +it(`reads metadata toJSON once with its object as the receiver`, () => { + let reads = 0 + let selfCalls = 0 + const value = { + marker: `receiver`, + get toJSON() { + reads++ + if (reads > 1) throw new Error(`toJSON read more than once`) + return function (this: { marker: string }, key: string) { + return `${this.marker}:${key}` + } + }, + } + const self = { + keep: `value`, + toJSON() { + selfCalls++ + return this + }, + } + const transaction = metadataTransaction({ value, self }) + + const wire = JSON.parse(new TransactionSerializer({}).serialize(transaction)) + + expect(wire.metadata).toEqual({ + value: `receiver:value`, + self: { keep: `value` }, + }) + expect(reads).toBe(1) + expect(selfCalls).toBe(1) +}) + it(`preserves native scalar identity across storage restart`, async () => { type NativeRow = { id: string From 942f7db0164bd76ba436611782bf1b0d68f9a18d Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 16:22:33 -0600 Subject: [PATCH 5/7] test(offline-transactions): validate Temporal tags once --- .../src/outbox/TransactionSerializer.ts | 2 +- .../transaction-serializer.property.test.ts | 31 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index 9ce003268b..d281990115 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -201,7 +201,7 @@ export class TransactionSerializer { requireTemporalConstructor(temporalConstructorName) return { __type: `Temporal`, - type: value[Symbol.toStringTag], + type: `Temporal.${temporalConstructorName}`, value: value.toString(), } } diff --git a/packages/offline-transactions/tests/transaction-serializer.property.test.ts b/packages/offline-transactions/tests/transaction-serializer.property.test.ts index d11ef8295e..b25a88adc4 100644 --- a/packages/offline-transactions/tests/transaction-serializer.property.test.ts +++ b/packages/offline-transactions/tests/transaction-serializer.property.test.ts @@ -388,6 +388,37 @@ it(`rejects native scalars before storage when global restoration is unavailable } }) +it(`uses one validated Temporal tag when writing a marker`, () => { + const temporalGlobal = globalThis as { Temporal?: Record } + const previousTemporal = temporalGlobal.Temporal + let reads = 0 + temporalGlobal.Temporal = { + PlainDate: { from: (value: string) => value }, + } + const value = { + get [Symbol.toStringTag]() { + reads++ + return reads === 1 ? `Temporal.PlainDate` : `Temporal.Invalid` + }, + toString: () => `2026-09-16`, + } + + try { + const wire = JSON.parse( + new TransactionSerializer({}).serialize(metadataTransaction({ value })), + ) + expect(wire.metadata.value).toEqual({ + __type: `Temporal`, + type: `Temporal.PlainDate`, + value: `2026-09-16`, + }) + expect(reads).toBe(1) + } finally { + if (previousTemporal === undefined) delete temporalGlobal.Temporal + else temporalGlobal.Temporal = previousTemporal + } +}) + it(`preserves metadata toJSON values without changing mutation value semantics`, () => { class JsonValue { toJSON(key: string) { From de726111c2f21dd37e73dd16eee02384fcc84558 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Wed, 16 Sep 2026 18:00:49 -0600 Subject: [PATCH 6/7] fix(offline-transactions): settle successful replay removals --- .changeset/fix-offline-runtime-correctness.md | 2 +- .../src/executor/TransactionExecutor.ts | 10 +- .../src/outbox/TransactionSerializer.ts | 72 ++++--- .../tests/leadership-replay.property.test.ts | 183 ++++++++++++++++++ 4 files changed, 224 insertions(+), 43 deletions(-) diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md index 568e4bcdd3..0879dd55a3 100644 --- a/.changeset/fix-offline-runtime-correctness.md +++ b/.changeset/fix-offline-runtime-correctness.md @@ -2,6 +2,6 @@ '@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, retry scheduling remains live when an outbox update or removal fails, and metadata keeps standard `toJSON(key)` replacement semantics. Recognized native scalars now fail before storage when the matching global constructor is unavailable. +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 an outbox update or 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. diff --git a/packages/offline-transactions/src/executor/TransactionExecutor.ts b/packages/offline-transactions/src/executor/TransactionExecutor.ts index 662d1bd178..8cf83b54e5 100644 --- a/packages/offline-transactions/src/executor/TransactionExecutor.ts +++ b/packages/offline-transactions/src/executor/TransactionExecutor.ts @@ -267,10 +267,14 @@ export class TransactionExecutor { }) if (removedIds.length > 0) { - await this.outbox.removeMany(removedIds) const error = new NonRetriableError(`Transaction excluded by beforeRetry`) - for (const id of removedIds) - this.offlineExecutor.rejectTransaction(id, error) + await Promise.all( + removedIds.map((id) => + this.outbox + .remove(id) + .then(() => this.offlineExecutor.rejectTransaction(id, error)), + ), + ) } } diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index d281990115..9344d9aba4 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -185,16 +185,14 @@ export class TransactionSerializer { } private serializeValue(value: any, jsonKey?: string | false): any { - if (value === null || value === undefined) { - return value - } + if (value === null || typeof value !== `object`) return value if (jsonKey !== false && value instanceof Date) { return { __type: `Date`, value: value.toISOString() } } const temporalConstructorName = - jsonKey !== false && typeof value === `object` + jsonKey !== false ? getTemporalConstructorName(value[Symbol.toStringTag]) : undefined if (temporalConstructorName) { @@ -206,42 +204,38 @@ export class TransactionSerializer { } } - if (typeof value === `object`) { - const toJSON = typeof jsonKey === `string` && value.toJSON - if (typeof toJSON === `function`) - return this.serializeValue(toJSON.call(value, jsonKey), false) - if ( - jsonKey !== undefined && - (value instanceof Boolean || - value instanceof BigInt || - value instanceof Number || - value instanceof String) - ) { - return value.valueOf() - } - const isArray = Array.isArray(value) - const result: any = isArray ? [] : {} - const keys = isArray - ? Array.from({ length: value.length }, (_, index) => String(index)) - : Object.keys(value) - for (const key of keys) { - setDataProperty( - result, - key, - this.serializeValue( - value[key], - jsonKey === undefined ? undefined : key, - ), - ) - } - if (jsonKey === false && typeof result.toJSON === `function`) - delete result.toJSON - return !isArray && Object.prototype.hasOwnProperty.call(value, `__type`) - ? { __type: `Object`, value: result } - : result + const toJSON = typeof jsonKey === `string` && value.toJSON + if (typeof toJSON === `function`) + return this.serializeValue(toJSON.call(value, jsonKey), false) + if ( + jsonKey !== undefined && + (value instanceof Boolean || + value instanceof BigInt || + value instanceof Number || + value instanceof String) + ) { + return value.valueOf() } - - return value + const isArray = Array.isArray(value) + const result: any = isArray ? [] : {} + const keys = isArray + ? Array.from({ length: value.length }, (_, index) => String(index)) + : Object.keys(value) + for (const key of keys) { + setDataProperty( + result, + key, + this.serializeValue( + value[key], + jsonKey === undefined ? undefined : key, + ), + ) + } + if (jsonKey === false && typeof result.toJSON === `function`) + delete result.toJSON + return !isArray && Object.prototype.hasOwnProperty.call(value, `__type`) + ? { __type: `Object`, value: result } + : result } private deserializeValue(value: any, valueEncoding: 2 | 3 | undefined): any { diff --git a/packages/offline-transactions/tests/leadership-replay.property.test.ts b/packages/offline-transactions/tests/leadership-replay.property.test.ts index e769b9370e..1c9249b785 100644 --- a/packages/offline-transactions/tests/leadership-replay.property.test.ts +++ b/packages/offline-transactions/tests/leadership-replay.property.test.ts @@ -243,6 +243,189 @@ it(`settles replay work discarded by the retry hook`, async () => { } }) +it.each([0, 1])( + `settles each discarded replay after its durable removal succeeds when deletion %i fails`, + async (failedIndex) => { + const successfulIndex = 1 - failedIndex + const persisted = gate() + const removing = gate() + const releaseRemoval = gate() + const removed = gate() + const failed = gate() + const retried = gate() + const storageError = new Error(`discard removal failed`) + let writes = 0 + let failedKey = `` + let failRemoval = false + let failedOnce = false + class Storage extends FakeStorageAdapter { + override async set(key: string, value: string) { + await super.set(key, value) + if (++writes === 2) persisted.resolve() + } + + override async delete(key: string) { + if (failRemoval && key !== failedKey) { + removing.resolve() + await releaseRemoval.promise + } + if (failRemoval && key === failedKey && !failedOnce) { + failedOnce = true + failed.resolve() + throw storageError + } + await super.delete(key) + if (key === failedKey) retried.resolve() + else removed.resolve() + } + } + const onlineDetector: OnlineDetector = { + subscribe: () => () => {}, + notifyOnline: () => {}, + isOnline: () => false, + dispose: () => {}, + } + let discardReplay = false + const storage = new Storage() + const env = createTestOfflineEnvironment({ + storage, + config: { + onlineDetector, + beforeRetry: (transactions) => (discardReplay ? [] : transactions), + }, + }) + const ids: Array = [] + const commitStatuses: Array = [`pending`, `pending`] + const waitStatuses: Array = [`pending`, `pending`] + const commitObserved: Array> = [] + const waitObserved: Array> = [] + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + let hasPrimaryFailure = false + try { + await env.waitForLeader() + for (let index = 0; index < 2; index++) { + const transaction = env.executor.createOfflineTransaction({ + mutationFnName: env.mutationFnName, + autoCommit: false, + }) + ids.push(transaction.id) + waitObserved.push( + env.executor.waitForTransactionCompletion(transaction.id).then( + () => { + waitStatuses[index] = `fulfilled` + }, + (error: unknown) => { + waitStatuses[index] = error + }, + ), + ) + transaction.mutate(() => { + env.collection.insert({ + id: `discarded-${index}`, + value: `optimistic-${index}`, + completed: false, + updatedAt: new Date(index), + }) + }) + commitObserved.push( + transaction.commit().then( + () => { + commitStatuses[index] = `fulfilled` + }, + (error: unknown) => { + commitStatuses[index] = error + }, + ), + ) + } + await atOracleCheckpoint(persisted.promise, `discarded work persisted`) + + env.leader.setLeader(false) + discardReplay = true + failedKey = `tx:${ids[failedIndex]}` + failRemoval = true + env.leader.setLeader(true) + await atOracleCheckpoint( + Promise.all([removing.promise, failed.promise]), + `mixed discard removals started`, + ) + await turn() + + expect(commitStatuses).toEqual([`pending`, `pending`]) + expect(waitStatuses).toEqual([`pending`, `pending`]) + expect(warning).toHaveBeenCalledWith( + `Failed to load and replay transactions:`, + storageError, + ) + expect(env.executor.getPendingCount()).toBe(0) + expect(storage.snapshot()).toHaveProperty(`tx:${ids[0]}`) + expect(storage.snapshot()).toHaveProperty(failedKey) + expect(env.collection.get(`discarded-0`)?.value).toBe(`optimistic-0`) + expect(env.collection.get(`discarded-1`)?.value).toBe(`optimistic-1`) + expect(env.mutationCalls).toHaveLength(0) + + releaseRemoval.resolve() + await atOracleCheckpoint(removed.promise, `successful discard removed`) + await turn() + + expect(commitStatuses[successfulIndex]).toBeInstanceOf(NonRetriableError) + expect(waitStatuses[successfulIndex]).toBe( + commitStatuses[successfulIndex], + ) + expect(commitStatuses[failedIndex]).toBe(`pending`) + expect(waitStatuses[failedIndex]).toBe(`pending`) + expect({ + queued: env.executor.getPendingCount(), + durable: Object.keys(storage.snapshot()), + optimistic: [ + env.collection.get(`discarded-${successfulIndex}`), + env.collection.get(`discarded-${failedIndex}`)?.value, + ], + calls: env.mutationCalls.length, + }).toEqual({ + queued: 0, + durable: [failedKey], + optimistic: [undefined, `optimistic-${failedIndex}`], + calls: 0, + }) + + env.leader.setLeader(false) + env.leader.setLeader(true) + await atOracleCheckpoint(retried.promise, `failed discard retried`) + await turn() + + expect(commitStatuses[failedIndex]).toBeInstanceOf(NonRetriableError) + expect(waitStatuses[failedIndex]).toBe(commitStatuses[failedIndex]) + expect({ + queued: env.executor.getPendingCount(), + durable: storage.snapshot(), + optimistic: env.collection.get(`discarded-${failedIndex}`), + calls: env.mutationCalls.length, + }).toEqual({ queued: 0, durable: {}, optimistic: undefined, calls: 0 }) + } catch (error) { + hasPrimaryFailure = true + throw error + } finally { + releaseRemoval.resolve() + for (let index = 0; index < ids.length; index++) + if (commitStatuses[index] === `pending`) + env.executor.rejectTransaction( + ids[index]!, + new NonRetriableError(`oracle cleanup`), + ) + await cleanupOfflineOracle( + [ + () => Promise.all([...commitObserved, ...waitObserved]), + () => env.executor.dispose(), + () => env.collection.cleanup(), + ], + hasPrimaryFailure, + ) + warning.mockRestore() + } + }, +) + it(`keeps retry timers live when a retry record update fails`, async () => { vi.useFakeTimers() vi.setSystemTime(0) From 83fc42f3ef98c11d49c097ec07ca8040259ac147 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 17 Sep 2026 07:26:29 -0600 Subject: [PATCH 7/7] docs: clarify offline retry liveness scope --- .changeset/fix-offline-runtime-correctness.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md index 0879dd55a3..e45018a110 100644 --- a/.changeset/fix-offline-runtime-correctness.md +++ b/.changeset/fix-offline-runtime-correctness.md @@ -2,6 +2,6 @@ '@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 an outbox update or removal fails, and metadata keeps standard `toJSON(key)` replacement semantics. Recognized native scalars now fail before storage when the matching global constructor is unavailable. +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.