diff --git a/.changeset/fix-offline-runtime-correctness.md b/.changeset/fix-offline-runtime-correctness.md new file mode 100644 index 0000000000..22027a11ea --- /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, wait for React Native's subscribed connectivity snapshot before reporting online, and preserve Temporal scalar identity through storage and restart when the runtime provides `globalThis.Temporal`. Filtered replay work now settles and rolls back after each successful durable removal even when a sibling removal fails, while a failed filtered-work cleanup no longer aborts unrelated startup replay. Successful and permanently failed provider work settles its own caller even when durable cleanup also fails, retry scheduling remains live when a retry update or permanent-failure removal fails, and metadata keeps standard `toJSON(key)` replacement semantics. Recognized native scalars now fail before storage when the matching global constructor is unavailable. + +Offline storage compatibility: new records use `valueEncoding: 3`. Older clients cannot read these records, so do not run old and new clients against the same pending outbox or downgrade while new records remain pending. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 06eb175c2a..a95f4b280c 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -82,6 +82,7 @@ does not award oracle credit for a filename alone. | [#1833](https://github.com/TanStack/db/pull/1833) | Explicit oracle | `packages/db/tests/query/pagination-oracle.property.test.ts` owns inherited collection collation, actual `item2`/`item10` order, exact request options, hostile lexical/numeric controls, and both scan and auto-index paths. It runs in `@tanstack/db`'s `test:oracles` campaign. | | [#1834](https://github.com/TanStack/db/pull/1834) | Explicit oracle | `packages/db/tests/query/cold-join-reconciliation-oracle.test.ts` owns join/predicate equality equivalence across the established value domains, binary/string and nullish controls, replacement histories, raw lazy demand, and scan/auto-index paths. It runs in `@tanstack/db`'s `test:oracles` campaign. | | [#1835](https://github.com/TanStack/db/pull/1835) | Explicit oracle | The existing `packages/db/tests/collection-state-retention-oracle.property.test.ts` and `packages/db/tests/optimistic-transaction-oracle.property.test.ts` owners cover separate collection-state and transaction-history laws. Both were already registered in `@tanstack/db`'s `test:oracles` campaign; focused storage/local-only tests remain collateral. | +| [#1837](https://github.com/TanStack/db/pull/1837) | Explicit oracle | The offline scheduler, leadership replay, and serializer owners cover selective replay retirement, durable per-ID settlement, stale-read fencing, lifecycle recovery, native scalar encoding, and prior wire compatibility. They run in the package test campaign; generated owners expose `OFFLINE_ORACLE_{SEED,PATH,RUNS}` or the scheduler's `TANSTACK_DB_OFFLINE_ORACLE_*` replay interface. | | [#1842](https://github.com/TanStack/db/pull/1842) | No shipped-law case | The PR changed only focused observer tests and introduced no production behavior. `packages/db/tests/live-query-observer.test.ts` remains the correct evidence; no synthetic oracle or campaign claim is added. | [PR #1816](https://github.com/TanStack/db/pull/1816) preserves existing witnesses, 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..eee79640d2 100644 --- a/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts +++ b/packages/offline-transactions/src/connectivity/ReactNativeOnlineDetector.ts @@ -14,7 +14,7 @@ export class ReactNativeOnlineDetector implements OnlineDetector { private netInfoUnsubscribe: (() => void) | null = null private appStateSubscription: NativeEventSubscription | null = null private isListening = false - private wasConnected = true + private wasConnected = false constructor() { this.startListening() @@ -27,16 +27,6 @@ export class ReactNativeOnlineDetector implements OnlineDetector { this.isListening = true - if (typeof NetInfo.fetch === `function`) { - void NetInfo.fetch() - .then((state) => { - this.wasConnected = this.toConnectivityState(state) - }) - .catch(() => { - // Ignore initial fetch failures and rely on subscription updates. - }) - } - // Subscribe to network state changes this.netInfoUnsubscribe = NetInfo.addEventListener((state) => { const isConnected = this.toConnectivityState(state) diff --git a/packages/offline-transactions/src/executor/KeyScheduler.ts b/packages/offline-transactions/src/executor/KeyScheduler.ts index 20fdd1dbf3..794930371f 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,23 @@ export class KeyScheduler { } getRunningCount(): number { - return this.isRunning ? 1 : 0 + return this.activeTransactionId === undefined ? 0 : 1 } clear(): void { this.pendingTransactions = [] - this.isRunning = false + this.activeTransactionId = undefined + } + + /** @internal Reconcile one replay snapshot without canceling issued work. */ + removePendingTransactions(transactionIds: Iterable): 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 { diff --git a/packages/offline-transactions/src/executor/TransactionExecutor.ts b/packages/offline-transactions/src/executor/TransactionExecutor.ts index b6bfbd964a..88b90fc2ea 100644 --- a/packages/offline-transactions/src/executor/TransactionExecutor.ts +++ b/packages/offline-transactions/src/executor/TransactionExecutor.ts @@ -56,6 +56,7 @@ export class TransactionExecutor { } finally { this.isExecuting = false this.executionPromise = null + this.scheduleNextRetry() } } @@ -73,9 +74,6 @@ export class TransactionExecutor { await this.executeTransaction(transaction) } - - // Schedule next retry after execution completes - this.scheduleNextRetry() } private async executeTransaction( @@ -97,18 +95,9 @@ export class TransactionExecutor { span.setAttribute(`retry.attempt`, transaction.retryCount) } + let result: void try { - const result = await this.runMutationFn(transaction) - - try { - // Replay can still see this ID until durable deletion settles. - await this.outbox.remove(transaction.id) - } finally { - this.scheduler.markCompleted(transaction) - } - - span.setAttribute(`result`, `success`) - this.offlineExecutor.resolveTransaction(transaction.id, result) + result = await this.runMutationFn(transaction) } catch (error) { const err = error instanceof Error ? error : new Error(String(error)) @@ -119,6 +108,20 @@ export class TransactionExecutor { ;(err as any)[HANDLED_EXECUTION_ERROR] = true throw err } + + let removalError: unknown + try { + // Replay can still see this ID until durable deletion settles. + await this.outbox.remove(transaction.id) + } catch (error) { + removalError = error + } finally { + this.scheduler.markCompleted(transaction) + } + + span.setAttribute(`result`, `success`) + this.offlineExecutor.resolveTransaction(transaction.id, result) + if (removalError !== undefined) throw removalError }, ) } catch (error) { @@ -180,8 +183,14 @@ export class TransactionExecutor { span.setAttribute(`shouldRetry`, shouldRetry) if (!shouldRetry) { - this.scheduler.markCompleted(transaction) - await this.outbox.remove(transaction.id) + let removalError: unknown + try { + await this.outbox.remove(transaction.id) + } catch (cleanupError) { + removalError = cleanupError + } finally { + this.scheduler.markCompleted(transaction) + } console.warn( `Transaction ${transaction.id} failed permanently:`, error, @@ -190,6 +199,7 @@ export class TransactionExecutor { span.setAttribute(`result`, `permanent_failure`) // Signal permanent failure to the waiting transaction this.offlineExecutor.rejectTransaction(transaction.id, error) + if (removalError !== undefined) throw removalError return } @@ -211,7 +221,6 @@ export class TransactionExecutor { span.setAttribute(`retryDelay`, delay) span.setAttribute(`nextRetryCount`, updatedTransaction.retryCount) - this.scheduler.markFailed(transaction) this.scheduler.updateTransaction(updatedTransaction) try { @@ -221,10 +230,9 @@ export class TransactionExecutor { span.recordException(persistError as Error) span.setAttribute(`result`, `persist_failed`) throw persistError + } finally { + this.scheduler.markFailed(transaction) } - - // Schedule retry timer - this.scheduleNextRetry() }, ) } @@ -240,13 +248,21 @@ export class TransactionExecutor { filteredTransactions = this.config.beforeRetry(transactions) } - // The outbox read or retry hook may outlive this owner's right to replay. + // The retry hook is user code and may synchronously revoke replay rights. if (!this.offlineExecutor.isOfflineEnabled) return const newlyLoaded = filteredTransactions.filter((transaction) => this.scheduler.schedule(transaction), ) + removedIds = transactions + .filter( + (tx) => + !filteredTransactions.some((filtered) => filtered.id === tx.id), + ) + .map(({ id }) => id) + removedIds = this.scheduler.removePendingTransactions(removedIds) + // Restore optimistic state for loaded transactions // This ensures the UI shows the optimistic data while transactions are pending this.restoreOptimisticState(newlyLoaded) @@ -256,17 +272,24 @@ export class TransactionExecutor { // Schedule retry timer for loaded transactions this.scheduleNextRetry() - - removedIds = transactions - .filter( - (tx) => - !filteredTransactions.some((filtered) => filtered.id === tx.id), - ) - .map(({ id }) => id) }) if (removedIds.length > 0) { - await this.outbox.removeMany(removedIds) + const error = new NonRetriableError(`Transaction excluded by beforeRetry`) + await Promise.all( + removedIds.map(async (id) => { + try { + await this.outbox.remove(id) + this.offlineExecutor.rejectTransaction(id, error) + } catch (cleanupError) { + console.warn( + `Failed to remove transaction excluded by beforeRetry:`, + id, + cleanupError, + ) + } + }), + ) } } 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 89c2ef8706..c89dd7ddc1 100644 --- a/packages/offline-transactions/tests/KeyScheduler.property.test.ts +++ b/packages/offline-transactions/tests/KeyScheduler.property.test.ts @@ -18,6 +18,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` } @@ -64,22 +65,33 @@ const { /** * # Which offline transaction may run next? * - * The scheduler is globally serial in creation order. Equal creation times keep - * scheduling order. A delayed FIFO head blocks younger work. At most one entry - * is active. Failure makes that entry retryable; retry updates its deadline and - * payload without changing its place. Clear retires active and pending work and - * leaves the scheduler reusable. + * Contract and source: the established KeyScheduler FIFO tests and + * TransactionExecutor calling order require one globally serial queue. Equal + * creation times retain scheduling order. A delayed FIFO head blocks younger + * work. Failure makes the active transaction retryable without changing its + * place. Replay reconciliation may retire only unissued IDs. Clear retires all + * scheduler work and leaves the scheduler reusable. * - * A declarative ledger, fake clock, and stable sequence form the model. Legal - * commands are schedule, inspect, start, complete, fail, retry, bulk update, - * advance time, and clear. The driver calls only executor-facing scheduler - * methods, then compares returned identity and payload, ordered pending records, - * counts, active state, and eligibility after every command. + * Model: a declarative ledger, fake clock, stable creation sequence, and active + * ID predict scheduler observations. The model does not import scheduler state + * or production classifiers. * - * Fixed histories cover every transition and deadline relation. Generated - * histories add shrinking and replay; four injected faults calibrate the path. - * Persistence, promise settlement, leadership, and real timers have separate - * owners. + * History grammar: legal commands schedule, inspect, start, fulfill, reject, + * update one or all pending records, reconcile one replay snapshot, advance + * the clock, and clear. A reject is followed immediately by its retry update. + * IDs are unique while pending. At most five IDs exist in generated histories. + * + * Production driver and refinement check: the driver calls the real + * executor-facing scheduler methods. After every command, it compares the next + * eligible ID and payload, ordered pending records, counts, and active state. + * + * Reach and controls: one fixed history reaches every transition. Two fixed + * histories cross retry-deadline order. Five injected observation faults prove + * the comparison rejects bypass, double issue, stale payload, stale clear, and + * failed selective retirement. The generated lane supports seed/path replay. + * + * Limits: persistence, caller promise settlement, leadership, and real timers + * have separate owners. The model does not promise fairness beyond FIFO order. */ type CommandToken = { @@ -161,6 +173,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 @@ -183,6 +200,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({ @@ -259,6 +292,8 @@ function ordered(model: Model): Array { ) } +// Model law: only the oldest pending transaction can become eligible. An +// active transaction or a delayed FIFO head makes the next result empty. function expectedNext(model: Model): OfflineTransaction | undefined { if (model.activeId) return undefined const first = ordered(model)[0]?.transaction @@ -420,6 +455,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(scheduler.removePendingTransactions(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 = [] @@ -459,6 +506,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` }, @@ -477,6 +525,7 @@ describe(`KeyScheduler generated lifecycle`, () => { `fail`, `retry`, `bulkUpdate`, + `remove`, `advance`, `complete`, `clear`, @@ -595,4 +644,54 @@ 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( + scheduler.removePendingTransactions([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..d719f88fe5 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(`does not claim online before the initial subscription state arrives`, async () => { + ;(NetInfo as any).__setLatestState({ + isConnected: false, + isInternetReachable: false, + }) + ;(NetInfo as any).__deliverNextSubscriptionAsync() + const detector = new ReactNativeOnlineDetector() + try { + expect(detector.isOnline()).toBe(false) + 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 a762c20ad3..700ea438c1 100644 --- a/packages/offline-transactions/tests/leadership-replay.property.test.ts +++ b/packages/offline-transactions/tests/leadership-replay.property.test.ts @@ -8,22 +8,43 @@ 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 { readOfflineOracleConfig } from './oracle-config' +import type { OfflineTransaction, OnlineDetector } from '../src/types' /** * # May leadership replay an offline transaction more than once? * - * Only the current leader may read and schedule the outbox. Losing leadership - * or disposing fences startup, provider work, acknowledgement, retry hooks, and - * stale reads. Regaining leadership may replay durable pending rows, but an ID - * already pending, running, completed, or permanently rejected in the current - * boundary must not execute twice. + * Contract and source: OfflineExecutor leadership, TransactionExecutor serial + * scheduling, and OutboxManager durability define this boundary. Only the + * current leader may admit stored work. Issued work may settle after leadership + * loss, but loss or disposal forbids new provider work. A durable row remains + * owned until its acknowledgement or permanent rejection is durably removed. * - * A fake durable outbox, explicit leadership callbacks, and gated provider and - * delete acknowledgements form the history grammar. The real OfflineExecutor, - * TransactionExecutor, scheduler, and transaction path run unchanged. Checks - * cover mutation calls and idempotency keys, outbox contents, scheduler counts, - * leadership ownership, retry records, restoration, disposal, and cleanup. + * Model: this file is a partial relational oracle, not a second executor. Each + * history relates three independent projections: durable outbox IDs, scheduler + * lifecycle, and caller settlement with optimistic rows. Expected relations + * come from those contracts rather than production queues or classifiers. + * + * History grammar: leadership may be acquired, lost, regained, or disposed at + * construction, outbox read, retry hook, provider, acknowledgement, or retry + * boundaries. Controlled storage may hold, reject, or finish individual reads, + * writes, and deletes. Histories include concurrent scans, repeated leadership + * reports, filtered replay snapshots, mixed deletion outcomes, and later work. + * + * Production driver and refinement check: the real OfflineExecutor, + * TransactionExecutor, KeyScheduler, OutboxManager, and transaction path run + * unchanged. Gates control causal events. Named checkpoints compare provider + * calls and idempotency keys, outbox contents, scheduler counts, leadership, + * retry records, caller promise settlement, optimistic rows, and cleanup. + * + * Reach and controls: pinned histories force each lifecycle boundary and both + * mixed-deletion positions. Fixed and random campaigns cover legal adjacent + * histories and support seed/path replay. Delayed-read, stale-admission, and + * mixed-removal witnesses reject the recorded pre-fix behaviors. + * + * Limits: the fake storage adapter proves ordering and ownership, not a native + * storage engine. Exactly-once network execution across independent leaders is + * not promised. Provider side effects after an issued call remain provider-owned. */ function gate() { @@ -48,6 +69,736 @@ const storedTransaction = (id: string): OfflineTransaction => ({ version: 1, }) +// These campaigns share the package replay variables. Target this file and one +// test name when replaying a shrink path, because paths are property-specific. +const serialWorkOracle = readOfflineOracleConfig({ + prefix: `OFFLINE_ORACLE`, + defaultRuns: 20, +}) +const leadershipReportOracle = readOfflineOracleConfig({ + prefix: `OFFLINE_ORACLE`, + defaultRuns: 40, +}) +const delayedReadOracle = readOfflineOracleConfig({ + prefix: `OFFLINE_ORACLE`, + defaultRuns: 30, +}) + +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 remove transaction excluded by beforeRetry:`, + ids[failedIndex], + 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) => { @@ -348,7 +1099,15 @@ it.each( } }, ), - { seed, numRuns: 20 }, + { + numRuns: serialWorkOracle.runs, + ...((seed ?? serialWorkOracle.seed) === undefined + ? {} + : { seed: seed ?? serialWorkOracle.seed }), + ...(seed === undefined && serialWorkOracle.path !== undefined + ? { path: serialWorkOracle.path } + : {}), + }, ) }, ) @@ -529,8 +1288,13 @@ it.each( }, ), { - seed, - numRuns: 40, + numRuns: leadershipReportOracle.runs, + ...((seed ?? leadershipReportOracle.seed) === undefined + ? {} + : { seed: seed ?? leadershipReportOracle.seed }), + ...(seed === undefined && leadershipReportOracle.path !== undefined + ? { path: leadershipReportOracle.path } + : {}), examples: [ [ { @@ -882,8 +1646,13 @@ it.each([20260919, undefined])( }, ), { - seed, - numRuns: 30, + numRuns: delayedReadOracle.runs, + ...((seed ?? delayedReadOracle.seed) === undefined + ? {} + : { seed: seed ?? delayedReadOracle.seed }), + ...(seed === undefined && delayedReadOracle.path !== undefined + ? { path: delayedReadOracle.path } + : {}), examples: [ [ { @@ -964,6 +1733,72 @@ it.each([`keys`, `get`] as const)( }, ) +it(`keeps startup replay available when discarded-work cleanup fails`, async () => { + const discarded = storedTransaction(`discarded-at-startup`) + const retained = { + ...storedTransaction(`retained-at-startup`), + createdAt: new Date(1), + } + const cleanupError = new Error(`discarded cleanup unavailable`) + class Storage extends FakeStorageAdapter { + override async delete(key: string): Promise { + if (key === `tx:${discarded.id}`) throw cleanupError + await super.delete(key) + } + } + const storage = new Storage() + const outbox = new OutboxManager(storage, {}) + await outbox.add(discarded) + await outbox.add(retained) + const replayed = gate() + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const env = createTestOfflineEnvironment({ + storage, + mutationFn: ({ transaction }) => { + if (transaction.id === retained.id) replayed.resolve() + }, + config: { + beforeRetry: (transactions) => + transactions.filter(({ id }) => id !== discarded.id), + }, + }) + let hasPrimaryFailure = false + try { + await expect( + atOracleCheckpoint( + env.executor.waitForInit(), + `startup replay admitted despite discarded cleanup failure`, + ), + ).resolves.toBeUndefined() + await atOracleCheckpoint(replayed.promise, `retained startup work replayed`) + await turn() + + expect(env.mutationCalls.map(({ transaction }) => transaction.id)).toEqual([ + retained.id, + ]) + expect((await env.executor.peekOutbox()).map(({ id }) => id)).toEqual([ + discarded.id, + ]) + expect(warning).toHaveBeenCalledWith( + `Failed to remove transaction excluded by beforeRetry:`, + discarded.id, + cleanupError, + ) + } catch (error) { + hasPrimaryFailure = true + throw error + } finally { + await cleanupOfflineOracle( + [ + () => env.executor.dispose(), + () => env.collection.cleanup(), + () => warning.mockRestore(), + ], + hasPrimaryFailure, + ) + } +}) + it.each([false, true])( `fences each successful clear deletion while a peer is pending or fails, failure=%s`, async (failure) => { diff --git a/packages/offline-transactions/tests/transaction-serializer.property.test.ts b/packages/offline-transactions/tests/transaction-serializer.property.test.ts index d328ea828c..1a3115b169 100644 --- a/packages/offline-transactions/tests/transaction-serializer.property.test.ts +++ b/packages/offline-transactions/tests/transaction-serializer.property.test.ts @@ -1,27 +1,51 @@ 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 { readOfflineOracleConfig } from './oracle-config' import type { OfflineTransaction } from '../src/types' +import type { PendingMutation } from '@tanstack/db' /** * # Does an offline transaction survive durable serialization exactly? * - * The wire format supports JSON trees plus Date values. User keys that resemble - * codec markers remain data. Insert, update, and delete mutations retain their - * original, modified, changes, collection registry, timestamps, and order across - * restart. Unknown encodings fail rather than creating a zombie transaction. + * Contract and source: SerializedOfflineTransaction and the established restart + * contract preserve mutation meaning across process boundaries. Encoding v3 + * supports JSON trees, Date, and the eight restorable Temporal scalar types. + * Marker-shaped user objects remain data. Unversioned and v2 records retain + * their prior meanings. Unsupported or malformed encodings fail visibly. * - * The generator builds semantic runtime and wire pairs from leaves; it never - * walks a production value with a copy of the serializer. A restarted set of - * Collections with different object identities decodes the record and replays - * it. Encoder and decoder fault modes prove Date/string confusion, wrong - * registry, omitted changes, and unknown versions are observed. + * Model: each generated Pair derives runtime and wire values from one semantic + * leaf. It never walks production output or imports serializer helpers. Focused + * metadata witnesses use JSON's toJSON, boxed-scalar, and array-index rules. + * Mutation values deliberately retain the package's structural encoding rules. * - * Cycles, undefined, non-finite numbers, and arbitrary native objects are - * outside this declared durable format. + * History grammar: insert, update, and delete edits span two collection registry + * keys. Trees contain bounded arrays, objects, hostile keys, Date values, and + * date-like strings. Separate histories cover unversioned, v2, and v3 records, + * metadata replacements, Temporal constructors, and corrupted wire markers. + * + * Production driver and refinement check: real Collections create mutations. + * TransactionSerializer writes the record. New Collection identities then read + * it through the same registry keys. Each checkpoint compares the wire tree, + * decoded transaction fields, registry binding, order, errors, and retained + * storage after a failed read. + * + * Reach and controls: pinned histories force every edit kind and hostile marker. + * Fixed and random lanes support seed/path replay. Encoder and decoder faults + * inject Date/string confusion, wrong registry, omitted changes, and unknown + * versions. Malformed markers and missing Temporal constructors test failure + * paths before durable data can be replaced. + * + * Limits: cycles, undefined, non-finite numbers, arbitrary native objects, and + * cross-realm boxed values are outside current evidence. This oracle does not + * claim byte stability for object key order beyond JSON's established rules. */ type Value = @@ -118,12 +142,15 @@ function objectWire(fields: { [key: string]: Value }): Value { : fields } +// Model law: v3 wire values escape marker-shaped objects exactly once. The +// expected tree is built from semantic Pairs, independently of production. async function checkRoundtrip( edits: Array, time: number, fault: Fault = `none`, boundary: `encoder` | `decoder` = `encoder`, legacy = false, + versionTwo = false, ) { const row = (index: number, revision: number, payload: Value): Row => ({ id: `row:${index}`, @@ -222,7 +249,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, @@ -253,7 +280,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) @@ -273,6 +300,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 @@ -315,6 +344,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)( @@ -351,7 +1006,7 @@ it.each([`modified`, `original`, `changes`] as const)( JSON.stringify({ id: `bad`, createdAt: new Date(0).toISOString(), - valueEncoding: 2, + valueEncoding: 3, mutations: [mutation], }), ), @@ -479,6 +1134,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( ( [ @@ -507,7 +1191,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), diff --git a/packages/offline-transactions/tests/transaction-settlement.property.test.ts b/packages/offline-transactions/tests/transaction-settlement.property.test.ts index 539d77741d..fcc46a5d58 100644 --- a/packages/offline-transactions/tests/transaction-settlement.property.test.ts +++ b/packages/offline-transactions/tests/transaction-settlement.property.test.ts @@ -623,3 +623,147 @@ it(`rejects only the transaction whose durable admission fails`, async () => { { seed: 20260916, numRuns: 10 }, ) }) + +it(`fulfills successful provider work when durable acknowledgement cleanup fails`, async () => { + const deletionAttempted = gate() + const storageError = new Error(`acknowledgement cleanup failed`) + class Storage extends FakeStorageAdapter { + override async delete(key: string): Promise { + if (key.startsWith(`tx:`)) { + deletionAttempted.resolve() + throw storageError + } + await super.delete(key) + } + } + const env = createTestOfflineEnvironment({ storage: new Storage() }) + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + let status: unknown = `pending` + let transactionId = `` + let observed: Promise | undefined + let hasPrimaryFailure = false + try { + await env.waitForLeader() + const transaction = env.executor.createOfflineTransaction({ + mutationFnName: env.mutationFnName, + autoCommit: false, + }) + transactionId = transaction.id + transaction.mutate(() => + env.collection.insert({ + id: `successful-cleanup-failure`, + value: `provider-applied`, + completed: false, + updatedAt: new Date(0), + }), + ) + observed = transaction.commit().then( + () => { + status = `fulfilled` + }, + (error: unknown) => { + status = error + }, + ) + + await atOracleCheckpoint( + deletionAttempted.promise, + `successful acknowledgement cleanup attempted`, + ) + await turn() + + expect(status).toBe(`fulfilled`) + expect((await env.executor.peekOutbox()).map(({ id }) => id)).toEqual([ + transaction.id, + ]) + } catch (error) { + hasPrimaryFailure = true + throw error + } finally { + if (status === `pending` && transactionId) + env.executor.resolveTransaction(transactionId, undefined) + await cleanupOfflineOracle( + [ + () => observed, + () => env.executor.dispose(), + () => env.collection.cleanup(), + () => warning.mockRestore(), + ], + hasPrimaryFailure, + ) + } +}) + +it(`preserves permanent provider failure when rejection cleanup also fails`, async () => { + const deletionAttempted = gate() + const primaryError = new NonRetriableError(`provider rejected permanently`) + const storageError = new Error(`rejection cleanup failed`) + class Storage extends FakeStorageAdapter { + override async delete(key: string): Promise { + if (key.startsWith(`tx:`)) { + deletionAttempted.resolve() + throw storageError + } + await super.delete(key) + } + } + const env = createTestOfflineEnvironment({ + storage: new Storage(), + mutationFn: () => Promise.reject(primaryError), + }) + const warning = vi.spyOn(console, `warn`).mockImplementation(() => {}) + let status: unknown = `pending` + let transactionId = `` + let observed: Promise | undefined + let hasPrimaryFailure = false + try { + await env.waitForLeader() + const transaction = env.executor.createOfflineTransaction({ + mutationFnName: env.mutationFnName, + autoCommit: false, + }) + transactionId = transaction.id + transaction.mutate(() => + env.collection.insert({ + id: `permanent-cleanup-failure`, + value: `optimistic`, + completed: false, + updatedAt: new Date(0), + }), + ) + observed = transaction.commit().then( + () => { + status = `fulfilled` + }, + (error: unknown) => { + status = error + }, + ) + + await atOracleCheckpoint( + deletionAttempted.promise, + `permanent rejection cleanup attempted`, + ) + await turn() + + expect(status).toBe(primaryError) + expect((await env.executor.peekOutbox()).map(({ id }) => id)).toEqual([ + transaction.id, + ]) + } catch (error) { + hasPrimaryFailure = true + throw error + } finally { + if (status === `pending` && transactionId) + env.executor.rejectTransaction(transactionId, primaryError) + await cleanupOfflineOracle( + [ + () => observed, + () => env.executor.dispose(), + () => env.collection.cleanup(), + () => warning.mockRestore(), + ], + hasPrimaryFailure, + ) + } +})