From 5c3ec1c772242538d14a832b4e5c5c9c6485e6dc Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 20:43:04 -0600 Subject: [PATCH 1/4] fix(db): reconcile optimistic fields and queued derived updates --- .../fix-optimistic-field-reconciliation.md | 5 + packages/db/src/collection/state.ts | 44 +++- packages/db/src/query/live/ARCHITECTURE.md | 11 + .../query/live/collection-config-builder.ts | 7 +- .../tests/collection-sync-reentrancy.test.ts | 9 +- .../db/tests/optimistic-composition.test.ts | 237 ++++++++++++++++++ .../derived-delete-reconciliation.test.ts | 110 ++++++++ 7 files changed, 417 insertions(+), 6 deletions(-) create mode 100644 .changeset/fix-optimistic-field-reconciliation.md create mode 100644 packages/db/tests/optimistic-composition.test.ts create mode 100644 packages/db/tests/query/derived-delete-reconciliation.test.ts diff --git a/.changeset/fix-optimistic-field-reconciliation.md b/.changeset/fix-optimistic-field-reconciliation.md new file mode 100644 index 0000000000..140b570cf1 --- /dev/null +++ b/.changeset/fix-optimistic-field-reconciliation.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +Compose optimistic updates from their changed fields so concurrent updates and rollbacks preserve unrelated changes. Keep source updates beneath an optimistic live-query delete when queued sync batches apply, without changing sync queue timing. diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index a74bb261c8..0515cb723b 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -14,6 +14,7 @@ import type { ChangeMessage, CollectionConfig, OptimisticChangeMessage, + PendingMutation, } from '../types' import type { CollectionImpl } from './index.js' import type { CollectionLifecycleManager } from './lifecycle' @@ -88,6 +89,11 @@ export class CollectionStateManager< // Optimistic state tracking - make public for testing public optimisticUpserts = new Map() public optimisticDeletes = new Set() + + private composedUpserts = new WeakMap< + object, + { base: TOutput; composed: TOutput } + >() public pendingOptimisticUpserts = new Map() public pendingOptimisticDeletes = new Set() public pendingOptimisticDirectUpserts = new Set() @@ -644,7 +650,7 @@ export class CollectionStateManager< case `update`: this.optimisticUpserts.set( mutation.key, - mutation.modified as TOutput, + this.resolveOptimisticUpsert(mutation), ) this.optimisticDeletes.delete(mutation.key) break @@ -819,6 +825,40 @@ export class CollectionStateManager< } } + // Updates own only their changed top-level fields. Inserts keep the full + // validated row, including schema defaults absent from `changes`. + private resolveOptimisticUpsert(mutation: PendingMutation): TOutput { + if (mutation.type !== `update`) return mutation.modified + + const key = mutation.key as TKey + const base = this.optimisticUpserts.get(key) ?? this.syncedData.get(key) + if (base === undefined) return mutation.modified + + // Reuse an unchanged composition so unrelated transactions do not publish + // new row identities. A new changes object or base invalidates the cache. + const cached = this.composedUpserts.get(mutation.changes) + if (cached?.base === base) return cached.composed + const composed = { ...base, ...mutation.changes } + this.composedUpserts.set(mutation.changes, { base, composed }) + return composed + } + + /** Authoritative membership after committed sync writes, without optimistic edits. */ + hasSyncedKey(key: TKey): boolean { + for (let i = this.pendingSyncedTransactions.length - 1; i >= 0; i--) { + const transaction = this.pendingSyncedTransactions[i]! + if (!transaction.committed) continue + for (let j = transaction.operations.length - 1; j >= 0; j--) { + const operation = transaction.operations[j]! + if (operation.key === key) { + return operation.type !== `delete` + } + } + if (transaction.truncate) return false + } + return this.syncedData.has(key) + } + /** * Get the previous value for a key given previous optimistic state */ @@ -1261,7 +1301,7 @@ export class CollectionStateManager< case `update`: this.optimisticUpserts.set( mutation.key, - mutation.modified as TOutput, + this.resolveOptimisticUpsert(mutation), ) this.optimisticDeletes.delete(mutation.key) break diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index 4e5e8e9e73..3680357e24 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -1001,6 +1001,17 @@ has already done that work. The public Collection is an output, never scratch state. Placeholder rows, in-place include repair, and forced secondary events are forbidden. +Classify root deltas against authoritative membership, including earlier queued +sync writes, not the optimistic public view. An optimistic delete must not turn +a balanced graph update into an authoritative delete. This does not bypass the +normal sync queue or publish part of a graph-output transaction early. + +At the Collection boundary, each active optimistic update owns only its changed +top-level fields. Compose those fields in transaction order over the current +base and earlier optimistic work. Inserts retain their full validated rows, +including schema defaults. Removing one update must not revive its fields from +another update's whole-row snapshot. + Installed state, synchronous reads, change-event payloads, and downstream queries must all observe the same fully materialized commit. The facade adapter may defer event delivery across its Collection transactions, but it must not diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts index 11966c14ba..ab9478c874 100644 --- a/packages/db/src/query/live/collection-config-builder.ts +++ b/packages/db/src/query/live/collection-config-builder.ts @@ -1126,9 +1126,10 @@ export class CollectionConfigBuilder< } else if ( // Insert & update(s) (updates are a delete & insert) inserts > deletes || - // Just update(s) but the item is already in the collection (so - // was inserted previously). - (inserts === deletes && collection.has(collection.getKeyFromItem(value))) + // A balanced delta updates an existing authoritative row, even if an + // optimistic delete hides it or its earlier insert is still queued. + (inserts === deletes && + collection._state.hasSyncedKey(collection.getKeyFromItem(value))) ) { write({ value, diff --git a/packages/db/tests/collection-sync-reentrancy.test.ts b/packages/db/tests/collection-sync-reentrancy.test.ts index 48bd077d48..9d7bc1e66c 100644 --- a/packages/db/tests/collection-sync-reentrancy.test.ts +++ b/packages/db/tests/collection-sync-reentrancy.test.ts @@ -700,7 +700,14 @@ describe(`sync publication reentrancy`, () => { `one`, ]) expect(collection._layoutRevision).toBe(revisionBeforeDrain + 1) - expect(callbacks).toEqual([ + // Deltas form one batch; their order is not the collection's row order. + // Assert exact membership while retaining ordered public-read assertions. + expect( + callbacks.map((callback) => ({ + ...callback, + changes: [...callback.changes].sort((a, b) => a - b), + })), + ).toEqual([ { changes: [1, 2], keys: [2, 1], diff --git a/packages/db/tests/optimistic-composition.test.ts b/packages/db/tests/optimistic-composition.test.ts new file mode 100644 index 0000000000..1e555e88d6 --- /dev/null +++ b/packages/db/tests/optimistic-composition.test.ts @@ -0,0 +1,237 @@ +import { describe, expect, it } from 'vitest' +import { z } from 'zod' +import { createCollection } from '../src/collection/index.js' +import { createDeferred } from '../src/deferred.js' +import { createTransaction } from '../src/transactions.js' +import { stripVirtualProps } from './utils.js' +import type { SyncConfig } from '../src/types.js' + +type Row = { id: number; a: string; b: string; c: string } +const initial: Row = { id: 1, a: `a0`, b: `b0`, c: `c0` } +const orders = [ + [0, 1, 2], + [0, 2, 1], + [1, 0, 2], + [1, 2, 0], + [2, 0, 1], + [2, 1, 0], +] + +function source() { + let sync!: Parameters[`sync`]>[0] + const collection = createCollection({ + getKey: (row) => row.id, + onUpdate: () => Promise.resolve(), + sync: { + sync: (params) => { + sync = params + params.begin() + params.write({ type: `insert`, value: initial }) + params.commit() + params.markReady() + }, + }, + }) + return { + collection, + get sync() { + return sync + }, + } +} + +function pendingTransaction() { + const done = createDeferred() + const tx = createTransaction({ + autoCommit: false, + mutationFn: () => done.promise, + }) + const settled = tx.isPersisted.promise.catch((error: unknown) => error) + return { tx, done, settled } +} + +const cases = orders.flatMap((order) => + ([`pending`, `persisting`] as const).flatMap((phase) => + [0, 1, 2].flatMap((removed) => + ([`disjoint`, `overlapping`] as const).map((fields) => ({ + order, + phase, + removed, + fields, + })), + ), + ), +) + +describe(`optimistic field composition`, () => { + it.each(cases)( + `$order / $phase / rollback $removed / $fields`, + async ({ order, phase, removed, fields }) => { + const fixture = source() + const pending = [ + pendingTransaction(), + pendingTransaction(), + pendingTransaction(), + ] + const patches: Array> = + fields === `disjoint` + ? [{ a: `a1` }, { b: `b2` }, { c: `c3` }] + : [{ a: `a1` }, { a: `a2`, b: `b2` }, { c: `c3` }] + const expected = (excluded = -1) => + Object.assign( + {}, + initial, + ...patches.filter((_, index) => index !== excluded), + ) + try { + await fixture.collection.preload() + for (const index of order) { + pending[index]!.tx.mutate(() => + fixture.collection.update(1, (row) => { + Object.assign(row, patches[index]) + }), + ) + } + if (phase === `persisting`) { + for (const entry of pending) void entry.tx.commit().catch(() => {}) + } + expect(stripVirtualProps(fixture.collection.get(1))).toEqual(expected()) + // Isolate one rollback's projection; conflict-cascade policy is unchanged. + pending[removed]!.tx.rollback({ isSecondaryRollback: true }) + expect(stripVirtualProps(fixture.collection.get(1))).toEqual( + expected(removed), + ) + } finally { + for (const entry of pending) { + entry.tx.rollback({ isSecondaryRollback: true }) + entry.done.resolve() + } + await Promise.all(pending.map((entry) => entry.settled)) + await fixture.collection.cleanup() + } + }, + ) + + it.each([`pending`, `persisting`] as const)( + `preserves untouched synced fields during $phase work`, + async (phase) => { + const fixture = source() + const entry = pendingTransaction() + try { + await fixture.collection.preload() + entry.tx.mutate(() => + fixture.collection.update(1, (row) => { + row.a = `local` + }), + ) + if (phase === `persisting`) void entry.tx.commit().catch(() => {}) + // Exercise the existing explicit immediate path, not a new queue policy. + fixture.sync.begin({ immediate: phase === `persisting` }) + fixture.sync.write({ + type: `update`, + value: { ...initial, b: `remote` }, + }) + expect(fixture.sync.commit()).toBe(true) + expect(stripVirtualProps(fixture.collection.get(1))).toEqual({ + ...initial, + a: `local`, + b: `remote`, + }) + } finally { + entry.tx.rollback() + entry.done.resolve() + await entry.settled + await fixture.collection.cleanup() + } + }, + ) + + it(`keeps a settled direct update beneath an older pending update`, async () => { + const fixture = source() + const entry = pendingTransaction() + try { + await fixture.collection.preload() + entry.tx.mutate(() => + fixture.collection.update(1, (row) => { + row.a = `local` + }), + ) + void entry.tx.commit().catch(() => {}) + const settled = fixture.collection.update(1, (row) => { + row.b = `settled` + }) + await settled.isPersisted.promise + fixture.sync.begin() + fixture.sync.write({ type: `insert`, value: { ...initial, id: 2 } }) + expect(fixture.sync.commit()).not.toBe(true) + expect(stripVirtualProps(fixture.collection.get(1))).toEqual({ + ...initial, + a: `local`, + b: `settled`, + }) + } finally { + entry.tx.rollback() + entry.done.resolve() + await entry.settled + await fixture.collection.cleanup() + } + }) + + it(`preserves insert defaults and suppresses unchanged overlay publications`, async () => { + const collection = createCollection({ + schema: z.object({ + id: z.number(), + title: z.string(), + priority: z.number().default(3), + }), + getKey: (row) => row.id, + sync: { + sync: ({ begin, commit, markReady }) => { + begin() + commit() + markReady() + }, + }, + }) + const entries = [ + pendingTransaction(), + pendingTransaction(), + pendingTransaction(), + ] + try { + await collection.preload() + entries[0]!.tx.mutate(() => collection.insert({ id: 1, title: `first` })) + entries[1]!.tx.mutate(() => + collection.update(1, (row) => { + row.title = `second` + }), + ) + expect(stripVirtualProps(collection.get(1))).toEqual({ + id: 1, + title: `second`, + priority: 3, + }) + const before = collection.get(1) + const seen: Array = [] + const subscription = collection.subscribeChanges((changes) => { + seen.push(...changes.filter((change) => change.key === 1)) + }) + try { + entries[2]!.tx.mutate(() => + collection.insert({ id: 2, title: `unrelated` }), + ) + expect(collection.get(1)).toBe(before) + expect(seen).toEqual([]) + } finally { + subscription.unsubscribe() + } + } finally { + for (const entry of entries) { + entry.tx.rollback({ isSecondaryRollback: true }) + entry.done.resolve() + } + await Promise.all(entries.map((entry) => entry.settled)) + await collection.cleanup() + } + }) +}) diff --git a/packages/db/tests/query/derived-delete-reconciliation.test.ts b/packages/db/tests/query/derived-delete-reconciliation.test.ts new file mode 100644 index 0000000000..5ba5f3b0cf --- /dev/null +++ b/packages/db/tests/query/derived-delete-reconciliation.test.ts @@ -0,0 +1,110 @@ +import { describe, expect, it } from 'vitest' +import { createCollection } from '../../src/collection/index.js' +import { createDeferred } from '../../src/deferred.js' +import { createLiveQueryCollection } from '../../src/query/index.js' +import { createTransaction } from '../../src/transactions.js' +import { stripVirtualProps } from '../utils.js' +import type { SyncConfig } from '../../src/types.js' + +type Row = { id: number; value: number } +const cases = ([`pass-through`, `order`, `select`] as const).flatMap((shape) => + [false, true].flatMap((layered) => + ([`acknowledge`, `rollback`] as const).flatMap((outcome) => + [1, 2].map((batches) => ({ shape, layered, outcome, batches })), + ), + ), +) + +describe(`derived updates beneath optimistic deletes`, () => { + it.each(cases)( + `$shape / layered=$layered / $outcome / $batches batches`, + async ({ shape, layered, outcome, batches }) => { + let sync!: Parameters[`sync`]>[0] + const source = createCollection({ + getKey: (row) => row.id, + sync: { + sync: (params) => { + sync = params + params.begin() + params.write({ type: `insert`, value: { id: 1, value: 10 } }) + params.commit() + params.markReady() + }, + }, + }) + const middle = createLiveQueryCollection({ + query: (q) => q.from({ row: source }), + }) + const derived = createLiveQueryCollection({ + query: (q) => { + const query = q.from({ row: layered ? middle : source }) + if (shape === `order`) return query.orderBy(({ row }) => row.value) + if (shape === `select`) + return query.fn.select(({ row }) => ({ ...row })) + return query + }, + }) + const downstream = createLiveQueryCollection({ + query: (q) => q.from({ row: derived }), + }) + const done = createDeferred() + const tx = createTransaction({ mutationFn: () => done.promise }) + const settled = tx.isPersisted.promise.catch((error: unknown) => error) + const read = (collection: { values: () => Iterable }) => + [...collection.values()] + .map((row) => ({ id: row.id, value: row.value })) + .sort((a, b) => a.id - b.id) + try { + await downstream.preload() + tx.mutate(() => derived.delete(1)) + const reconstructed = new Map() + const subscription = derived.subscribeChanges((changes) => { + for (const change of changes) { + if (change.type === `delete`) reconstructed.delete(change.key) + else reconstructed.set(change.key, stripVirtualProps(change.value)) + } + expect( + [...reconstructed.values()].sort((a, b) => a.id - b.id), + ).toEqual(read(derived)) + }) + try { + for (let index = 0; index < batches; index++) { + sync.begin() + sync.write({ type: `update`, value: { id: 1, value: 20 + index } }) + if (index > 0) + sync.write({ type: `update`, value: { id: 2, value: 40 } }) + sync.write({ + type: `insert`, + value: { id: 2 + index, value: 30 + index }, + }) + expect(sync.commit()).toBe(true) + // The whole graph-output batch remains queued, including the insert. + expect(read(derived)).toEqual([]) + expect(read(downstream)).toEqual([]) + } + if (outcome === `rollback`) done.reject(new Error(`Rejected delete`)) + else done.resolve() + await settled + const expected = [ + { id: 1, value: 19 + batches }, + ...Array.from({ length: batches }, (_, index) => ({ + id: 2 + index, + value: batches > 1 && index === 0 ? 40 : 30 + index, + })), + ] + expect(read(derived)).toEqual(expected) + expect(read(downstream)).toEqual(expected) + } finally { + subscription.unsubscribe() + } + } finally { + done.resolve() + await settled + await downstream.cleanup() + await derived.cleanup() + await middle.cleanup() + await source.cleanup() + } + }, + ) +}) From 297f453791fd43cae8a3cd492a06f82969f5c61d Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 21:15:59 -0600 Subject: [PATCH 2/4] fix(db): preserve optimistic field ownership through settlement --- packages/db/src/collection/state.ts | 101 +++++------ packages/db/src/query/live/ARCHITECTURE.md | 8 +- .../query/live/collection-config-builder.ts | 9 +- .../db/tests/optimistic-composition.test.ts | 157 +++++++++++++++++- .../derived-delete-reconciliation.test.ts | 138 +++++++++++++++ 5 files changed, 358 insertions(+), 55 deletions(-) diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index 0515cb723b..f448d124e2 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -52,6 +52,11 @@ interface PendingSyncedTransaction< type PendingMetadataWrite = { type: `set`; value: unknown } | { type: `delete` } +type OptimisticUpsert = Pick< + PendingMutation, + `key` | `type` | `modified` | `changes` +> + type InternalChangeMessage< T extends object = Record, TKey extends string | number = string | number, @@ -94,7 +99,7 @@ export class CollectionStateManager< object, { base: TOutput; composed: TOutput } >() - public pendingOptimisticUpserts = new Map() + public pendingOptimisticUpserts = new Map>() public pendingOptimisticDeletes = new Set() public pendingOptimisticDirectUpserts = new Set() public pendingOptimisticDirectDeletes = new Set() @@ -540,11 +545,26 @@ export class CollectionStateManager< } switch (mutation.type) { case `insert`: - case `update`: - this.pendingOptimisticUpserts.set( - mutation.key, - mutation.modified as TOutput, - ) + case `update`: { + // Retain only this completed mutation's contribution. Other + // active updates may roll back after this one has settled. + const previous = this.pendingOptimisticUpserts.get(mutation.key) + const changes = + mutation.type === `update` + ? { ...previous?.changes, ...mutation.changes } + : mutation.changes + this.pendingOptimisticUpserts.set(mutation.key, { + key: mutation.key, + type: + mutation.type === `update` + ? (previous?.type ?? `update`) + : `insert`, + modified: + mutation.type === `update` && previous?.type === `insert` + ? { ...previous.modified, ...mutation.changes } + : mutation.modified, + changes, + }) this.pendingOptimisticDeletes.delete(mutation.key) if (isDirectTransaction) { this.pendingOptimisticDirectUpserts.add(mutation.key) @@ -554,6 +574,7 @@ export class CollectionStateManager< this.pendingOptimisticDirectDeletes.delete(mutation.key) } break + } case `delete`: this.pendingOptimisticUpserts.delete(mutation.key) this.pendingOptimisticDeletes.add(mutation.key) @@ -572,12 +593,13 @@ export class CollectionStateManager< if (!this.isThisCollection(mutation.collection)) { continue } - this.pendingLocalOrigins.delete(mutation.key) - if (mutation.optimistic) { - this.pendingOptimisticUpserts.delete(mutation.key) - this.pendingOptimisticDeletes.delete(mutation.key) - this.pendingOptimisticDirectUpserts.delete(mutation.key) - this.pendingOptimisticDirectDeletes.delete(mutation.key) + // Failed transactions never enter retained state. A same-key entry + // belongs to a successful sibling and must survive this rollback. + if ( + !this.pendingOptimisticUpserts.has(mutation.key) && + !this.pendingOptimisticDeletes.has(mutation.key) + ) { + this.pendingLocalOrigins.delete(mutation.key) } } } @@ -601,7 +623,7 @@ export class CollectionStateManager< pendingSyncKeys.has(key) || this.pendingOptimisticDirectUpserts.has(key) ) { - this.optimisticUpserts.set(key, value) + this.optimisticUpserts.set(key, this.resolveOptimisticUpsert(value)) } else { staleOptimisticUpserts.push(key) } @@ -827,7 +849,9 @@ export class CollectionStateManager< // Updates own only their changed top-level fields. Inserts keep the full // validated row, including schema defaults absent from `changes`. - private resolveOptimisticUpsert(mutation: PendingMutation): TOutput { + private resolveOptimisticUpsert( + mutation: OptimisticUpsert, + ): TOutput { if (mutation.type !== `update`) return mutation.modified const key = mutation.key as TKey @@ -843,20 +867,23 @@ export class CollectionStateManager< return composed } - /** Authoritative membership after committed sync writes, without optimistic edits. */ - hasSyncedKey(key: TKey): boolean { - for (let i = this.pendingSyncedTransactions.length - 1; i >= 0; i--) { - const transaction = this.pendingSyncedTransactions[i]! + /** Build once per output flush; queued membership excludes optimistic edits. */ + createSyncedKeyLookup(): (key: TKey) => boolean { + if (this.pendingSyncedTransactions.length === 0) + return (key) => this.syncedData.has(key) + const queued = new Map() + let truncated = false + for (const transaction of this.pendingSyncedTransactions) { if (!transaction.committed) continue - for (let j = transaction.operations.length - 1; j >= 0; j--) { - const operation = transaction.operations[j]! - if (operation.key === key) { - return operation.type !== `delete` - } + if (transaction.truncate) { + queued.clear() + truncated = true + } + for (const operation of transaction.operations) { + queued.set(operation.key as TKey, operation.type !== `delete`) } - if (transaction.truncate) return false } - return this.syncedData.has(key) + return (key) => queued.get(key) ?? (!truncated && this.syncedData.has(key)) } /** @@ -1369,36 +1396,12 @@ export class CollectionStateManager< ) : undefined - // Check if this sync operation is redundant with a completed optimistic operation - const completedOp = completedOptimisticOps.get(key) - let isRedundantSync = false - - if (completedOp) { - if ( - completedOp.type === `delete` && - previousVisibleValue !== undefined && - newVisibleValue === undefined && - deepEquals(completedOp.value, previousVisibleValue) - ) { - isRedundantSync = true - } else if ( - newVisibleValue !== undefined && - deepEquals(completedOp.value, newVisibleValue) - ) { - isRedundantSync = true - } - } - const shouldEmitVirtualUpdate = virtualChanged && previousVisibleValue !== undefined && newVisibleValue !== undefined && deepEquals(previousVisibleValue, newVisibleValue) - if (isRedundantSync && !shouldEmitVirtualUpdate) { - continue - } - if ( previousVisibleValue === undefined && newVisibleValue !== undefined diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index 3680357e24..00f18f0ea0 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -1005,12 +1005,18 @@ Classify root deltas against authoritative membership, including earlier queued sync writes, not the optimistic public view. An optimistic delete must not turn a balanced graph update into an authoritative delete. This does not bypass the normal sync queue or publish part of a graph-output transaction early. +Build queued membership once per output flush, preserving committed last-write +and truncate semantics, rather than scanning the queue again for each row. -At the Collection boundary, each active optimistic update owns only its changed +At the Collection boundary, each active or retained optimistic update owns only its changed top-level fields. Compose those fields in transaction order over the current base and earlier optimistic work. Inserts retain their full validated rows, including schema defaults. Removing one update must not revive its fields from another update's whole-row snapshot. +Completed contributions remain beneath active transactions under the existing +retention policy until sync retires them. A failed sibling cannot clear those +contributions. Sync publication compares actual previous and next visible rows, +not captured mutation snapshots that may predate a rebase. Installed state, synchronous reads, change-event payloads, and downstream queries must all observe the same fully materialized commit. The facade adapter diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts index ab9478c874..4eb30d73ad 100644 --- a/packages/db/src/query/live/collection-config-builder.ts +++ b/packages/db/src/query/live/collection-config-builder.ts @@ -1058,7 +1058,10 @@ export class CollectionConfigBuilder< facadePublication.prepare() if (hasParentChanges) { begin() - changesToApply.forEach(this.applyChanges.bind(this, config)) + const hasSyncedKey = config.collection._state.createSyncedKeyLookup() + changesToApply.forEach( + this.applyChanges.bind(this, config, hasSyncedKey), + ) if (hasOrderOnlyMove(changesToApply)) { markLayoutChange(config.collection) } @@ -1097,6 +1100,7 @@ export class CollectionConfigBuilder< private applyChanges( config: SyncMethods, + hasSyncedKey: (key: string | number) => boolean, changes: { deletes: number inserts: number @@ -1128,8 +1132,7 @@ export class CollectionConfigBuilder< inserts > deletes || // A balanced delta updates an existing authoritative row, even if an // optimistic delete hides it or its earlier insert is still queued. - (inserts === deletes && - collection._state.hasSyncedKey(collection.getKeyFromItem(value))) + (inserts === deletes && hasSyncedKey(collection.getKeyFromItem(value))) ) { write({ value, diff --git a/packages/db/tests/optimistic-composition.test.ts b/packages/db/tests/optimistic-composition.test.ts index 1e555e88d6..0d674b56e2 100644 --- a/packages/db/tests/optimistic-composition.test.ts +++ b/packages/db/tests/optimistic-composition.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest' import { z } from 'zod' import { createCollection } from '../src/collection/index.js' import { createDeferred } from '../src/deferred.js' +import { createLiveQueryCollection } from '../src/query/index.js' import { createTransaction } from '../src/transactions.js' import { stripVirtualProps } from './utils.js' import type { SyncConfig } from '../src/types.js' @@ -17,11 +18,11 @@ const orders = [ [2, 1, 0], ] -function source() { +function source(onUpdate = () => Promise.resolve()) { let sync!: Parameters[`sync`]>[0] const collection = createCollection({ getKey: (row) => row.id, - onUpdate: () => Promise.resolve(), + onUpdate, sync: { sync: (params) => { sync = params @@ -64,6 +65,158 @@ const cases = orders.flatMap((order) => ) describe(`optimistic field composition`, () => { + it.each([false, true])( + `settles a rebased update with queued confirmation=%s`, + async (confirm) => { + const done = createDeferred() + const fixture = source(() => done.promise) + const downstream = createLiveQueryCollection({ + query: (q) => q.from({ row: fixture.collection }), + }) + await downstream.preload() + const replica = new Map([[1, initial]]) + const subscription = fixture.collection.subscribeChanges((changes) => { + for (const change of changes) { + if (change.type === `delete`) replica.delete(change.key) + else replica.set(change.key, stripVirtualProps(change.value)) + } + }) + const check = (expected: Row) => { + expect(stripVirtualProps(fixture.collection.get(1))).toEqual(expected) + expect(replica.get(1)).toEqual(expected) + expect(stripVirtualProps(downstream.get(1))).toEqual(expected) + } + try { + const tx = fixture.collection.update(1, (row) => { + row.a = `a1` + }) + check({ ...initial, a: `a1` }) + fixture.sync.begin({ immediate: true }) + fixture.sync.write({ type: `update`, value: { ...initial, b: `b1` } }) + expect(fixture.sync.commit()).toBe(true) + check({ ...initial, a: `a1`, b: `b1` }) + let applied: true | Promise = true + if (confirm) { + fixture.sync.begin() + fixture.sync.write({ type: `update`, value: { ...initial, a: `a1` } }) + applied = fixture.sync.commit() + expect(applied).not.toBe(true) + } + done.resolve() + await tx.isPersisted.promise + await applied + check({ ...initial, a: `a1`, b: confirm ? `b0` : `b1` }) + } finally { + done.resolve() + subscription.unsubscribe() + await downstream.cleanup() + await fixture.collection.cleanup() + } + }, + ) + + it.each(orders)( + `retains disjoint direct fields across settlement order %j`, + async (...order) => { + const done = [ + createDeferred(), + createDeferred(), + createDeferred(), + ] + let calls = 0 + const fixture = source(() => done[calls++]!.promise) + const downstream = createLiveQueryCollection({ + query: (q) => q.from({ row: fixture.collection }), + }) + await downstream.preload() + const replica = new Map([[1, initial]]) + const subscription = fixture.collection.subscribeChanges((changes) => { + for (const change of changes) { + if (change.type === `delete`) replica.delete(change.key) + else replica.set(change.key, stripVirtualProps(change.value)) + } + }) + try { + const patches = [{ a: `a1` }, { b: `b2` }, { c: `c3` }] + const transactions = patches.map((patch) => + fixture.collection.update(1, (row) => { + Object.assign(row, patch) + }), + ) + const expected = { ...initial, a: `a1`, b: `b2`, c: `c3` } + for (const index of order) { + done[index]!.resolve() + await transactions[index]!.isPersisted.promise + expect(stripVirtualProps(fixture.collection.get(1))).toEqual(expected) + expect(replica.get(1)).toEqual(expected) + expect(stripVirtualProps(downstream.get(1))).toEqual(expected) + } + } finally { + for (const pending of done) pending.resolve() + subscription.unsubscribe() + await downstream.cleanup() + await fixture.collection.cleanup() + } + }, + ) + + it.each([`before`, `after`] as const)( + `retains only a direct survivor's fields when it completes %s sibling rollback`, + async (completion) => { + const done = [createDeferred(), createDeferred()] + let calls = 0 + const fixture = source(() => done[calls++]!.promise) + const downstream = createLiveQueryCollection({ + query: (q) => q.from({ row: fixture.collection }), + }) + await downstream.preload() + const replica = new Map([[1, initial]]) + const subscription = fixture.collection.subscribeChanges((changes) => { + for (const change of changes) { + if (change.type === `delete`) replica.delete(change.key) + else replica.set(change.key, stripVirtualProps(change.value)) + } + }) + const check = (expected: Row) => { + expect(stripVirtualProps(fixture.collection.get(1))).toEqual(expected) + expect(replica.get(1)).toEqual(expected) + expect(stripVirtualProps(downstream.get(1))).toEqual(expected) + } + try { + const first = fixture.collection.update(1, (row) => { + row.a = `a1` + }) + const firstSettled = first.isPersisted.promise.catch(() => {}) + const second = fixture.collection.update(1, (row) => { + row.b = `b2` + }) + check({ ...initial, a: `a1`, b: `b2` }) + if (completion === `before`) { + done[1]!.resolve() + await second.isPersisted.promise + check({ ...initial, a: `a1`, b: `b2` }) + } + done[0]!.reject(new Error(`first update failed`)) + await firstSettled + check({ ...initial, b: `b2` }) + if (completion === `after`) { + done[1]!.resolve() + await second.isPersisted.promise + check({ ...initial, b: `b2` }) + } + fixture.sync.begin() + fixture.sync.write({ type: `update`, value: { ...initial, b: `b2` } }) + await fixture.sync.commit() + check({ ...initial, b: `b2` }) + } finally { + for (const pending of done) pending.resolve() + subscription.unsubscribe() + await downstream.cleanup() + await fixture.collection.cleanup() + } + }, + ) + it.each(cases)( `$order / $phase / rollback $removed / $fields`, async ({ order, phase, removed, fields }) => { diff --git a/packages/db/tests/query/derived-delete-reconciliation.test.ts b/packages/db/tests/query/derived-delete-reconciliation.test.ts index 5ba5f3b0cf..eaacfad1b5 100644 --- a/packages/db/tests/query/derived-delete-reconciliation.test.ts +++ b/packages/db/tests/query/derived-delete-reconciliation.test.ts @@ -16,6 +16,144 @@ const cases = ([`pass-through`, `order`, `select`] as const).flatMap((shape) => ) describe(`derived updates beneath optimistic deletes`, () => { + it.each([false, true])( + `uses committed last writes and truncate=%s for membership`, + async (truncate) => { + const collection = createCollection({ + getKey: (row) => row.id, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ type: `insert`, value: { id: 1, value: 0 } }) + write({ type: `insert`, value: { id: 2, value: 0 } }) + commit() + markReady() + }, + }, + }) + try { + await collection.preload() + const batch = ( + committed: boolean, + ): (typeof collection._state.pendingSyncedTransactions)[number] => ({ + committed, + applicationStarted: false, + layoutChanged: false, + operations: [], + deletedKeys: new Set(), + rowMetadataWrites: new Map(), + collectionMetadataWrites: new Map(), + applied: createDeferred(), + }) + const earlier = batch(true) + earlier.operations.push({ + type: `insert`, + key: 3, + value: { id: 3, value: 0 }, + }) + const later = batch(true) + later.truncate = truncate + later.operations.push( + { type: `delete`, key: 1, value: { id: 1, value: 0 } }, + { type: `insert`, key: 4, value: { id: 4, value: 0 } }, + { type: `delete`, key: 4, value: { id: 4, value: 0 } }, + { type: `insert`, key: 1, value: { id: 1, value: 1 } }, + ) + const uncommitted = batch(false) + uncommitted.truncate = true + uncommitted.operations.push({ + type: `insert`, + key: 5, + value: { id: 5, value: 0 }, + }) + // Directly exercise the membership snapshot; a real truncate normally + // drains immediately and must not be delayed just to construct this case. + collection._state.pendingSyncedTransactions.push( + earlier, + later, + uncommitted, + ) + const hasKey = collection._state.createSyncedKeyLookup() + expect([1, 2, 3, 4, 5].map(hasKey)).toEqual([ + true, + !truncate, + !truncate, + false, + false, + ]) + } finally { + for (const batch of collection._state.pendingSyncedTransactions) + batch.applied.resolve() + collection._state.pendingSyncedTransactions.length = 0 + await collection.cleanup() + } + }, + ) + + it.each([64, 256])( + `classifies %i queued updates with linear membership work`, + async (count) => { + let sync!: Parameters[`sync`]>[0] + const source = createCollection({ + getKey: (row) => row.id, + sync: { + sync: (params) => { + sync = params + params.begin() + for (let id = 0; id < count; id++) + params.write({ type: `insert`, value: { id, value: 0 } }) + params.commit() + params.markReady() + }, + }, + }) + const derived = createLiveQueryCollection({ + query: (q) => q.from({ row: source }), + }) + const done = createDeferred() + const tx = createTransaction({ mutationFn: () => done.promise }) + const settled = tx.isPersisted.promise.catch(() => {}) + try { + await derived.preload() + tx.mutate(() => derived.delete(0)) + const publish = (value: number) => { + sync.begin() + for (let id = 0; id < count; id++) + sync.write({ type: `update`, value: { id, value } }) + expect(sync.commit()).toBe(true) + } + publish(1) + let keyReads = 0 + const queued = derived._state.pendingSyncedTransactions.flatMap( + (batch) => batch.operations, + ) + expect(queued).toHaveLength(count) + for (const operation of queued) { + const key = operation.key + Object.defineProperty(operation, `key`, { + configurable: true, + get: () => { + keyReads++ + return key + }, + }) + } + publish(2) + expect(keyReads).toBeLessThanOrEqual(count * 4) + done.resolve() + await settled + expect( + [...derived.values()].map((row) => stripVirtualProps(row)), + ).toEqual(Array.from({ length: count }, (_, id) => ({ id, value: 2 }))) + } finally { + done.resolve() + await settled + await derived.cleanup() + await source.cleanup() + } + }, + ) + it.each(cases)( `$shape / layered=$layered / $outcome / $batches batches`, async ({ shape, layered, outcome, batches }) => { From 41291d4b138e0f8450d4726c80480ed48b7eefd6 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 21:54:28 -0600 Subject: [PATCH 3/4] fix(db): keep successful mutation origins separate from active changes Provisional direct-handler attribution uses active local changes. Only successful mutations retain origin evidence; failure and queued sync writes no longer erase or prematurely promote it. Cover nonoptimistic sibling completion, failed-only writes, synchronous acknowledgements and the retained rollback publication law. --- packages/db/src/collection/mutations.ts | 12 +- packages/db/src/collection/state.ts | 18 +-- packages/db/src/collection/sync.ts | 4 - .../db/tests/optimistic-composition.test.ts | 107 ++++++++++++++++++ 4 files changed, 117 insertions(+), 24 deletions(-) diff --git a/packages/db/src/collection/mutations.ts b/packages/db/src/collection/mutations.ts index 9c91789780..328d481657 100644 --- a/packages/db/src/collection/mutations.ts +++ b/packages/db/src/collection/mutations.ts @@ -174,11 +174,13 @@ export class CollectionMutationsManager< return `KEY::${this.id}/${key}` } - private markPendingLocalOrigins( + private markPendingLocalChanges( mutations: Array>, ): void { for (const mutation of mutations) { - this.state.pendingLocalOrigins.add(mutation.key as TKey) + // The handler can sync synchronously before its transaction is registered. + // This is provisional; only completed mutations retain a local origin. + this.state.pendingLocalChanges.add(mutation.key as TKey) } } @@ -267,7 +269,7 @@ export class CollectionMutationsManager< // Apply mutations to the new transaction directOpTransaction.applyMutations(mutations) - this.markPendingLocalOrigins(mutations) + this.markPendingLocalChanges(mutations) // Errors still reject tx.isPersisted.promise; this catch only prevents global unhandled rejections directOpTransaction.commit().catch(() => undefined) @@ -464,7 +466,7 @@ export class CollectionMutationsManager< // Apply mutations to the new transaction directOpTransaction.applyMutations(mutations) - this.markPendingLocalOrigins(mutations) + this.markPendingLocalChanges(mutations) // Errors still hit tx.isPersisted.promise; avoid leaking an unhandled rejection from the fire-and-forget commit directOpTransaction.commit().catch(() => undefined) @@ -568,7 +570,7 @@ export class CollectionMutationsManager< // Apply mutations to the new transaction directOpTransaction.applyMutations(mutations) - this.markPendingLocalOrigins(mutations) + this.markPendingLocalChanges(mutations) // Errors still reject tx.isPersisted.promise; silence the internal commit promise to prevent test noise directOpTransaction.commit().catch(() => undefined) diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index f448d124e2..55210b0b59 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -122,6 +122,8 @@ export class CollectionStateManager< * When sync confirms data for a key with pending local changes, it keeps 'local' origin. */ public pendingLocalChanges = new Set() + // Successful mutations retain attribution until sync applies. Active or + // failed mutations must not add to, or erase a sibling's entry in, this set. public pendingLocalOrigins = new Set() private virtualPropsCache = new WeakMap< @@ -521,7 +523,7 @@ export class CollectionStateManager< const previousDeletes = new Set(this.optimisticDeletes) const previousRowOrigins = this.rowOrigins - // Update pending optimistic state for completed/failed transactions + // Retain successful contributions; failed/active work is recomputed below. for (const transaction of this.transactions.values()) { const isDirectTransaction = transaction.metadata[DIRECT_TRANSACTION_METADATA_KEY] === true @@ -588,20 +590,6 @@ export class CollectionStateManager< break } } - } else if (transaction.state === `failed`) { - for (const mutation of transaction.mutations) { - if (!this.isThisCollection(mutation.collection)) { - continue - } - // Failed transactions never enter retained state. A same-key entry - // belongs to a successful sibling and must survive this rollback. - if ( - !this.pendingOptimisticUpserts.has(mutation.key) && - !this.pendingOptimisticDeletes.has(mutation.key) - ) { - this.pendingLocalOrigins.delete(mutation.key) - } - } } } diff --git a/packages/db/src/collection/sync.ts b/packages/db/src/collection/sync.ts index a5b02f7a6c..615427d14d 100644 --- a/packages/db/src/collection/sync.ts +++ b/packages/db/src/collection/sync.ts @@ -175,10 +175,6 @@ export class CollectionSyncManager< key = this.config.getKey(messageWithOptionalKey.value) } - if (this.state.pendingLocalChanges.has(key)) { - this.state.pendingLocalOrigins.add(key) - } - let messageType = messageWithOptionalKey.type // Check if an item with this key already exists when inserting diff --git a/packages/db/tests/optimistic-composition.test.ts b/packages/db/tests/optimistic-composition.test.ts index 0d674b56e2..fddcf23477 100644 --- a/packages/db/tests/optimistic-composition.test.ts +++ b/packages/db/tests/optimistic-composition.test.ts @@ -65,6 +65,113 @@ const cases = orders.flatMap((order) => ) describe(`optimistic field composition`, () => { + it.each([false, true])( + `does not attribute a later remote write to a failed-only update, optimistic=%s`, + async (optimistic) => { + const done = createDeferred() + const fixture = source(() => done.promise) + await fixture.collection.preload() + const tx = fixture.collection.update(1, { optimistic }, (row) => { + row.a = `failed` + }) + const settled = tx.isPersisted.promise.catch(() => {}) + try { + done.reject(new Error(`failed update`)) + await settled + fixture.sync.begin() + fixture.sync.write({ + type: `update`, + value: { ...initial, a: `remote` }, + }) + await fixture.sync.commit() + expect(fixture.collection.get(1)?.$origin).toBe(`remote`) + } finally { + done.resolve() + await fixture.collection.cleanup() + } + }, + ) + + it.each([false, true])( + `attributes synchronous handler acknowledgement to its local update, optimistic=%s`, + async (optimistic) => { + const fixture = source(() => { + fixture.sync.begin() + fixture.sync.write({ + type: `update`, + value: { ...initial, a: `accepted` }, + }) + void fixture.sync.commit() + return Promise.resolve() + }) + await fixture.collection.preload() + try { + const tx = fixture.collection.update(1, { optimistic }, (row) => { + row.a = `accepted` + }) + await tx.isPersisted.promise + expect(fixture.collection.get(1)?.$origin).toBe(`local`) + expect(stripVirtualProps(fixture.collection.get(1))).toEqual({ + ...initial, + a: `accepted`, + }) + } finally { + await fixture.collection.cleanup() + } + }, + ) + + it.each([false, true])( + `keeps completed nonoptimistic origin across sibling rollback, completed first=%s`, + async (completedFirst) => { + const done = [createDeferred(), createDeferred()] + let calls = 0 + const fixture = source(() => done[calls++]!.promise) + await fixture.collection.preload() + const first = fixture.collection.update( + 1, + { optimistic: false }, + (row) => { + row.a = `accepted` + }, + ) + const second = fixture.collection.update(1, (row) => { + row.b = `rejected` + }) + const secondSettled = second.isPersisted.promise.catch(() => {}) + try { + if (completedFirst) { + done[0]!.resolve() + await first.isPersisted.promise + } + done[1]!.reject(new Error(`sibling failure`)) + await secondSettled + if (!completedFirst) { + done[0]!.resolve() + await first.isPersisted.promise + } + fixture.sync.begin() + fixture.sync.write({ + type: `update`, + value: { ...initial, a: `accepted` }, + }) + await fixture.sync.commit() + expect(fixture.collection.get(1)?.$origin).toBe(`local`) + expect(stripVirtualProps(fixture.collection.get(1))).toEqual({ + ...initial, + a: `accepted`, + }) + } finally { + done.forEach((entry) => entry.resolve()) + await Promise.allSettled([ + first.isPersisted.promise, + second.isPersisted.promise, + ]) + await fixture.collection.cleanup() + } + }, + ) + it.each([false, true])( `settles a rebased update with queued confirmation=%s`, async (confirm) => { From fee70f9ecf6cbf052c4149fbc205bdd830b55081 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Fri, 11 Sep 2026 07:28:25 -0600 Subject: [PATCH 4/4] fix(db): preserve optimistic snapshots across settlement and sync Keep captured whole-row snapshots instead of rebasing their fields onto synced rows. Track the exact creating-insert dependency for retained updates, preserve local attribution through rollback and truncate, and retire completed snapshots before rebuilding the active overlay. Generalize regressions into shared model-driven histories that compare reads, event reconstruction, downstream queries, request snapshots, and publication counts. Build queued membership lazily for balanced graph deltas. Verified 5,263 core tests and a 100x campaign of 36,000 generated histories, plus standalone types, lint, and formatting. Sync queue timing is unchanged. --- .../fix-optimistic-field-reconciliation.md | 2 +- AGENTS.md | 17 +- packages/db/src/collection/state.ts | 275 ++++++----- packages/db/src/query/live/ARCHITECTURE.md | 33 +- .../query/live/collection-config-builder.ts | 6 +- ...on-state-retention-oracle.property.test.ts | 114 ++++- .../tests/collection-sync-reentrancy.test.ts | 2 + .../db/tests/optimistic-composition.test.ts | 41 +- .../db/tests/optimistic-history-oracle.ts | 428 ++++++++++++++++++ .../optimistic-settlement-boundaries.test.ts | 150 ++++++ packages/db/tests/oracle-config.ts | 2 + .../derived-delete-reconciliation.test.ts | 86 +++- 12 files changed, 1000 insertions(+), 156 deletions(-) create mode 100644 packages/db/tests/optimistic-history-oracle.ts create mode 100644 packages/db/tests/optimistic-settlement-boundaries.test.ts diff --git a/.changeset/fix-optimistic-field-reconciliation.md b/.changeset/fix-optimistic-field-reconciliation.md index 140b570cf1..0d96935419 100644 --- a/.changeset/fix-optimistic-field-reconciliation.md +++ b/.changeset/fix-optimistic-field-reconciliation.md @@ -2,4 +2,4 @@ '@tanstack/db': patch --- -Compose optimistic updates from their changed fields so concurrent updates and rollbacks preserve unrelated changes. Keep source updates beneath an optimistic live-query delete when queued sync batches apply, without changing sync queue timing. +Preserve whole-row optimistic snapshots through sync and truncate. Fix insert-dependent update settlement, local origin tracking, and rollback publication while sibling requests remain pending. Keep source updates beneath an optimistic live-query delete when queued sync batches apply, without changing sync queue timing. diff --git a/AGENTS.md b/AGENTS.md index 1d6fbf182b..23285c3c9a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -362,7 +362,8 @@ const dependentBuilders = [] // Accurately describes dependents ### Always Add Tests for Bugs -**Key Principle:** If you're fixing a bug, add a unit test that reproduces the bug before fixing it. This ensures: +**Key Principle:** Reproduce a bug in a test before fixing it. Prefer extending +an oracle as described below over adding an isolated unit test. This ensures: - The bug is actually fixed - The bug doesn't regress in the future @@ -389,6 +390,20 @@ classifier, fixture, or assertion that let it pass. Use that analysis to suggest the smallest test or oracle improvement that would catch the same class of bug, not only the reported example. +### Prefer Oracle Coverage Over Isolated Regressions + +An oracle that checks general laws across generated states and histories is a +stronger form of coverage than a unit test for one specific example. Prefer +extending an existing oracle when it can cover the behavior. Add the missing +model rule, generator dimension, state transition, or observable assertion; +adding more pinned examples alone does not generalize the oracle. + +Use a focused regression to isolate and shrink a failure, then keep it as a +replay example for the broader oracle where possible. Verify that the expanded +oracle fails without the fix and passes with it. Keep valuable unit tests, but +do not treat them as a substitute for applicable oracle coverage. If an oracle +is not practical for the behavior, explain why a focused test is sufficient. + ### Name Tests After Behavior Test names should state the behavior they prove. Do not put issue or pull diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index 55210b0b59..aac2e7c84f 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -54,8 +54,8 @@ type PendingMetadataWrite = { type: `set`; value: unknown } | { type: `delete` } type OptimisticUpsert = Pick< PendingMutation, - `key` | `type` | `modified` | `changes` -> + `key` | `modified` +> & { insert?: object } type InternalChangeMessage< T extends object = Record, @@ -95,10 +95,6 @@ export class CollectionStateManager< public optimisticUpserts = new Map() public optimisticDeletes = new Set() - private composedUpserts = new WeakMap< - object, - { base: TOutput; composed: TOutput } - >() public pendingOptimisticUpserts = new Map>() public pendingOptimisticDeletes = new Set() public pendingOptimisticDirectUpserts = new Set() @@ -523,6 +519,9 @@ export class CollectionStateManager< const previousDeletes = new Set(this.optimisticDeletes) const previousRowOrigins = this.rowOrigins + // A retained update can depend on an unconfirmed insert, not just its key. + // Keep that exact dependency so settlement and key reuse cannot conflate rows. + const pendingInserts = new Map() // Retain successful contributions; failed/active work is recomputed below. for (const transaction of this.transactions.values()) { const isDirectTransaction = @@ -548,24 +547,20 @@ export class CollectionStateManager< switch (mutation.type) { case `insert`: case `update`: { - // Retain only this completed mutation's contribution. Other - // active updates may roll back after this one has settled. + // Retain whole snapshots, not fields rebased over newer synced + // values. A slow insert must not replace its accepted dependent + // update with the older insertion snapshot. const previous = this.pendingOptimisticUpserts.get(mutation.key) - const changes = - mutation.type === `update` - ? { ...previous?.changes, ...mutation.changes } - : mutation.changes + const confirmsInsert = previous?.insert === mutation this.pendingOptimisticUpserts.set(mutation.key, { key: mutation.key, - type: + modified: confirmsInsert + ? previous.modified + : mutation.modified, + insert: mutation.type === `update` - ? (previous?.type ?? `update`) - : `insert`, - modified: - mutation.type === `update` && previous?.type === `insert` - ? { ...previous.modified, ...mutation.changes } - : mutation.modified, - changes, + ? (previous?.insert ?? pendingInserts.get(mutation.key)) + : undefined, }) this.pendingOptimisticDeletes.delete(mutation.key) if (isDirectTransaction) { @@ -590,6 +585,28 @@ export class CollectionStateManager< break } } + } else { + for (const mutation of transaction.mutations) { + if ( + !this.isThisCollection(mutation.collection) || + mutation.type !== `insert` || + !mutation.optimistic + ) + continue + if ( + transaction.state !== `failed` && + !this.acknowledgedInserts.has(mutation) + ) { + pendingInserts.set(mutation.key, mutation) + } else if ( + this.pendingOptimisticUpserts.get(mutation.key)?.insert === mutation + ) { + // Drop only the dependent row, never a later same-key insertion or + // successful sibling attribution. Failed transactions remain listed. + this.pendingOptimisticUpserts.delete(mutation.key) + this.pendingOptimisticDirectUpserts.delete(mutation.key) + } + } } } @@ -705,7 +722,7 @@ export class CollectionStateManager< // Filter out redundant delete events if there are pending sync transactions // that will immediately restore the same data, but only for completed transactions // IMPORTANT: Skip complex filtering for user-triggered actions to prevent UI blocking - if (this.pendingSyncedTransactions.length > 0 && !triggeredByUserAction) { + if (this.changes.shouldBatchEvents && !triggeredByUserAction) { const pendingSyncKeysForFilter = new Set() // Collect keys from pending sync operations @@ -819,7 +836,9 @@ export class CollectionStateManager< } else if ( previousValue !== undefined && currentValue !== undefined && - previousValue !== currentValue + (!deepEquals(previousValue, currentValue) || + previousVirtualProps.$origin !== nextVirtualProps.$origin || + previousVirtualProps.$synced !== nextVirtualProps.$synced) ) { events.push({ type: `update`, @@ -835,24 +854,16 @@ export class CollectionStateManager< } } - // Updates own only their changed top-level fields. Inserts keep the full - // validated row, including schema defaults absent from `changes`. + // Optimistic mutations keep their full validated snapshot. Mixing in newer + // synced fields could produce a row neither the user nor the server created. private resolveOptimisticUpsert( mutation: OptimisticUpsert, ): TOutput { - if (mutation.type !== `update`) return mutation.modified - const key = mutation.key as TKey - const base = this.optimisticUpserts.get(key) ?? this.syncedData.get(key) - if (base === undefined) return mutation.modified - - // Reuse an unchanged composition so unrelated transactions do not publish - // new row identities. A new changes object or base invalidates the cache. - const cached = this.composedUpserts.get(mutation.changes) - if (cached?.base === base) return cached.composed - const composed = { ...base, ...mutation.changes } - this.composedUpserts.set(mutation.changes, { base, composed }) - return composed + const dependent = this.pendingOptimisticUpserts.get(key) + return dependent?.insert === mutation + ? dependent.modified + : mutation.modified } /** Build once per output flush; queued membership excludes optimistic edits. */ @@ -1118,22 +1129,27 @@ export class CollectionStateManager< }) } + // Attribution belongs to the whole atomic batch. A repeated write must + // not forget the local acknowledgement consumed by its first operation. + const localKeys = new Set() for (const operation of transaction.operations) { const key = operation.key as TKey // Determine origin: 'local' for local-only collections or pending local changes const retainedLocalOrigin = - (truncatePendingLocalChanges?.has(key) === true || - truncatePendingLocalOrigins?.has(key) === true) && - !completedDirectUpserts.has(key) && - !completedDirectDeletes.has(key) + truncatePendingLocalChanges?.has(key) === true || + (truncatePendingLocalOrigins?.has(key) === true && + !completedDirectUpserts.has(key) && + !completedDirectDeletes.has(key)) const origin: VirtualOrigin = this.isLocalOnly || this.pendingLocalChanges.has(key) || this.pendingLocalOrigins.has(key) || + localKeys.has(key) || retainedLocalOrigin ? 'local' : 'remote' + if (origin === `local`) localKeys.add(key) // Update synced data switch (operation.type) { @@ -1208,69 +1224,45 @@ export class CollectionStateManager< } } - // After applying synced operations, if this commit included a truncate, - // re-apply optimistic mutations on top of the fresh synced base. This ensures - // the UI preserves local intent while respecting server rebuild semantics. - // Ordering: deletes (above) -> server ops (just applied) -> optimistic upserts. - if (hasTruncateSync) { - // Build re-apply sets from the snapshot taken at the start of this function. - // This prevents losing optimistic state if transactions complete during truncate processing. - const reapplyUpserts = new Map( - truncateOptimisticSnapshot!.upserts, - ) - const reapplyDeletes = new Set( - truncateOptimisticSnapshot!.deletes, + // A completed optimistic insert may have used a temporary client key while + // the sync confirmation used a different server-generated key. Once a + // sync commit has been applied, stop retaining completed optimistic keys + // that were not confirmed by this commit so the temporary row is removed. + for (const key of this.pendingOptimisticDirectUpserts) { + // Truncate republishes this captured snapshot. Keep its existing + // retention marker so the next sync can also publish its removal. + if ( + hasTruncateSync && + truncateOptimisticSnapshot?.upserts.has(key) && + !changedKeys.has(key) ) - // A same-key authoritative row confirms a completed direct mutation. - // Keep active optimistic work, but do not restore a completed client - // value over the row that just replaced it. - for (const key of completedDirectUpserts) { - if (changedKeys.has(key)) reapplyUpserts.delete(key) - } - for (const key of completedDirectDeletes) { - if (changedKeys.has(key)) reapplyDeletes.delete(key) - } - - // Emit inserts for re-applied upserts, skipping any keys that have an optimistic delete. - // If the server also inserted/updated the same key in this batch, override that value - // with the optimistic value to preserve local intent. - for (const [key, value] of reapplyUpserts) { - if (reapplyDeletes.has(key)) continue - if (syncedInsertedOrUpdatedKeys.has(key)) { - let foundInsert = false - for (let i = events.length - 1; i >= 0; i--) { - const evt = events[i]! - if (evt.key === key && evt.type === `insert`) { - evt.value = value - foundInsert = true - break - } - } - if (!foundInsert) { - events.push({ type: `insert`, key, value }) - } - } else { - events.push({ type: `insert`, key, value }) - } - } - - // Finally, ensure we do NOT insert keys that have an outstanding optimistic delete. - if (events.length > 0 && reapplyDeletes.size > 0) { - const filtered: Array> = [] - for (const evt of events) { - if (evt.type === `insert` && reapplyDeletes.has(evt.key)) { - continue + continue + if (!changedKeys.has(key)) { + changedKeys.add(key) + if (!currentVisibleState.has(key)) { + const previousValue = previousOptimisticUpserts.get(key) + if (previousValue !== undefined) { + currentVisibleState.set(key, previousValue) } - filtered.push(evt) } - events.length = 0 - events.push(...filtered) + this.pendingOptimisticUpserts.delete(key) + this.pendingLocalOrigins.delete(key) } - - // Ensure listeners are active before emitting this critical batch - if (this.lifecycle.status !== `ready`) { - this.lifecycle.markReady() + this.pendingOptimisticDirectUpserts.delete(key) + } + for (const key of this.pendingOptimisticDirectDeletes) { + if ( + hasTruncateSync && + truncateOptimisticSnapshot?.deletes.has(key) && + !changedKeys.has(key) + ) + continue + if (!changedKeys.has(key)) { + changedKeys.add(key) } + this.pendingOptimisticDeletes.delete(key) + this.pendingLocalOrigins.delete(key) + this.pendingOptimisticDirectDeletes.delete(key) } // Maintain optimistic state appropriately @@ -1300,6 +1292,10 @@ export class CollectionStateManager< for (const transaction of this.transactions.values()) { if (![`completed`, `failed`].includes(transaction.state)) { for (const mutation of transaction.mutations) { + // Truncate clears attribution with the old base, not the still-live + // local requests. Preserve them for later source acknowledgements. + if (this.isThisCollection(mutation.collection)) + this.pendingLocalChanges.add(mutation.key) if ( this.isThisCollection(mutation.collection) && mutation.type === `insert` && @@ -1330,32 +1326,56 @@ export class CollectionStateManager< } } - // A completed optimistic insert may have used a temporary client key while - // the sync confirmation used a different server-generated key. Once a - // sync commit has been applied, stop retaining completed optimistic keys - // that were not confirmed by this commit so the temporary row is removed. - for (const key of this.pendingOptimisticDirectUpserts) { - if (!changedKeys.has(key)) { - changedKeys.add(key) - if (!currentVisibleState.has(key)) { - const previousValue = previousOptimisticUpserts.get(key) - if (previousValue !== undefined) { - currentVisibleState.set(key, previousValue) + // After applying synced operations, if this commit included a truncate, + // re-apply optimistic mutations on top of the fresh synced base. This ensures + // the UI preserves local intent while respecting server rebuild semantics. + // Ordering: deletes (above) -> server ops (just applied) -> optimistic upserts. + if (hasTruncateSync) { + // Events use the same rebuilt overlay as synchronous reads. + const reapplyUpserts = this.optimisticUpserts + const reapplyDeletes = this.optimisticDeletes + + // Emit inserts for re-applied upserts, skipping any keys that have an optimistic delete. + // If the server also inserted/updated the same key in this batch, override that value + // with the optimistic value to preserve local intent. + for (const [key, value] of reapplyUpserts) { + if (reapplyDeletes.has(key)) continue + if (syncedInsertedOrUpdatedKeys.has(key)) { + let foundInsert = false + for (let i = events.length - 1; i >= 0; i--) { + const evt = events[i]! + if (evt.key === key && evt.type === `insert`) { + evt.value = value + foundInsert = true + break + } } + if (!foundInsert) { + events.push({ type: `insert`, key, value }) + } + } else { + events.push({ type: `insert`, key, value }) } - this.pendingOptimisticUpserts.delete(key) - this.pendingLocalOrigins.delete(key) } - } - for (const key of this.pendingOptimisticDirectDeletes) { - if (!changedKeys.has(key)) { - changedKeys.add(key) + + // Finally, ensure we do NOT insert keys that have an outstanding optimistic delete. + if (events.length > 0 && reapplyDeletes.size > 0) { + const filtered: Array> = [] + for (const evt of events) { + if (evt.type === `insert` && reapplyDeletes.has(evt.key)) { + continue + } + filtered.push(evt) + } + events.length = 0 + events.push(...filtered) + } + + // Ensure listeners are active before emitting this critical batch + if (this.lifecycle.status !== `ready`) { + this.lifecycle.markReady() } - this.pendingOptimisticDeletes.delete(key) - this.pendingLocalOrigins.delete(key) } - this.pendingOptimisticDirectUpserts.clear() - this.pendingOptimisticDirectDeletes.clear() // Now check what actually changed in the final visible state for (const key of changedKeys) { @@ -1589,12 +1609,21 @@ export class CollectionStateManager< * This method should be called by the Transaction class when state changes */ public onTransactionStateChange(): void { - // Check if commitPendingTransactions will be called after this - // by checking if there are pending sync transactions (same logic as in transactions.ts) - this.changes.shouldBatchEvents = this.pendingSyncedTransactions.length > 0 + // Batch only when the next sync drain can actually publish. A persisting + // sibling can keep normal sync queued; it must not hide this rollback. + const hasPersistingTransaction = [...this.transactions.values()].some( + (transaction) => transaction.state === `persisting`, + ) + this.changes.shouldBatchEvents = this.pendingSyncedTransactions.some( + (transaction) => + transaction.committed && + (!hasPersistingTransaction || + transaction.immediate || + transaction.truncate), + ) // CRITICAL: Capture visible state BEFORE clearing optimistic state - this.capturePreSyncVisibleState() + if (this.changes.shouldBatchEvents) this.capturePreSyncVisibleState() this.recomputeOptimisticState(false) } diff --git a/packages/db/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index 00f18f0ea0..99a94be323 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -1005,18 +1005,27 @@ Classify root deltas against authoritative membership, including earlier queued sync writes, not the optimistic public view. An optimistic delete must not turn a balanced graph update into an authoritative delete. This does not bypass the normal sync queue or publish part of a graph-output transaction early. -Build queued membership once per output flush, preserving committed last-write -and truncate semantics, rather than scanning the queue again for each row. - -At the Collection boundary, each active or retained optimistic update owns only its changed -top-level fields. Compose those fields in transaction order over the current -base and earlier optimistic work. Inserts retain their full validated rows, -including schema defaults. Removing one update must not revive its fields from -another update's whole-row snapshot. -Completed contributions remain beneath active transactions under the existing -retention policy until sync retires them. A failed sibling cannot clear those -contributions. Sync publication compares actual previous and next visible rows, -not captured mutation snapshots that may predate a rebase. +Build queued membership lazily on the first balanced delta in an output flush, +preserving committed last-write and truncate semantics. Insert-only flushes do +not scan the queue, and balanced rows share that flush's lookup. + +At the Collection boundary, optimistic mutations own whole validated row +snapshots, including fields they did not change and insert schema defaults. +Do not merge newer synced fields into those snapshots: that could publish a +combination neither the mutation nor the server created. This applies to both +ordinary sync and truncate. The mutation payload stays unchanged as well. +Active snapshots are selected in transaction order. Completed snapshots remain +beneath active transactions under the existing retention policy until sync +retires them. A later snapshot may contain values seen from an earlier sibling; +rolling back that sibling does not rewrite the later snapshot. Sync publication +compares actual previous and next visible rows, not just mutation identities. +An update made over an unconfirmed insert retains that exact insert dependency, +not just its key. Insert success preserves the later completed snapshot; insert +failure removes the already-retained dependent row. An independently submitted +update accepted after that failure still retains its own snapshot. An +acknowledged insert or a later same-key +insertion is not the failed insertion. Truncate replay derives events and reads +from the same snapshot overlay, without merging in its new authoritative fields. Installed state, synchronous reads, change-event payloads, and downstream queries must all observe the same fully materialized commit. The facade adapter diff --git a/packages/db/src/query/live/collection-config-builder.ts b/packages/db/src/query/live/collection-config-builder.ts index 4eb30d73ad..dd29f95636 100644 --- a/packages/db/src/query/live/collection-config-builder.ts +++ b/packages/db/src/query/live/collection-config-builder.ts @@ -1058,7 +1058,11 @@ export class CollectionConfigBuilder< facadePublication.prepare() if (hasParentChanges) { begin() - const hasSyncedKey = config.collection._state.createSyncedKeyLookup() + let lookup: ((key: string | number) => boolean) | undefined + const hasSyncedKey = (key: string | number) => { + lookup ??= config.collection._state.createSyncedKeyLookup() + return lookup(key) + } changesToApply.forEach( this.applyChanges.bind(this, config, hasSyncedKey), ) diff --git a/packages/db/tests/collection-state-retention-oracle.property.test.ts b/packages/db/tests/collection-state-retention-oracle.property.test.ts index 587def1a7e..cd091d6b21 100644 --- a/packages/db/tests/collection-state-retention-oracle.property.test.ts +++ b/packages/db/tests/collection-state-retention-oracle.property.test.ts @@ -3,7 +3,9 @@ import { expect, it } from 'vitest' import { createCollection } from '../src/collection/index.js' import { DuplicateKeySyncError } from '../src/errors.js' import { createTransaction } from '../src/transactions.js' -import { oraclePropertyOptions } from './oracle-config.js' +import { oraclePropertyOptions, oracleRuns } from './oracle-config.js' +import { runOptimisticHistory } from './optimistic-history-oracle.js' +import type { OptimisticStep } from './optimistic-history-oracle.js' import type { Collection } from '../src/collection/index.js' import type { SyncConfig, TransactionState } from '../src/types.js' @@ -733,3 +735,113 @@ fcTest.prop( await runRetentionHistory(actions) }, ) + +const historyRow = fc.record({ + id: fc.integer({ min: 1, max: 3 }), + a: fc.integer({ min: -2, max: 2 }), + b: fc.integer({ min: -2, max: 2 }), + c: fc.integer({ min: -2, max: 2 }), +}) +const optimisticStep: fc.Arbitrary = fc.oneof( + { + weight: 4, + arbitrary: fc.record({ + type: fc.constant(`edit` as const), + key: fc.integer({ min: 1, max: 3 }), + fields: fc + .record( + { + a: fc.integer({ min: -2, max: 2 }), + b: fc.integer({ min: -2, max: 2 }), + c: fc.integer({ min: -2, max: 2 }), + }, + { requiredKeys: [] }, + ) + .filter((fields) => Object.keys(fields).length > 0), + optimistic: fc.boolean(), + }), + }, + { + weight: 4, + arbitrary: fc.record({ + type: fc.constant(`settle` as const), + slot: fc.nat(5), + success: fc.boolean(), + cascade: fc.boolean(), + }), + }, + { + weight: 3, + arbitrary: fc.record({ + type: fc.constant(`sync` as const), + rows: fc.uniqueArray(historyRow, { + selector: (row) => row.id, + maxLength: 3, + }), + truncate: fc.boolean(), + immediate: fc.boolean(), + copies: fc.integer({ min: 1, max: 2 }), + }), + }, +) +const optimisticHistory = fc.record({ + initial: fc.uniqueArray(historyRow, { + selector: (row) => row.id, + maxLength: 3, + }), + steps: fc.array(optimisticStep, { minLength: 2, maxLength: 24 }), +}) + +// These are replay programs for the same model and driver as randomized runs, +// not separate assertions that only know the reported final state. +const insertionPrefix: Array = [ + { type: `edit`, key: 1, fields: { a: 1 }, optimistic: true }, + { type: `edit`, key: 1, fields: { b: 2 }, optimistic: true }, + { type: `settle`, slot: 1, success: true, cascade: false }, +] +it.each([true, false])( + `replays insert dependency settlement, accepted=%s`, + async (success) => { + await runOptimisticHistory( + [], + [ + ...insertionPrefix, + { type: `settle`, slot: 0, success, cascade: false }, + ], + ) + }, +) +it.each([true, false])( + `preserves a whole-row mutation snapshot across sync, truncate=%s`, + async (truncate) => { + await runOptimisticHistory( + [{ id: 1, a: 0, b: 0, c: 0 }], + [ + { type: `edit`, key: 1, fields: { a: 1 }, optimistic: true }, + { + type: `sync`, + rows: [{ id: 1, a: 0, b: 2, c: 3 }], + immediate: !truncate, + truncate, + copies: 1, + }, + { type: `settle`, slot: 0, success: true, cascade: false }, + ], + ) + }, +) +fcTest.prop([optimisticHistory], { numRuns: oracleRuns(60), seed: 86103 })( + `matches optimistic ownership and publication histories with a fixed seed`, + async ({ initial, steps }) => { + await runOptimisticHistory(initial, steps) + }, +) +fcTest.prop( + [optimisticHistory], + oraclePropertyOptions(100, `collection-state.optimistic-history`), +)( + `matches optimistic ownership and publication histories with a random or replayed seed`, + async ({ initial, steps }) => { + await runOptimisticHistory(initial, steps) + }, +) diff --git a/packages/db/tests/collection-sync-reentrancy.test.ts b/packages/db/tests/collection-sync-reentrancy.test.ts index 9d7bc1e66c..9ed3b80cf0 100644 --- a/packages/db/tests/collection-sync-reentrancy.test.ts +++ b/packages/db/tests/collection-sync-reentrancy.test.ts @@ -939,6 +939,8 @@ describe(`sync publication reentrancy`, () => { expect(collection._layoutRevision).toBe(revisionBeforeDrain + 1) expect(callbacks).toEqual([ { + // Reapply whole snapshots in transaction order. Public + // layout, batch membership/count and receipt timing stay unchanged. changes: [2, 1, 3, 1, 3, 1, 2], keys: [2, 1, 3], values: [`two`, `optimistic-one`, `optimistic-three`], diff --git a/packages/db/tests/optimistic-composition.test.ts b/packages/db/tests/optimistic-composition.test.ts index fddcf23477..1ad03a2ebc 100644 --- a/packages/db/tests/optimistic-composition.test.ts +++ b/packages/db/tests/optimistic-composition.test.ts @@ -64,7 +64,7 @@ const cases = orders.flatMap((order) => ), ) -describe(`optimistic field composition`, () => { +describe(`optimistic snapshot ownership`, () => { it.each([false, true])( `does not attribute a later remote write to a failed-only update, optimistic=%s`, async (optimistic) => { @@ -173,7 +173,7 @@ describe(`optimistic field composition`, () => { ) it.each([false, true])( - `settles a rebased update with queued confirmation=%s`, + `keeps its captured snapshot until queued confirmation=%s`, async (confirm) => { const done = createDeferred() const fixture = source(() => done.promise) @@ -201,7 +201,7 @@ describe(`optimistic field composition`, () => { fixture.sync.begin({ immediate: true }) fixture.sync.write({ type: `update`, value: { ...initial, b: `b1` } }) expect(fixture.sync.commit()).toBe(true) - check({ ...initial, a: `a1`, b: `b1` }) + check({ ...initial, a: `a1` }) let applied: true | Promise = true if (confirm) { fixture.sync.begin() @@ -212,7 +212,7 @@ describe(`optimistic field composition`, () => { done.resolve() await tx.isPersisted.promise await applied - check({ ...initial, a: `a1`, b: confirm ? `b0` : `b1` }) + check({ ...initial, a: `a1` }) } finally { done.resolve() subscription.unsubscribe() @@ -223,7 +223,7 @@ describe(`optimistic field composition`, () => { ) it.each(orders)( - `retains disjoint direct fields across settlement order %j`, + `selects whole direct snapshots across settlement order %j`, async (...order) => { const done = [ createDeferred(), @@ -250,10 +250,15 @@ describe(`optimistic field composition`, () => { Object.assign(row, patch) }), ) - const expected = { ...initial, a: `a1`, b: `b2`, c: `c3` } + const snapshots = patches.map((_, index) => + Object.assign({}, initial, ...patches.slice(0, index + 1)), + ) + const active = new Set([0, 1, 2]) for (const index of order) { done[index]!.resolve() await transactions[index]!.isPersisted.promise + active.delete(index) + const expected = snapshots[active.size ? Math.max(...active) : index] expect(stripVirtualProps(fixture.collection.get(1))).toEqual(expected) expect(replica.get(1)).toEqual(expected) expect(stripVirtualProps(downstream.get(1))).toEqual(expected) @@ -268,7 +273,7 @@ describe(`optimistic field composition`, () => { ) it.each([`before`, `after`] as const)( - `retains only a direct survivor's fields when it completes %s sibling rollback`, + `retains a direct survivor's captured snapshot when it completes %s sibling rollback`, async (completion) => { const done = [createDeferred(), createDeferred()] let calls = 0 @@ -301,15 +306,15 @@ describe(`optimistic field composition`, () => { if (completion === `before`) { done[1]!.resolve() await second.isPersisted.promise - check({ ...initial, a: `a1`, b: `b2` }) + check({ ...initial, a: `a1` }) } done[0]!.reject(new Error(`first update failed`)) await firstSettled - check({ ...initial, b: `b2` }) + check({ ...initial, a: `a1`, b: `b2` }) if (completion === `after`) { done[1]!.resolve() await second.isPersisted.promise - check({ ...initial, b: `b2` }) + check({ ...initial, a: `a1`, b: `b2` }) } fixture.sync.begin() fixture.sync.write({ type: `update`, value: { ...initial, b: `b2` } }) @@ -337,15 +342,21 @@ describe(`optimistic field composition`, () => { fields === `disjoint` ? [{ a: `a1` }, { b: `b2` }, { c: `c3` }] : [{ a: `a1` }, { a: `a2`, b: `b2` }, { c: `c3` }] + const snapshots: Array = [ + undefined, + undefined, + undefined, + ] const expected = (excluded = -1) => - Object.assign( - {}, + snapshots.reduce( + (last, row, index) => + row !== undefined && index !== excluded ? row : last, initial, - ...patches.filter((_, index) => index !== excluded), ) try { await fixture.collection.preload() for (const index of order) { + snapshots[index] = { ...expected(), ...patches[index] } pending[index]!.tx.mutate(() => fixture.collection.update(1, (row) => { Object.assign(row, patches[index]) @@ -373,7 +384,7 @@ describe(`optimistic field composition`, () => { ) it.each([`pending`, `persisting`] as const)( - `preserves untouched synced fields during $phase work`, + `does not merge new synced fields into a $phase mutation snapshot`, async (phase) => { const fixture = source() const entry = pendingTransaction() @@ -395,7 +406,6 @@ describe(`optimistic field composition`, () => { expect(stripVirtualProps(fixture.collection.get(1))).toEqual({ ...initial, a: `local`, - b: `remote`, }) } finally { entry.tx.rollback() @@ -427,7 +437,6 @@ describe(`optimistic field composition`, () => { expect(stripVirtualProps(fixture.collection.get(1))).toEqual({ ...initial, a: `local`, - b: `settled`, }) } finally { entry.tx.rollback() diff --git a/packages/db/tests/optimistic-history-oracle.ts b/packages/db/tests/optimistic-history-oracle.ts new file mode 100644 index 0000000000..44bcc6a20b --- /dev/null +++ b/packages/db/tests/optimistic-history-oracle.ts @@ -0,0 +1,428 @@ +import { expect } from 'vitest' +import { createCollection } from '../src/collection/index.js' +import { createDeferred } from '../src/deferred.js' +import { createLiveQueryCollection } from '../src/query/index.js' +import type { SyncConfig } from '../src/types.js' + +export type HistoryRow = { id: number; a: number; b: number; c: number } +type Fields = Partial> +export type OptimisticStep = + | { type: `edit`; key: number; fields: Fields; optimistic: boolean } + | { type: `settle`; slot: number; success: boolean; cascade: boolean } + | { + type: `sync` + rows: Array + truncate: boolean + immediate: boolean + copies: number + } + +type Intent = { + key: number + kind: `insert` | `update` + fields: Fields + snapshot: HistoryRow + optimistic: boolean + dependency?: number + state: `active` | `accepted` | `failed` + settled: number + retired: boolean + acknowledged: boolean + originPending: boolean +} +type ObservedRow = HistoryRow & { + $origin: `local` | `remote` + $synced: boolean +} + +/** Specification state is an event history, never a copy of production caches. + * Mutations own whole-row snapshots, never patches over changing synced rows. + * Accepted snapshots precede active snapshots. An insert supplies row existence; + * accepted updates dependent on it survive its success, but not its failed birth. + */ +class HistoryModel { + base = new Map() + origins = new Map() + intents: Array = [] + queue: Array> = [] + clock = 0 + + constructor(rows: Array) { + for (const row of rows) { + this.base.set(row.id, row) + this.origins.set(row.id, `remote`) + } + } + + private retained(intent: Intent) { + return ( + intent.state === `accepted` && + !intent.retired && + intent.optimistic && + (intent.dependency === undefined || + this.intents[intent.dependency]!.state !== `failed` || + this.intents[intent.dependency]!.acknowledged) + ) + } + + visible(): Map { + const result = new Map( + [...this.base].map(([key, row]) => [ + key, + { ...row, $origin: this.origins.get(key)!, $synced: true }, + ]), + ) + const accepted = this.intents + .filter((intent) => this.retained(intent)) + .sort((a, b) => a.settled - b.settled) + const active = this.intents.filter( + (intent) => intent.state === `active` && intent.optimistic, + ) + const apply = (intent: Intent) => { + result.set(intent.key, { + ...intent.snapshot, + $origin: `local`, + $synced: false, + }) + if (intent.kind === `insert` && !intent.acknowledged) { + // An accepted dependent snapshot belongs after its creating insert even + // when transport completion occurs in the opposite order. + for (const child of accepted) { + if (child.dependency === this.intents.indexOf(intent)) { + result.set(intent.key, { + ...child.snapshot, + $origin: `local`, + $synced: false, + }) + } + } + } + } + for (const intent of [...accepted, ...active]) apply(intent) + return result + } + + edit(step: Extract): number | undefined { + const row = this.visible().get(step.key) + if ( + row && + Object.entries(step.fields).every( + ([key, value]) => row[key as keyof Fields] === value, + ) + ) + return + const kind = row ? `update` : `insert` + const snapshot = { + ...(row ?? { id: step.key, a: 0, b: 0, c: 0 }), + ...step.fields, + } + const dependency = this.intents.reduce( + (previous, intent, index) => + kind === `update` && + intent.key === step.key && + intent.kind === `insert` && + intent.optimistic && + intent.state === `active` && + !intent.acknowledged + ? index + : previous, + -1, + ) + this.intents.push({ + key: step.key, + kind, + fields: step.fields, + snapshot, + optimistic: step.optimistic, + dependency: dependency < 0 ? undefined : dependency, + state: `active`, + settled: 0, + retired: false, + acknowledged: false, + originPending: false, + }) + return this.intents.length - 1 + } + + settle(index: number, success: boolean) { + const intent = this.intents[index]! + // A separately submitted update may succeed after the insert has already + // failed. That later server acceptance is not undone by an earlier failure. + if ( + success && + intent.dependency !== undefined && + this.intents[intent.dependency]!.state === `failed` + ) + intent.dependency = undefined + intent.state = success ? `accepted` : `failed` + if (intent.kind === `insert` && intent.acknowledged) intent.retired = true + intent.settled = ++this.clock + // A synced insert has already spent its acknowledgement. Completing its + // transport cannot turn the next unrelated remote write into a local one. + if (success && !(intent.kind === `insert` && intent.acknowledged)) + intent.originPending = true + // This grammar submits direct operations immediately. Rollback cascades + // affect pending (not already persisting) peer transactions, so none of + // these independently submitted requests is canceled by a sibling failure. + if (!this.intents.some((entry) => entry.state === `active`)) this.drain() + } + + sync(step: Extract) { + this.queue.push(step) + if ( + step.immediate || + step.truncate || + !this.intents.some((entry) => entry.state === `active`) + ) + this.drain() + } + + private drain() { + if (!this.queue.length) return + const localKeys = new Set( + this.intents + .filter((intent) => intent.state === `active`) + .map((intent) => intent.key), + ) + const replaced = this.queue.some((batch) => batch.truncate) + const written = new Set( + this.queue.flatMap((batch) => batch.rows.map((row) => row.id)), + ) + const retainedKeys = new Set( + this.intents + .filter((intent) => this.retained(intent)) + .map((intent) => intent.key), + ) + for (const batch of this.queue) { + if (batch.truncate) { + this.base.clear() + this.origins.clear() + // Origin is row-level attribution, not per-mutation acknowledgement. + // A truncate replacement of a retained optimistic row is remote unless + // a still-active request also owns that key. Do not invent finer + // acknowledgement matching between completed same-key requests. + for (const intent of this.intents) + if (intent.state === `accepted` && retainedKeys.has(intent.key)) + intent.originPending = false + } + for (const row of batch.rows) { + const local = + this.intents.some( + (intent) => intent.key === row.id && intent.originPending, + ) || localKeys.has(row.id) + this.base.set(row.id, row) + this.origins.set(row.id, local ? `local` : `remote`) + localKeys.delete(row.id) + for (const intent of this.intents) { + if (intent.key === row.id) intent.originPending = false + if ( + intent.key === row.id && + intent.kind === `insert` && + intent.state === `active` + ) + intent.acknowledged = true + } + } + // Truncate retains attribution only for rows in its own replacement, + // not for an unrelated future write after the old source was cleared. + if (batch.truncate) + for (const intent of this.intents) intent.originPending = false + } + // Ordinary source publication retires completed direct snapshots, including + // temporary keys. Truncate preserves snapshots omitted from its replacement. + for (const intent of this.intents) + if (intent.state === `accepted`) { + intent.retired ||= !replaced || written.has(intent.key) + if (retainedKeys.has(intent.key)) intent.originPending = false + } + this.queue = [] + } +} + +const plain = ({ id, a, b, c }: HistoryRow): HistoryRow => ({ id, a, b, c }) +const observed = ( + row: HistoryRow & { $origin: `local` | `remote`; $synced: boolean }, +): ObservedRow => ({ + ...plain(row), + $origin: row.$origin, + $synced: row.$synced, +}) +const sorted = (rows: Iterable) => + [...rows].sort((a, b) => a.id - b.id) + +export async function runOptimisticHistory( + initial: Array, + steps: ReadonlyArray, +) { + const model = new HistoryModel(initial) + let sync!: Parameters[`sync`]>[0] + let starting: ReturnType> | undefined + const handler = () => starting!.promise + const collection = createCollection({ + getKey: (row) => row.id, + onInsert: handler, + onUpdate: handler, + sync: { + rowUpdateMode: `full`, + sync: (actions) => { + sync = actions + actions.begin() + for (const row of initial) + actions.write({ type: `insert`, value: { ...row } }) + actions.commit() + actions.markReady() + }, + }, + }) + const downstream = createLiveQueryCollection({ + query: (q) => q.from({ row: collection }), + }) + await downstream.preload() + const replica = new Map( + [...collection.values()].map((row) => [row.id, observed(row)]), + ) + let deliveries = 0 + const sub = collection.subscribeChanges( + (batch) => { + for (const change of batch) { + deliveries++ + if (change.type === `delete`) replica.delete(Number(change.key)) + else replica.set(Number(change.key), observed(change.value)) + } + }, + { includeInitialState: true }, + ) + const operations: Array<{ + tx: + | ReturnType + | ReturnType + done: ReturnType> + outcome: Promise + }> = [] + const receipts: Array> = [] + const counts = { + edits: 0, + settlements: 0, + replacements: 0, + queued: 0, + dependencies: 0, + failures: 0, + snapshotOverrides: 0, + } + const check = (label: string) => { + for (const [index, operation] of operations.entries()) { + expect( + operation.tx.mutations[0]!.modified, + `${label}: immutable request ${index}`, + ).toMatchObject(plain(model.intents[index]!.snapshot)) + } + const expected = sorted(model.visible().values()) + const actual = sorted([...collection.values()].map(observed)) + expect( + actual, + `${label}: reads ${JSON.stringify(actual)} expected ${JSON.stringify(expected)}`, + ).toEqual(expected) + expect(sorted(replica.values()), `${label}: event replica`).toEqual( + expected, + ) + expect( + sorted([...downstream.values()].map(plain)), + `${label}: downstream`, + ).toEqual(expected.map(plain)) + } + try { + check(`initial`) + for (const [position, step] of steps.entries()) { + const before = sorted(model.visible().values()) + const deliveredBefore = deliveries + if (step.type === `edit`) { + const index = model.edit(step) + if (index === undefined) continue + const intent = model.intents[index]! + const done = createDeferred() + starting = done + const tx = + intent.kind === `insert` + ? collection.insert(plain(intent.snapshot), { + optimistic: step.optimistic, + }) + : collection.update( + step.key, + { optimistic: step.optimistic }, + (draft) => Object.assign(draft, step.fields), + ) + operations.push({ + tx, + done, + outcome: tx.isPersisted.promise.catch((error: unknown) => error), + }) + expect( + tx.mutations[0]!.modified, + `captured request snapshot`, + ).toMatchObject(plain(intent.snapshot)) + counts.edits++ + if (intent.dependency !== undefined) counts.dependencies++ + } else if (step.type === `settle`) { + const active = model.intents.flatMap((intent, index) => + intent.state === `active` ? [index] : [], + ) + if (!active.length) continue + const index = active[step.slot % active.length]! + const op = operations[index]! + model.settle(index, step.success) + if (!step.success) + op.tx.rollback({ isSecondaryRollback: !step.cascade }) + op.done.resolve() + await op.outcome + await Promise.resolve() + counts.settlements++ + if (!step.success) counts.failures++ + } else { + model.sync(step) + sync.begin({ immediate: step.immediate }) + if (step.truncate) { + sync.truncate() + counts.replacements++ + } + for (let copy = 0; copy < step.copies; copy++) { + for (const row of step.rows) + sync.write({ type: `update`, value: { ...row } }) + } + const receipt = sync.commit() + if (receipt !== true) { + receipts.push(receipt.catch((error: unknown) => error)) + counts.queued++ + } + if ( + (step.immediate || step.truncate) && + model.intents.some((intent) => intent.state === `active`) + ) + counts.snapshotOverrides++ + } + check(`${position}: ${JSON.stringify(step)}`) + // Count events as well as final values; value-only oracles miss redundant + // publications when a mutation moves into completed retention. + if ( + step.type === `settle` && + JSON.stringify(before) === + JSON.stringify(sorted(model.visible().values())) + ) { + expect(deliveries, `unchanged settlement ${position}`).toBe( + deliveredBefore, + ) + } + } + return counts + } finally { + for (const op of operations) { + if (op.tx.state === `pending` || op.tx.state === `persisting`) + op.tx.rollback({ isSecondaryRollback: true }) + op.done.resolve() + } + await Promise.all(operations.map((op) => op.outcome)) + sub.unsubscribe() + await downstream.cleanup() + await collection.cleanup() + await Promise.all(receipts) + } +} diff --git a/packages/db/tests/optimistic-settlement-boundaries.test.ts b/packages/db/tests/optimistic-settlement-boundaries.test.ts new file mode 100644 index 0000000000..409c58930b --- /dev/null +++ b/packages/db/tests/optimistic-settlement-boundaries.test.ts @@ -0,0 +1,150 @@ +import { it } from 'vitest' +import { runOptimisticHistory } from './optimistic-history-oracle.js' +import type { HistoryRow, OptimisticStep } from './optimistic-history-oracle.js' + +const row: HistoryRow = { id: 1, a: 0, b: 0, c: 0 } +const edit = ( + fields: Partial>, + optimistic = true, +): OptimisticStep => ({ type: `edit`, key: 1, fields, optimistic }) +const settle = (slot: number, success = true): OptimisticStep => ({ + type: `settle`, + slot, + success, + cascade: false, +}) +const sync = ( + rows: Array, + truncate = false, + immediate = false, + copies = 1, +): OptimisticStep => ({ type: `sync`, rows, truncate, immediate, copies }) + +// Every regression is a program for the same model, driver and checkpoint +// assertions used by generated histories. Membership work has its own generated +// law in query/derived-delete-reconciliation.test.ts. +const cases: Array<{ + name: string + initial: Array + steps: Array +}> = [ + { + name: `one confirmation for repeated queued writes`, + initial: [row], + steps: [ + edit({ a: 1 }), + sync([{ ...row, a: 1 }], false, false, 2), + settle(0), + ], + }, + { + name: `an acknowledged insert cannot remove an accepted update`, + initial: [], + steps: [ + edit({ a: 1 }), + sync([{ ...row, a: 1 }], false, true), + edit({ b: 2 }), + settle(1), + settle(0, false), + ], + }, + { + name: `failed insertion cannot remove later same-key snapshots`, + initial: [], + steps: [ + edit({ a: 1 }), + edit({ b: 1 }), + settle(1), + settle(0, false), + edit({ a: 2 }), + edit({ b: 2 }), + settle(1), + settle(0), + ], + }, + ...[true, false].map((success) => ({ + name: `dependent snapshot follows insert settlement: ${success}`, + initial: [], + steps: [edit({ a: 1 }), edit({ b: 2 }), settle(1), settle(0, success)], + })), + ...[true, false].map((truncate) => ({ + name: `failed birth does not erase accepted sibling attribution: ${truncate}`, + initial: [], + steps: [ + edit({ a: 1 }), + edit({ b: 2 }), + settle(1), + settle(0, false), + sync([row], truncate), + ], + })), + { + name: `a later accepted update survives an earlier failed insertion`, + initial: [], + steps: [edit({ a: 1 }), edit({ b: 2 }), settle(0, false), settle(0)], + }, + ...[true, false].map((truncate) => ({ + name: `sync retirement precedes rebuilding active snapshots: ${truncate}`, + initial: [], + steps: [ + edit({ a: 1 }), + edit({ b: 2 }), + settle(1), + sync([], truncate, true), + settle(0, false), + ], + })), + { + name: `unchanged persistence completion does not publish`, + initial: [row], + steps: [edit({ a: 1 }), settle(0)], + }, + { + name: `truncate preserves the captured whole row`, + initial: [row], + steps: [edit({ a: 1 }), sync([{ ...row, b: 2 }], true), settle(0)], + }, + { + name: `rollback does not rewrite a later captured request`, + initial: [row], + steps: [edit({ a: 1 }), edit({ b: 2 }), settle(0, false), settle(0)], + }, + { + name: `duplicate writes retain a local acknowledgement within a batch`, + initial: [], + steps: [edit({ a: 1 }, false), sync([{ ...row, a: 1 }], false, true, 2)], + }, + { + name: `a persisting peer does not hide rollback publication`, + initial: [], + steps: [edit({ a: 1 }, false), edit({ a: 2 }), sync([]), settle(1, false)], + }, + { + name: `truncate retains pending nonoptimistic attribution`, + initial: [], + steps: [ + edit({ a: 1 }, false), + sync([], true), + sync([{ ...row, a: 1 }], false, true), + ], + }, + { + name: `completed sibling does not erase active truncate attribution`, + initial: [], + steps: [ + edit({ a: 1 }, false), + edit({ a: 2 }), + settle(1), + sync([{ ...row, a: 1 }], true), + ], + }, + { + name: `removal of a truncate-retained snapshot reaches subscribers`, + initial: [], + steps: [edit({ a: 1 }), settle(0), sync([], true), sync([])], + }, +] + +it.each(cases)(`oracle replay: $name`, async ({ initial, steps }) => { + await runOptimisticHistory(initial, steps) +}) diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index ec7731bbdb..ecceb262a2 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -3,6 +3,8 @@ type OracleEnvironment = Record const staticOracleProperties = [ `collection-sync.reentrant-drain`, `collection-state.retention`, + `collection-state.optimistic-history`, + `derived-publication.membership-work`, `collection-publication.metadata-cancellation`, `collection-publication.metadata-only`, `collection-publication.metadata-rollback`, diff --git a/packages/db/tests/query/derived-delete-reconciliation.test.ts b/packages/db/tests/query/derived-delete-reconciliation.test.ts index eaacfad1b5..8a52552c30 100644 --- a/packages/db/tests/query/derived-delete-reconciliation.test.ts +++ b/packages/db/tests/query/derived-delete-reconciliation.test.ts @@ -1,9 +1,11 @@ -import { describe, expect, it } from 'vitest' +import { fc, test as fcTest } from '@fast-check/vitest' +import { describe, expect, it, vi } from 'vitest' import { createCollection } from '../../src/collection/index.js' import { createDeferred } from '../../src/deferred.js' import { createLiveQueryCollection } from '../../src/query/index.js' import { createTransaction } from '../../src/transactions.js' import { stripVirtualProps } from '../utils.js' +import { oraclePropertyOptions, oracleRuns } from '../oracle-config.js' import type { SyncConfig } from '../../src/types.js' type Row = { id: number; value: number } @@ -15,6 +17,88 @@ const cases = ([`pass-through`, `order`, `select`] as const).flatMap((shape) => ), ) +// The model owns source rows; production owns graph deltas and queued output. +// Work is checked at each flush, not just for one final batch size. +const flushHistory = fc.array( + fc.record({ + inserts: fc.integer({ min: 1, max: 5 }), + update: fc.boolean(), + }), + { minLength: 1, maxLength: 12 }, +) +async function runFlushHistory( + steps: Array<{ inserts: number; update: boolean }>, +) { + let sync!: Parameters[`sync`]>[0] + const expected = new Map([[1, { id: 1, value: 0 }]]) + const source = createCollection({ + getKey: (row) => row.id, + sync: { + sync: (actions) => { + sync = actions + actions.begin() + actions.write({ type: `insert`, value: expected.get(1)! }) + actions.commit() + actions.markReady() + }, + }, + }) + const derived = createLiveQueryCollection({ + query: (q) => q.from({ row: source }), + }) + const done = createDeferred() + const tx = createTransaction({ mutationFn: () => done.promise }) + const settled = tx.isPersisted.promise.catch((error: unknown) => error) + const lookup = vi.spyOn(derived._state, `createSyncedKeyLookup`) + try { + await derived.preload() + tx.mutate(() => derived.delete(1)) + for (const [index, step] of steps.entries()) { + lookup.mockClear() + sync.begin() + if (step.update) { + const row = { id: 1, value: index + 1 } + expected.set(1, row) + sync.write({ type: `update`, value: row }) + } + for (let offset = 0; offset < step.inserts; offset++) { + const row = { id: expected.size + 1, value: index } + expected.set(row.id, row) + sync.write({ type: `insert`, value: row }) + } + expect(sync.commit()).toBe(true) + expect(lookup, `flush ${index}`).toHaveBeenCalledTimes( + step.update ? 1 : 0, + ) + expect(derived._state.pendingSyncedTransactions).toHaveLength(index + 1) + expect([...derived.values()]).toEqual([]) + } + done.resolve() + await settled + expect([...derived.values()].map((row) => stripVirtualProps(row))).toEqual([ + ...expected.values(), + ]) + } finally { + done.resolve() + await settled + lookup.mockRestore() + await derived.cleanup() + await source.cleanup() + } +} + +fcTest.prop([flushHistory], { numRuns: oracleRuns(40), seed: 41703 })( + `shares membership work only for balanced deltas across fixed queue histories`, + runFlushHistory, +) +fcTest.prop( + [flushHistory], + oraclePropertyOptions(60, `derived-publication.membership-work`), +)( + `shares membership work only for balanced deltas across random queue histories`, + runFlushHistory, +) + describe(`derived updates beneath optimistic deletes`, () => { it.each([false, true])( `uses committed last writes and truncate=%s for membership`,