diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md new file mode 100644 index 0000000000..e45018a110 --- /dev/null +++ b/.changeset/fix-offline-runtime-correctness.md @@ -0,0 +1,7 @@ +--- +'@tanstack/offline-transactions': patch +--- + +Fix offline replay filtering so it preserves concurrently admitted and issued transactions, use React Native's subscribed connectivity snapshot as the initial authority, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. Filtered replay work now settles and rolls back after each successful durable removal even when a sibling removal fails, retry scheduling remains live when a retry update or permanent-failure removal fails, and metadata keeps standard `toJSON(key)` replacement semantics. Recognized native scalars now fail before storage when the matching global constructor is unavailable. + +Offline storage compatibility: new records use `valueEncoding: 3`. Older clients cannot read these records, so do not run old and new clients against the same pending outbox or downgrade while new records remain pending. 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 5c38c06911..627f733c36 100644 --- a/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts +++ b/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts @@ -27,16 +27,6 @@ export class ReactNativeOnlineDetector implements OnlineDetector { this.isListening = true - if (typeof NetInfo.fetch === `function`) { - void NetInfo.fetch() - .then((state) => { - this.wasConnected = this.toConnectivityState(state) - }) - .catch(() => { - // Ignore initial fetch failures and rely on subscription updates. - }) - } - // Subscribe to network state changes this.netInfoUnsubscribe = NetInfo.addEventListener((state) => { const isConnected = this.toConnectivityState(state) 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..8cf83b54e5 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 { @@ -56,6 +57,7 @@ export class TransactionExecutor { } finally { this.isExecuting = false this.executionPromise = null + this.scheduleNextRetry() } } @@ -73,9 +75,6 @@ export class TransactionExecutor { await this.executeTransaction(transaction) } - - // Schedule next retry after execution completes - this.scheduleNextRetry() } private async executeTransaction( @@ -180,8 +179,11 @@ export class TransactionExecutor { span.setAttribute(`shouldRetry`, shouldRetry) if (!shouldRetry) { - this.scheduler.markCompleted(transaction) - await this.outbox.remove(transaction.id) + try { + await this.outbox.remove(transaction.id) + } finally { + this.scheduler.markCompleted(transaction) + } console.warn( `Transaction ${transaction.id} failed permanently:`, error, @@ -211,7 +213,6 @@ export class TransactionExecutor { span.setAttribute(`retryDelay`, delay) span.setAttribute(`nextRetryCount`, updatedTransaction.retryCount) - this.scheduler.markFailed(transaction) this.scheduler.updateTransaction(updatedTransaction) try { @@ -221,10 +222,9 @@ export class TransactionExecutor { span.recordException(persistError as Error) span.setAttribute(`result`, `persist_failed`) throw persistError + } finally { + this.scheduler.markFailed(transaction) } - - // Schedule retry timer - this.scheduleNextRetry() }, ) } @@ -247,6 +247,14 @@ export class TransactionExecutor { this.scheduler.schedule(transaction), ) + removedIds = transactions + .filter( + (tx) => + !filteredTransactions.some((filtered) => filtered.id === tx.id), + ) + .map(({ id }) => id) + removedIds = reconcilePendingTransactions(this.scheduler, removedIds) + // Restore optimistic state for loaded transactions // This ensures the UI shows the optimistic data while transactions are pending this.restoreOptimisticState(newlyLoaded) @@ -256,17 +264,17 @@ export class TransactionExecutor { // Schedule retry timer for loaded transactions this.scheduleNextRetry() - - removedIds = transactions - .filter( - (tx) => - !filteredTransactions.some((filtered) => filtered.id === tx.id), - ) - .map(({ id }) => id) }) if (removedIds.length > 0) { - await this.outbox.removeMany(removedIds) + const error = new NonRetriableError(`Transaction excluded by beforeRetry`) + await Promise.all( + removedIds.map((id) => + this.outbox + .remove(id) + .then(() => this.offlineExecutor.rejectTransaction(id, error)), + ), + ) } } diff --git a/packages/offline-transactions/src/outbox/OutboxManager.ts b/packages/offline-transactions/src/outbox/OutboxManager.ts index 82cccd3780..3b9dc366ee 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,10 @@ export class OutboxManager { span.setAttribute(`result`, `found`) return transaction } catch (error) { + if (error instanceof MissingTemporalConstructorError) { + error.message = `transaction ${id}: ${error.message}` + throw error + } console.warn(`Failed to deserialize transaction ${id}:`, error) span.setAttribute(`result`, `deserialize_error`) return null @@ -108,6 +115,10 @@ export class OutboxManager { const transaction = this.serializer.deserialize(data) transactions.push(transaction) } catch (error) { + if (error instanceof MissingTemporalConstructorError) { + error.message = `transaction ${key.slice(this.keyPrefix.length)}: ${error.message}` + throw error + } console.warn( `Failed to deserialize transaction from key ${key}:`, error, diff --git a/packages/offline-transactions/src/outbox/TransactionSerializer.ts b/packages/offline-transactions/src/outbox/TransactionSerializer.ts index 0eeb9f65df..9344d9aba4 100644 --- a/packages/offline-transactions/src/outbox/TransactionSerializer.ts +++ b/packages/offline-transactions/src/outbox/TransactionSerializer.ts @@ -6,6 +6,49 @@ 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] +type TemporalConstructor = { from: (value: string) => unknown } + +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 +} + +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( object: Record, key: string, @@ -39,8 +82,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, `metadata`), mutations: transaction.mutations.map((mutation) => this.serializeMutation(mutation), ), @@ -57,7 +101,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 +121,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 +151,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 +170,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 @@ -132,32 +184,61 @@ export class TransactionSerializer { } as PendingMutation } - private serializeValue(value: any): any { - if (value === null || value === undefined) { - return value - } + private serializeValue(value: any, jsonKey?: string | false): any { + if (value === null || typeof value !== `object`) return value - if (value instanceof Date) { + if (jsonKey !== false && value instanceof Date) { return { __type: `Date`, value: value.toISOString() } } - if (typeof value === `object`) { - 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])) - } + const temporalConstructorName = + jsonKey !== false + ? getTemporalConstructorName(value[Symbol.toStringTag]) + : undefined + if (temporalConstructorName) { + requireTemporalConstructor(temporalConstructorName) + return { + __type: `Temporal`, + type: `Temporal.${temporalConstructorName}`, + value: value.toString(), } - return !Array.isArray(value) && - Object.prototype.hasOwnProperty.call(value, `__type`) - ? { __type: `Object`, value: result } - : result } - return value + 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 } - 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 +256,22 @@ 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`) + return requireTemporalConstructor(constructorName).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 +287,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..21506beb4e 100644 --- a/packages/offline-transactions/tests/OfflineExecutor.test.ts +++ b/packages/offline-transactions/tests/OfflineExecutor.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { LocalStorageAdapter, startOfflineExecutor } from '../src/index' +import { FakeStorageAdapter } from './harness' import type { OfflineConfig } from '../src/types' describe(`OfflineExecutor`, () => { @@ -73,4 +74,146 @@ describe(`OfflineExecutor`, () => { expect(() => executor.dispose()).not.toThrow() }) + + 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() + const record = (id: string, metadata: Record) => + JSON.stringify({ + valueEncoding: 3, + id, + mutationFnName: `syncData`, + 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 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 { + 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 { + 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 f1bd98218c..43463ed428 100644 --- a/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts +++ b/packages/offline-transactions/tests/ReactNativeOnlineDetector.test.ts @@ -38,36 +38,49 @@ 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 latestState: NetworkState = { + isConnected: true, + isInternetReachable: true, + } + let deliverNextSubscriptionAsync = false 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(() => 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) { + listeners.splice(index, 1) } - }, - ), + } + }), + __setLatestState: (state: NetworkState) => { + latestState = state + }, + __deliverNextSubscriptionAsync: () => { + deliverNextSubscriptionAsync = true + }, + __resetSubscriptionDelivery: () => { + deliverNextSubscriptionAsync = false + }, // Expose for testing __listeners: listeners, - __triggerState: (state: { - isConnected: boolean - isInternetReachable: boolean | null - }) => { + __triggerState: (state: NetworkState) => { + latestState = state for (const listener of listeners) { listener(state) } @@ -82,6 +95,11 @@ describe(`ReactNativeOnlineDetector`, () => { // Clear internal listener arrays ;(AppState as any).__listeners.length = 0 ;(NetInfo as any).__listeners.length = 0 + ;(NetInfo as any).__resetSubscriptionDelivery() + ;(NetInfo as any).__setLatestState({ + isConnected: true, + isInternetReachable: true, + }) }) describe(`initialization`, () => { @@ -110,6 +128,60 @@ describe(`ReactNativeOnlineDetector`, () => { }) describe(`network connectivity changes`, () => { + it(`uses the initial state delivered by the network subscription`, () => { + ;(NetInfo as any).__setLatestState({ + isConnected: false, + isInternetReachable: false, + }) + const detector = new ReactNativeOnlineDetector() + try { + expect(detector.isOnline()).toBe(false) + expect(NetInfo.fetch).not.toHaveBeenCalled() + } finally { + detector.dispose() + } + }) + + 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() + try { + expect(detector.isOnline()).toBe(true) + await Promise.resolve() + expect(detector.isOnline()).toBe(false) + expect(NetInfo.fetch).not.toHaveBeenCalled() + } finally { + detector.dispose() + } + }) + + 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).__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() diff --git a/packages/offline-transactions/tests/leadership-replay.property.test.ts b/packages/offline-transactions/tests/leadership-replay.property.test.ts index 6168e59c15..1c9249b785 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 @@ -32,6 +32,720 @@ 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(`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.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) + 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() + 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..b25a88adc4 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,632 @@ 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 + } +} + +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 + 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(`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) { + 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(`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 + 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 with the retained native scalar transaction id`, 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( + /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 + 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 +970,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 +1106,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 +1163,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),