diff --git a/.changeset/confirm-write-hook.md b/.changeset/confirm-write-hook.md new file mode 100644 index 0000000000..1cb30bc461 --- /dev/null +++ b/.changeset/confirm-write-hook.md @@ -0,0 +1,11 @@ +--- +'@tanstack/offline-transactions': minor +--- + +Add an opt-in `OfflineConfig.confirmWrite` hook that holds optimistic state across the post-commit confirmation window **off** the serial drain path. + +Previously the only way to keep a row painted until an async sync stream (e.g. ElectricSQL's `awaitTxId`) echoed the write back was to `await` that confirmation inside the `mutationFn` — which serializes the whole outbox and collapses drain throughput. `confirmWrite` runs after the write commits and its outbox entry is removed, while the library keeps the just-committed mutations' optimistic overlay painted (reusing the same hold primitive as `restoreOptimisticState`) and releases it when the hook settles. The serial chain still serializes the POSTs (preserving create-then-update ordering); only the confirmation moved off it. + +The hook applies consistently to leader, non-leader, and storage-degraded online-only writes. It is never expected to roll back — the write is already durably committed, so a rejection just drops the overlay early (a possible brief flicker), never data loss. A `maxConfirmationHolds` cap (default 1000) bounds concurrent holds to avoid O(n²) optimistic recompute on a large, fast drain, and `getActiveConfirmationHoldCount()` exposes the live count for diagnostics. Setting the cap to 0 still runs the hook without retaining an overlay. + +As part of this, the `mutationFn`'s return value is now threaded through to the completion promise and to `confirmWrite` (e.g. a server-assigned txid); previously it was awaited and discarded. diff --git a/packages/offline-transactions/README.md b/packages/offline-transactions/README.md index d73c1b20ca..b55cf341ae 100644 --- a/packages/offline-transactions/README.md +++ b/packages/offline-transactions/README.md @@ -113,6 +113,41 @@ Transactions are processed one at a time in the order they were created: - **Dependency safety**: Avoids conflicts between transactions that may reference each other - **Predictable behavior**: Transactions complete in the exact order they were created +### Post-commit Confirmation + +Some sync engines do not echo a committed write back immediately. Waiting for +that echo inside a `mutationFn` keeps optimistic state visible, but also blocks +the serial outbox. Use `confirmWrite` to move that wait off the drain path: + +```typescript +const electricById = new Map([[todoCollection.id, todoCollection]]) + +const offline = startOfflineExecutor({ + collections: { todos: todoCollection }, + mutationFns: { + syncTodos: async ({ transaction, idempotencyKey }) => { + // Return the server result needed by confirmWrite. + return api.saveBatch(transaction.mutations, { idempotencyKey }) + }, + }, + confirmWrite: async ({ mutations, result }) => { + const { txid } = result as { txid: number } + const collectionIds = new Set(mutations.map((m) => m.collection.id)) + + await Promise.all( + [...collectionIds].map((id) => + electricById.get(id)?.utils.awaitTxId(txid), + ), + ) + }, +}) +``` + +The hook runs for leader, non-leader, and online-only writes. While it is +pending, the affected optimistic rows remain visible; its completion never +blocks the next outbox write. Set `maxConfirmationHolds: 0` to run the hook +without retaining overlays. + ## API Reference ### startOfflineExecutor(config) @@ -129,6 +164,8 @@ interface OfflineConfig { beforeRetry?: (transactions: OfflineTransaction[]) => OfflineTransaction[] onUnknownMutationFn?: (name: string, tx: OfflineTransaction) => void onLeadershipChange?: (isLeader: boolean) => void + confirmWrite?: (context: ConfirmWriteContext) => void | Promise + maxConfirmationHolds?: number onlineDetector?: OnlineDetector } ``` @@ -145,6 +182,7 @@ interface OfflineConfig { - `waitForTransactionCompletion(id)` - Wait for a specific transaction to complete - `removeFromOutbox(id)` - Manually remove transaction from outbox - `peekOutbox()` - View all pending transactions +- `getActiveConfirmationHoldCount()` - Count post-commit optimistic holds - `dispose()` - Clean up resources ### Error Handling diff --git a/packages/offline-transactions/skills/offline/SKILL.md b/packages/offline-transactions/skills/offline/SKILL.md index 1295640dc6..5161c7ea0d 100644 --- a/packages/offline-transactions/skills/offline/SKILL.md +++ b/packages/offline-transactions/skills/offline/SKILL.md @@ -112,7 +112,8 @@ If the executor is not the leader tab, falls back to `createOptimisticAction` di 1. Mutation applied optimistically to collection (instant UI update) 2. Transaction serialized and persisted to storage (outbox) 3. Leader tab picks up transaction and executes `mutationFn` -4. On success: removed from outbox, optimistic state resolved +4. On success: removed from outbox; optional `confirmWrite` runs off the drain + while a temporary hold keeps optimistic state visible 5. On failure: retried with exponential backoff 6. On page reload: outbox replayed, optimistic state restored @@ -165,6 +166,8 @@ interface OfflineConfig { maxConcurrency?: number // Parallel execution limit jitter?: boolean // Add jitter to retry delays beforeRetry?: (txs) => txs // Transform/filter before retry + confirmWrite?: (context) => void | Promise // Post-commit sync confirmation + maxConfirmationHolds?: number // Optimistic hold cap (default: 1000) onUnknownMutationFn?: (name, tx) => void // Handle orphaned transactions onLeadershipChange?: (isLeader) => void // Leadership state callback onStorageFailure?: (diagnostic) => void // Storage probe failure callback @@ -173,6 +176,32 @@ interface OfflineConfig { } ``` +### Post-commit sync confirmation + +Use `confirmWrite` when a server write commits before an asynchronous sync +stream echoes it back. The hook receives the `mutationFn` result, runs for +leader and online-only paths, and does not block the serial outbox: + +```ts +const executor = startOfflineExecutor({ + // ... + mutationFns: { + syncTodos: ({ transaction, idempotencyKey }) => + api.sync(transaction.mutations, { idempotencyKey }), + }, + confirmWrite: async ({ mutations, result }) => { + const { txid } = result as { txid: number } + const collectionIds = new Set(mutations.map((m) => m.collection.id)) + await Promise.all( + [...collectionIds].map((id) => collections[id].utils.awaitTxId(txid)), + ) + }, +}) +``` + +The library holds the affected optimistic rows until the hook settles. A +rejection releases the hold but never rolls back the already-committed write. + ### Custom storage adapter ```ts diff --git a/packages/offline-transactions/src/OfflineExecutor.ts b/packages/offline-transactions/src/OfflineExecutor.ts index a6140cfebc..122b1d9aff 100644 --- a/packages/offline-transactions/src/OfflineExecutor.ts +++ b/packages/offline-transactions/src/OfflineExecutor.ts @@ -11,6 +11,7 @@ import { LocalStorageAdapter } from './storage/LocalStorageAdapter' import { OutboxManager } from './outbox/OutboxManager' import { KeyScheduler } from './executor/KeyScheduler' import { TransactionExecutor } from './executor/TransactionExecutor' +import { ConfirmationManager } from './executor/ConfirmationManager' // Coordination import { WebLocksLeader } from './coordination/WebLocksLeader' @@ -48,6 +49,7 @@ export class OfflineExecutor { private outbox: OutboxManager | null private scheduler: KeyScheduler private executor: TransactionExecutor | null + private confirmationManager: ConfirmationManager private leaderElection: LeaderElection | null private onlineDetector: OnlineDetector private isLeaderState = false @@ -74,11 +76,15 @@ export class OfflineExecutor { > = new Map() // Track restoration transactions for cleanup when offline transactions complete - private restorationTransactions: Map = new Map() + private restorationTransactions: Map< + string, + { transaction: Transaction; release: () => void } + > = new Map() constructor(config: OfflineConfig) { this.config = config this.scheduler = new KeyScheduler() + this.confirmationManager = new ConfirmationManager(config) this.onlineDetector = config.onlineDetector ?? new WebOnlineDetector() // Initialize as pending - will be set by async initialization @@ -290,6 +296,7 @@ export class OfflineExecutor { this.outbox, this.config, this, + this.confirmationManager, ) this.leaderElection = this.createLeaderElection() @@ -366,13 +373,25 @@ export class OfflineExecutor { if (!this.isOfflineEnabled) { // Non-leader: use createTransaction directly with the resolved mutation function // We need to wrap it to add the idempotency key + const idempotencyKey = options.idempotencyKey ?? safeRandomUUID() return createTransaction({ + id: options.id, autoCommit: options.autoCommit ?? true, - mutationFn: (params) => - mutationFn({ + mutationFn: async (params) => { + const result = await mutationFn({ ...params, - idempotencyKey: options.idempotencyKey || safeRandomUUID(), - }), + idempotencyKey, + }) + this.confirmationManager.schedule({ + transactionId: params.transaction.id, + mutationFnName: options.mutationFnName, + idempotencyKey, + mutations: params.transaction.mutations, + result, + metadata: params.transaction.metadata, + }) + return result + }, metadata: options.metadata, }) } @@ -398,13 +417,24 @@ export class OfflineExecutor { // Check leadership when action is called, not when it's created if (!this.isOfflineEnabled) { // Non-leader: use createOptimisticAction directly + const idempotencyKey = safeRandomUUID() const action = createOptimisticAction({ - mutationFn: (vars, params) => - mutationFn({ + mutationFn: async (vars, params) => { + const result = await mutationFn({ ...vars, ...params, - idempotencyKey: safeRandomUUID(), - }), + idempotencyKey, + }) + this.confirmationManager.schedule({ + transactionId: params.transaction.id, + mutationFnName: options.mutationFnName, + idempotencyKey, + mutations: params.transaction.mutations, + result, + metadata: params.transaction.metadata, + }) + return result + }, onMutate: options.onMutate, }) return action(variables) @@ -498,56 +528,48 @@ export class OfflineExecutor { this.pendingTransactionPromises.delete(transactionId) } - // Clean up the restoration transaction and rollback optimistic state - this.cleanupRestorationTransaction(transactionId, true) + // Drop the synthetic restoration hold. The real transaction handles its + // own failure state; the hold must not cascade a rollback. + this.cleanupRestorationTransaction(transactionId) } // Method for TransactionExecutor to register restoration transactions registerRestorationTransaction( offlineTransactionId: string, restorationTransaction: Transaction, + releaseRestorationTransaction: () => void, ): void { - this.restorationTransactions.set( - offlineTransactionId, - restorationTransaction, - ) + // Replaying the same outbox id twice must not orphan the previous hold. + this.cleanupRestorationTransaction(offlineTransactionId) + this.restorationTransactions.set(offlineTransactionId, { + transaction: restorationTransaction, + release: releaseRestorationTransaction, + }) } - private cleanupRestorationTransaction( - transactionId: string, - shouldRollback = false, - ): void { - const restorationTx = this.restorationTransactions.get(transactionId) - if (!restorationTx) { + private cleanupRestorationTransaction(transactionId: string): void { + const restoration = this.restorationTransactions.get(transactionId) + if (!restoration) { return } this.restorationTransactions.delete(transactionId) - - if (shouldRollback) { - restorationTx.rollback() - return + try { + // A restoration transaction is only an optimistic display hold. Releasing + // it is correct for both success and permanent failure and cannot cascade + // rollback into unrelated user transactions. + restoration.release() + } catch (error) { + console.warn( + `Failed to release restoration transaction ${restoration.transaction.id}:`, + error, + ) } + } - // Mark as completed so recomputeOptimisticState removes it from consideration. - // The actual data will come from the sync. - restorationTx.setState(`completed`) - - // Remove from each collection's transaction map and recompute - const touchedCollections = new Set() - for (const mutation of restorationTx.mutations) { - // Defensive check for corrupted deserialized data - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (!mutation.collection) { - continue - } - const collectionId = mutation.collection.id - if (touchedCollections.has(collectionId)) { - continue - } - touchedCollections.add(collectionId) - mutation.collection._state.transactions.delete(restorationTx.id) - mutation.collection._state.recomputeOptimisticState(false) + private releaseRestorationTransactions(): void { + for (const transactionId of [...this.restorationTransactions.keys()]) { + this.cleanupRestorationTransaction(transactionId) } } @@ -566,6 +588,8 @@ export class OfflineExecutor { } async clearOutbox(): Promise { + this.confirmationManager.releaseAll() + this.releaseRestorationTransactions() if (!this.outbox || !this.executor) { return } @@ -587,6 +611,14 @@ export class OfflineExecutor { return this.executor.getRunningCount() } + /** + * Number of optimistic holds currently kept alive by `confirmWrite` (see + * `OfflineConfig.confirmWrite`). Returns 0 when the hook is unused. + */ + getActiveConfirmationHoldCount(): number { + return this.confirmationManager.getActiveHoldCount() + } + getOnlineDetector(): OnlineDetector { return this.onlineDetector } @@ -596,6 +628,11 @@ export class OfflineExecutor { } dispose(): void { + // Drop any optimistic holds still waiting on confirmation so they don't + // outlive the executor. + this.confirmationManager.dispose() + this.releaseRestorationTransactions() + for (const collection of Object.values(this.config.collections)) { collection.deferDataRefresh = null } diff --git a/packages/offline-transactions/src/api/OfflineTransaction.ts b/packages/offline-transactions/src/api/OfflineTransaction.ts index af13484a75..2ce8ffeeb6 100644 --- a/packages/offline-transactions/src/api/OfflineTransaction.ts +++ b/packages/offline-transactions/src/api/OfflineTransaction.ts @@ -23,7 +23,7 @@ export class OfflineTransaction { persistTransaction: (tx: OfflineTransactionType) => Promise, executor: any, ) { - this.offlineId = safeRandomUUID() + this.offlineId = options.id ?? safeRandomUUID() this.mutationFnName = options.mutationFnName this.autoCommit = options.autoCommit ?? true this.idempotencyKey = options.idempotencyKey ?? safeRandomUUID() diff --git a/packages/offline-transactions/src/executor/ConfirmationManager.ts b/packages/offline-transactions/src/executor/ConfirmationManager.ts new file mode 100644 index 0000000000..8489510204 --- /dev/null +++ b/packages/offline-transactions/src/executor/ConfirmationManager.ts @@ -0,0 +1,117 @@ +import { createOptimisticHold } from './OptimisticHold' +import type { OptimisticHold } from './OptimisticHold' +import type { ConfirmWriteContext, OfflineConfig } from '../types' + +const DEFAULT_MAX_CONFIRMATION_HOLDS = 1000 + +type ConfirmationConfig = Pick< + OfflineConfig, + `confirmWrite` | `maxConfirmationHolds` +> + +function getMaxConfirmationHolds(configured: number | undefined): number { + if (configured === undefined) { + return DEFAULT_MAX_CONFIRMATION_HOLDS + } + + if (!Number.isFinite(configured) || configured < 0) { + return DEFAULT_MAX_CONFIRMATION_HOLDS + } + + return Math.floor(configured) +} + +/** + * Runs post-commit confirmation hooks outside the serial outbox path while + * keeping the committed mutations visible until each hook settles. + */ +export class ConfirmationManager { + private readonly holds = new Set() + private disposed = false + + constructor(private readonly config: ConfirmationConfig) {} + + /** + * Schedule confirmation without throwing into the caller's already-committed + * write path. The hook still runs when a hold cannot be created or is capped. + */ + schedule(context: ConfirmWriteContext): void { + if (this.disposed) { + return + } + + const confirmWrite = this.config.confirmWrite + if (!confirmWrite) { + return + } + + const hold = this.tryCreateHold(context) + + // Start on a microtask so confirmation never blocks the serial drain. A + // synchronous throw and an async rejection have the same guarded behavior. + void Promise.resolve() + .then(() => confirmWrite(context)) + .catch((error) => { + console.warn( + `confirmWrite rejected for ${context.transactionId}:`, + error, + ) + }) + .finally(() => { + this.releaseHold(hold) + }) + } + + /** Release every optimistic hold, for example during clear or dispose. */ + releaseAll(): void { + for (const hold of [...this.holds]) { + this.releaseHold(hold) + } + } + + /** Permanently stop accepting confirmations and release every active hold. */ + dispose(): void { + this.disposed = true + this.releaseAll() + } + + /** Number of active holds currently keeping optimistic state visible. */ + getActiveHoldCount(): number { + return this.holds.size + } + + private tryCreateHold(context: ConfirmWriteContext): OptimisticHold | null { + if (context.mutations.length === 0) { + return null + } + + const maxHolds = getMaxConfirmationHolds(this.config.maxConfirmationHolds) + if (this.holds.size >= maxHolds) { + return null + } + + try { + const hold = createOptimisticHold(context.mutations) + this.holds.add(hold) + return hold + } catch (error) { + // The write is already committed. Failure to paint a hold may cause a + // brief flicker, but must never retry or roll back the durable write. + console.warn(`Failed to create confirmation hold:`, error) + return null + } + } + + private releaseHold(hold: OptimisticHold | null): void { + if (!hold) { + return + } + + this.holds.delete(hold) + try { + hold.release() + } catch (error) { + console.warn(`Failed to release confirmation hold:`, error) + } + } +} diff --git a/packages/offline-transactions/src/executor/OptimisticHold.ts b/packages/offline-transactions/src/executor/OptimisticHold.ts new file mode 100644 index 0000000000..62e6533f5a --- /dev/null +++ b/packages/offline-transactions/src/executor/OptimisticHold.ts @@ -0,0 +1,122 @@ +import { createTransaction } from '@tanstack/db' +import type { Collection, PendingMutation, Transaction } from '@tanstack/db' + +/** + * A standalone, never-committed transaction whose only job is to keep an + * optimistic overlay painted on the affected collections for a bounded window. + * + * This is the same primitive `restoreOptimisticState` uses to re-show pending + * writes after a reload. It is factored out here so the post-commit + * confirmation window (see `OfflineConfig.confirmWrite`) can reuse it without + * duplicating the `_state` bookkeeping. + */ +export interface OptimisticHold { + /** The underlying hold transaction. Never auto-commits. */ + transaction: Transaction + /** Tear the hold down. Idempotent. */ + release: () => void +} + +/** + * Create an optimistic hold for `mutations` and register it on every touched + * collection synchronously (before returning), so the overlay is painted with + * no gap. The returned `release` removes it again. + * + * Mirrors the lifecycle the offline executor already drives for restoration + * transactions: `setState("completed")` + delete + `recomputeOptimisticState` + * on a normal release, or `rollback()` to discard. + */ +export function createOptimisticHold( + mutations: Array, + options: { id?: string } = {}, +): OptimisticHold { + // `autoCommit: false` + an inert mutationFn means it never POSTs or settles on + // its own — the caller drives its lifecycle by hand via `release`. + const transaction = createTransaction({ + ...(options.id === undefined ? {} : { id: options.id }), + autoCommit: false, + mutationFn: async () => {}, + }) + + // It never commits, so `isPersisted` never resolves through the normal flow; + // swallow so a stray rejection on teardown can't surface as an unhandled + // rejection. Mirrors `restoreOptimisticState`. + transaction.isPersisted.promise.catch(() => { + // Intentionally ignored - holds are torn down via `release`, not commit. + }) + + // Register with each affected collection's state manager. Dedup by collection + // reference (the same collection can be touched by several mutations). + const touchedCollections = new Set>() + let released = false + const release = (): void => { + if (released) { + return + } + released = true + + const errors: Array = [] + try { + // A hold is synthetic: completing it only removes it from TanStack DB's + // module-global registry. It must never rollback and cascade into real + // user transactions that happen to touch the same keys. + transaction.setState(`completed`) + } catch (error) { + errors.push(error) + } + + // Delete every registration before recomputing. If one collection throws, + // later collections are still cleaned up and cannot retain a leaked hold. + for (const collection of touchedCollections) { + try { + collection._state.transactions.delete(transaction.id) + } catch (error) { + errors.push(error) + } + } + for (const collection of touchedCollections) { + try { + collection._state.recomputeOptimisticState(false) + } catch (error) { + errors.push(error) + } + } + + if (errors.length > 0) { + throw errors[0] + } + } + + try { + transaction.applyMutations(mutations) + + for (const mutation of mutations) { + // Defensive check for corrupted deserialized data + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (!mutation.collection) { + continue + } + if (touchedCollections.has(mutation.collection)) { + continue + } + // Track before registration so a throwing set/recompute is unwound too. + touchedCollections.add(mutation.collection) + mutation.collection._state.transactions.set(transaction.id, transaction) + // `recomputeOptimisticState(true)` forces the recompute through even when + // a sync commit is in flight (the "triggered by user action" path), so the + // overlay always applies. + mutation.collection._state.recomputeOptimisticState(true) + } + } catch (error) { + // Failure-atomic creation: remove any registrations installed before the + // throw and complete the synthetic transaction so no global state leaks. + try { + release() + } catch { + // Preserve the original registration error after best-effort cleanup. + } + throw error + } + + return { transaction, release } +} diff --git a/packages/offline-transactions/src/executor/TransactionExecutor.ts b/packages/offline-transactions/src/executor/TransactionExecutor.ts index 71328443fa..c2686c5d2e 100644 --- a/packages/offline-transactions/src/executor/TransactionExecutor.ts +++ b/packages/offline-transactions/src/executor/TransactionExecutor.ts @@ -1,7 +1,8 @@ -import { createTransaction } from '@tanstack/db' import { DefaultRetryPolicy } from '../retry/RetryPolicy' import { NonRetriableError } from '../types' import { withNestedSpan } from '../telemetry/tracer' +import { createOptimisticHold } from './OptimisticHold' +import type { ConfirmationManager } from './ConfirmationManager' import type { KeyScheduler } from './KeyScheduler' import type { OutboxManager } from '../outbox/OutboxManager' import type { @@ -20,6 +21,7 @@ export class TransactionExecutor { private isExecuting = false private executionPromise: Promise | null = null private offlineExecutor: TransactionSignaler + private confirmationManager: ConfirmationManager private retryTimer: ReturnType | null = null constructor( @@ -27,6 +29,7 @@ export class TransactionExecutor { outbox: OutboxManager, config: OfflineConfig, offlineExecutor: TransactionSignaler, + confirmationManager: ConfirmationManager, ) { this.scheduler = scheduler this.outbox = outbox @@ -36,6 +39,7 @@ export class TransactionExecutor { config.jitter ?? true, ) this.offlineExecutor = offlineExecutor + this.confirmationManager = confirmationManager } async execute(transaction: OfflineTransaction): Promise { @@ -104,7 +108,7 @@ export class TransactionExecutor { await this.outbox.remove(transaction.id) span.setAttribute(`result`, `success`) - this.offlineExecutor.resolveTransaction(transaction.id, result) + this.resolveCommittedTransaction(transaction, result) } catch (error) { const err = error instanceof Error ? error : new Error(String(error)) @@ -129,7 +133,9 @@ export class TransactionExecutor { } } - private async runMutationFn(transaction: OfflineTransaction): Promise { + private async runMutationFn( + transaction: OfflineTransaction, + ): Promise { const mutationFn = this.config.mutationFns[transaction.mutationFnName] if (!mutationFn) { @@ -150,12 +156,44 @@ export class TransactionExecutor { metadata: transaction.metadata ?? {}, } - await mutationFn({ + // Return the result so it can be surfaced to the waiting transaction and to + // `confirmWrite` (e.g. a server-assigned txid). Previously this value was + // awaited and discarded. + return await mutationFn({ transaction: transactionWithMutations as any, idempotencyKey: transaction.idempotencyKey, }) } + /** + * Hand a committed write to the confirmation manager before resolving its + * original transaction, so the optimistic overlay is owned continuously. + */ + private resolveCommittedTransaction( + transaction: OfflineTransaction, + result: unknown, + ): void { + this.confirmationManager.schedule({ + transactionId: transaction.id, + mutationFnName: transaction.mutationFnName, + idempotencyKey: transaction.idempotencyKey, + mutations: transaction.mutations, + result, + metadata: transaction.metadata, + }) + + try { + this.offlineExecutor.resolveTransaction(transaction.id, result) + } catch (error) { + // The outbox entry is already removed and the server write committed. + // Cleanup failures must not re-enter retry handling and resend the write. + console.warn( + `Failed to resolve committed transaction ${transaction.id}:`, + error, + ) + } + } + private async handleError( transaction: OfflineTransaction, error: Error, @@ -270,47 +308,18 @@ export class TransactionExecutor { } try { - // Create a restoration transaction that holds mutations for optimistic state display. - // It will never commit - the real mutation is handled by the offline executor. - const restorationTx = createTransaction({ + // Hold the mutations for optimistic display while the write is pending. + // It will never commit - the real mutation is handled by the offline + // executor, which tears the hold down via cleanupRestorationTransaction + // (keyed by the offline transaction id) once the write resolves. + const hold = createOptimisticHold(offlineTx.mutations, { id: offlineTx.id, - autoCommit: false, - mutationFn: async () => {}, }) - // Prevent unhandled promise rejection when cleanup calls rollback() - // We don't care about this promise - it's just for holding mutations - restorationTx.isPersisted.promise.catch(() => { - // Intentionally ignored - restoration transactions are cleaned up - // via cleanupRestorationTransaction, not through normal commit flow - }) - - restorationTx.applyMutations(offlineTx.mutations) - - // Register with each affected collection's state manager - const touchedCollections = new Set() - for (const mutation of offlineTx.mutations) { - // Defensive check for corrupted deserialized data - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (!mutation.collection) { - continue - } - const collectionId = mutation.collection.id - if (touchedCollections.has(collectionId)) { - continue - } - touchedCollections.add(collectionId) - - mutation.collection._state.transactions.set( - restorationTx.id, - restorationTx, - ) - mutation.collection._state.recomputeOptimisticState(true) - } - this.offlineExecutor.registerRestorationTransaction( offlineTx.id, - restorationTx, + hold.transaction, + hold.release, ) } catch (error) { console.warn( @@ -324,6 +333,7 @@ export class TransactionExecutor { clear(): void { this.scheduler.clear() this.clearRetryTimer() + this.confirmationManager.releaseAll() } getPendingCount(): number { diff --git a/packages/offline-transactions/src/index.ts b/packages/offline-transactions/src/index.ts index f43f4633a1..8edb6a3669 100644 --- a/packages/offline-transactions/src/index.ts +++ b/packages/offline-transactions/src/index.ts @@ -5,6 +5,7 @@ export { OfflineExecutor, startOfflineExecutor } from './OfflineExecutor' export type { OfflineTransaction, OfflineConfig, + ConfirmWriteContext, OfflineMode, StorageAdapter, StorageDiagnostic, diff --git a/packages/offline-transactions/src/react-native/index.ts b/packages/offline-transactions/src/react-native/index.ts index 977b1e93f5..0726281458 100644 --- a/packages/offline-transactions/src/react-native/index.ts +++ b/packages/offline-transactions/src/react-native/index.ts @@ -3,6 +3,7 @@ export { // Types type OfflineTransaction, type OfflineConfig, + type ConfirmWriteContext, type OfflineMode, type StorageAdapter, type StorageDiagnostic, diff --git a/packages/offline-transactions/src/types.ts b/packages/offline-transactions/src/types.ts index e18a287cbe..12e5c690d3 100644 --- a/packages/offline-transactions/src/types.ts +++ b/packages/offline-transactions/src/types.ts @@ -89,6 +89,24 @@ export interface StorageDiagnostic { error?: Error } +export interface ConfirmWriteContext { + /** Id of the offline transaction whose write just committed. */ + transactionId: string + /** Name of the mutation function that committed the write. */ + mutationFnName: string + /** Stable idempotency key used for the committed write. */ + idempotencyKey: string + /** + * The mutations that were committed. One optimistic overlay is held per + * touched collection until the hook settles. + */ + mutations: Array + /** Whatever the matching mutationFn resolved with (e.g. a server txid). */ + result: unknown + /** The transaction's metadata, if any was supplied when it was created. */ + metadata?: Record +} + export interface OfflineConfig { collections: Record> mutationFns: Record @@ -101,6 +119,34 @@ export interface OfflineConfig { onUnknownMutationFn?: (name: string, tx: OfflineTransaction) => void onLeadershipChange?: (isLeader: boolean) => void onStorageFailure?: (diagnostic: StorageDiagnostic) => void + /** + * Optional post-commit confirmation hook. Runs AFTER a transaction's + * mutationFn resolves and its outbox entry is removed, but OFF the serial + * drain path — it does NOT block the next transaction's mutationFn, so a slow + * confirmation never throttles drain throughput. + * + * While the returned promise is pending, the library keeps the just-committed + * mutations' optimistic state painted (via an internal hold transaction), then + * releases the hold when it settles (resolve OR reject). Use it to wait for an + * asynchronous sync stream to echo the write back — e.g. ElectricSQL's + * `awaitTxId` — so the affected rows don't flicker (disappear then reappear) + * in the gap between server commit and sync. + * + * The hook is never expected to roll back: the write is already durably + * committed server-side, so a rejection only means the optimistic overlay is + * dropped early (a possible brief flicker), never data loss. Implement any + * timeout / verify-by-state logic inside the hook and resolve when done. + */ + confirmWrite?: (context: ConfirmWriteContext) => void | Promise + /** + * Safety cap on simultaneously-held confirmation holds (see `confirmWrite`). + * Each hold adds one transaction to every touched collection's optimistic + * recompute, which is O(transactions). Beyond the cap the hold is skipped (the + * overlay drops at commit instead) to avoid O(n^2) churn on a large, fast + * drain. Set to 0 to run the hook without retaining optimistic holds. + * Non-finite or negative values fall back to the default of 1000. + */ + maxConfirmationHolds?: number leaderElection?: LeaderElection /** * Custom online detector implementation. @@ -136,6 +182,7 @@ export interface TransactionSignaler { registerRestorationTransaction: ( offlineTransactionId: string, restorationTransaction: Transaction, + releaseRestorationTransaction: () => void, ) => void isOnline: () => boolean } diff --git a/packages/offline-transactions/tests/confirm-write.test.ts b/packages/offline-transactions/tests/confirm-write.test.ts new file mode 100644 index 0000000000..f75fb1933e --- /dev/null +++ b/packages/offline-transactions/tests/confirm-write.test.ts @@ -0,0 +1,352 @@ +import { describe, expect, it } from 'vitest' +import { createTestOfflineEnvironment } from './harness' +import type { + ConfirmWriteContext, + CreateOfflineTransactionOptions, + OfflineConfig, +} from '../src/types' + +const flushMicrotasks = () => new Promise((resolve) => setTimeout(resolve, 0)) + +function deferred(): { + promise: Promise + resolve: (value: T) => void + reject: (error: unknown) => void +} { + let resolve!: (value: T) => void + let reject!: (error: unknown) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + return { promise, resolve, reject } +} + +/** + * A mutationFn that "commits on the server" but never feeds the row back into + * the collection's synced stream. This reproduces the real-world gap the + * `confirmWrite` hook targets: the write is durable server-side, but the sync + * stream hasn't echoed it back yet — so the optimistic overlay is the ONLY + * thing keeping the row visible. With the hook the row stays until the hook + * settles; without it the row vanishes the instant the transaction resolves. + */ +const committedButNotSynced = async () => ({ txid: 42 }) + +async function insertAndCommit( + env: ReturnType, + id: string, + options: Partial = {}, +) { + const offlineTx = env.executor.createOfflineTransaction({ + ...options, + mutationFnName: env.mutationFnName, + autoCommit: false, + }) + offlineTx.mutate(() => { + env.collection.insert({ + id, + value: id, + completed: false, + updatedAt: new Date(), + }) + }) + await offlineTx.commit() + return offlineTx +} + +describe(`OfflineConfig.confirmWrite`, () => { + it(`holds optimistic state past commit until the hook settles, then releases`, async () => { + const gate = deferred() + const calls: Array = [] + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: (context) => { + calls.push(context) + return gate.promise + }, + }, + }) + await env.waitForLeader() + + const offlineTx = await insertAndCommit(env, `item-1`, { + id: `offline-item-1`, + idempotencyKey: `idempotency-item-1`, + metadata: { source: `test` }, + }) + await flushMicrotasks() + + // The server committed but the sync stream never delivered the row, so the + // ONLY thing keeping it visible is the confirmation hold. It must be there. + expect(env.executor.getActiveConfirmationHoldCount()).toBe(1) + expect(env.collection.get(`item-1`)?.value).toBe(`item-1`) + + // The hook received the committed mutations and the mutationFn's result. + expect(calls).toHaveLength(1) + expect(calls[0]!.transactionId).toBe(offlineTx.id) + expect(calls[0]!.mutationFnName).toBe(env.mutationFnName) + expect(calls[0]!.idempotencyKey).toBe(`idempotency-item-1`) + expect(calls[0]!.mutations).toHaveLength(1) + expect(calls[0]!.result).toEqual({ txid: 42 }) + expect(calls[0]!.metadata).toEqual({ source: `test` }) + + // Settle the hook → hold released. With nothing in the synced stream, the + // optimistic row now drops (in production the sync stream would have it). + gate.resolve() + await flushMicrotasks() + + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`item-1`)).toBeUndefined() + + env.executor.dispose() + }) + + it(`does not block the serial drain: a hung hook still lets the next write POST`, async () => { + const never = deferred() // confirmWrite that never settles + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: () => never.promise, + }, + }) + await env.waitForLeader() + + await insertAndCommit(env, `item-1`) + await insertAndCommit(env, `item-2`) + await flushMicrotasks() + + // Both mutationFns ran even though the first hook never settled — the + // confirmation runs OFF the serial path. Pre-fix, awaiting confirmation + // inline would have parked the queue on the first write. + expect(env.mutationCalls).toHaveLength(2) + expect(env.executor.getActiveConfirmationHoldCount()).toBe(2) + + env.executor.dispose() + }) + + it(`releases the hold even when the hook rejects (write is already committed)`, async () => { + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: () => Promise.reject(new Error(`shape never confirmed`)), + }, + }) + await env.waitForLeader() + + await insertAndCommit(env, `item-1`) + await flushMicrotasks() + + // A rejection is not a rollback: the hold is released, not retried, and the + // drain is unaffected. + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`item-1`)).toBeUndefined() + + env.executor.dispose() + }) + + it(`without the hook, optimistic state drops at commit (the gap the hook fills)`, async () => { + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + }) + await env.waitForLeader() + + await insertAndCommit(env, `item-1`) + await flushMicrotasks() + + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`item-1`)).toBeUndefined() + + env.executor.dispose() + }) + + it(`skips the hold past maxConfirmationHolds (O(n^2) safety valve)`, async () => { + const gate = deferred() + let confirmCalls = 0 + const config: Partial = { + confirmWrite: () => { + confirmCalls += 1 + return gate.promise + }, + maxConfirmationHolds: 0, + } + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config, + }) + await env.waitForLeader() + + await insertAndCommit(env, `item-1`) + await flushMicrotasks() + + // Cap is 0, so no hold is created — the write still succeeds, the overlay + // just drops at commit as it would without the hook. + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`item-1`)).toBeUndefined() + expect(confirmCalls).toBe(1) + + gate.resolve() + env.executor.dispose() + }) + + it(`runs confirmation for a non-leader direct transaction`, async () => { + const gate = deferred() + const calls: Array = [] + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: (context) => { + calls.push(context) + return gate.promise + }, + }, + }) + await env.waitForLeader() + env.leader.setLeader(false) + + await insertAndCommit(env, `direct-item`, { + id: `direct-transaction`, + idempotencyKey: `direct-idempotency`, + metadata: { mode: `direct` }, + }) + await flushMicrotasks() + + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ + transactionId: `direct-transaction`, + mutationFnName: env.mutationFnName, + idempotencyKey: `direct-idempotency`, + result: { txid: 42 }, + metadata: { mode: `direct` }, + }) + expect(env.executor.getActiveConfirmationHoldCount()).toBe(1) + expect(env.collection.get(`direct-item`)?.value).toBe(`direct-item`) + + gate.resolve() + await flushMicrotasks() + + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`direct-item`)).toBeUndefined() + env.executor.dispose() + }) + + it(`runs confirmation for a non-leader optimistic action`, async () => { + const gate = deferred() + const calls: Array = [] + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: (context) => { + calls.push(context) + return gate.promise + }, + }, + }) + await env.waitForLeader() + env.leader.setLeader(false) + + const action = env.executor.createOfflineAction<{ id: string }>({ + mutationFnName: env.mutationFnName, + onMutate: ({ id }) => { + env.collection.insert({ + id, + value: id, + completed: false, + updatedAt: new Date(), + }) + }, + }) + const tx = action({ id: `action-item` }) + await tx.isPersisted.promise + await flushMicrotasks() + + expect(calls).toHaveLength(1) + expect(calls[0]!.mutationFnName).toBe(env.mutationFnName) + expect(calls[0]!.result).toEqual({ txid: 42 }) + expect(env.executor.getActiveConfirmationHoldCount()).toBe(1) + expect(env.collection.get(`action-item`)?.value).toBe(`action-item`) + + gate.resolve() + await flushMicrotasks() + + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`action-item`)).toBeUndefined() + env.executor.dispose() + }) + + it(`releases confirmation holds when the outbox is cleared`, async () => { + const never = deferred() + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: () => never.promise, + }, + }) + await env.waitForLeader() + + await insertAndCommit(env, `item-1`) + await flushMicrotasks() + expect(env.executor.getActiveConfirmationHoldCount()).toBe(1) + + await env.executor.clearOutbox() + + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + expect(env.collection.get(`item-1`)).toBeUndefined() + env.executor.dispose() + }) + + it(`does not install a hold after the executor is disposed`, async () => { + const server = deferred<{ txid: number }>() + let confirmCalls = 0 + const env = createTestOfflineEnvironment({ + mutationFn: () => server.promise, + config: { + confirmWrite: () => { + confirmCalls += 1 + }, + }, + }) + await env.waitForLeader() + env.leader.setLeader(false) + + const tx = env.executor.createOfflineTransaction({ + mutationFnName: env.mutationFnName, + autoCommit: false, + }) + tx.mutate(() => { + env.collection.insert({ + id: `late-item`, + value: `late-item`, + completed: false, + updatedAt: new Date(), + }) + }) + const commit = tx.commit() + await flushMicrotasks() + + env.executor.dispose() + server.resolve({ txid: 42 }) + await commit + await flushMicrotasks() + + expect(confirmCalls).toBe(0) + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + }) + + it(`releases all holds on dispose`, async () => { + const never = deferred() + const env = createTestOfflineEnvironment({ + mutationFn: committedButNotSynced, + config: { + confirmWrite: () => never.promise, + }, + }) + await env.waitForLeader() + + await insertAndCommit(env, `item-1`) + await flushMicrotasks() + expect(env.executor.getActiveConfirmationHoldCount()).toBe(1) + + env.executor.dispose() + expect(env.executor.getActiveConfirmationHoldCount()).toBe(0) + }) +}) diff --git a/packages/offline-transactions/tests/offline-e2e.test.ts b/packages/offline-transactions/tests/offline-e2e.test.ts index 981315c267..d65a31e835 100644 --- a/packages/offline-transactions/tests/offline-e2e.test.ts +++ b/packages/offline-transactions/tests/offline-e2e.test.ts @@ -87,7 +87,10 @@ describe(`offline executor end-to-end`, () => { await offlineTx.commit() - await expect(waitPromise).resolves.toBeUndefined() + // waitForTransactionCompletion now resolves with the mutationFn's return + // value (previously this value was awaited and discarded, so it always + // resolved `undefined`). The default test mutationFn returns { ok, mutations }. + await expect(waitPromise).resolves.toMatchObject({ ok: true }) const outboxEntries = await env.executor.peekOutbox() expect(outboxEntries).toEqual([]) diff --git a/packages/offline-transactions/tests/optimistic-hold.test.ts b/packages/offline-transactions/tests/optimistic-hold.test.ts new file mode 100644 index 0000000000..28be9af14c --- /dev/null +++ b/packages/offline-transactions/tests/optimistic-hold.test.ts @@ -0,0 +1,103 @@ +import { createCollection, createTransaction } from '@tanstack/db' +import { describe, expect, it, vi } from 'vitest' +import { createOptimisticHold } from '../src/executor/OptimisticHold' +import type { Collection, PendingMutation } from '@tanstack/db' + +interface TestItem { + id: string + value: string +} + +function makeCollection(id: string): Collection { + return createCollection({ + id, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ markReady }) => { + markReady() + }, + }, + }) +} + +async function makeCommittedMutations( + first: Collection, + second: Collection, +): Promise> { + const source = createTransaction({ + autoCommit: false, + mutationFn: async () => {}, + }) + source.mutate(() => { + first.insert({ id: `first`, value: `first` }) + second.insert({ id: `second`, value: `second` }) + }) + await source.commit() + await Promise.resolve() + return source.mutations +} + +describe(`createOptimisticHold`, () => { + it(`unwinds every registration when creation fails partway`, async () => { + const first = makeCollection(`first-collection`) + const second = makeCollection(`second-collection`) + const mutations = await makeCommittedMutations(first, second) + const firstTransactions = new Set(first._state.transactions.keys()) + const secondTransactions = new Set(second._state.transactions.keys()) + + const originalRecompute = second._state.recomputeOptimisticState.bind( + second._state, + ) + const recompute = vi + .spyOn(second._state, `recomputeOptimisticState`) + .mockImplementation((triggeredByUserAction) => { + if (triggeredByUserAction) { + throw new Error(`registration failed`) + } + return originalRecompute(triggeredByUserAction) + }) + + expect(() => createOptimisticHold(mutations)).toThrow(`registration failed`) + + expect(new Set(first._state.transactions.keys())).toEqual(firstTransactions) + expect(new Set(second._state.transactions.keys())).toEqual( + secondTransactions, + ) + expect(first.get(`first`)).toBeUndefined() + expect(second.get(`second`)).toBeUndefined() + recompute.mockRestore() + }) + + it(`cleans later collections when one release recompute throws`, async () => { + const first = makeCollection(`first-release-collection`) + const second = makeCollection(`second-release-collection`) + const mutations = await makeCommittedMutations(first, second) + const hold = createOptimisticHold(mutations) + + expect(first.get(`first`)?.value).toBe(`first`) + expect(second.get(`second`)?.value).toBe(`second`) + + const originalRecompute = first._state.recomputeOptimisticState.bind( + first._state, + ) + const recompute = vi + .spyOn(first._state, `recomputeOptimisticState`) + .mockImplementation((triggeredByUserAction) => { + if (!triggeredByUserAction) { + throw new Error(`release failed`) + } + return originalRecompute(triggeredByUserAction) + }) + + expect(() => hold.release()).toThrow(`release failed`) + + expect(first._state.transactions.has(hold.transaction.id)).toBe(false) + expect(second._state.transactions.has(hold.transaction.id)).toBe(false) + expect(second.get(`second`)).toBeUndefined() + + recompute.mockRestore() + first._state.recomputeOptimisticState(false) + expect(first.get(`first`)).toBeUndefined() + }) +})