diff --git a/.changeset/fix-facade-retire-leftover-rows.md b/.changeset/fix-facade-retire-leftover-rows.md new file mode 100644 index 0000000000..dd5967c6ca --- /dev/null +++ b/.changeset/fix-facade-retire-leftover-rows.md @@ -0,0 +1,11 @@ +--- +'@tanstack/db': patch +--- + +A live query with included child collections now throws when the query graph retires a child collection that still has rows the graph never removed. The graph always removes a child collection's rows before it retires it, so this is an internal invariant violation. Before, the rows were removed silently, which hid the fault. The failed flush restores the child collections, as for any failed flush. + +Calling `cleanup()` on a child collection of a live query now returns a rejected promise. The live query owns its child collections, and several parent rows can share one. Before, a cleanup stopped the child collection for every parent that shared it, so they never saw its rows again. Clean up the live query instead. Ordinary Collections are not affected. + +An optimistic update of a row in a child collection no longer throws "Bucket facade row has no public key". Each child row now carries its key in `$key`, so the copy that an update makes keeps it. + +Also fix a lost child row behind a persisting transaction. When a user transaction on a child collection was persisting, an update to a child row that an earlier flush had written was applied as a delete, so the row disappeared when the transaction settled. A rollback in that state could also delete such a row. The adapter now decides each write from the child collection's synced rows, including writes that a persisting transaction holds, not from the rows the child collection shows. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 6bec7ddc3a..2b8d191ff6 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -56,7 +56,7 @@ that test identifiers must copy production's private data structures. | ---------------------------------------- | -------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Ordered relations and BTree | Complete | The signed top-K relation, BTree/Map refinement model, and DBSP incrementalization laws are literate. | | D2 operator work | `packages/db-ivm/tests/d2-work-oracle.test.ts` | A finalized 65-branch graph receives one direct input at the first, middle, or last branch, then a second input on a different branch. The oracle compares exact forwarded messages and operator invocations after each `run()`: branches without queued input perform no operator work. A connected reversed-registration pair must drain in the same run, and an empty unfinalized graph rejects. The old unconditional loop fails with 64 idle calls per turn. This bounded work check does not establish elapsed latency, fused operators, arbitrary graph topologies, or provider-level live-query work. | -| Includes and publication | Complete, [facade rollback](https://github.com/TanStack/db/blob/main/packages/db/tests/query/bucket-facade-rollback-oracle.property.test.ts) ([review](oracle-reviews/perf-lazy-bucket-rollback-copy.md)) | The central recomputation model plus cross-formulation, temporal demand, layered publication, Collection facade lifecycle and space bounds, route-context transport, functional projection, query-shape, optimistic, and source-work owners are literate. They keep the architecture document as their contract source. A failed facade flush restores every facade it wrote, disposes the facades it created, publishes no event or layout revision, and reads the stored rows of no other facade; nested facades across two edges have a pinned witness. The grammar adds optimistic facade rows, and retirement retracts every key a facade still shows. Every change pending at a failed root commit, child rows included, stays pending, and the next flush publishes it once. Open: the flush's edge and bucket map copy is unmeasured bookkeeping; disposal between a flush and its rollback has pinned witnesses only; held facade commits are generated one hold per step, settled before the comparison, and a hold across several flushes is not generated. | +| Includes and publication | Complete, [facade rollback](https://github.com/TanStack/db/blob/main/packages/db/tests/query/bucket-facade-rollback-oracle.property.test.ts) ([review](oracle-reviews/perf-lazy-bucket-rollback-copy.md)) | The central recomputation model plus cross-formulation, temporal demand, layered publication, Collection facade lifecycle and space bounds, route-context transport, functional projection, query-shape, optimistic, and source-work owners are literate. They keep the architecture document as their contract source. A failed facade flush restores every facade it wrote, disposes the facades it created, publishes no event or layout revision, and reads the stored rows of no other facade; nested facades across two edges have a pinned witness. The grammar adds optimistic facade inserts and optimistic deletes of shown graph rows; a retired facade must show exactly its visible optimistic rows, and every history ends by settling them and comparing all rows again. A facade commit that throws before committing is generated, and the restore must commit the open sync transaction. Every change pending at a failed root commit, child rows included, stays pending, and the next flush publishes it once. Open: the flush's edge and bucket map copy is unmeasured bookkeeping; disposal between a flush and its rollback has pinned witnesses only; held facade commits are generated one hold at a time for up to three flushes, and a release that covered earlier flushes checks rows only; the grammar also cleans up a shown facade from outside, lets optimistic rows share graph ids and later rolls them back, after which those keys are compared by value; a holder's cleanup of a shown facade rejects (code 241) and the facade keeps receiving the graph's writes; retiring a shown bucket without the graph's retractions is generated (`contradict`) and throws code 240; optimistic deletes and updates of shown graph rows are generated, and an update's copy keeps the row's key; a facade that `resolve()` created before activation can stay ready and empty after a failed flush (recorded limit). | | Collection lifecycle | Complete | The shared logical-owner/acquisition-attempt/sync-run grammar plus mutation admission, lifecycle trace, publication, replay, disposal, and transaction-refinement boundaries are literate. | | Paced mutations | Partial | The virtual-clock owner checks queue order, bounded admission, cleanup draining in FIFO/LIFO order at the configured wait, admitted writes after separate Collection cleanup, post-cleanup queue rejection with a disposal reason, pending debounce drain at the last call's quiet edge, pending trailing throttle drain at its regular edge, debounce and throttle edge options including omitted options, immediate rejection of dropped leading-only calls and both-disabled calls, caller-option immutability, and settlement. New debounce/throttle admission after cleanup, failed persistence, and broader generated schedules remain open. | | Optimistic state | Complete | The independent base/intent/source-queue graph plus outcome, transaction-payload, and publication drivers are literate. | diff --git a/docs/errors.md b/docs/errors.md index 56caba2d89..bd3c0116ef 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -2419,3 +2419,23 @@ Development builds show this message, with `${...}` replaced by values: ```text Include query is missing a captured parent projection for "${ref.path.join(`.`)}". ``` + + + +## Error 240: `query/live/bucket-facade-adapter.ts` + +Development builds show this message, with `${...}` replaced by values: + +```text +Bucket facade retired with rows the graph did not retract +``` + + + +## Error 241: `query/live/bucket-facade-adapter.ts` + +Development builds show this message, with `${...}` replaced by values: + +```text +A live query's child collection cannot be cleaned up directly. Clean up the live query instead. +``` diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index 5723483aaf..9f47f0261a 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -879,6 +879,11 @@ export class CollectionStateManager< this.commitPendingTransactions() } + /** Whether the newest sync transaction is still open for writes. */ + hasOpenSyncTransaction(): boolean { + return this.pendingSyncedTransactions.at(-1)?.committed === false + } + /** Every synced row once every accepted sync transaction applies. */ *acceptedSyncedEntries(): IterableIterator<[TKey, TOutput]> { const { states, truncated } = this.pendingSyncedProjection diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index f1a26be847..61e4670f38 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -610,23 +610,37 @@ Route + BucketFacade -> CellValue(cell, Collection) Parents sharing a bucket share its facade. Child changes update that Collection without re-emitting every parent, and moving a route changes the parent field to -the destination bucket's facade. A facade is never retargeted to another -bucket. The D2 join retains inactive bucket rows and emits their current -snapshot when the bucket becomes active; the facade adapter does not buffer -discarded deltas. It keeps only the deltas a flush consumed until that flush -publishes: if the root commit fails, the adapter restores them, so the next -successful flush publishes each pending child change exactly once. The adapter retains a facade only while at least one parent -route uses its bucket. When the last route leaves, it deletes through sync every -key the facade still shows and drops its strong reference. The graph has -already retracted the bucket's rows, but a facade is a Collection: a user -transaction can show an optimistic row in it, and a sync commit can be held -behind a persisting transaction. Retirement is therefore a legal write to a -non-empty facade, not an invariant violation. It is a facade write of the -flush, so a failed flush restores it. A restore deletes every key the flush -wrote, including a write that a persisting transaction still holds, so the held -write cannot land after the rollback. An external holder may keep the retired -Collection alive, but a later active interval gets a new facade. Inline modes -do not create child Collections. +the destination bucket's facade. A facade is never retargeted to another bucket. +The D2 join retains inactive bucket rows and emits their current snapshot when +the bucket becomes active; the facade adapter does not buffer discarded deltas. +It keeps only the deltas a flush consumed until that flush publishes: if the +root commit fails, the adapter restores them, so the next successful flush +publishes each pending child change exactly once. The adapter retains a facade +only while at least one parent route uses its bucket. When the last route +leaves, it drops its strong reference. The graph has already retracted the +bucket's rows, but a facade is a Collection: a user transaction can show an +optimistic row in it, and a sync commit can be held behind a persisting +transaction. Retiring a facade that still shows such rows is legal, not an +invariant violation. Those rows have no synced row, so they leave when their +transactions settle. A row that the graph sent and never retracted is +different. The adapter reads each facade's projected synced rows: the synced +rows once every queued sync transaction applies, including a transaction a +persisting one holds and a transaction still open. It decides inserts, updates +and deletes from them rather than from the visible rows. A projected row left +at retirement is a +contradictory graph signal: the retirement throws, and the flush aborts and +restores every facade it wrote. The live query owns its facades: +several parent rows can share one, so a holder's `cleanup()` rejects (code 241). +The adapter disposes facades itself, when the live query is cleaned up and +when a failed flush removes the facades it created. Ordinary Collections keep +their cleanup and restart. A +restore undoes only the keys the +flush wrote, including writes that a persisting transaction still holds, so a +held write cannot land after the rollback. A facade that `resolve()` created +before its bucket activated can stay ready and empty after a failed flush; the +next flush writes its rows (a recorded limit). An external holder may keep the +retired Collection alive, but a later active interval gets a new facade. Inline +modes do not create child Collections. Composition is pure. It constructs a new result along changed paths and does not mutate a previously published row or use public routing metadata: diff --git a/packages/db/src/query/live/bucket-facade-adapter.ts b/packages/db/src/query/live/bucket-facade-adapter.ts index b4081e5ce4..ec6fc31af5 100644 --- a/packages/db/src/query/live/bucket-facade-adapter.ts +++ b/packages/db/src/query/live/bucket-facade-adapter.ts @@ -29,10 +29,10 @@ type PendingRow = { type FacadeEntry = { collection: Collection + /** Cleans up the facade; holders cannot, because its live query owns it. */ + dispose: () => Promise sync: FacadeSync | undefined - keys: WeakMap order: WeakMap - currentOrder: Map } type SnapshotRow = { @@ -153,12 +153,7 @@ export class BucketFacadeAdapter { for (const [bucketKey, multiplicity] of activity ?? []) { if (multiplicity >= 0) continue active.delete(bucketKey) - this.retireEntry( - compilation.edgeId, - bucketKey, - snapshot, - publications, - ) + this.retireEntry(compilation.edgeId, bucketKey) } } } catch (error) { @@ -227,7 +222,7 @@ export class BucketFacadeAdapter { this.cleanedUp = true for (const byBucket of this.entries.values()) { for (const entry of byBucket.values()) { - void entry.collection.cleanup() + void entry.dispose() } } this.entries.clear() @@ -262,11 +257,11 @@ export class BucketFacadeAdapter { } private copyRows(entry: FacadeEntry): Array { - return [...entry.collection._state.syncedData].map(([key, value]) => ({ - key, - value, - order: entry.currentOrder.get(key), - })) + // The projected synced rows include every queued sync write, held or + // still open; the visible rows can lag behind them. + return [...entry.collection._state.acceptedSyncedEntries()].map( + ([key, value]) => ({ key, value, order: entry.order.get(value) }), + ) } private snapshot(): FacadeSnapshot { @@ -315,24 +310,27 @@ export class BucketFacadeAdapter { if (!previousEntries.has(entry)) continue const sync = entry.sync if (!sync) continue - const restoredKeys = new Set(rows.map((row) => row.key)) - const synced = entry.collection._state.syncedData - sync.begin() - // A write held behind a persisting transaction is not yet in the synced - // rows, so delete every key the flush wrote as well. - for (const key of new Set([ - ...synced.keys(), - ...snapshot.written.get(entry)!, - ])) { - if (!restoredKeys.has(key)) sync.write({ type: `delete`, key }) - } - entry.currentOrder.clear() - for (const row of rows) { - entry.keys.set(row.value, row.key) + const before = new Map(rows.map((row) => [row.key, row])) + // A failed write or commit can leave the facade's sync transaction + // open with staged writes. Restore into it and commit it, so its + // writes cannot reappear in the projected synced rows later. + if (!entry.collection._state.hasOpenSyncTransaction()) sync.begin() + // Undo only the keys the flush wrote, held writes included; a row it did + // not touch is still the one the facade showed before the flush. + for (const key of snapshot.written.get(entry)!) { + const row = before.get(key) + if (!row) { + // A key that netted to no change has no row to remove. + if (entry.collection._state.getAcceptedSyncedRow(key) !== undefined) + sync.write({ type: `delete`, key }) + continue + } if (row.order !== undefined) entry.order.set(row.value, row.order) - entry.currentOrder.set(row.key, row.order) sync.write({ - type: synced.has(row.key) ? `update` : `insert`, + type: + entry.collection._state.getAcceptedSyncedRow(key) === undefined + ? `insert` + : `update`, value: row.value, }) } @@ -350,7 +348,7 @@ export class BucketFacadeAdapter { this.resolvedValues = new WeakMap() for (const entry of currentEntries) { - if (!previousEntries.has(entry)) void entry.collection.cleanup() + if (!previousEntries.has(entry)) void entry.dispose() } } @@ -370,7 +368,7 @@ export class BucketFacadeAdapter { /** * Begin a facade write. Copy the facade's rows and defer its events before * its first write in the flush, so a rollback reads only the facades the - * flush wrote. A bucket written and then retired in one flush copies once. + * flush wrote. */ private beginWrite( entry: FacadeEntry, @@ -388,27 +386,25 @@ export class BucketFacadeAdapter { sync.begin() } - private retireEntry( - edgeId: string, - bucketKey: string, - snapshot: FacadeSnapshot, - publications: Array, - ): void { + private retireEntry(edgeId: string, bucketKey: string): void { const byBucket = this.entries.get(edgeId) const entry = byBucket?.get(bucketKey) if (!entry) return - - // The graph retracts a bucket's rows when it retires it, but the facade - // can still show rows it never sent: an optimistic row from a pending - // transaction, or a sync commit held behind a persisting one. Retract - // whatever it still holds. - const sync = entry.sync - const keys = [...entry.collection.keys()] - if (sync && keys.length > 0) { - this.beginWrite(entry, sync, snapshot, publications, keys) - for (const key of keys) sync.write({ type: `delete`, key }) - sync.commit() + // The graph retracts every row it sent before it retires a bucket, and + // the projected synced rows include that retraction even while a + // persisting transaction holds it. A row left there is a contradictory + // graph signal. + if (!entry.collection._state.acceptedSyncedEntries().next().done) { + throw new Error( + devBuild() && process.env.NODE_ENV !== `production` + ? `Bucket facade retired with rows the graph did not retract` + : codedMessage(240), + ) } + + // The facade can still show rows the graph never sent, such as an + // optimistic row or a held sync commit. It has no synced row to retract, + // so those rows leave when their transactions settle. byBucket!.delete(bucketKey) if (byBucket!.size === 0) this.entries.delete(edgeId) const retired = getOrCreate(this.retiredEntries, edgeId, () => new Map()) @@ -420,13 +416,14 @@ export class BucketFacadeAdapter { const existing = byBucket.get(bucketKey) if (existing) return existing - const keys = new WeakMap() const order = new WeakMap() let sync: FacadeSync | undefined const collection = createCollection({ id: `__bucket-facade:${this.parentId}:${edgeId}:${bucketKey}`, getKey: (row) => { - const key = keys.get(row) ?? row?.$key + // Every stored row carries its key, so a copy that core makes for an + // update keeps it. + const key = row?.$key if (typeof key !== `string` && typeof key !== `number`) { throw new Error( devBuild() && process.env.NODE_ENV !== `production` @@ -456,14 +453,23 @@ export class BucketFacadeAdapter { startSync: true, gcTime: 0, }) + const dispose = collection.cleanup.bind(collection) + // Other parent rows can share this facade, so a holder must not end it. + collection.cleanup = () => + Promise.reject( + new Error( + devBuild() && process.env.NODE_ENV !== `production` + ? `A live query's child collection cannot be cleaned up directly. Clean up the live query instead.` + : codedMessage(241), + ), + ) const entry: FacadeEntry = { collection, + dispose, get sync() { return sync }, - keys, order, - currentOrder: new Map(), } byBucket.set(bucketKey, entry) return entry @@ -476,39 +482,26 @@ export class BucketFacadeAdapter { hasOrderBy: boolean, ): void { const key = change.value.publicKey as string | number - const previousOrder = entry.currentOrder.get(key) - const nextOrder = change.value.order - const orderChanged = sync.collection.has(key) && previousOrder !== nextOrder - const resolvedRow = this.resolve(change.value.value) - const row = orderChanged ? { ...resolvedRow } : resolvedRow - entry.keys.set(row, key) - if (nextOrder !== undefined) { - entry.order.set(row, nextOrder) - } - - if (change.inserts > change.deletes) { - sync.write({ - type: sync.collection.has(key) ? `update` : `insert`, - value: row, - }) - } else if (change.inserts === change.deletes && sync.collection.has(key)) { - sync.write({ type: `update`, value: row }) - } else if (change.deletes > 0) { + if (change.deletes > change.inserts) { sync.write({ type: `delete`, key }) - entry.currentOrder.delete(key) return } - - entry.currentOrder.set(key, nextOrder) + const accepted = entry.collection._state.getAcceptedSyncedRow(key) + const present = accepted !== undefined + // A key that nets to no change while absent gets no write. + if (change.inserts === change.deletes && !present) return + const nextOrder = change.value.order + const orderChanged = present && entry.order.get(accepted) !== nextOrder + const row = { ...this.resolve(change.value.value), $key: key } + if (nextOrder !== undefined) entry.order.set(row, nextOrder) + sync.write({ type: present ? `update` : `insert`, value: row }) if (hasOrderBy && orderChanged) sync.collection._markLayoutChange() } /** Resolve and validate every public key before opening a sync transaction. */ private prepareChange(entry: FacadeEntry, change: PendingRow): void { - const key = change.value.publicKey as string | number - const row = this.resolve(change.value.value) - entry.keys.set(row, key) - entry.collection.getKeyFromItem(row) + this.resolve(change.value.value) + entry.collection.getKeyFromItem({ $key: change.value.publicKey }) } private resolveValue(value: unknown): unknown { @@ -539,7 +532,7 @@ export class BucketFacadeAdapter { private cleanupRetiredEntries(): void { for (const byBucket of this.retiredEntries.values()) { for (const entry of byBucket.values()) { - void entry.collection.cleanup() + void entry.dispose() } } this.retiredEntries.clear() diff --git a/packages/db/tests/fixtures/error-site-messages.json b/packages/db/tests/fixtures/error-site-messages.json index fe62728485..c990f1a3e2 100644 --- a/packages/db/tests/fixtures/error-site-messages.json +++ b/packages/db/tests/fixtures/error-site-messages.json @@ -906,5 +906,17 @@ "file": "query/compiler/index.ts", "template": "Include query is missing a captured parent projection for \"${ref.path.join(`.`)}\".", "literals": ["Include query is missing a captured parent projection for \""] + }, + "240": { + "file": "query/live/bucket-facade-adapter.ts", + "template": "Bucket facade retired with rows the graph did not retract", + "literals": ["Bucket facade retired with rows the graph did not retract"] + }, + "241": { + "file": "query/live/bucket-facade-adapter.ts", + "template": "A live query's child collection cannot be cleaned up directly. Clean up the live query instead.", + "literals": [ + "A live query's child collection cannot be cleaned up directly. Clean up the live query instead." + ] } } diff --git a/packages/db/tests/query/bucket-facade-adapter.test.ts b/packages/db/tests/query/bucket-facade-adapter.test.ts index 51db58daba..0c72556cd4 100644 --- a/packages/db/tests/query/bucket-facade-adapter.test.ts +++ b/packages/db/tests/query/bucket-facade-adapter.test.ts @@ -33,6 +33,27 @@ class ThrowingBuildIndex extends BasicIndex { } } +/** + * Each projected synced row's order in a facade entry, keyed by row key, so + * iteration order is not compared. + */ +function projectedOrders( + facadeEntry: unknown, +): Map { + const entry = facadeEntry as { + collection: { + _state: { acceptedSyncedEntries: () => Iterable<[number, object]> } + } + order: WeakMap + } + return new Map( + [...entry.collection._state.acceptedSyncedEntries()].map(([key, value]) => [ + key, + entry.order.get(value), + ]), + ) +} + describe(`BucketFacadeAdapter`, () => { it.each( [false, true].flatMap((present) => @@ -337,16 +358,12 @@ describe(`BucketFacadeAdapter`, () => { expect(() => adapter.flush()).toThrow(`facade flush failed`) expect(facade.toArray.map(stripVirtualProps)).toEqual([original, fixed]) - expect([ - ...( - entries.get(`children`)?.get(bucketKey) as unknown as { - currentOrder: Map - } - ).currentOrder, - ]).toEqual([ - [original.id, `0`], - [fixed.id, `1`], - ]) + expect(projectedOrders(entries.get(`children`)?.get(bucketKey))).toEqual( + new Map([ + [original.id, `0`], + [fixed.id, `1`], + ]), + ) expect(publications).toEqual([]) expect(layoutPublications).toBe(0) expect(statusChanges).toBe(0) @@ -382,17 +399,13 @@ describe(`BucketFacadeAdapter`, () => { replacement.id, added.id, ]) - expect([ - ...( - entries.get(`children`)?.get(bucketKey) as unknown as { - currentOrder: Map - } - ).currentOrder, - ]).toEqual([ - [original.id, `2`], - [fixed.id, `1`], - [added.id, `3`], - ]) + expect(projectedOrders(entries.get(`children`)?.get(bucketKey))).toEqual( + new Map([ + [original.id, `2`], + [fixed.id, `1`], + [added.id, `3`], + ]), + ) rows.sendData( new MultiSet([ diff --git a/packages/db/tests/query/bucket-facade-rollback-oracle.property.test.ts b/packages/db/tests/query/bucket-facade-rollback-oracle.property.test.ts index 959c06b7d9..a5cd412bc6 100644 --- a/packages/db/tests/query/bucket-facade-rollback-oracle.property.test.ts +++ b/packages/db/tests/query/bucket-facade-rollback-oracle.property.test.ts @@ -67,14 +67,24 @@ import type { BucketRow } from '../../src/query/live/materialized-pipeline.js' * a facade (leaving its sync transaction open), throws from the commit of a * facade it creates, or is rolled back after `prepare()`. Before a step's * operations, the step may insert an optimistic row into a held facade - * through a user transaction that stays pending until cleanup. The model + * through a user transaction that stays pending until cleanup. That row may + * take an id the graph does not hold in the bucket now, so a later graph row + * can share its key (`shareId`); the comparison sets such ids aside until a + * step rolls back the open optimistic rows (`settleLocal`), after which those + * keys are compared by value. A step + * may also try to clean up a shown facade from outside the adapter + * (`cleanup`), as a holder might. The live query owns its facades, so the + * call throws and the facade keeps receiving the graph's writes. The model * keeps only synced rows, so the comparison sets those optimistic rows aside * and checks that each stays visible while its facade is held. A step may * also hold one shown facade's sync commits behind a persisting user - * transaction during the flush, and settle it before the comparison. The - * flush's writes and a rollback's restoring writes then land only when the - * transaction settles, so every law is checked after they land. The settled - * transaction's own row and its events are set aside. + * transaction during the flush and up to two later flushes (`holdFor`). The + * flushes' writes and a rollback's restoring writes then land only when the + * transaction settles. Steps that end with the hold still open compare only + * reads, because nothing they wrote has landed. The step that settles it + * compares rows, and also events and layout when the hold covered only that + * step. The settled transaction's own row and its events are set aside. A + * hold still open at the end settles and its rows are compared. * * The driver runs the real adapter over a D2 graph. After each flush it * compares the set of facades the adapter holds with the model's buckets, @@ -83,7 +93,13 @@ import type { BucketRow } from '../../src/query/live/materialized-pipeline.js' * revision. For a facade that existed before a successful flush, it replays * the flush's events on the facade's previous rows and checks which rows the * events name. After a failed flush it checks that every facade the flush - * created reports `status === 'cleaned-up'`. + * created reports `status === 'cleaned-up'`. Every facade the model shows + * must report `status === 'ready'`, and every change message a facade + * delivers must be valid for its subscriber: an insert names an absent key, + * and an update or delete names a present one. + * While a hold settles after a failed flush, a restoring sync update can + * still publish the value the facade already shows; that permitted false + * positive is the only event a failed flush may publish. * The work counter wraps the iteration, `forEach`, `get` and `has` methods of * each facade's stored rows during the flush and records which facades the * flush read. A pinned case covers nested facades across two edges: the child @@ -99,17 +115,25 @@ import type { BucketRow } from '../../src/query/live/materialized-pipeline.js' * - The graph retracts a bucket's synced rows before it retires the bucket, * but a retired facade can still show rows the graph never sent: an * optimistic row from a user transaction on the facade, or a sync commit - * held behind a persisting transaction. The grammar adds optimistic rows - * (the `local` step field) and the oracle requires the flush to keep them - * visible while the facade is held. Retirement retracts every key the facade - * still shows through an ordinary facade write, so a failed flush restores - * it. Synced rows left at retirement have pinned witnesses only. Held - * commits are in the grammar (the `hold` step field), with one hold per - * step that settles before the comparison. A hold that stays open across - * several flushes is not generated. + * held behind a persisting transaction. The grammar adds optimistic + * inserts and optimistic deletes and updates of shown graph rows (the + * `local` and `changeShown` step fields). An update builds a copy of the + * stored row, which must keep the row's key. The oracle requires a retired facade to show + * exactly its visible optimistic rows. Those rows have no synced row, so + * they leave when their transactions settle, and every history ends by + * settling them and comparing all rows again. A row the graph sent and + * never retracted is a contradictory graph + * signal, so retiring its bucket throws and the flush restores the facade. + * The `contradict` outcome generates it for a shown facade with rows; every + * other retirement follows the graph's retractions, including under a + * hold, and must not throw. Held + * commits are in the grammar (the `hold` and `holdFor` step fields), one + * hold at a time, for up to three flushes. A step that releases a hold + * opened earlier checks rows only, not events or layout. Optimistic rows + * that share a graph id are not compared by value. * - After a failed flush, each facade the flush created must report - * `status === 'cleaned-up'`. The model does not otherwise follow facades - * after the adapter drops them. + * `status === 'cleaned-up'`. A facade the adapter retired or replaced is + * followed only for its rows. Its events after retirement are not checked. * - The adapter is internal. Its public surface is `flush`, `resolve` and * `cleanup`, and `resolve` creates a facade for a bucket it does not hold. * So the driver reads the adapter's private facade map to see which facades @@ -137,7 +161,19 @@ type Op = | { type: `delete`; bucket: BucketKey; id: number } | { type: `retire`; bucket: BucketKey } -type Outcome = `publish` | `throw` | `throwWrite` | `throwNew` | `rollback` +/** + * How a flush ends. `contradict` retires a shown bucket without the graph's + * retractions, a contradictory graph signal: the flush throws code 240 and + * restores every facade, as for any failed flush. + */ +type Outcome = + | `publish` + | `throw` + | `throwWrite` + | `throwStaged` + | `throwNew` + | `rollback` + | `contradict` type FacadeSync = Parameters>[`sync`]>[0] type FacadeEntry = { @@ -184,6 +220,26 @@ function layoutExpectation( : `permitted` } +/** + * The layout owed to a reader whose facade hides graph rows behind optimistic + * deletes. The reader's rows decide what is required. The adapter orders only + * graph rows, so a graph reorder that the hidden rows make invisible permits + * one revision: a false-positive revision leaves the reader's rows unchanged. + */ +function shownLayoutExpectation( + before: Map, + after: Map, + local: Map, +): LayoutExpectation { + const shown = (rows: Map) => + new Map([...rows].filter(([id]) => local.get(id) !== false)) + const expectation = layoutExpectation(shown(before), shown(after)) + return expectation === `forbidden` && + layoutExpectation(before, after) !== `forbidden` + ? `permitted` + : expectation +} + /** * Run `body`, then `cleanup`. When both throw, keep the body's failure as the * primary error: an AggregateError whose `cause` is that failure and whose @@ -295,7 +351,20 @@ function kidsOf(row: unknown): unknown { * failure leaves the facade's sync transaction open, as a validation error * from a real write would. */ -function injectFailure(sync: FacadeSync, method: `commit` | `write`): void { +function injectFailure( + sync: FacadeSync, + method: `commit` | `write` | `staged`, +): void { + if (method === `staged`) { + // The facade's writes are staged, then the commit throws before it + // commits, so the sync transaction stays open with those writes. + const commit = sync.commit + sync.commit = () => { + sync.commit = commit + throw new Error(`injected facade write failure`) + } + return + } if (method === `commit`) { const commit = sync.commit sync.commit = () => { @@ -324,7 +393,8 @@ class Driver { private readonly sent = new Map>() // Rows a user transaction inserted into a facade. They are optimistic, so // the graph never sent them and the model's synced rows do not hold them. - private readonly localIds = new Map>() + /** Optimistic row ids per facade, and whether the row is visible. */ + private localIds = new WeakMap>() private readonly localTransactions: Array = [] private nextLocalId = 100 // Rows of a persisting user transaction that holds a facade's sync commits. @@ -401,25 +471,46 @@ class Driver { this.graph.run() } + /** Cancel a contradictory retirement that a failed flush kept pending. */ + cancelRetirement(bucket: BucketKey): void { + this.activity.sendData(new MultiSet([[[bucket, true], 1]])) + this.graph.run() + } + /** * Insert a row into a shown facade inside a user transaction that stays * pending, as an application may with a Collection-valued include. The * facade then shows an optimistic row the graph never sent. */ - addLocal(bucket: BucketKey): void { + addLocal( + bucket: BucketKey, + sharedId?: number, + shown?: { id: number; type: `delete` | `update` }, + ): void { const facade = this.entry(bucket)?.collection if (!facade) return - const id = this.nextLocalId++ + if (shown && !facade.has(shown.id)) return + const id = shown?.id ?? sharedId ?? this.nextLocalId++ const transaction = createTransaction({ autoCommit: false, mutationFn: () => new Promise(() => {}), }) + // An optimistic delete or update of a graph row makes the visible rows + // differ from the projected synced rows without any hold. transaction.mutate(() => { - facade.insert({ id, v: 0, $key: id } as unknown as Row) + if (shown?.type === `delete`) facade.delete(id) + else if (shown?.type === `update`) + // A fresh value, so the update always records a mutation: an empty + // transaction is tracked by no Collection, and its rollback would + // reject with no handler (core behavior, outside this oracle). + facade.update(id, (draft) => { + draft.v = 100 + this.nextLocalId++ + }) + else facade.insert({ id, v: 0, $key: id } as unknown as Row) }) this.localTransactions.push(transaction) - const ids = this.localIds.get(facade) ?? new Set() - ids.add(id) + const ids = this.localIds.get(facade) ?? new Map() + ids.set(id, shown?.type !== `delete`) this.localIds.set(facade, ids) } @@ -457,9 +548,20 @@ class Driver { const facade = entry.collection if (!this.events.has(facade)) { this.events.set(facade, 0) + const seen = new Set() facade.subscribeChanges( (messages) => { this.events.set(facade, (this.events.get(facade) ?? 0) + 1) + // Change-message protocol: an insert names a key this subscriber + // does not have, and an update or delete names one it has. + for (const message of messages) { + if (seen.has(message.key) === (message.type === `insert`)) + this.protocolViolations.push( + `${facade.id}: ${message.type} of ${message.key} while ${seen.has(message.key) ? `present` : `absent`}`, + ) + if (message.type === `delete`) seen.delete(message.key) + else seen.add(message.key) + } const changes = this.changes.get(facade) ?? [] changes.push(...messages) this.changes.set(facade, changes) @@ -476,6 +578,18 @@ class Driver { > if (this.wrapped.has(stored)) continue this.wrapped.add(stored) + // The adapter also reads the projected synced rows, held writes included. + const state = facade._state as unknown as Record< + string, + (...args: Array) => unknown + > + for (const method of [`acceptedSyncedEntries`, `getAcceptedSyncedRow`]) { + const original = state[method]!.bind(state) + state[method] = (...args: Array) => { + this.reads.set(facade, (this.reads.get(facade) ?? 0) + 1) + return original(...args) + } + } // Every read path of the stored rows: iteration, forEach, and lookups. for (const method of [ Symbol.iterator, @@ -575,31 +689,63 @@ class Driver { return reads } + /** The adapter holds exactly the facades the model shows. */ + checkHeld(model: Model, label: string): void { + expect( + [...this.entries().keys()].sort(), + `${label}: facades held by the adapter`, + ).toEqual([...model.keys()].sort()) + } + check(model: Model, label: string): void { + expect( + this.protocolViolations, + `${label}: change-message protocol`, + ).toEqual([]) // The adapter holds exactly the facades the model shows: a failed flush // removes the facades it created, and a retired bucket has no facade. expect( [...this.entries().keys()].sort(), `${label}: facades held by the adapter`, ).toEqual([...model.keys()].sort()) + // A retired or replaced facade shows only its visible optimistic rows, + // including one retired while a hold spanned several flushes. + const held = new Set([...this.entries().values()].map((e) => e.collection)) + for (const facade of held) this.shownFacades.add(facade) + for (const facade of this.shownFacades) { + if (held.has(facade)) continue + const visible = [...this.local(facade)] + .filter(([, shown]) => shown) + .map(([id]) => id) + expect( + [...facade.keys()].sort(), + `${label}: rows of a retired facade`, + ).toEqual(visible.sort()) + } for (const bucket of BUCKETS) { const rows = model.get(bucket) const facade = this.entry(bucket)?.collection if (!rows) continue expect(facade, `${label}: facade ${bucket}`).toBeDefined() - const local = this.localIds.get(facade!) ?? new Set() + // A facade the model shows has been made ready by the flush that + // created it. + expect(facade!.status, `${label}: status of ${bucket}`).toBe(`ready`) + // An optimistic row overlays a graph row with the same id, so neither is + // compared by value. + const local = this.local(facade!) expect( facade!.toArray .map(stripVirtualProps) .filter((row) => !local.has(row.id)), `${label}: rows of ${bucket}`, - ).toEqual(expectedRows(rows)) - for (const id of local) { + ).toEqual(expectedRows(rows).filter((row) => !local.has(row.id))) + for (const [id, visible] of local) { expect(facade!.has(id), `${label}: local row ${bucket}/${id}`).toBe( - true, + visible, ) } for (const id of rows.keys()) { + if (local.has(id)) continue const stored = facade!.get(id) expect(stored, `${label}: row ${bucket}/${id}`).toBeDefined() expect(facade!.getKeyFromItem(stored!), `${label}: key ${id}`).toBe(id) @@ -607,6 +753,41 @@ class Driver { } } + /** Change messages that were invalid for the subscriber that got them. */ + readonly protocolViolations: Array = [] + + /** Every facade a check has seen the adapter hold. */ + private readonly shownFacades = new Set>() + + /** + * Try to clean up a shown facade from outside the adapter, as a holder + * might. Its live query owns it, so the call throws and the facade keeps + * receiving the graph's writes. + */ + async attemptCleanup(bucket: BucketKey): Promise { + const facade = this.entry(bucket)?.collection + if (!facade) return + await expect(facade.cleanup(), `holder cleanup`).rejects.toThrow( + /cannot be cleaned up directly/, + ) + expect(facade.status, `status after a refused cleanup`).toBe(`ready`) + } + + /** + * Roll back every open optimistic transaction. Its rows leave, so a graph + * row that shared one of their ids is compared by value from now on. + */ + settleLocals(): void { + for (const transaction of this.localTransactions.splice(0)) + transaction.rollback() + this.localIds = new WeakMap() + } + + /** Optimistic row ids on a facade. */ + local(facade: object): Map { + return this.localIds.get(facade) ?? new Map() + } + async cleanup(): Promise { for (const transaction of this.localTransactions) transaction.rollback() this.adapter.cleanup() @@ -650,8 +831,9 @@ function checkEvents( for (const bucket of BUCKETS) { const facade = facadesBefore.get(bucket) if (!facade) continue + const local = driver.local(facade) const messages = (changes.get(facade) ?? []).filter( - (message) => !driver.heldIds.has(message.key), + (message) => !driver.heldIds.has(message.key) && !local.has(message.key), ) const touched = touchedIds(bucket, previous, ops) const named = messages.map((message) => message.key) @@ -682,6 +864,10 @@ function checkEvents( ]), ) : new Map() + for (const id of local.keys()) { + replayed.delete(id) + expected.delete(id) + } expect( Object.fromEntries(replayed), `${label}: replayed events of ${bucket}`, @@ -702,35 +888,113 @@ async function runHistory( outcome: Outcome throwPick: number local?: number + shareId?: boolean + changeShown?: `delete` | `update` hold?: number + holdFor?: number + cleanup?: number + settleLocal?: boolean }>, ): Promise { const driver = new Driver() let published: Model = new Map() let pending: Array = [] + // A hold can stay open for later steps. Until it settles, its facade's + // writes have not landed, so those steps compare nothing but reads. + let open: + | { settle: () => Promise; remaining: number; openedAt: number } + | undefined await withCleanup( async () => { for (const [index, step] of steps.entries()) { const label = `step ${index}` + // Before anything else, the step may roll back the open optimistic + // rows, outside any hold, so their events precede this flush. + if (step.settleLocal) driver.settleLocals() // A user transaction may first add an optimistic row to a shown facade. const localBucket = step.local === undefined ? undefined : BUCKETS[step.local] if (localBucket && published.has(localBucket)) { - driver.addLocal(localBucket) + // The optimistic row may take a graph id the bucket does not hold + // now, so a later graph row can share its key. + // A rolled-back delete keeps its row shown, so skip ids shown now + // as well as ids the pending operations leave. + const shown = published.get(localBucket) ?? new Map() + const after = + applyOps(published, pending).get(localBucket) ?? new Map() + const facade = driver.entry(localBucket)?.collection + const taken = facade + ? driver.local(facade) + : new Map() + // Under an open hold the facade can still show a row the model has + // already removed. + const free = IDS.filter( + (id) => + !shown.has(id) && + !after.has(id) && + !taken.has(id) && + !facade?.has(id), + ) + const shownIds = [...shown.keys()] + driver.addLocal( + localBucket, + step.shareId && free.length > 0 + ? free[index % free.length] + : undefined, + step.changeShown && shownIds.length > 0 + ? { + id: shownIds[index % shownIds.length]!, + type: step.changeShown, + } + : undefined, + ) } + // A holder may try to clean up a shown facade. The call is refused, + // and the facade keeps receiving the graph's writes. + const cleanupBucket = + step.cleanup === undefined ? undefined : BUCKETS[step.cleanup] + if (cleanupBucket && published.has(cleanupBucket)) + await driver.attemptCleanup(cleanupBucket) // A persisting user transaction may hold one shown facade's sync - // commits through the flush. It settles before the comparison. + // commits through this flush and up to two more. It settles before + // the comparison of its last step. const holdBucket = step.hold === undefined ? undefined : BUCKETS[step.hold] - const settle = - holdBucket && published.has(holdBucket) - ? driver.hold(holdBucket) - : undefined + if (!open && holdBucket && published.has(holdBucket)) { + const settleHold = driver.hold(holdBucket) + if (settleHold) + open = { + settle: settleHold, + remaining: step.holdFor ?? 0, + openedAt: index, + } + } + const settle = open?.remaining === 0 ? open.settle : undefined + // The events and layout of a step that releases a hold opened earlier + // mix several flushes, so only its rows are compared. + const spanning = settle !== undefined && open!.openedAt < index const ops = legalize(applyOps(published, pending), step.choices) driver.send(ops) pending = [...pending, ...ops] driver.instrument() const written = writtenBuckets(pending) + // A contradictory retirement targets a shown, live facade that keeps + // rows after the pending operations. + const afterOps = applyOps(published, pending) + const contradictTargets = BUCKETS.filter( + (bucket) => + driver.entry(bucket) !== undefined && + (afterOps.get(bucket)?.size ?? 0) > 0, + ) + const contradicted = + step.outcome === `contradict` && contradictTargets.length > 0 + ? contradictTargets[step.throwPick % contradictTargets.length]! + : undefined + if (contradicted) { + driver.retireWithoutRetractions(contradicted) + // The retirement check reads that facade's rows. + written.add(contradicted) + } const before = driver.eventCounts() const layoutsBefore = driver.layoutCounts() const facadesBefore = new Map( @@ -740,13 +1004,21 @@ async function runHistory( // Throw from the commit of one facade the flush writes. A facade commits // only when it has a pending row operation, so choose among existing // facades with one. Retiring an empty facade writes nothing. + // A key inserted and retracted again within the flush, absent before + // and after, nets to nothing, so the flush writes nothing for it. const rowBuckets = new Set( pending - .filter((op) => op.type !== `activate` && op.type !== `retire`) + .filter( + (op) => + op.type !== `activate` && + op.type !== `retire` && + (published.get(op.bucket)?.has(op.id) === true || + afterOps.get(op.bucket)?.has(op.id) === true), + ) .map((op) => op.bucket), ) - const throwTargets = [...rowBuckets].filter((bucket) => - driver.entry(bucket), + const throwTargets = [...rowBuckets].filter( + (bucket) => driver.entry(bucket) !== undefined, ) // A facade created in this flush: no facade before the flush, and the // pending operations leave its bucket active with rows to commit. @@ -757,17 +1029,31 @@ async function runHistory( ) let outcome = step.outcome if ( - (outcome === `throw` || outcome === `throwWrite`) && + (outcome === `throw` || + outcome === `throwWrite` || + outcome === `throwStaged`) && throwTargets.length === 0 ) outcome = `publish` if (outcome === `throwNew` && newTargets.length === 0) outcome = `publish` - if (outcome === `throw` || outcome === `throwWrite`) { + if (outcome === `contradict` && !contradicted) outcome = `publish` + if ( + outcome === `throw` || + outcome === `throwWrite` || + outcome === `throwStaged` + ) { const target = driver.entry( throwTargets[step.throwPick % throwTargets.length]!, )! - injectFailure(target.sync!, outcome === `throw` ? `commit` : `write`) + injectFailure( + target.sync!, + outcome === `throw` + ? `commit` + : outcome === `throwWrite` + ? `write` + : `staged`, + ) } else if (outcome === `throwNew`) { driver.failNewFacade(newTargets[step.throwPick % newTargets.length]!) } @@ -777,8 +1063,11 @@ async function runHistory( const created = driver.recordCreated() if (outcome !== `publish` && outcome !== `rollback`) { expect(() => driver.adapter.flush(), label).toThrow( - `injected facade write failure`, + outcome === `contradict` + ? `Bucket facade retired with rows the graph did not retract` + : `injected facade write failure`, ) + if (contradicted) driver.cancelRetirement(contradicted) } else { const publication = driver.adapter.flush() if (outcome === `rollback`) { @@ -802,11 +1091,40 @@ async function runHistory( ).toBe(0) } + if (open && open.remaining > 0) { + open.remaining-- + // Laws that do not wait for held writes still hold: a failed flush + // disposes of the facades it created, and the adapter holds exactly + // the facades the model shows. + if (outcome !== `publish`) + for (const facade of created) + expect(facade.status, `${label}: created facade disposed`).toBe( + `cleaned-up`, + ) + driver.checkHeld( + outcome === `publish` ? applyOps(published, pending) : published, + label, + ) + driver.takeChanges() + if (outcome === `publish`) { + published = applyOps(published, pending) + pending = [] + } + continue + } // The held commits land when the transaction settles. Every law below // is checked after that, so a write the flush held cannot hide. await settle?.() + open = undefined const layoutsAfter = driver.layoutCounts() + if (outcome === `publish` && spanning) { + published = applyOps(published, pending) + pending = [] + driver.takeChanges() + driver.check(published, `${label} (hold released)`) + continue + } if (outcome === `publish`) { const next = applyOps(published, pending) // Layout: the revision advances once for each facade whose shown rows @@ -819,7 +1137,11 @@ async function runHistory( if (!previous || !following) continue const advanced = (layoutsAfter.get(facade) ?? 0) - (layoutsBefore.get(facade) ?? 0) - const expectation = layoutExpectation(previous, following) + const expectation = shownLayoutExpectation( + previous, + following, + driver.local(facade), + ) if (expectation === `permitted`) { expect( advanced, @@ -847,10 +1169,36 @@ async function runHistory( ) } driver.check(published, `${label} (${outcome})`) + if (spanning) { + driver.takeChanges() + continue + } // Settling the hold publishes only its own row's events. - for (const messages of driver.takeChanges().values()) { + // Under a hold, the + // restoring writes land when the transaction settles, and a sync update + // that restores the shown value can still publish. That repeats the + // value the facade shows, a permitted false positive; any other event + // is a change the failed flush must not publish. + const shownRow = (facade: object, key: number) => { + const bucket = BUCKETS.find( + (candidate) => driver.entry(candidate)?.collection === facade, + ) + return bucket === undefined + ? undefined + : published.get(bucket)?.get(key) + } + for (const [facade, messages] of driver.takeChanges()) { expect( messages + .filter( + (message) => + !( + settle && + message.type === `update` && + JSON.stringify(stripVirtualProps(message.value)) === + JSON.stringify(shownRow(facade, message.key)?.value) + ), + ) .map((message) => message.key) .filter((key) => !driver.heldIds.has(key)), `${label}: events after ${outcome}`, @@ -869,6 +1217,15 @@ async function runHistory( ).toBe(layoutsBefore.get(facade)) } } + // A hold still open at the end settles, and its writes must then match. + if (open) { + await open.settle() + driver.check(published, `end (hold released)`) + } + // Open optimistic rows roll back, so keys they hid are compared too. + driver.settleLocals() + await Promise.resolve() + driver.check(published, `end (optimistic rows settled)`) }, () => driver.cleanup(), ) @@ -888,8 +1245,10 @@ const step = fc.record({ `publish`, `throw`, `throwWrite`, + `throwStaged`, `throwNew`, `rollback`, + `contradict`, ), throwPick: fc.nat(3), // About two steps in five add an optimistic row to a facade first. @@ -901,6 +1260,24 @@ const step = fc.record({ hold: fc .integer({ min: -3, max: BUCKETS.length - 1 }) .map((bucket) => (bucket < 0 ? undefined : bucket)), + // A hold stays open for this step and up to two more. + holdFor: fc.integer({ min: 0, max: 2 }), + // An optimistic row sometimes takes a graph id. + shareId: fc.boolean(), + // The optimistic change is sometimes a delete or an update of a shown + // graph row. An update copies the stored row, so the copy must keep its key. + changeShown: fc.constantFrom<`delete` | `update` | undefined>( + undefined, + undefined, + `delete`, + `update`, + ), + // About one step in eight cleans up a shown facade from outside. + cleanup: fc + .integer({ min: -28, max: BUCKETS.length - 1 }) + .map((bucket) => (bucket < 0 ? undefined : bucket)), + // About one step in six first rolls back the open optimistic rows. + settleLocal: fc.integer({ min: 0, max: 5 }).map((n) => n === 0), }) const history = fc.array(step, { minLength: 1, maxLength: 8 }) @@ -1279,12 +1656,10 @@ describe(`bucket facade rollback`, () => { ) }) - // The graph retracts a bucket's rows before it retires the bucket, so synced - // rows left at retirement are a contradictory graph signal. The adapter - // retracts every key the facade still shows, as it does for optimistic and - // held rows, so the facade's synced state empties instead of failing the - // flush. - it(`retracts a retired facade's remaining synced rows`, async () => { + // The graph retracts a bucket's rows before it retires the bucket, so a + // row the graph sent and never retracted is a contradictory graph signal. + // The flush throws and restores the facade, as for any failed flush. + it(`throws when a retired bucket still has rows the graph sent`, async () => { const driver = new Driver() await withCleanup( () => { @@ -1295,8 +1670,11 @@ describe(`bucket facade rollback`, () => { driver.adapter.flush().publish() const facade = driver.entry(`b0`)!.collection driver.retireWithoutRetractions(`b0`) - driver.adapter.flush().publish() - expect(facade.toArray).toEqual([]) + expect(() => driver.adapter.flush()).toThrow( + `Bucket facade retired with rows the graph did not retract`, + ) + expect(driver.entry(`b0`)?.collection).toBe(facade) + expect(facade.toArray.map(stripVirtualProps)).toEqual([{ id: 1, v: 1 }]) }, () => driver.cleanup(), ) @@ -1313,7 +1691,7 @@ describe(`bucket facade rollback`, () => { ]) driver.adapter.flush().publish() const facade = driver.entry(`b0`)!.collection - driver.retireWithoutRetractions(`b0`) + driver.send([{ type: `retire`, bucket: `b0` }]) const publication = driver.adapter.flush() publication.prepare() publication.rollback() @@ -1454,4 +1832,99 @@ describe(`bucket facade rollback`, () => { () => driver.cleanup(), ) }) + + // Pinned replay of a generated failure: an update to a row that an earlier + // flush wrote behind a still-persisting transaction keeps the row. + it(`updates a row that a persisting transaction still holds`, async () => { + await runHistory([ + { + choices: [{ kind: 0, bucket: 0, id: 0, v: 0, rank: 0 }], + outcome: `publish`, + throwPick: 0, + }, + { + choices: [{ kind: 0, bucket: 0, id: 0, v: 1, rank: 0 }], + outcome: `publish`, + throwPick: 0, + hold: 0, + holdFor: 1, + }, + { + choices: [{ kind: 0, bucket: 0, id: 0, v: 2, rank: 0 }], + outcome: `publish`, + throwPick: 0, + }, + ]) + }) + + // Pinned replay of a CI failure: a rollback restores a row that an earlier + // flush wrote behind a still-persisting transaction. The restore must write + // it as an update, because the held insert lands first. + it(`restores a held row after a rolled-back update`, async () => { + await runHistory([ + { + choices: [{ kind: 0, bucket: 3, id: 0, v: 0, rank: 0 }], + outcome: `publish`, + throwPick: 0, + }, + { + choices: [ + { kind: 0, bucket: 3, id: 5, v: 0, rank: 0 }, + { kind: 0, bucket: 0, id: 0, v: 0, rank: 0 }, + ], + outcome: `publish`, + throwPick: 0, + hold: 3, + holdFor: 1, + }, + { + choices: [{ kind: 0, bucket: 3, id: 5, v: 1, rank: 0 }], + outcome: `rollback`, + throwPick: 0, + }, + ]) + }) + + // Pinned replay of a generated failure: an optimistic row shares a graph + // id, is rolled back, and then a flush fails while writing the facade. The + // failed flush publishes nothing. + it(`publishes nothing from a failed flush after a shared optimistic row leaves`, async () => { + const plain = { + throwPick: 0, + local: 0, + hold: 0, + holdFor: 0, + shareId: false, + settleLocal: true, + } + await runHistory([ + { + ...plain, + cleanup: 0, + choices: [{ kind: 0, bucket: 1, id: 0, v: 0, rank: 0 }], + outcome: `publish`, + }, + { + ...plain, + cleanup: 0, + choices: [{ kind: 0, bucket: 0, id: 0, v: 0, rank: 0 }], + outcome: `publish`, + }, + { + ...plain, + shareId: true, + choices: [ + { kind: 0, bucket: 0, id: 2, v: 0, rank: 0 }, + { kind: 0, bucket: 0, id: 0, v: 0, rank: 0 }, + ], + outcome: `publish`, + }, + { + ...plain, + cleanup: 1, + choices: [{ kind: 0, bucket: 0, id: 0, v: 0, rank: 0 }], + outcome: `throwWrite`, + }, + ]) + }) }) diff --git a/packages/db/tests/query/includes-collection-oracle.property.test.ts b/packages/db/tests/query/includes-collection-oracle.property.test.ts index 131f04fdc6..b162ba7e5e 100644 --- a/packages/db/tests/query/includes-collection-oracle.property.test.ts +++ b/packages/db/tests/query/includes-collection-oracle.property.test.ts @@ -1770,7 +1770,7 @@ describe(`Collection-valued includes oracle`, () => { }) fcTest( - `cleanup during root publication suppresses the prepared facade callback`, + `a holder's cleanup during root publication throws and keeps the facade`, async () => { type NodeRow = { id: number @@ -1800,11 +1800,91 @@ describe(`Collection-valued includes oracle`, () => { const facade = live.get(1)!.children const rootSnapshots: Array> = [] const facadeSnapshots: Array> = [] + // The live query owns its child collections, so a holder's cleanup + // throws, even inside the root's publication. + const cleanupErrors: Array = [] + const rootSubscription = live.subscribeChanges( + () => { + rootSnapshots.push(facade.toArray.map(({ value }) => value)) + facade.cleanup().catch((error: unknown) => cleanupErrors.push(error)) + }, + { includeInitialState: false }, + ) + const facadeSubscription = facade.subscribeChanges( + () => facadeSnapshots.push(facade.toArray.map(({ value }) => value)), + { includeInitialState: false }, + ) + + try { + nodes.writeBatch([ + { + type: `update`, + value: { id: 1, kind: `parent`, group: 1, value: 2 }, + }, + { + type: `update`, + value: { id: 10, kind: `child`, group: 1, value: 2 }, + }, + ]) + await Promise.resolve() + expect(rootSnapshots).toEqual([[2]]) + expect(cleanupErrors).toHaveLength(1) + expect(String(cleanupErrors[0])).toMatch( + /cannot be cleaned up directly/, + ) + expect(facadeSnapshots).toEqual([[2]]) + expect(facade.status).toBe(`ready`) + expect(facade.toArray.map(({ value }) => value)).toEqual([2]) + } finally { + rootSubscription.unsubscribe() + facadeSubscription.unsubscribe() + await Promise.all([live.cleanup(), nodes.collection.cleanup()]) + } + }, + ) + + fcTest( + `live query cleanup during root publication suppresses the prepared facade callback`, + async () => { + type NodeRow = { + id: number + kind: `parent` | `child` + group: number + value: number + } + const nodes = createControlledCollection( + `publication-live-cleanup`, + [ + { id: 1, kind: `parent`, group: 1, value: 1 }, + { id: 10, kind: `child`, group: 1, value: 1 }, + ], + ) + const live = createLiveQueryCollection((q) => + q + .from({ parent: nodes.collection }) + .where(({ parent }) => eq(parent.kind, `parent`)) + .select(({ parent }) => ({ + id: parent.id, + value: parent.value, + children: q + .from({ child: nodes.collection }) + .where(({ child }) => eq(child.kind, `child`)) + .where(({ child }) => eq(child.group, parent.group)), + })), + ) + + await live.preload() + const facade = live.get(1)!.children + const rootSnapshots: Array> = [] + const facadeSnapshots: Array> = [] + // The owner ends its child collections while the root publishes. The + // facade's prepared publication must not reach its subscribers after + // that. let cleanup: Promise | undefined const rootSubscription = live.subscribeChanges( () => { rootSnapshots.push(facade.toArray.map(({ value }) => value)) - cleanup = facade.cleanup() + cleanup = live.cleanup() }, { includeInitialState: false }, ) @@ -1833,7 +1913,7 @@ describe(`Collection-valued includes oracle`, () => { } finally { rootSubscription.unsubscribe() facadeSubscription.unsubscribe() - await Promise.all([live.cleanup(), nodes.collection.cleanup()]) + await nodes.collection.cleanup() } }, )