diff --git a/.changeset/fix-direct-write-subscriber-throw.md b/.changeset/fix-direct-write-subscriber-throw.md new file mode 100644 index 0000000000..fc95df0668 --- /dev/null +++ b/.changeset/fix-direct-write-subscriber-throw.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +A direct write (`insert`, `update` or `delete` outside a transaction) whose change subscriber throws no longer leaves a pending transaction behind. Before, the write's transaction stayed pending with its optimistic row visible, its handler never ran, and the Collection kept tracking it. The write now rolls back its own transaction, without rolling back other transactions, and rethrows the subscriber's error. A local-only direct write that takes the fast path (the Collection has no handler for that operation and no other pending or persisting transaction) stores its rows when it publishes them, so it completes its transaction and rethrows. On the transaction path it rolls back and rethrows, as any other direct write does. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index edf8e2e63b..d8252eeab4 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -418,7 +418,7 @@ generic visitor behavior outside the named collectors. | Includes and publication | [cross-formulation](https://github.com/TanStack/db/blob/main/packages/db/tests/query/includes-cross-formulation-oracle.property.test.ts), [temporal](https://github.com/TanStack/db/blob/main/packages/db/tests/query/includes-temporal-oracle.test.ts), [Collection includes](https://github.com/TanStack/db/blob/main/packages/db/tests/query/includes-collection-oracle.property.test.ts), [architecture and complete suite map](https://github.com/TanStack/db/blob/main/packages/db/src/query/live/ARCHITECTURE.md#executable-contracts) | Per-parent/flat-join/partition relations, callback-time rows, nested values, and route histories. Fragmented lazy-demand consolidation covers delta growth and successful, rejected/retried, and obsolete replacements at the Collection request boundary; it observes exact key unions and abort checkpoints. A compiled-includes fixture makes adapter unload evict owned child rows and proves a rejected union cannot remove established public child rows. Retry after that terminal live-query failure remains covered at the acquisition boundary; no automatic in-place recovery is claimed. Observe raw promised order; fresh queries do not establish continuous publication safety. | | Collection lifecycle | [mutation startup](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-mutation-startup-oracle.test.ts), [change-event history](https://github.com/TanStack/db/blob/main/packages/db/tests/change-event-history-oracle.test.ts), [sync publication reentrancy](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-sync-reentrancy-oracle.test.ts), [history](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-subscription-lifecycle-history-oracle.property.test.ts), [publication](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-subscription-lifecycle-publication-oracle.property.test.ts), [replay](https://github.com/TanStack/db/blob/main/packages/db/tests/collection-subscription-replay-oracle.property.test.ts), [effect disposal](https://github.com/TanStack/db/blob/main/packages/db/tests/effect-disposal-oracle.test.ts), [PR #1902 review](oracle-reviews/pr-1902-change-event-history.md), [batch-order decision review](oracle-reviews/change-event-batch-order.md), [absent-key insert review](oracle-reviews/2026-10-02-insert-for-absent-key.md) | Core Collection `insert`/`update`/`delete` admission while `startSync:false` is idle; complete public-row/change-message agreement across bounded one- and two-key histories plus replayable four-key campaigns, where every message must be valid for a strict consumer (insert only for an absent key, update or delete only for a present one; [absent-key insert review](oracle-reviews/2026-10-02-insert-for-absent-key.md)); a two-key deferred sync history checks each key's causal insert/update/delete trace while allowing any cross-key interleaving. Reversing all deferred messages fails that check; reversing each sync transaction's distinct-key messages passes. Queued duplicate admission and cancellation, ownership and phase histories, exact caller/error/publication evidence, and late completion and restart have separate witnesses. Generated deferred histories with cancellation or reentrant callbacks need a witness in the change-event history owner before claiming general batch-shape coverage. Query write utilities and effect self-dependent disposal remain separate contracts. During a truncate, a ready callback can write to any key after a prior optimistic request. Each subscriber, with or without initial state or a filter, must receive valid messages after the Collection is ready and end at the Collection's rows. A throwing ready callback during sync entry surfaces once the sync function returns, whether it calls markReady or commits a truncate ([round 3 follow-ups](oracle-reviews/2026-10-05-round-3-followups.md)). Callback rollbacks, non-optimistic callback writes, and several writing callbacks remain open ([ready-callback truncate review](oracle-reviews/2026-10-05-truncate-ready-callback-delete.md)). | | Paced mutations | [virtual-clock oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/paced-mutations-oracle.test.ts) | `createPacedMutations` with queue front/back options, bounded waiting capacity at zero and one, cleanup draining in FIFO/LIFO order after a held write, cleanup timing at the configured wait for immediate writes, admitted writes after separate Collection cleanup, debounce and throttle schedules with explicit and omitted options, and first-leading throttle execution at epoch zero. Finite histories compare optimistic rows, returned transaction identity and receipt outcome, execution order and virtual-clock time. Held writes verify queue serialization. Leading-only throttle and debounce witnesses reject skipped calls immediately with their named dropped-call errors, including omitted edges, both edges disabled, and same-row rollback while a prior write is held. Capacity witnesses cover overflow rejection, distinct-key optimistic rollback, same-key admitted-write survival, and in-flight versus waiting admission. Cleanup witnesses check repeated queue cleanup, eventual admitted receipt settlement, direct queue strategy rejection of new callbacks, public post-cleanup rollback with `QueueDisposedError`, pending debounce transactions that wait until the last call's quiet edge after separate Collection cleanup, and a trailing throttle timer that runs at its regular edge after separate Collection cleanup. Deferred custom queue and batch callbacks check compatibility with `void` and `false` execute results, respectively. Frozen options verify factory non-mutation; a mutable option witness checks that debounce uses its construction-time trailing setting. Throwing `onMutate` histories check leading-edge reuse before any row change, failed-row restoration when a pending transaction already has admitted work, eligible leading writes behind a held handler, and throttle pending merges at the throw, start, receipt, and later-timer cuts. New debounce/throttle admission after cleanup, manual commit of a strategy-owned transaction, longer throwing/reentrant bursts, failed persistence, broader capacities and schedules remain outside this owner. | -| Optimistic state | [same-key composition](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-composition-oracle.test.ts), [history model](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-history-oracle.ts), [generated histories](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-transaction-oracle.property.test.ts), [outcomes](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-history-outcomes-oracle.test.ts), [publication](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-history-publication-oracle.test.ts) | Optimistic state drops at settlement and held sync transactions publish with it ([settlement-drop review](oracle-reviews/2026-10-03-settlement-drop.md)); whole-row active overlays, metadata and prior-value events. A raw subscriber without initial state checks that each truncate batch inserts a key once ([2026-10-02 review](oracle-reviews/2026-10-02-truncate-raw-subscriber.md)). Never rebase a pending snapshot merely to simplify the model. Generated histories also write source batches inside mutation handlers, before they return; the Collection must own the request by then ([handler source batch review](oracle-reviews/2026-10-01-handler-source-batch.md)). The same-key transaction lane also authors a delete followed by a reinsert, which must be visible optimistically and net to the truth-table request ([state mutation round 2](oracle-reviews/2026-10-02-state-mutation-round-2.md)). Source rows are written as `insert` for a key the source lacks, and batches may delete keys the source holds ([state-stack mutation gaps](oracle-reviews/2026-10-02-state-stack-mutant-gaps.md)). Batches may also delete a key the source never held. That delete leaves an active request in place ([absent-key source delete review](oracle-reviews/2026-10-02-absent-key-source-delete.md)). A separate partial campaign keeps the default partial row update mode, where source updates may omit a field and merge into the source row, not into an optimistic row ([round 3 review](oracle-reviews/2026-10-05-state-mutation-round-3.md)). An aborted open batch restores the driver's own source rows as well as its keys, so a later partial update still omits the field ([aborted-batch review](oracle-reviews/2026-10-08-partial-update-aborted-batch.md)). A failed synchronous manual `mutate` callback is now checked after same-key update, distinct insert, and same-key delete, with and without earlier intent, plus two two-Collection cases, one with a restoration listener failure. The throw cut compares public rows and pending payload; the later commit checks only surviving intent. A restoration listener error retains the callback error as the aggregate cause and does not stop other Collection restoration. These witnesses allow transient notifications during the callback and do not cover nested throwing callbacks or async work after the synchronous scope. After every step the driver also checks, by transaction identity, that the Collection tracks exactly its unsettled transactions, so a failed or rolled-back transaction leaves the map ([failed transaction retention review](oracle-reviews/failed-transaction-retention.md)). The [transaction ownership oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/transaction-ownership-oracle.property.test.ts) owns release across two Collections: merged-away pairs, conflicting rollbacks with a throwing subscriber, asynchronous success and failure settlements with a throwing subscriber, rollbacks from a `truncate` listener during a sync commit, a reused transaction id, and a `mutate()` callback that throws, checked in the same call stack and after promises flush. It also checks the settlement error shape and that a settled transaction's set of tracking Collections is empty; its mutations still name their Collection, so neither owner claims the Collection is collectable. Focused [restoration ownership](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/restoration-ownership.test.ts) witnesses check that repeated offline `mutate()` calls add to one transaction, that a completed restoration settles `isPersisted`, and that a restoration one Collection cannot track leaves no Collection tracking it. Open: the generated ownership histories read every receipt themselves, so only a focused witness (rolling back an empty manual transaction) checks that an unread receipt reaches no `unhandledrejection`; two Collection instances with one id, a Collection cleaned up before its transaction settles, and offline restoration have focused witnesses only; no generated history combines a held completed row with a second Collection; and direct writes (`commitDirect`) with a throwing subscriber leave an orphaned pending transaction on `main` and on this branch, a settlement law still under review ([targeted round](oracle-reviews/failed-transaction-retention.md#targeted-review-head-9e7ebb0a8)). | +| Optimistic state | [same-key composition](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-composition-oracle.test.ts), [history model](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-history-oracle.ts), [generated histories](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-transaction-oracle.property.test.ts), [outcomes](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-history-outcomes-oracle.test.ts), [publication](https://github.com/TanStack/db/blob/main/packages/db/tests/optimistic-history-publication-oracle.test.ts) | Optimistic state drops at settlement and held sync transactions publish with it ([settlement-drop review](oracle-reviews/2026-10-03-settlement-drop.md)); whole-row active overlays, metadata and prior-value events. A raw subscriber without initial state checks that each truncate batch inserts a key once ([2026-10-02 review](oracle-reviews/2026-10-02-truncate-raw-subscriber.md)). Never rebase a pending snapshot merely to simplify the model. Generated histories also write source batches inside mutation handlers, before they return; the Collection must own the request by then ([handler source batch review](oracle-reviews/2026-10-01-handler-source-batch.md)). The same-key transaction lane also authors a delete followed by a reinsert, which must be visible optimistically and net to the truth-table request ([state mutation round 2](oracle-reviews/2026-10-02-state-mutation-round-2.md)). Source rows are written as `insert` for a key the source lacks, and batches may delete keys the source holds ([state-stack mutation gaps](oracle-reviews/2026-10-02-state-stack-mutant-gaps.md)). Batches may also delete a key the source never held. That delete leaves an active request in place ([absent-key source delete review](oracle-reviews/2026-10-02-absent-key-source-delete.md)). A separate partial campaign keeps the default partial row update mode, where source updates may omit a field and merge into the source row, not into an optimistic row ([round 3 review](oracle-reviews/2026-10-05-state-mutation-round-3.md)). An aborted open batch restores the driver's own source rows as well as its keys, so a later partial update still omits the field ([aborted-batch review](oracle-reviews/2026-10-08-partial-update-aborted-batch.md)). A failed synchronous manual `mutate` callback is now checked after same-key update, distinct insert, and same-key delete, with and without earlier intent, plus two two-Collection cases, one with a restoration listener failure. The throw cut compares public rows and pending payload; the later commit checks only surviving intent. A restoration listener error retains the callback error as the aggregate cause and does not stop other Collection restoration. These witnesses allow transient notifications during the callback and do not cover nested throwing callbacks or async work after the synchronous scope. After every step the driver also checks, by transaction identity, that the Collection tracks exactly its unsettled transactions, so a failed or rolled-back transaction leaves the map ([failed transaction retention review](oracle-reviews/failed-transaction-retention.md)). The [transaction ownership oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/transaction-ownership-oracle.property.test.ts) owns release across two Collections: merged-away pairs, conflicting rollbacks with a throwing subscriber, asynchronous success and failure settlements with a throwing subscriber, rollbacks from a `truncate` listener during a sync commit, a reused transaction id, and a `mutate()` callback that throws, checked in the same call stack and after promises flush. It also checks the settlement error shape and that a settled transaction's set of tracking Collections is empty; its mutations still name their Collection, so neither owner claims the Collection is collectable. Focused [restoration ownership](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/restoration-ownership.test.ts) witnesses check that repeated offline `mutate()` calls add to one transaction, that a completed restoration settles `isPersisted`, and that a restoration one Collection cannot track leaves no Collection tracking it. Its grammar also has direct writes (`insert`, `update` and `delete` outside a transaction), optionally with a throwing subscriber that may have been sent the rows first; it captures each direct write's own transaction and requires it to fail when the admission throws and to complete otherwise, counts handler calls (none for a rejected write), and keeps a mirror subscriber on each Collection that must match the rows after every step. A write whose admission throws rolls back only its own transaction and rethrows. The [change-event history oracle](https://github.com/TanStack/db/blob/main/packages/db/tests/change-event-history-oracle.test.ts) generates local-only direct writes that meet a throwing subscriber, registered after the mirror, and requires the call to rethrow with its transaction completed and the rows following the model. Focused witnesses cover a local-only subscriber that writes the key it was told about, and a subscriber that throws a value that is not an Error; whether a throwing subscriber may stop a later subscriber is not decided. Open (approved limits): a direct write made inside another write's change subscriber reports its subscriber failure to the outer write; on a local-only Collection a direct write whose subscriber throws completes on the fast path but rolls back on the transaction path, and the change-event oracle generates only the fast path; a direct write admitted from inside a sync-commit listener, where the rollback's recompute is deferred, is not generated; the generated ownership histories read every receipt themselves, so only a focused witness (rolling back an empty manual transaction) checks that an unread receipt reaches no `unhandledrejection`; two Collection instances with one id, a Collection cleaned up before its transaction settles, and offline restoration have focused witnesses only; no generated history combines a held completed row with a second Collection. | | Drafts and native values | [proxy](https://github.com/TanStack/db/blob/main/packages/db/tests/proxy-oracle.test.ts), [detachment](https://github.com/TanStack/db/blob/main/packages/db/tests/proxy-detachment-contract-oracle.test.ts), [iteration](https://github.com/TanStack/db/blob/main/packages/db/tests/proxy-iteration-contract-oracle.test.ts), [revert](https://github.com/TanStack/db/blob/main/packages/db/tests/proxy-revert-oracle.property.test.ts), [native methods](https://github.com/TanStack/db/blob/main/packages/db/tests/proxy-native-methods-oracle.property.test.ts) | Native-operation controls, exact patches and actual stored rows; alias/cycle/adversarial-key histories. Set and Map reorder histories check draft patches and stored iteration order after replacement and clear-and-readd; RegExp replacement/reversion histories check `lastIndex`. General native-mutator and symbol-write support is not established by a plain-object oracle. The revert owner generates assignment, revert, delete, nested, symbol-key, and `for...of` histories and checks `getChanges()` and the draft against an independent draft-equality model: ordered Map and Set contents (including Sets inside Map values), RegExp `lastIndex`, array holes, and symbol keys inside nested values. Map and Set mutators and top-level symbol writes are outside its grammar; array mutators appear only in the mixed block below. Its round-trip grammar also replaces a field with a new object and restores the original value through a nested write. The law requires every changed key, and permits a written key whose final value equals its original to be reported with that value: a false positive repeats the stored value, and a false negative loses a write ([review](oracle-reviews/proxy-revert-replaced-object.md)). A further revert block mixes array `push`, `pop`, and `reverse` with assignments, a nested object replacement, and a retained handle, and checks that a mutator through a replaced handle changes nothing; pinned witnesses require a native write through a Map or Set value, a typed-array view, or a frozen draft to be reported, and a nested revert inside a Map or Set to keep the container's other write; Map and Set mutators are not mixed with assignments, and an equal-valued assignment keeps a retained handle attached (declared). The proxy owner's stored-function law checks, against a native row, that a function stored as data reads back as stored by every read path and that a stored method sees the draft as `this`, so its writes are tracked. The detachment owner pins which key classes a draft copies: enumerable string keys and every symbol key. The revert owner also generates typed arrays (including a subclass that does not forward its constructor argument), URLs, and a class with only private state as written values, with a property that writes two of them over one field. The native-methods owner generates rows (holes, `undefined`, duplicates, nested arrays, empty rows, typed `-0` and `NaN`) and calls every method of `Array.prototype` and `TypedArray.prototype`, in fixed and random campaigns with replay; it compares the result, whether the method returned the array itself, the identity of each original object, and the published row with native values, and a new built-in method fails until it has arguments. The proxy owner's `defineProperty`, stored-function, and frozen-draft laws are bounded tables against native rows. Its `defineProperty` law compares result, value, descriptor, and changes, for values and accessors. The iteration owner checks that array iterators see a push, removal, index write, or length cut made while they are open. Open: a draft reads a class instance of the original row as a plain object, so private-field getters read `undefined`, and a built-in method called through `this` on a nested Map or Set draft rejects the Proxy receiver; a mutator that changes nothing publishes an equal value; writing back a keyless class instance of the original row is a change, and `DataView` setters are untracked ([draft proxy review](oracle-reviews/code-weight-draft-proxy.md)). | | Query DB and observer | [ownership](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/ownership-lifecycle.oracle.test.ts), [persisted direct-write window](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/persisted-direct-write-window.oracle.test.ts) ([review](oracle-reviews/issue-2029-direct-write-lock-window.md)), [load lifecycle](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/load-subset-lifecycle-oracle.test.ts), [observer histories](https://github.com/TanStack/db/blob/main/packages/db/tests/live-query-observer-history-oracle.property.test.ts), [retained snapshots](https://github.com/TanStack/db/blob/main/packages/db/tests/live-query-observer-snapshot-oracle.property.test.ts) ([review](oracle-reviews/pr-2043-retained-observer-snapshots.md)) | Real QueryClient boundary, applied-settlement barriers for existing and cached demand, mutation-handler Collection identity, terminal fetch-record lifecycle, and a per-listener eligibility ledger, not a duplicate dispatch queue. The bounded handler grammar crosses insert, update, delete, parameter and mutation-alias access, refetch, and clearError. A held clearError retry retains the prior public error through initial, background, and Collection application failures; success clears it, failure advances the count, including a repeated error at clock time zero. An intermediate Query retry attempt does not recount the previous background error; a held final attempt can succeed or fail. A held application after successful Query fetch keeps the error visible until the applied result. A mocked persisted-baseline scan holds an older success while a newer terminal Query failure arrives; the newer public error and count survive application, including when the Error object and clock tick repeat. Native SQLite persistence, concurrent retries across multiple tracked Queries, mutation-handler fetch-boundary settlement, and deferred result application remain outside this witness. The mutation overlap grammar crosses both start orders. Check reentry, peer survival, FIFO and disposal independently of final rows. A retained observer snapshot shows its capture-time rows on a later first read of `data` or `state`. | | Query DB SSR dehydration | `packages/query-db-collection/tests/ssr-dehydration-oracle.test.ts`, [review record](oracle-reviews/issue-1950-ssr-dehydration.md) | A plain Query cache control, a live on-demand compound predicate, a signal-only request, and an ordered cursor with a custom comparator all retain their row through QueryClient dehydration, reconstruction of Seroval's emitted stream payload, and Query Core hydration from that payload. The on-demand query functions still receive the original IR, comparator, and signal; serialized metadata retains enumerable user metadata but excludes request options. The original implementation fails the compound and signal cases at the stream checkpoint. A serializable stream-only request-options leak passes the former chunk-count check but fails the current payload assertion. This owner covers successful Query cache entries at the installed Query Core 5.90.20 and Seroval 1.5.0 versions. A TanStack Start integration owner still needs the reporter's Router 1.171.33 and Seroval 1.6.8 path, browser Collection resume/refetch, and request cancellation across SSR. Query subset and pagination oracles remain owners of predicate/order/cursor meaning and supported value types. The issue's `shouldDehydrateQuery` workaround replaces Query Core's success-only default and admits pending/error queries; any documented workaround must preserve that default filter. | diff --git a/packages/db/src/collection/mutations.ts b/packages/db/src/collection/mutations.ts index cf93b11098..4f87f8810e 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -3,6 +3,7 @@ import { withChangeTracking, withFlatChangeTracking, } from '../proxy' +import { normalizeError } from '../utils/error.js' import { safeRandomUUID } from '../utils/uuid' import { createTransaction, getActiveTransaction } from '../transactions' import { @@ -219,9 +220,14 @@ export class CollectionMutationsManager< mutationFn: () => Promise.resolve(), }) transaction.applyMutations(mutations) - direct.write(mutations) - transaction.setState(`completed`) - transaction.isPersisted.resolve(transaction) + // The rows are stored when `write` publishes them, so a throwing + // subscriber cannot undo the write; the transaction still completes. + try { + direct.write(mutations) + } finally { + transaction.setState(`completed`) + transaction.isPersisted.resolve(transaction) + } return transaction } @@ -268,7 +274,22 @@ export class CollectionMutationsManager< >, }), }) - this.applyOwnedMutations(transaction, mutations) + try { + this.applyOwnedMutations(transaction, mutations) + } catch (error) { + // A subscriber threw before the handler could run. Roll back only this + // write, so no Collection keeps a transaction that nothing will settle. + // The admission error is the reported cause. + try { + transaction.rollback({ + isSecondaryRollback: true, + error: normalizeError(error), + }) + } catch { + // Settlement still ran every step. + } + throw error + } // Errors still reject `isPersisted.promise`. This catch only prevents an // unhandled rejection from the fire-and-forget commit. transaction.commit().catch(() => undefined) diff --git a/packages/db/tests/change-event-history-oracle.test.ts b/packages/db/tests/change-event-history-oracle.test.ts index 949f6ca3cb..005b450e22 100644 --- a/packages/db/tests/change-event-history-oracle.test.ts +++ b/packages/db/tests/change-event-history-oracle.test.ts @@ -10,6 +10,7 @@ import { import { localOnlyCollectionOptions } from '../src/local-only' import { PropRef } from '../src/query/ir' import { oraclePropertyOptions, oracleRuns } from './oracle-config' +import { captureCreatedTransactions } from './utils' import type { LocalOnlyCollectionUtils } from '../src/local-only' import type { Collection } from '../src/index' import type { BaseIndex } from '../src/indexes/base-index' @@ -47,6 +48,13 @@ import type { ChangeMessage, SyncConfig } from '../src/types' * repeated changes and complete row values, while leaving cross-key * interleaving unconstrained. * + * About one generated history in four meets a second subscriber, registered + * after the mirror, that throws while one operation publishes. A local-only + * direct write stores its rows as it publishes, so the call rethrows, its + * transaction is already completed, and the rows and mirror still follow the + * model. Whether a throwing subscriber may stop a later subscriber of the same + * publication is not decided here; the throwing subscriber is always last. + * * This partial oracle does not cover failed persistence, sync-transaction * cancellation, general callback batch shape, callback-time index agreement, or * publications without change messages. @@ -84,6 +92,8 @@ type GeneratedHistory = { initialKeys: ReadonlyArray sequence: ReadonlyArray mode: `batched` | `sequential` + /** The operation whose publication meets a throwing subscriber. */ + throwOn?: number } type GeneratedStep = { key: Key; deleteWhenPresent: boolean } @@ -359,6 +369,7 @@ async function runHistory( mode: `batched` | `sequential`, id: string, indexType?: IndexType, + throwOn?: number, ): Promise { const initialRows = [...expectedRowsAfter(initialKeys, [])].map( ([, row]) => row, @@ -436,9 +447,46 @@ async function runHistory( ? collection.subscribeChanges(onChanges, { includeInitialState: false }) : collection.subscribeChanges(onChanges) + // A second subscriber, after the mirror, throws while one operation + // publishes. A local-only direct write stores its rows as it publishes + // them, so the throw cannot undo the write: the call rethrows, and its + // transaction still settles. The rows and the mirror follow the model. + const injected = new Error(`subscriber failed`) + let armed = false + const throwing = + throwOn === undefined + ? undefined + : collection.subscribeChanges( + () => { + if (armed) throw injected + }, + { includeInitialState: true }, + ) + const { created, restore: restoreCreate } = + captureCreatedTransactions(collection) + const applyAt = (op: Op, operationIndex: number) => { + if (operationIndex !== throwOn) return applyOp(collection, op) + armed = true + const before = created.length + try { + applyOp(collection, op) + } catch (error) { + expect(error, `operation ${operationIndex} rethrows`).toBe(injected) + expect(created.length, `one transaction`).toBe(before + 1) + expect( + created[before]!.state, + `operation ${operationIndex} settles`, + ).toBe(`completed`) + return created[before]! + } finally { + armed = false + } + throw new Error(`operation ${operationIndex} did not meet the throw`) + } + if (mode === `batched`) { const transactions = sequence.map((op, operationIndex) => { - const transaction = applyOp(collection, op) + const transaction = applyAt(op, operationIndex) if (index) { const expected = expectedRowsAfter( initialKeys, @@ -455,8 +503,8 @@ async function runHistory( if (index) assertIndexAgreement(index, expected) } else { const settled: Array = [] - for (const op of sequence) { - await settle(applyOp(collection, op)) + for (const [operationIndex, op] of sequence.entries()) { + await settle(applyAt(op, operationIndex)) settled.push(op) const expected = expectedRowsAfter(initialKeys, settled) assertFinalAgreement(collection, mirror, expected) @@ -473,6 +521,8 @@ async function runHistory( for (const publication of publications) { expect(publication.mirrorRows).toEqual(publication.publicRows) } + throwing?.unsubscribe() + restoreCreate() expect(protocolViolations, `change-message protocol`).toEqual([]) } catch (error) { primaryFailure = error @@ -509,12 +559,15 @@ async function verifyGeneratedHistory({ initialKeys, sequence, mode, + throwOn, }: GeneratedHistory): Promise { await runHistory( sequence, initialKeys, mode, `history-oracle-generated-${initialKeys.join(``)}-${mode}-${describeSequence(sequence)}`, + undefined, + throwOn === undefined ? undefined : throwOn % sequence.length, ) } @@ -660,10 +713,13 @@ describe(`change-event history oracle`, () => { { minLength: 1, maxLength: 20 }, ), mode: fc.constantFrom(`batched` as const, `sequential` as const), + // About two histories in three arm a throwing subscriber; about one in + // four reaches the operation it throws on. + throwOn: fc.option(fc.nat(19), { nil: undefined, freq: 2 }), }) - .map(({ initialKeys, steps, mode }) => { + .map(({ initialKeys, steps, mode, throwOn }) => { const sequence = opsFromPresence(initialKeys, steps, 1) - return { initialKeys, sequence, mode } + return { initialKeys, sequence, mode, throwOn } }) // The fixed prefix forces reused keys through an already-built index. diff --git a/packages/db/tests/local-only-direct-write-oracle.test.ts b/packages/db/tests/local-only-direct-write-oracle.test.ts index bea6f239f6..dc2b58559e 100644 --- a/packages/db/tests/local-only-direct-write-oracle.test.ts +++ b/packages/db/tests/local-only-direct-write-oracle.test.ts @@ -4,6 +4,7 @@ import { createCollection } from '../src/index' import { localOnlyCollectionOptions } from '../src/local-only' import { createTransaction } from '../src/transactions' import { createDeferred } from '../src/deferred' +import { captureCreatedTransactions } from './utils' /** * # When does a local-only direct write skip the optimistic stage? @@ -138,6 +139,88 @@ describe(`local-only direct writes`, () => { expect(orders.get(1)?.value).toBe(`a`) }) + // The direct write stores its rows when it publishes them, so a throwing + // subscriber cannot undo the write. The call rethrows, and the transaction + // still completes instead of staying pending. + it(`completes a direct write whose subscriber throws`, async () => { + const orders = createOrders() + const { created, restore: restoreCreate } = + captureCreatedTransactions(orders) + const failure = new Error(`subscriber failed`) + const subscription = orders.subscribeChanges(() => { + throw failure + }) + expect(() => orders.insert({ id: 2, value: `b` })).toThrow(failure) + subscription.unsubscribe() + restoreCreate() + + expect(orders.get(2)?.value).toBe(`b`) + expect(created.map((transaction) => transaction.state)).toEqual([ + `completed`, + ]) + await expect(created[0]!.isPersisted.promise).resolves.toBe(created[0]) + // The next direct write still takes the direct path. + expect(orders.insert({ id: 3, value: `c` }).state).toBe(`completed`) + }) + + // A subscriber may write the key it was told about, as a normalizing or + // rejecting subscriber does. The first write stored its rows before that, + // so it completes and does not throw. + for (const kind of [`update`, `delete`] as const) { + it(`completes a direct insert whose subscriber ${kind}s the same key`, async () => { + const orders = createOrders() + const { created, restore: restoreCreate } = + captureCreatedTransactions(orders) + let reacted = false + const subscription = orders.subscribeChanges((changes) => { + if (reacted || changes.every((change) => change.key !== 2)) return + reacted = true + if (kind === `update`) + orders.update(2, (draft) => { + draft.value = `normalized` + }) + else orders.delete(2) + }) + try { + expect(() => orders.insert({ id: 2, value: `b` })).not.toThrow() + } finally { + subscription.unsubscribe() + restoreCreate() + } + expect(orders.get(2)?.value).toBe( + kind === `update` ? `normalized` : undefined, + ) + await Promise.all( + created.map((transaction) => transaction.isPersisted.promise), + ) + expect(created.map((transaction) => transaction.state)).toEqual([ + `completed`, + `completed`, + ]) + }) + } + + // A subscriber may throw a value that is not an Error. The rolled-back + // transaction still records an Error, so its receipt names the failure. + it(`records an Error when a direct write's subscriber throws a non-Error`, async () => { + const orders = createOrders({ onInsert: async () => {} }) + const { created, restore: restoreCreate } = + captureCreatedTransactions(orders) + const subscription = orders.subscribeChanges(() => { + throw `bad` + }) + try { + expect(() => orders.insert({ id: 2, value: `b` })).toThrow(`bad`) + } finally { + subscription.unsubscribe() + restoreCreate() + } + expect(created.map((transaction) => transaction.state)).toEqual([`failed`]) + expect(created[0]!.error?.error).toBeInstanceOf(Error) + expect(created[0]!.error?.message).toBe(`bad`) + await expect(created[0]!.isPersisted.promise).rejects.toBeInstanceOf(Error) + }) + it(`runs the user's handler for that operation type`, async () => { let calls = 0 const orders = createOrders({ diff --git a/packages/db/tests/transaction-ownership-oracle.property.test.ts b/packages/db/tests/transaction-ownership-oracle.property.test.ts index 8289d93e3c..7c23a54102 100644 --- a/packages/db/tests/transaction-ownership-oracle.property.test.ts +++ b/packages/db/tests/transaction-ownership-oracle.property.test.ts @@ -4,7 +4,10 @@ import { createCollection } from '../src/collection/index.js' import { createTransaction } from '../src/transactions.js' import { DuplicateTransactionIdError } from '../src/errors.js' import { oraclePropertyOptions, oracleRuns } from './oracle-config.js' -import { mockSyncCollectionOptionsNoInitialState } from './utils.js' +import { + captureCreatedTransactions, + mockSyncCollectionOptionsNoInitialState, +} from './utils.js' import type { Transaction } from '../src/transactions.js' /** @@ -45,6 +48,14 @@ import type { Transaction } from '../src/transactions.js' * transaction. * - A failed `mutate()` callback undoes its writes, but the Collection that * took one still tracked the transaction. + * - A direct write (`insert`, `update` or `delete` outside a transaction) runs + * in its own transaction, which settles within its step: its handler + * resolves at once and writes no sync row, so afterwards it shows nothing + * and no Collection tracks it. A subscriber that throws while the write is + * admitted makes the write fail: the call rethrows that error, the write's + * transaction rolls back without rolling back other transactions, and no + * Collection keeps it. Either way the write's own transaction has settled, + * with its `isPersisted`. So a direct write never changes the model. * - Error shape: a settling call that ran no throwing subscriber reports no * subscriber error. Otherwise it throws one of the subscriber errors, as * is. The contract does not say which one, because these failures are rare @@ -57,7 +68,10 @@ import type { Transaction } from '../src/transactions.js' * state allows that step, so generated steps rarely skip. Steps write a value, write and remove key 9 in one call * (a pair that merges away), commit, settle a commit as success or failure, * roll back, roll back from a `truncate` listener during a sync commit, and - * run a `mutate()` callback that writes and then throws. A settling step may + * run a `mutate()` callback that writes and then throws. A direct write + * inserts key 3, or updates or deletes key 1 or 2, outside any transaction, + * optionally with a throwing subscriber on its Collection. The driver + * captures the transaction each direct write creates. A settling step may * install a throwing subscriber on one Collection or on both. It stays * installed until promises flush, because a commit settles in a later * microtask, and it must have run whenever the model predicts that the step @@ -107,6 +121,16 @@ type Step = key: Key value: number } + | { + type: `direct` + op: `insert` | `update` | `delete` + on: CollectionName + key: Key + value: number + throws: boolean + /** The throwing subscriber was sent the rows before the write. */ + seen: boolean + } type ModelState = `pending` | `persisting` | `completed` | `failed` type ModelTx = { @@ -246,6 +270,22 @@ const step: fc.Arbitrary = fc.oneof( on: name, }), }, + { + weight: 2, + arbitrary: fc.record({ + type: fc.constant(`direct` as const), + op: fc.constantFrom( + `insert` as const, + `update` as const, + `delete` as const, + ), + on: name, + key: fc.constantFrom(1, 2), + value: fc.integer({ min: 1, max: 9 }), + throws: fc.boolean(), + seen: fc.boolean(), + }), + }, { weight: 1, arbitrary: fc.record({ @@ -306,7 +346,20 @@ async function makeCollection(id: string) { id: number v: number }>({ id, getKey: (row) => row.id, startSync: true }) - const collection = createCollection(options) + // A direct write's handler resolves at once, so its transaction settles + // within the step instead of waiting for a sync commit. + // Each handler call counts, so a rolled-back direct write can be told from + // one that reached its handler. + const handled = { calls: 0 } + const handler = async () => { + handled.calls++ + } + const collection = createCollection({ + ...options, + onInsert: handler, + onUpdate: handler, + onDelete: handler, + }) const writeBase = () => { options.utils.write({ type: `insert`, value: { id: 1, v: 0 } }) options.utils.write({ type: `insert`, value: { id: 2, v: 0 } }) @@ -316,7 +369,19 @@ async function makeCollection(id: string) { options.utils.commit() options.utils.markReady() await collection.stateWhenReady() - return { collection, utils: options.utils, writeBase } + // A mirror subscriber applies every change message, so a rollback that + // restores rows without publishing the revert leaves it behind. + const mirror = new Map() + collection.subscribeChanges( + (changes) => { + for (const change of changes) { + if (change.type === `delete`) mirror.delete(change.key as number) + else mirror.set(change.key as number, change.value.v) + } + }, + { includeInitialState: true }, + ) + return { collection, utils: options.utils, writeBase, handled, mirror } } /** Runs one history against the model and two Collections. */ @@ -346,20 +411,29 @@ async function runHistory(steps: ReadonlyArray): Promise { throwOn: ThrowOn, run: () => void, applyModel: () => void, + seen = false, ) => { const targets: Array = throwOn === `both` ? [`A`, `B`] : throwOn ? [throwOn] : [] const before = targets.map((on) => JSON.stringify(rows(on))) const raised: Array = [] const calls = new Map() + // A subscriber that was sent the rows first is told of every later + // change, including a delete; it throws only once armed. + let armed = false const subscriptions = targets.map((on) => - collections[on].collection.subscribeChanges(() => { - calls.set(on, (calls.get(on) ?? 0) + 1) - const error = new Error(`subscriber on ${on} failed`) - raised.push(error) - throw error - }), + collections[on].collection.subscribeChanges( + () => { + if (!armed) return + calls.set(on, (calls.get(on) ?? 0) + 1) + const error = new Error(`subscriber on ${on} failed`) + raised.push(error) + throw error + }, + seen ? { includeInitialState: true } : undefined, + ), ) + armed = true let result: { ok: true } | { ok: false; error: unknown } = { ok: true } try { run() @@ -411,6 +485,12 @@ async function runHistory(steps: ReadonlyArray): Promise { ) } expect(collection.has(9), `${label}: ${on} merged-away row`).toBe(false) + expect(collection.has(3), `${label}: ${on} direct insert`).toBe(false) + const { mirror } = collections[on] + for (const key of [1, 2, 3, 9] as const) + expect(mirror.get(key), `${label}: ${on} mirror of row ${key}`).toBe( + collection.get(key)?.v, + ) } for (const [index, entry] of driven.entries()) { const settled = !model.unsettled(model.transactions[index]!) @@ -453,6 +533,74 @@ async function runHistory(steps: ReadonlyArray): Promise { model.open( reuse === undefined ? undefined : model.transactions[reuse]!.id, ) + } else if (current.type === `direct`) { + // A direct update that leaves the visible row unchanged writes nothing. + if ( + current.op === `update` && + model.row(current.on, current.key) === current.value + ) + continue + const { collection } = collections[current.on] + const { created, restore: restoreCreate } = + captureCreatedTransactions(collection) + const settled = new Set>() + const handledBefore = collections[current.on].handled.calls + const { result, raised } = await withThrowingSubscribers( + label, + current.throws ? current.on : undefined, + () => { + try { + if (current.op === `insert`) + collection.insert({ id: 3, v: current.value }) + else if (current.op === `delete`) collection.delete(current.key) + else + collection.update(current.key, (draft) => { + draft.v = current.value + }) + } finally { + restoreCreate() + for (const tx of created) + tx.isPersisted.promise.then( + () => settled.add(tx), + () => settled.add(tx), + ) + } + }, + () => {}, + current.seen, + ) + // The throwing subscriber must reach the admission of an insert or an + // update, so the write fails rather than passing unnoticed. A + // subscriber is not told of a delete of a row it was never sent, so + // such a delete's throwing subscriber runs only when the settled + // delete shows the row again, and the write itself succeeds. + const rejected = + current.throws && (current.op !== `delete` || current.seen) + if (rejected) { + expect( + raised.length, + `${label}: throwing subscriber ran`, + ).toBeGreaterThan(0) + expect(result.ok, `${label}: direct write rejected`).toBe(false) + } + if (current.throws && !rejected) + expect(result.ok, `${label}: direct delete succeeds`).toBe(true) + else expectErrorShape(`${label} direct write`, result, raised) + // A rejected write rolls back before its handler; an accepted one + // reaches its handler exactly once. + expect( + collections[current.on].handled.calls - handledBefore, + `${label}: handler calls`, + ).toBe(rejected ? 0 : 1) + expect(created.length, `${label}: one direct transaction`).toBe(1) + for (const tx of created) { + expect(tx.state, `${label}: direct transaction settled`).toBe( + rejected ? `failed` : `completed`, + ) + expect(settled.has(tx), `${label}: direct isPersisted settled`).toBe( + true, + ) + } } else { const states: ReadonlyArray = current.type === `edit` || diff --git a/packages/db/tests/utils.ts b/packages/db/tests/utils.ts index 15883824cb..c952b7674e 100644 --- a/packages/db/tests/utils.ts +++ b/packages/db/tests/utils.ts @@ -12,6 +12,8 @@ import type { } from '../src/index.js' import type { IndexConstructor } from '../src/indexes/base-index' import type { WithVirtualProps } from '../src/virtual-props.js' +import type { Transaction } from '../src/transactions.js' +import type { Collection } from '../src/collection/index.js' export type OutputWithVirtual< T extends object, @@ -678,3 +680,36 @@ export function resetCleanupQueue(): void { clearTimeout(holder.instance.timeoutId) holder.instance = null } + +/** + * Record each transaction a Collection's mutations manager creates, such as + * the one a direct write makes for itself, which the call may not return when + * it throws. Call `restore` when done. + */ +export function captureCreatedTransactions( + collection: Collection, +): { + created: Array> + restore: () => void +} { + const manager = ( + collection as unknown as { + _mutations: { + createTransaction: (...args: Array) => Transaction + } + } + )._mutations + const createTransaction = manager.createTransaction + const created: Array> = [] + manager.createTransaction = (...args) => { + const transaction = createTransaction.apply(manager, args) + created.push(transaction) + return transaction + } + return { + created, + restore: () => { + manager.createTransaction = createTransaction + }, + } +}