From 3be35ea50bfc960e5573fed34f0d85ddb79396e1 Mon Sep 17 00:00:00 2001 From: Isaac Date: Tue, 13 Oct 2026 07:55:30 -0600 Subject: [PATCH] fix(db): recover a live query's demand when its source recovers from error Co-authored-by: Isaac --- .changeset/source-error-demand-recovers.md | 5 + docs/contributing/oracle-coverage.md | 2 +- .../progressive-source-readiness.md | 37 ++-- packages/db/src/collection/subscription.ts | 88 ++++++++-- packages/db/src/query/live/ARCHITECTURE.md | 22 +++ .../query/live/collection-config-builder.ts | 7 +- .../src/query/live/collection-subscriber.ts | 9 +- .../query/includes-temporal-oracle.test.ts | 162 +++++++++++++++--- 8 files changed, 284 insertions(+), 48 deletions(-) create mode 100644 .changeset/source-error-demand-recovers.md diff --git a/.changeset/source-error-demand-recovers.md b/.changeset/source-error-demand-recovers.md new file mode 100644 index 0000000000..1a2ed0c29e --- /dev/null +++ b/.changeset/source-error-demand-recovers.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +A live query now recovers when a source Collection errors and then becomes ready again. Before, a subset load that rejected because its source entered `error` put a query with a join or include into a terminal error, and a direct query became ready on recovery without loading its subset again. The load is now requested again when the source recovers, and the query becomes ready when that load settles. A load that rejects while its source is healthy still puts the query in error. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index f2f2c7f873..60e09fee79 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -415,7 +415,7 @@ generic visitor behavior outside the named collectors. | Surface | Primary executable owners | Independent judgment and important limit | | -------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Ordered relations and BTree | [top-K relation oracle](https://github.com/TanStack/db/blob/main/packages/db-ivm/tests/operators/topk-relation-oracle.test.ts), [fractional-index window moves](https://github.com/TanStack/db/blob/main/packages/db-ivm/tests/operators/topKWithFractionalIndex-oracle.test.ts), [BTree/Map](https://github.com/TanStack/db/blob/main/packages/db/tests/btree-map-oracle.test.ts), [incrementalization laws](https://github.com/TanStack/db/blob/main/packages/db-ivm/tests/incrementalization-law-oracle.property.test.ts) | Independent ordered relations and cumulative signed output. Top-K consolidation compares same-key values without hashing, including cyclic replacements and fresh transient cancellation. The fractional-index test checks an empty output message for a window move after all rows retract; it does not prove downstream lazy acquisition. Other hash-based operators retain hashing's declared domain. Algebra does not specify client readiness. The BTree/Map oracle checks point operations, neighbor pairs, and full, partial, boundary, exclusive-high, and empty range scans for integer keys and node sizes 4–8. It does not check balance or space, and it assumes consistent comparators ([code-weight BTree review](oracle-reviews/code-weight-btree-trim.md)). The generated groupBy law compares JSON values; a pinned contract test in `packages/db-ivm/tests/operators/groupBy.test.ts` records that after a delete, `min` and `max` return a value hash-equal to a remaining member, not necessarily the remaining instance (`-0` for `0`, an equal Date), which is why the query compiler keeps its own exact-value keys. It checks both insertion orders, reads the group's row by integrating output multiplicities, deletes the minimum and the maximum, and shows that an aggregate input that tells the rows apart returns the remaining instance, for -0/0 in both directions and for both `min` and `max`. The contract holds while every member has positive multiplicity. It pins the current behavior separately: the delete's retraction is the last record a shared hash entry gets, so the deleted instance comes back in either insertion order, for -0/0 in either direction and for an equal Date. For -0/0 and equal Dates the contract assertions only show hash equality, which is all the contract promises. Open: the incrementalization law's value policy compares JSON and clones by `structuredClone`, so no generated history sees -0 or Date identity; a value-identity policy with -0/0 and equal-time Dates would generalize this pinned contract. | -| 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. The temporal owner's source-readiness lane enumerates every legal order of delivery, settlement, source readiness, and one later change for a source whose Collection readiness lags its settled subsets. It checks live-query readiness, preload settlement, and visible rows for direct, single-result, ordered-window, include-child, lazy-join, and eager-root sources; `queryOnce` results for the direct and eager-root shapes; `toArrayWhenReady()` and `stateWhenReady()` results for a read started after rows are visible but before readiness, over the direct, single-result, and eager-root shapes (and the lazy join for `toArrayWhenReady()`); and Effect `skipInitial` deltas for the direct, lazy-join, and eager-root shapes, and every on-demand cell of every consumer is crossed with a truncate or (join and include) a parent rekey before settlement. Each on-demand cell also compares its consumer with a metamorphic twin over a source that is ready from its start, which does not rely on the model's eager/on-demand rule. A live-query observer test checks that a hydration seed hands off at subset settlement, and an Electric test pins that a ready progressive query's rows update only at the full sync's swap. Effects over an ordered window or an include have no lane. It rejects both a design that waits for every source's Collection readiness and one that ignores source readiness. Electric live-query tests are the receiving witnesses for a real progressive source and an on-demand source that is ready before its subset loads; a React Suspense test checks that Suspense and `isReady` release together. Other framework adapters have no Suspense witness. | +| 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. The temporal owner's source-readiness lane enumerates every legal order of delivery, settlement, source readiness, and one later change for a source whose Collection readiness lags its settled subsets. It checks live-query readiness, preload settlement, and visible rows for direct, single-result, ordered-window, include-child, lazy-join, and eager-root sources; `queryOnce` results for the direct and eager-root shapes; `toArrayWhenReady()` and `stateWhenReady()` results for a read started after rows are visible but before readiness, over the direct, single-result, and eager-root shapes (and the lazy join for `toArrayWhenReady()`); live-query recovery from a source error before settlement (law 16) for the direct, single-result, lazy-join, and include shapes, with the same rejection from a healthy source as a terminal control; and Effect `skipInitial` deltas for the direct, lazy-join, and eager-root shapes, and every on-demand cell of every consumer is crossed with a truncate or (join and include) a parent rekey before settlement. Each on-demand cell also compares its consumer with a metamorphic twin over a source that is ready from its start, which does not rely on the model's eager/on-demand rule. A live-query observer test checks that a hydration seed hands off at subset settlement, and an Electric test pins that a ready progressive query's rows update only at the full sync's swap. Effects over an ordered window or an include have no lane. It rejects both a design that waits for every source's Collection readiness and one that ignores source readiness. Electric live-query tests are the receiving witnesses for a real progressive source and an on-demand source that is ready before its subset loads; a React Suspense test checks that Suspense and `isReady` release together. Other framework adapters have no Suspense witness. | | 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. 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. | diff --git a/docs/contributing/oracle-reviews/progressive-source-readiness.md b/docs/contributing/oracle-reviews/progressive-source-readiness.md index 0544eeb2ea..2041acb94e 100644 --- a/docs/contributing/oracle-reviews/progressive-source-readiness.md +++ b/docs/contributing/oracle-reviews/progressive-source-readiness.md @@ -297,20 +297,33 @@ outcome of every async hold. ## Open cells -- TrailBase on-demand: a change committed between a subset fetch and the - change subscription can be missed. The owner is - `packages/trailbase-db-collection/tests/ORACLE.md`. This predates the change. - +- **Fixed: TrailBase on-demand.** A load now waits for the change + subscription before it fetches, so a query is not ready before the + subscription and a change after the fetch is kept. + `packages/trailbase-db-collection/tests/ORACLE.md` owns the law and its + remaining limits. - Effect `skipInitial` over an ordered window or an include child has no lane. - Vue, Svelte, and Angular have no Suspense surface in the shared suite. -- A persisted follower tab's load may settle before the leader's rows apply - in that tab. Only the two-tab OPFS e2e suite can show this; it has not run. -- Electron IPC, two on-demand roots, and include children apart from lazy - joins are not exercised. -- Whether a load that was waiting at a retried stream error is a real failure - is a product decision. +- **Fixed: persisted follower tabs.** A follower's load settled at the + leader's reply, before the follower applied the leader's broadcast commit. + It now waits for that apply. The two-tab OPFS e2e suite fails 3 of 3 runs on + the original code and passes with the fix. +- **Exercised: Electron IPC.** Five witnesses pass in-process and through a + real Electron main process. A follower renderer uses the same coordinator as + a follower tab and has no two-renderer witness. +- Two on-demand roots and include children apart from lazy joins are not + exercised. +- **Decided: a retried stream error that succeeds is a success.** Electric + keeps waiting loads through an `onError` retry. A load that rejects while + its source is in `error` now detaches and is acquired again when the source + recovers (ARCHITECTURE.md law 16), for root and lazy demand alike. Before, + lazy demand entered terminal error, and a root query became ready on + recovery without reacquiring its demand. Model D's source-error histories + fail on the original code (80 cells). Open: an Effect still disposes on a + source error; one-shot reads reject with it; the ordered window has no + source-error history. - The Electric e2e progressive suite has no run receipt (see the coverage map). -- A rejected acquisition or a source error while a source lags is outside - Model D. +- A real rejection while a source lags is a Model D `fail` history, and a + source error is a `source-error` history. Both cover live queries only. - Before the full sync, an Electric progressive query's rows do not receive live changes. The source contract owns that behavior, not this law. diff --git a/packages/db/src/collection/subscription.ts b/packages/db/src/collection/subscription.ts index 76fbbad68a..b36e9a9403 100644 --- a/packages/db/src/collection/subscription.ts +++ b/packages/db/src/collection/subscription.ts @@ -113,6 +113,8 @@ type SubsetDemand = { acquisition: SubsetAcquisitionRecord acquisitionState: `starting` | `active` | `detached` initialResult?: Deferred + /** Detached by a source error; recovery acquires it again. */ + awaitsErrorRecovery?: true } type TruncateReplayAttempt = { @@ -145,6 +147,26 @@ function cancelAcquisition(acquisition: SubsetAcquisitionRecord): void { acquisition.removeRequestAbortListener?.() } +/** + * Wait for detached demand to be acquired again: the restart replay settles + * the wait, and release or abort rejects it with `AbortError`. + */ +function awaitRecovery(demand: SubsetDemand): Promise { + if (demand.initialResult) return demand.initialResult.promise + const initialResult = createDeferred() + demand.initialResult = initialResult + const signal = demand.requestOptions.signal + const abort = () => + initialResult.reject(new LoadSubsetOperationAbortedError()) + signal?.addEventListener(`abort`, abort, { once: true }) + const finish = () => { + signal?.removeEventListener(`abort`, abort) + demand.initialResult = undefined + } + void initialResult.promise.then(finish, finish) + return initialResult.promise +} + export class CollectionSubscription extends EventEmitter implements Subscription @@ -371,6 +393,8 @@ export class CollectionSubscription ) const attempt = replayState.currentAttempt this.truncateReplayState = replayState + // From here the replay's own loading status holds dependent readiness. + for (const demand of demands) delete demand.awaitsErrorRecovery this.setStatus(`loadingSubset`) if (this.truncateReplayState !== replayState) return this.startTruncateReplayAttempt(replayState, attempt, demands) @@ -895,6 +919,7 @@ export class CollectionSubscription if (!(syncResult instanceof Promise)) return const syncRunGeneration = this.collection._sync.getSyncRunGeneration() + const acquisition = demand.acquisition const participant = { demand, promise: syncResult } if (trackStatus) { @@ -912,6 +937,10 @@ export class CollectionSubscription } void syncResult.then(finish, (error: unknown) => { + if (this.detachForErrorRecovery(demand, acquisition)) { + finish() + return + } if ( this.isSyncRunGenerationCurrent(syncRunGeneration) && shouldReportError() @@ -925,6 +954,53 @@ export class CollectionSubscription }) } + /** + * A load that fails while its source Collection is in error belongs to that + * source error, not to the request. Its demand detaches, and recovery + * acquires it again, as for demand requested during the error. + */ + private detachForErrorRecovery( + demand: SubsetDemand, + acquisition: SubsetAcquisitionRecord, + ): boolean { + if ( + this.collection.status !== `error` || + demand.acquisition !== acquisition || + demand.acquisitionState !== `active` || + !this.isDemandActive(demand) || + !this.isSyncRunGenerationCurrent(acquisition.syncRunGeneration) || + demand.requestOptions.signal?.aborted + ) { + return false + } + cancelAcquisition(acquisition) + demand.acquisitionState = `detached` + demand.acquisition = { + options: demand.requestOptions, + syncRunGeneration: acquisition.syncRunGeneration, + } + demand.awaitsErrorRecovery = true + void awaitRecovery(demand) + return true + } + + /** + * Whether demand detached by a source error waits for its restart. A + * dependent's readiness check can run on the source's `ready` before this + * subscription starts the restart replay. + */ + public get awaitsErrorRecovery(): boolean { + return this.subsetDemands.some((demand) => demand.awaitsErrorRecovery) + } + + /** Settles when every demand detached by a source error is acquired again. */ + public get pendingErrorRecovery(): Promise | undefined { + const waits = this.subsetDemands.flatMap((demand) => + demand.awaitsErrorRecovery ? [awaitRecovery(demand)] : [], + ) + return waits.length > 0 ? Promise.all(waits).then(() => {}) : undefined + } + /** Give every logical observer of one transport rejection the same Error. */ private normalizeLoadSubsetPromiseError( promise: Promise, @@ -1049,17 +1125,7 @@ export class CollectionSubscription ) { demand.acquisitionState = `detached` this.subsetDemands.push(demand) - const initialResult = createDeferred() - demand.initialResult = initialResult - const abort = () => - initialResult.reject(new LoadSubsetOperationAbortedError()) - requestOptions.signal?.addEventListener(`abort`, abort, { once: true }) - const finish = () => { - requestOptions.signal?.removeEventListener(`abort`, abort) - demand.initialResult = undefined - } - void initialResult.promise.then(finish, finish) - return { demand, result: initialResult.promise, started: false } + return { demand, result: awaitRecovery(demand), started: false } } const acquisition = this.createSubsetAcquisitionRecord(demand) demand.acquisition = acquisition diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index b68b934caf..9ba232b67a 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -830,6 +830,22 @@ demand; releasing it before recovery creates no physical acquisition or unload. Queued reacquisition must not retry a failed attempt merely because both loading and ready notifications scheduled it. +A source Collection's `error` status is recoverable within its sync run, so a +load that rejects while its source is in `error` belongs to that source error, +not to the request. The subscription detaches that demand, and the source's +next `ready` in the same sync run acquires it again through a fresh replay +barrier. A source that reports a final stream error rejects its loads and then +enters `error` in the same synchronous turn, so the rejection is classified +after the status change. Root and lazy demand recover alike: neither puts the +live query in terminal error, and initial-query readiness after recovery waits +for the reacquired demand to settle. The live query's readiness check can run +on the source's `ready` before the subscription starts that replay, so a +subscription reports detached demand that awaits recovery until the replay +starts. A preload that already rejected with the source error stays rejected. +A load that rejects while its source is healthy is a query failure: it puts +the live query in terminal error, as before. An Effect still disposes on a +source error. + Requests waiting for a loader report one pending promise synchronously through `onLoadSubsetResult`, including requests made during initial error or after cleanup. The callback is not delayed until acquisition: query callers capture @@ -1555,6 +1571,12 @@ create recursive Collection machinery. counts as a preload. A deferring live query keeps its sources from garbage collection but is not counted in their `subscriberCount` or `subscribers:change`, which adapters read as a request for live data. +16. **Source-error recovery:** a load that rejects while its source Collection + is in `error` cannot put a live query in terminal error. When the source + becomes ready again in the same sync run, its root and lazy demand are + acquired again, and initial-query readiness waits for that demand to + settle. A load that rejects while its source is healthy remains a terminal + query failure. ## Glossary diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts index d683d262ac..89005ec720 100644 --- a/packages/db/src/query/live/collection-config-builder.ts +++ b/packages/db/src/query/live/collection-config-builder.ts @@ -1178,7 +1178,12 @@ export class CollectionConfigBuilder< const allDemandsSettled = [...this.activeDemands.values()].every( (demand) => demand.settled, ) - const isLoading = this.liveQueryCollection?.isLoadingSubset + const isLoading = + this.liveQueryCollection?.isLoadingSubset || + // A recovered source acquires detached demand again after this check. + Object.values(this.subscriptions).some( + (subscription) => subscription.awaitsErrorRecovery, + ) // Mark ready when: // 1. All subscriptions are set up (subscribedToAllCollections) // 2. All source collections are ready diff --git a/packages/db/src/query/live/collection-subscriber.ts b/packages/db/src/query/live/collection-subscriber.ts index b1e64861c9..cdbdeb6aea 100644 --- a/packages/db/src/query/live/collection-subscriber.ts +++ b/packages/db/src/query/live/collection-subscriber.ts @@ -216,9 +216,12 @@ export class CollectionSubscriber< void ready.then( () => this.collectionConfigBuilder.settleDemand(plan.id, generation), (error) => { - // A truncate replay aborts the acquisition it supersedes. Its - // replacement decides the demand; a real failure rejects it too. - const replacement = subscription.pendingTruncateReplacement + // A truncate replay aborts the acquisition it supersedes, and a + // source error detaches it until recovery. The replacement + // decides the demand; a real failure rejects it too. + const replacement = + subscription.pendingTruncateReplacement ?? + subscription.pendingErrorRecovery if (replacement) settleWith(replacement) else this.collectionConfigBuilder.failDemand( diff --git a/packages/db/tests/query/includes-temporal-oracle.test.ts b/packages/db/tests/query/includes-temporal-oracle.test.ts index 7c2fc9f24f..ee1f8e9dd3 100644 --- a/packages/db/tests/query/includes-temporal-oracle.test.ts +++ b/packages/db/tests/query/includes-temporal-oracle.test.ts @@ -2609,9 +2609,24 @@ async function expectProgressiveTraceMatches( // The driver uses real Collections, a live-query preload or `createEffect`, // and controlled source writes. Each step is followed by a timer turn. Each // checkpoint compares the consumer's observation and the number of -// acquisitions; the ordered shape records only that it acquired. A rejected acquisition, source error, and later demand changes -// are outside this lane; Models A to C, the Effect disposal oracle, and the -// subscription lifecycle oracle own them. The Electric live-query tests hold +// acquisitions; the ordered shape records only that it acquired. +// +// A source error is recoverable (ARCHITECTURE.md law 16). The live-query lane +// for the direct, single-result, lazy-join, and include shapes also runs a +// `source-error` before settlement: the pending load rejects and the source +// Collection enters `error`, in that order. `source-ready` then recovers the +// source, and the demand must be acquired again: the query is not ready until +// a later `settle`, and it then shows its rows. Its preload has already +// rejected with the source error. The hostile control, `fail`, rejects the +// same load while the source stays healthy: the query stays in terminal error +// and never becomes ready. Root and lazy shapes must agree. The restart +// replay barrier owns publication between the error and that settlement, and +// a failed query owes no rows, so the model judges neither; the twin compares +// them. Effects, one-shot reads, and the ordered window are outside these +// histories: an Effect disposes on a source error, and a read rejects with it. +// Other rejected acquisitions and later demand changes are outside this lane; +// Models A to C, the Effect disposal oracle, and the subscription lifecycle +// oracle own them. The Electric live-query tests hold // the receiving witnesses for a real progressive source and for an on-demand // source that is ready before its subset loads. type LaggingSourceMode = `on-demand` | `eager` @@ -2636,7 +2651,14 @@ const isReadConsumer = (consumer: LaggingSourceConsumer): boolean => consumer === `to-array-when-ready` || consumer === `state-when-ready` type LaggingSourceStep = - `deliver` | `settle` | `source-ready` | `change` | `rekey` | `truncate` + | `deliver` + | `settle` + | `source-ready` + | `change` + | `rekey` + | `truncate` + | `source-error` + | `fail` const LAGGING_SOURCE_SHAPES: Record = { direct: `on-demand`, @@ -2705,6 +2727,41 @@ function disturbedHistories( ] } +// A source error before settlement rejects the pending load and puts the +// source Collection in `error`, in that order, as Electric does when a stream +// error is final. `source-ready` then recovers the source. Each history moves +// the source's readiness after the error, so recovery is the source's first +// `ready` after it; a history may also settle before recovery, when no +// request is pending. A `fail` rejects the pending load while the source +// stays healthy: the hostile control for the same rejection. +function sourceErrorHistories( + failure: `source-error` | `fail`, +): Array> { + const histories = new Map>() + for (const history of laggingSourceHistories(`on-demand`)) { + const rest = history.filter((step) => step !== `source-ready`) + for (let at = 0; at <= rest.indexOf(`settle`); at++) { + const failed = [...rest.slice(0, at), failure, ...rest.slice(at)] + const recoveries = + failure === `fail` + ? [history.slice(0, at).includes(`source-ready`) ? 0 : failed.length] + : Array.from( + { length: failed.length - at }, + (_, offset) => at + 1 + offset, + ) + for (const ready of recoveries) { + const candidate = [ + ...failed.slice(0, ready), + `source-ready` as const, + ...failed.slice(ready), + ] + histories.set(candidate.join(), candidate) + } + } + } + return [...histories.values()] +} + const INITIAL_POST_ID = 2 const CHANGED_POST_ID = 4 @@ -2723,15 +2780,24 @@ type LaggingSourceFacts = { // it started before readiness and so owes `idsAtReadiness`. readOwes: Array | `at-readiness` | undefined replayPending: boolean + // The source Collection is in `error`: its rejected loads belong to that + // error, and recovery acquires their demand again. + sourceErrored: boolean + // A load rejected while its source was healthy: a terminal query failure. + failed: boolean + preloadFailed: boolean } function initiallyReady(mode: LaggingSourceMode, facts: LaggingSourceFacts) { - return mode === `on-demand` ? facts.settled : facts.sourceReady + return mode === `on-demand` + ? facts.settled && !facts.failed + : facts.sourceReady } type LiveQueryObservation = { ready: boolean preloadSettled: boolean + preloadFailed: boolean // `null` between a truncate and its settlement: the replay barrier owns // publication there, so only the metamorphic twin compares those rows. postIds: Array | null @@ -2769,14 +2835,19 @@ function expectedLiveQueryObservation( .sort((a, b) => a - b) return { ready, - preloadSettled: ready, - postIds: facts.replayPending - ? null - : shape !== `ordered` - ? heldIds - : facts.settled - ? heldIds.slice(-1) - : [], + // A preload settles at readiness, or rejects with the source error. + preloadSettled: ready || facts.preloadFailed, + preloadFailed: facts.preloadFailed, + // A failed query owes no rows; a recovering one publishes through the + // restart replay barrier. + postIds: + facts.replayPending || facts.failed + ? null + : shape !== `ordered` + ? heldIds + : facts.settled + ? heldIds.slice(-1) + : [], acquisitions: mode === `on-demand` ? 1 : 0, } } @@ -2847,6 +2918,8 @@ type LaggingPosts = { write: (post: { id: number; userId: number }) => void truncate: () => void markReady: () => void + markError: (error: unknown) => void + rejectPending: (error: unknown) => void } // The source never calls `markReady()` on its own. `write` inserts one post @@ -2865,6 +2938,7 @@ function createLaggingPosts( write: (change: { type: `insert`; value: ProgressivePost }) => void commit: () => unknown markReady: () => void + markError: (error: unknown) => void truncate: () => void } | undefined @@ -2873,8 +2947,8 @@ function createLaggingPosts( getKey: (post) => post.id, ...(explicitMode ? { syncMode: mode } : {}), sync: { - sync: ({ begin, write, commit, markReady, truncate }) => { - sync = { begin, write, commit, markReady, truncate } + sync: ({ begin, write, commit, markReady, markError, truncate }) => { + sync = { begin, write, commit, markReady, markError, truncate } if (readyAtStart) markReady() if (mode === `eager`) return undefined return { @@ -2904,6 +2978,11 @@ function createLaggingPosts( sync!.commit() }, markReady: () => sync!.markReady(), + markError: (error) => sync!.markError(error), + rejectPending: (error) => { + for (const acquisition of acquisitions) + if (acquisition.isPending()) acquisition.reject(error) + }, } } @@ -3101,11 +3180,31 @@ function applyLaggingSourceStep( facts.posts = [] facts.disturbed = true facts.replayPending = true + } else if (step === `source-error` || step === `fail`) { + const error = new Error(`subset load failed`) + for (const source of sources) { + source.rejectPending(error) + if (step === `source-error`) source.markError(error) + } + if (step === `source-error`) facts.sourceErrored = true + else facts.failed = true + facts.disturbed = true + facts.replayPending = step === `source-error` + facts.preloadFailed = true } else if (step === `settle`) { for (const source of sources) for (const acquisition of source.acquisitions) acquisition.resolve() - facts.settled = true - facts.replayPending = false + // During a source error no request is pending, and a failed query + // acquires nothing more. + if (!facts.sourceErrored && !facts.failed) { + facts.settled = true + facts.replayPending = false + } + } else if (facts.sourceErrored) { + // Recovery: both sources become ready and acquire their demand again. + for (const source of sources) source.markReady() + facts.sourceErrored = false + facts.sourceReady = true } else { // The twin's source has been ready since its start. main.source.markReady() @@ -3155,6 +3254,9 @@ function createLaggingSourceDriver( idsAtReadiness: undefined, readOwes: undefined, replayPending: false, + sourceErrored: false, + failed: false, + preloadFailed: false, }, main: startLaggingSourceConsumer( consumer, @@ -3209,7 +3311,7 @@ function createLaggingSourceDriver( await flushPromises() } }, - cleanup: async ({ users, main, twin }) => { + cleanup: async ({ users, main, twin, facts }) => { const states = [main, ...(twin ? [twin] : [])] for (const { source } of states) for (const acquisition of source.acquisitions) acquisition.resolve() @@ -3218,7 +3320,9 @@ function createLaggingSourceDriver( states .flatMap(({ live, effect, source, preload }) => [ () => live?.cleanup(), - () => (live ? finishPreload(preload) : undefined), + // Checkpoints already compared a preload failure with the model. + () => + live && !facts.preloadFailed ? finishPreload(preload) : undefined, () => effect?.dispose(), () => source.collection.cleanup(), ]) @@ -3255,6 +3359,7 @@ function observeLaggingSourceConsumer( ? { ready: state.live!.isReady(), preloadSettled: state.preload.preloadSettled, + preloadFailed: state.preload.preloadFailure !== undefined, postIds: (state.live!.toArray as Array>) .flatMap((row) => rowPostIds(shape, row)) .sort((a, b) => a - b), @@ -3292,7 +3397,7 @@ function laggingSourceProjection( twin && observeLaggingSourceConsumer(consumer, shape, facts, twin) return { observation: - facts.replayPending && `postIds` in observation + (facts.replayPending || facts.failed) && `postIds` in observation ? { ...observation, postIds: null } : observation, twinDisagreement: @@ -3542,6 +3647,23 @@ describe(`includes temporal oracle`, () => { it.each(cells)(`$consumer $shape: $history`, ({ consumer, shape, steps }) => expectLaggingSourceTraceMatches(consumer, shape, steps), ) + + // A source error is recoverable; the same rejection from a healthy + // source is terminal. Root and lazy shapes must agree. + const errorCells = ( + [`direct`, `find-one`, `join`, `include`] as const + ).flatMap((shape) => + ([`source-error`, `fail`] as const).flatMap((failure) => + sourceErrorHistories(failure).map((steps) => ({ + shape, + steps, + history: steps.join(` → `), + })), + ), + ) + it.each(errorCells)(`live-query $shape: $history`, ({ shape, steps }) => + expectLaggingSourceTraceMatches(`live-query`, shape, steps), + ) }) // Grammar: one parent receives 2..4 distinct-key children while its acquisition