diff --git a/.changeset/fix-optimistic-field-reconciliation.md b/.changeset/fix-optimistic-field-reconciliation.md new file mode 100644 index 0000000000..0d96935419 --- /dev/null +++ b/.changeset/fix-optimistic-field-reconciliation.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +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/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 a74bb261c8..aac2e7c84f 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' @@ -51,6 +52,11 @@ interface PendingSyncedTransaction< type PendingMetadataWrite = { type: `set`; value: unknown } | { type: `delete` } +type OptimisticUpsert = Pick< + PendingMutation, + `key` | `modified` +> & { insert?: object } + type InternalChangeMessage< T extends object = Record, TKey extends string | number = string | number, @@ -88,7 +94,8 @@ export class CollectionStateManager< // Optimistic state tracking - make public for testing public optimisticUpserts = new Map() public optimisticDeletes = new Set() - public pendingOptimisticUpserts = new Map() + + public pendingOptimisticUpserts = new Map>() public pendingOptimisticDeletes = new Set() public pendingOptimisticDirectUpserts = new Set() public pendingOptimisticDirectDeletes = new Set() @@ -111,6 +118,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< @@ -510,7 +519,10 @@ export class CollectionStateManager< const previousDeletes = new Set(this.optimisticDeletes) const previousRowOrigins = this.rowOrigins - // Update pending optimistic state for completed/failed transactions + // 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 = transaction.metadata[DIRECT_TRANSACTION_METADATA_KEY] === true @@ -534,11 +546,22 @@ export class CollectionStateManager< } switch (mutation.type) { case `insert`: - case `update`: - this.pendingOptimisticUpserts.set( - mutation.key, - mutation.modified as TOutput, - ) + case `update`: { + // 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 confirmsInsert = previous?.insert === mutation + this.pendingOptimisticUpserts.set(mutation.key, { + key: mutation.key, + modified: confirmsInsert + ? previous.modified + : mutation.modified, + insert: + mutation.type === `update` + ? (previous?.insert ?? pendingInserts.get(mutation.key)) + : undefined, + }) this.pendingOptimisticDeletes.delete(mutation.key) if (isDirectTransaction) { this.pendingOptimisticDirectUpserts.add(mutation.key) @@ -548,6 +571,7 @@ export class CollectionStateManager< this.pendingOptimisticDirectDeletes.delete(mutation.key) } break + } case `delete`: this.pendingOptimisticUpserts.delete(mutation.key) this.pendingOptimisticDeletes.add(mutation.key) @@ -561,17 +585,26 @@ export class CollectionStateManager< break } } - } else if (transaction.state === `failed`) { + } else { for (const mutation of transaction.mutations) { - if (!this.isThisCollection(mutation.collection)) { + if ( + !this.isThisCollection(mutation.collection) || + mutation.type !== `insert` || + !mutation.optimistic + ) continue - } - this.pendingLocalOrigins.delete(mutation.key) - if (mutation.optimistic) { + 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.pendingOptimisticDeletes.delete(mutation.key) this.pendingOptimisticDirectUpserts.delete(mutation.key) - this.pendingOptimisticDirectDeletes.delete(mutation.key) } } } @@ -595,7 +628,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) } @@ -644,7 +677,7 @@ export class CollectionStateManager< case `update`: this.optimisticUpserts.set( mutation.key, - mutation.modified as TOutput, + this.resolveOptimisticUpsert(mutation), ) this.optimisticDeletes.delete(mutation.key) break @@ -689,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 @@ -803,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`, @@ -819,6 +854,37 @@ export class CollectionStateManager< } } + // 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 { + const key = mutation.key as TKey + const dependent = this.pendingOptimisticUpserts.get(key) + return dependent?.insert === mutation + ? dependent.modified + : mutation.modified + } + + /** 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 + if (transaction.truncate) { + queued.clear() + truncated = true + } + for (const operation of transaction.operations) { + queued.set(operation.key as TKey, operation.type !== `delete`) + } + } + return (key) => queued.get(key) ?? (!truncated && this.syncedData.has(key)) + } + /** * Get the previous value for a key given previous optimistic state */ @@ -1063,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) { @@ -1153,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 @@ -1245,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` && @@ -1261,7 +1312,7 @@ export class CollectionStateManager< case `update`: this.optimisticUpserts.set( mutation.key, - mutation.modified as TOutput, + this.resolveOptimisticUpsert(mutation), ) this.optimisticDeletes.delete(mutation.key) break @@ -1275,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) { @@ -1329,36 +1404,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 @@ -1558,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/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/src/query/live/ARCHITECTURE.md b/packages/db/src/query/live/ARCHITECTURE.md index 4e5e8e9e73..99a94be323 100644 --- a/packages/db/src/query/live/ARCHITECTURE.md +++ b/packages/db/src/query/live/ARCHITECTURE.md @@ -1001,6 +1001,32 @@ 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. +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 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..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,14 @@ export class CollectionConfigBuilder< facadePublication.prepare() if (hasParentChanges) { begin() - changesToApply.forEach(this.applyChanges.bind(this, config)) + 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), + ) if (hasOrderOnlyMove(changesToApply)) { markLayoutChange(config.collection) } @@ -1097,6 +1104,7 @@ export class CollectionConfigBuilder< private applyChanges( config: SyncMethods, + hasSyncedKey: (key: string | number) => boolean, changes: { deletes: number inserts: number @@ -1126,9 +1134,9 @@ 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 && hasSyncedKey(collection.getKeyFromItem(value))) ) { write({ value, 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 48bd077d48..9ed3b80cf0 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], @@ -932,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 new file mode 100644 index 0000000000..1ad03a2ebc --- /dev/null +++ b/packages/db/tests/optimistic-composition.test.ts @@ -0,0 +1,506 @@ +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' + +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(onUpdate = () => Promise.resolve()) { + let sync!: Parameters[`sync`]>[0] + const collection = createCollection({ + getKey: (row) => row.id, + onUpdate, + 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 snapshot ownership`, () => { + 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])( + `keeps its captured snapshot until 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` }) + 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` }) + } finally { + done.resolve() + subscription.unsubscribe() + await downstream.cleanup() + await fixture.collection.cleanup() + } + }, + ) + + it.each(orders)( + `selects whole direct snapshots 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 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) + } + } finally { + for (const pending of done) pending.resolve() + subscription.unsubscribe() + await downstream.cleanup() + await fixture.collection.cleanup() + } + }, + ) + + it.each([`before`, `after`] as const)( + `retains a direct survivor's captured snapshot 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` }) + } + done[0]!.reject(new Error(`first update failed`)) + await firstSettled + check({ ...initial, a: `a1`, b: `b2` }) + if (completion === `after`) { + done[1]!.resolve() + await second.isPersisted.promise + check({ ...initial, a: `a1`, 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 }) => { + 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 snapshots: Array = [ + undefined, + undefined, + undefined, + ] + const expected = (excluded = -1) => + snapshots.reduce( + (last, row, index) => + row !== undefined && index !== excluded ? row : last, + initial, + ) + 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]) + }), + ) + } + 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)( + `does not merge new synced fields into a $phase mutation snapshot`, + 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`, + }) + } 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`, + }) + } 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/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 new file mode 100644 index 0000000000..8a52552c30 --- /dev/null +++ b/packages/db/tests/query/derived-delete-reconciliation.test.ts @@ -0,0 +1,332 @@ +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 } +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 })), + ), + ), +) + +// 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`, + 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 }) => { + 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() + } + }, + ) +})