diff --git a/.changeset/acknowledged-optimistic-state.md b/.changeset/acknowledged-optimistic-state.md new file mode 100644 index 0000000000..2703993f0f --- /dev/null +++ b/.changeset/acknowledged-optimistic-state.md @@ -0,0 +1,12 @@ +--- +'@tanstack/db': minor +--- + +Expose an `acknowledged` state for optimistic mutations, a virtual prop derived from existing internal state that sits between the optimistic write and the settled (synced-back) state. +Collections can use `tx.acknowledge()` from inside the mutation handler without exiting, so that people using the collection can decide to fire a transition or drop pending state upon `isAcknowledged` instead of waiting for `isSettled`. + +Additive and non-breaking; `isPersisted` / `$synced` are unchanged; no conflict with pending/planned behaviors for `isSettled`. + +- `Transaction.acknowledge()` — a setter called by a collection adapter when the server confirms a write. +- `Transaction.isAcknowledged` — resolves when `acknowledge()` is called, or with `isPersisted` when no adapter calls `acknowledge()`; rejects on failure. Never resolves later than `isPersisted`. +- `$acknowledged` virtual property — `true` once acknowledged, always `true` when `$synced` is `true`. Wired through row enrichment, the virtual-prop cache, and group-by aggregation, and emitted as a virtual-prop-only update when it flips mid-flight. diff --git a/docs/guides/live-queries.md b/docs/guides/live-queries.md index 9b6ee18b06..9c65d43354 100644 --- a/docs/guides/live-queries.md +++ b/docs/guides/live-queries.md @@ -34,6 +34,7 @@ The result types are automatically inferred from your query structure, providing Live query results include computed, read-only virtual properties on every row: - `$synced`: `true` when the row is confirmed by sync; `false` when it is still optimistic. +- `$acknowledged`: `false` at the start of an optimistic update; flips to `true` when `$synced` is flipped, or earlier if the mutation handler calls `tx.acknowledge()` - `$origin`: `"local"` if the last confirmed change came from this client, otherwise `"remote"`. - `$key`: the row key for the result. - `$collectionId`: the source collection ID. diff --git a/docs/guides/mutations.md b/docs/guides/mutations.md index 9a60c47dc5..a8ab020243 100644 --- a/docs/guides/mutations.md +++ b/docs/guides/mutations.md @@ -992,6 +992,27 @@ tx.isPersisted.promise.then(() => { console.log(tx.state) // 'pending', 'persisting', 'completed', or 'failed' ``` +### Reacting to acknowledgement vs. sync + +When a mutation handler acknowledges a server write before the sync comes back, the UI can choose to fire transitions or drop state based on this signal, either `transaction.isAcknowledged` or the `row.$acknowledged` + +```typescript +const tx = todoCollection.insert(draft) + +await tx.isAcknowledged.promise // resolves at acknowledgement — never later than isPersisted +toast.success("Profile created") // server has the write now; sync may still be catching up +router.navigate('/welcome') +``` + +Reactively, `$acknowledged` lets you show an intermediate "saved, syncing…" state: + +```tsx +const rowState = + !row.$acknowledged ? 'pending-spinner' // in flight + : !row.$synced ? 'saved-unsettled' // server has it; sync catching up + : 'saved' // fully settled +``` + ## Paced Mutations Paced mutations provide fine-grained control over **when and how** mutations are persisted to your backend. Instead of persisting every mutation immediately, you can use timing strategies to batch, delay, or queue mutations based on your application's needs. diff --git a/packages/db-sqlite-persistence-core/tests/persisted.test.ts b/packages/db-sqlite-persistence-core/tests/persisted.test.ts index 57c11d8442..a199f9b744 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted.test.ts @@ -244,6 +244,7 @@ const stripVirtualProps = | undefined>( if (!value || typeof value !== `object`) return value const { $synced: _synced, + $acknowledged: _acknowledged, $origin: _origin, $key: _key, $collectionId: _collectionId, diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index 9cbdebb234..a54e3f6955 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -84,6 +84,14 @@ export class CollectionStateManager< public pendingOptimisticDirectUpserts = new Set() public pendingOptimisticDirectDeletes = new Set() + /** + * Keys whose covering optimistic mutation belongs to a transaction that has + * been acknowledged by the server (via {@link Transaction.acknowledge}) but + * whose change has not yet synced back. Used for the `$acknowledged` virtual + * property. Rebuilt from transaction state on each recompute. + */ + public acknowledgedKeys = new Set() + /** * Tracks the origin of confirmed changes for each row. * 'local' = change originated from this client @@ -107,6 +115,7 @@ export class CollectionStateManager< object, { synced: boolean + acknowledged: boolean origin: VirtualOrigin key: TKey collectionId: string @@ -120,6 +129,16 @@ export class CollectionStateManager< // State used for computing the change events public syncedKeys = new Set() public preSyncVisibleState = new Map() + /** + * Whether each to-be-synced key was acknowledged in the visible state captured + * before the sync recompute, keyed alongside {@link preSyncVisibleState}. The + * recompute rebuilds `acknowledgedKeys` from active transactions only, so by + * the time a settle event is emitted the now-completed transaction's ack is + * gone; this preserves it so the settle event's `previousValue.$acknowledged` + * reflects what was actually emitted before (e.g. `false` when ack and settle + * coincide, `true` when the row was acknowledged first). + */ + public preSyncAcknowledged = new Map() public recentlySyncedKeys = new Set() public hasReceivedFirstCommit = false public isCommittingSyncTransactions = false @@ -164,6 +183,27 @@ export class CollectionStateManager< return !this.optimisticUpserts.has(key) && !this.optimisticDeletes.has(key) } + /** + * Whether the backend has acknowledged the pending write for a row without + * errors, independent of whether it has synced back. Used for `$acknowledged`. + * A synced row from a completed (settled) transaction, is always acknowledged. + */ + public isRowAcknowledged(key: TKey): boolean { + if (this.isLocalOnly) { + return true + } + if (this.isRowSynced(key)) { + return true + } + if ( + this.pendingOptimisticUpserts.has(key) || + this.pendingOptimisticDeletes.has(key) + ) { + return true + } + return this.acknowledgedKeys.has(key) + } + /** * Gets the origin of the last confirmed change to a row. * Returns 'local' if the row has optimistic mutations (optimistic changes are local). @@ -187,6 +227,7 @@ export class CollectionStateManager< ): VirtualRowProps { return { $synced: overrides?.$synced ?? this.isRowSynced(key), + $acknowledged: overrides?.$acknowledged ?? this.isRowAcknowledged(key), $origin: overrides?.$origin ?? this.getRowOrigin(key), $key: overrides?.$key ?? key, $collectionId: overrides?.$collectionId ?? this.collection.id, @@ -200,11 +241,13 @@ export class CollectionStateManager< optimisticUpserts?: Pick, 'has'> optimisticDeletes?: Pick, 'has'> completedOptimisticKeys?: Pick, 'has'> + acknowledgedKeys?: Pick, 'has'> }, ): VirtualRowProps { if (this.isLocalOnly) { return this.createVirtualPropsSnapshot(key, { $synced: true, + $acknowledged: true, $origin: 'local', }) } @@ -213,13 +256,22 @@ export class CollectionStateManager< options?.optimisticUpserts ?? this.optimisticUpserts const optimisticDeletes = options?.optimisticDeletes ?? this.optimisticDeletes + // `completedOptimisticKeys` are optimistic ops whose transaction has reached + // the `completed` state (its mutationFn returned) but whose overlay is still + // up while this cycle's sync is applied — so the row is not yet $synced. + // Whether it counts as $acknowledged is decided by the captured ack state + // below, not by the mere fact that the transaction completed. + const isCompletedOptimistic = + options?.completedOptimisticKeys?.has(key) === true const hasOptimisticChange = optimisticUpserts.has(key) || optimisticDeletes.has(key) || - options?.completedOptimisticKeys?.has(key) === true + isCompletedOptimistic + const acknowledgedKeys = options?.acknowledgedKeys ?? this.acknowledgedKeys return this.createVirtualPropsSnapshot(key, { $synced: !hasOptimisticChange, + $acknowledged: !hasOptimisticChange || acknowledgedKeys.has(key), $origin: hasOptimisticChange ? 'local' : ((options?.rowOrigins ?? this.rowOrigins).get(key) ?? 'remote'), @@ -232,6 +284,7 @@ export class CollectionStateManager< ): WithVirtualProps { const existingRow = row as Partial> const synced = existingRow.$synced ?? virtualProps.$synced + const acknowledged = existingRow.$acknowledged ?? virtualProps.$acknowledged const origin = existingRow.$origin ?? virtualProps.$origin const resolvedKey = existingRow.$key ?? virtualProps.$key const collectionId = existingRow.$collectionId ?? virtualProps.$collectionId @@ -240,6 +293,7 @@ export class CollectionStateManager< if ( cached && cached.synced === synced && + cached.acknowledged === acknowledged && cached.origin === origin && cached.key === resolvedKey && cached.collectionId === collectionId @@ -250,6 +304,7 @@ export class CollectionStateManager< const enriched = { ...row, $synced: synced, + $acknowledged: acknowledged, $origin: origin, $key: resolvedKey, $collectionId: collectionId, @@ -257,6 +312,7 @@ export class CollectionStateManager< this.virtualPropsCache.set(row as object, { synced, + acknowledged, origin, key: resolvedKey, collectionId, @@ -540,6 +596,8 @@ export class CollectionStateManager< this.optimisticUpserts.clear() this.optimisticDeletes.clear() this.pendingLocalChanges.clear() + // Rebuilt below from active transactions' acknowledged flags. + this.acknowledgedKeys.clear() // Seed optimistic state with pending optimistic mutations only when a sync is pending const pendingSyncKeys = new Set() @@ -612,6 +670,10 @@ export class CollectionStateManager< this.optimisticDeletes.add(mutation.key) break } + + if (transaction.acknowledged) { + this.acknowledgedKeys.add(mutation.key) + } } } } @@ -1187,6 +1249,15 @@ export class CollectionStateManager< this.pendingOptimisticDirectUpserts.clear() this.pendingOptimisticDirectDeletes.clear() + // The acknowledged state captured before the recompute (the recompute + // drops the now-completed transaction's ack), so each "previous" snapshot + // below reflects what was actually emitted before the settle — `false` + // when ack and settle coincide, `true` when the row was acknowledged + // first. Stable across the batch, so build the lookup once. + const preSyncAcknowledgedKeys = { + has: (k: TKey) => this.preSyncAcknowledged.get(k) === true, + } + // Now check what actually changed in the final visible state for (const key of changedKeys) { const previousVisibleValue = currentVisibleState.get(key) @@ -1196,10 +1267,13 @@ export class CollectionStateManager< optimisticUpserts: previousOptimisticUpserts, optimisticDeletes: previousOptimisticDeletes, completedOptimisticKeys: completedOptimisticOps, + acknowledgedKeys: preSyncAcknowledgedKeys, }) const nextVirtualProps = this.getVirtualPropsSnapshotForState(key) const virtualChanged = previousVirtualProps.$synced !== nextVirtualProps.$synced || + previousVirtualProps.$acknowledged !== + nextVirtualProps.$acknowledged || previousVirtualProps.$origin !== nextVirtualProps.$origin const previousValueWithVirtual = previousVisibleValue !== undefined @@ -1209,6 +1283,7 @@ export class CollectionStateManager< this.collection.id, () => previousVirtualProps.$synced, () => previousVirtualProps.$origin, + () => previousVirtualProps.$acknowledged, ) : undefined @@ -1256,6 +1331,7 @@ export class CollectionStateManager< this.collection.id, () => previousVirtualProps.$synced, () => previousVirtualProps.$origin, + () => previousVirtualProps.$acknowledged, ) events.push({ type: `update`, @@ -1309,6 +1385,7 @@ export class CollectionStateManager< // Clear the pre-sync state since sync operations are complete this.preSyncVisibleState.clear() + this.preSyncAcknowledged.clear() // Clear recently synced keys after a microtask to allow recomputeOptimisticState to see them Promise.resolve().then(() => { @@ -1375,6 +1452,68 @@ export class CollectionStateManager< this.preSyncVisibleState.set(key, currentValue) } } + // Capture the acknowledged state too, before the recompute clears + // acknowledgedKeys. Match the emitted $acknowledged: a still-optimistic + // row is acknowledged only if it was explicitly acknowledged or synced. + if (!this.preSyncAcknowledged.has(key)) { + this.preSyncAcknowledged.set( + key, + this.acknowledgedKeys.has(key) || this.isRowSynced(key), + ) + } + } + } + + /** + * Called when a transaction is acknowledged by the server (via + * `Transaction.acknowledge()`). Flips `$acknowledged` to true for the + * transaction's still-optimistic rows and emits virtual-prop-only update + * events so subscribers react, without touching the data or the overlay. + */ + public onTransactionAcknowledged(transaction: Transaction): void { + const events: Array> = [] + + for (const mutation of transaction.mutations) { + if (!this.isThisCollection(mutation.collection)) { + continue + } + const key = mutation.key as TKey + if (this.acknowledgedKeys.has(key)) { + continue + } + + const value = this.get(key) + const previousVirtualProps = this.getVirtualPropsSnapshotForState(key) + this.acknowledgedKeys.add(key) + + // Already settled / not visible: nothing to emit, the flag is enough. + if (value === undefined) { + continue + } + + const nextVirtualProps = this.getVirtualPropsSnapshotForState(key) + if ( + previousVirtualProps.$acknowledged === nextVirtualProps.$acknowledged + ) { + continue + } + + events.push({ + type: `update`, + key, + value, + previousValue: value, + __virtualProps: { + value: nextVirtualProps, + previousValue: previousVirtualProps, + }, + }) + } + + if (events.length > 0) { + this.indexes.updateIndexes(events) + // forceEmit: acknowledgement is a user-visible action, keep UI responsive. + this.changes.emitEvents(events, true) } } @@ -1408,6 +1547,8 @@ export class CollectionStateManager< this.pendingOptimisticDirectUpserts.clear() this.pendingOptimisticDirectDeletes.clear() this.clearOriginTrackingState() + this.acknowledgedKeys.clear() + this.preSyncAcknowledged.clear() this.isLocalOnly = false this.size = 0 this.pendingSyncedTransactions = [] diff --git a/packages/db/src/query/compiler/group-by.ts b/packages/db/src/query/compiler/group-by.ts index c670de9649..b0d0a08778 100644 --- a/packages/db/src/query/compiler/group-by.ts +++ b/packages/db/src/query/compiler/group-by.ts @@ -35,17 +35,32 @@ import type { NamespacedAndKeyedStream, NamespacedRow } from '../../types.js' import type { VirtualOrigin } from '../../virtual-props.js' const VIRTUAL_SYNCED_KEY = `__virtual_synced__` +const VIRTUAL_ACKNOWLEDGED_KEY = `__virtual_acknowledged__` const VIRTUAL_HAS_LOCAL_KEY = `__virtual_has_local__` const GROUP_KEY_REF_PREFIX = `__group_key_` type RowVirtualMetadata = { synced: boolean + acknowledged: boolean hasLocal: boolean } +/** + * Aggregates the virtual-property metadata for a namespaced group-by row. + * + * Scans each source object on the row and folds their virtual props into a + * single summary for the group: `synced`/`acknowledged` are true only if every + * source is (with `$acknowledged` falling back to `$synced` when absent), and + * `hasLocal` is true if any source has a `local` origin. Rows with no + * virtual-prop-bearing sources are treated as synced and acknowledged. + * + * @param row - The namespaced row whose source objects carry virtual props + * @returns The aggregated `{ synced, acknowledged, hasLocal }` metadata + */ function getRowVirtualMetadata(row: NamespacedRow): RowVirtualMetadata { let found = false let allSynced = true + let allAcknowledged = true let hasLocal = false for (const [alias, value] of Object.entries(row as Record)) { @@ -61,6 +76,14 @@ function getRowVirtualMetadata(row: NamespacedRow): RowVirtualMetadata { if (asRecord.$synced === false) { allSynced = false } + // If absent, $acknowledged falls back to $synced + const rowAcknowledged = + `$acknowledged` in asRecord + ? asRecord.$acknowledged !== false + : asRecord.$synced !== false + if (!rowAcknowledged) { + allAcknowledged = false + } if (asRecord.$origin === `local`) { hasLocal = true } @@ -68,6 +91,7 @@ function getRowVirtualMetadata(row: NamespacedRow): RowVirtualMetadata { return { synced: found ? allSynced : true, + acknowledged: found ? allAcknowledged : true, hasLocal, } } @@ -146,6 +170,18 @@ export function processGroupBy( return true }, }, + [VIRTUAL_ACKNOWLEDGED_KEY]: { + preMap: ([, row]: [string, NamespacedRow]) => + getRowVirtualMetadata(row).acknowledged, + reduce: (values: Array<[boolean, number]>) => { + for (const [isAcknowledged, multiplicity] of values) { + if (!isAcknowledged && multiplicity > 0) { + return false + } + } + return true + }, + }, [VIRTUAL_HAS_LOCAL_KEY]: { preMap: ([, row]: [string, NamespacedRow]) => getRowVirtualMetadata(row).hasLocal, @@ -242,10 +278,14 @@ export function processGroupBy( const groupSynced = (aggregatedRow as Record)[ VIRTUAL_SYNCED_KEY ] + const groupAcknowledged = (aggregatedRow as Record)[ + VIRTUAL_ACKNOWLEDGED_KEY + ] const groupHasLocal = (aggregatedRow as Record)[ VIRTUAL_HAS_LOCAL_KEY ] resultRow.$synced = groupSynced ?? true + resultRow.$acknowledged = groupAcknowledged ?? groupSynced ?? true resultRow.$origin = ( groupHasLocal ? `local` : `remote` ) satisfies VirtualOrigin @@ -421,10 +461,14 @@ export function processGroupBy( const groupSynced = (aggregatedRow as Record)[ VIRTUAL_SYNCED_KEY ] + const groupAcknowledged = (aggregatedRow as Record)[ + VIRTUAL_ACKNOWLEDGED_KEY + ] const groupHasLocal = (aggregatedRow as Record)[ VIRTUAL_HAS_LOCAL_KEY ] resultRow.$synced = groupSynced ?? true + resultRow.$acknowledged = groupAcknowledged ?? groupSynced ?? true resultRow.$origin = ( groupHasLocal ? `local` : `remote` ) satisfies VirtualOrigin diff --git a/packages/db/src/transactions.ts b/packages/db/src/transactions.ts index fe2f61c0fd..8ed18ec7b2 100644 --- a/packages/db/src/transactions.ts +++ b/packages/db/src/transactions.ts @@ -212,6 +212,8 @@ class Transaction> { public mutationFn: MutationFn public mutations: Array> public isPersisted: Deferred> + public isAcknowledged: Deferred> + public acknowledged: boolean public autoCommit: boolean public createdAt: Date public sequenceNumber: number @@ -230,6 +232,11 @@ class Transaction> { this.state = `pending` this.mutations = [] this.isPersisted = createDeferred>() + this.isAcknowledged = createDeferred>() + this.acknowledged = false + // Prevent unhandled-rejection noise when nobody awaits isAcknowledged on a + // failing transaction; explicit awaiters still observe the rejection. + void this.isAcknowledged.promise.catch(() => {}) this.autoCommit = config.autoCommit ?? true this.createdAt = new Date() this.sequenceNumber = sequenceNumber++ @@ -415,7 +422,10 @@ class Transaction> { } } - // Reject the promise + // Reject the promises + if (this.isAcknowledged.isPending()) { + this.isAcknowledged.reject(this.error?.error) + } this.isPersisted.reject(this.error?.error) this.touchCollection() @@ -439,6 +449,50 @@ class Transaction> { } } + /** + * Mark this transaction as having received some acknowledgement and + * confirmation from the server, so the UI can move ahead without waiting + * for the transaction to sync back ("settle"). @returns This transaction. + * + * @example: + * // Acknowledge before full sync/settle + * const wrappedOnInsert = async (params) => { + * const result = await config.onInsert!(params) + * + * params.transaction.acknowledge() // 🥳 UI can move on + * await processMatchingStrategy(result) // overlay stays open + * + * return result + * } + * + * Resolves {@link Transaction.isAcknowledged} and flips the `$acknowledged` + * virtual property on the affected rows. + */ + acknowledge(): Transaction { + if ( + this.acknowledged || + this.state === `completed` || + this.state === `failed` + ) { + return this + } + + this.acknowledged = true + this.isAcknowledged.resolve(this) + + // Notify each affected collection so it flips $acknowledged and emits updates. + const notified = new Set() + for (const mutation of this.mutations) { + if (notified.has(mutation.collection.id)) { + continue + } + notified.add(mutation.collection.id) + mutation.collection._state.onTransactionAcknowledged(this) + } + + return this + } + /** * Commit the transaction and execute the mutation function * @returns Promise that resolves to this transaction when complete @@ -487,6 +541,10 @@ class Transaction> { if (this.mutations.length === 0) { this.setState(`completed`) + if (this.isAcknowledged.isPending()) { + this.acknowledged = true + this.isAcknowledged.resolve(this) + } this.isPersisted.resolve(this) return this @@ -504,6 +562,11 @@ class Transaction> { this.setState(`completed`) this.touchCollection() + // A settling a transaction always flips acknowledged, if it wasn't already done + if (this.isAcknowledged.isPending()) { + this.acknowledged = true + this.isAcknowledged.resolve(this) + } this.isPersisted.resolve(this) } catch (error) { // Preserve the original error for rethrowing diff --git a/packages/db/src/virtual-props.ts b/packages/db/src/virtual-props.ts index 3205d31c2d..576bc840e3 100644 --- a/packages/db/src/virtual-props.ts +++ b/packages/db/src/virtual-props.ts @@ -68,6 +68,17 @@ export interface VirtualRowProps< */ readonly $synced: boolean + /** + * Whether this row has either been affirmatively marked as acknowledged, or it + * is $synced. + * + * - `true`: row is $synced, or `transaction.acknowledge() has been called + * - `false`: the write is still in flight; no confirmation yet. + * + * When mutations lack a separate acknowledgement signal, this coincides with `$synced`. + */ + readonly $acknowledged: boolean + /** * Origin of the last confirmed change to this row, from the current client's perspective. * @@ -171,9 +182,11 @@ export function createVirtualProps( collectionId: string, isSynced: boolean, origin: VirtualOrigin, + isAcknowledged: boolean = isSynced, ): VirtualRowProps { return { $synced: isSynced, + $acknowledged: isAcknowledged, $origin: origin, $key: key, $collectionId: collectionId, @@ -207,6 +220,7 @@ export function enrichRowWithVirtualProps< collectionId: string, computeSynced: () => boolean, computeOrigin: () => VirtualOrigin, + computeAcknowledged: () => boolean = computeSynced, ): WithVirtualProps { // Use nullish coalescing to preserve existing virtual properties (pass-through) // This is the "add-if-missing" pattern described in the RFC @@ -215,6 +229,7 @@ export function enrichRowWithVirtualProps< return { ...row, $synced: existingRow.$synced ?? computeSynced(), + $acknowledged: existingRow.$acknowledged ?? computeAcknowledged(), $origin: existingRow.$origin ?? computeOrigin(), $key: existingRow.$key ?? key, $collectionId: existingRow.$collectionId ?? collectionId, @@ -243,11 +258,17 @@ export function computeAggregateVirtualProps( // $synced = true only if ALL rows are synced (false if ANY is optimistic) const allSynced = rows.every((row) => row.$synced ?? true) + // $acknowledged = true only if ALL rows are acknowledged (and/or synced) + const allAcknowledged = rows.every( + (row) => row.$acknowledged ?? row.$synced ?? true, + ) + // $origin = 'local' if ANY row is local (consistent with "local influence" semantics) const hasLocal = rows.some((row) => row.$origin === 'local') return { $synced: allSynced, + $acknowledged: allAcknowledged, $origin: hasLocal ? 'local' : 'remote', $key: groupKey, $collectionId: collectionId, @@ -260,6 +281,7 @@ export function computeAggregateVirtualProps( */ export const VIRTUAL_PROP_NAMES = [ '$synced', + '$acknowledged', '$origin', '$key', '$collectionId', diff --git a/packages/db/tests/collection-subscribe-changes.test.ts b/packages/db/tests/collection-subscribe-changes.test.ts index 4f851f08a7..3f3f02f20d 100644 --- a/packages/db/tests/collection-subscribe-changes.test.ts +++ b/packages/db/tests/collection-subscribe-changes.test.ts @@ -2677,6 +2677,7 @@ describe(`Virtual properties`, () => { expect( hasVirtualProps({ $synced: true, + $acknowledged: true, $origin: `remote`, $key: `row-1`, $collectionId: `collection-1`, diff --git a/packages/db/tests/transaction-acknowledge.test.ts b/packages/db/tests/transaction-acknowledge.test.ts new file mode 100644 index 0000000000..3dd5230b07 --- /dev/null +++ b/packages/db/tests/transaction-acknowledge.test.ts @@ -0,0 +1,328 @@ +import { describe, expect, it } from 'vitest' +import { createCollection } from '../src/collection/index.js' +import type { ChangeMessage, SyncConfig } from '../src/types' +import type { OutputWithVirtual } from './utils' + +const waitForChanges = () => new Promise((resolve) => setTimeout(resolve, 10)) + +type Row = { id: string; value: string } +type Change = ChangeMessage> + +function createGate() { + let release!: () => void + const promise = new Promise((resolve) => { + release = resolve + }) + return { promise, release } +} + +describe(`Transaction.acknowledge() — the ack layer`, () => { + it(`flips $acknowledged and resolves isAcknowledged while still persisting, without settling`, async () => { + const changes: Array = [] + const ackGate = createGate() + const settleGate = createGate() + + const collection = createCollection({ + id: `ack-flip`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => markReady(), + }, + onInsert: async ({ transaction }) => { + // Wait until the test lets the "server" ack, then mark acknowledged but + // keep the handler open (settle hasn't happened yet). + await ackGate.promise + transaction.acknowledge() + await settleGate.promise + }, + }) + + const subscription = collection.subscribeChanges( + (events) => changes.push(...events), + { includeInitialState: false }, + ) + + const tx = collection.insert({ id: `r1`, value: `v` }) + await waitForChanges() + + // Optimistic insert: not acknowledged, not synced. + const optimistic = changes.find( + (c) => c.type === `insert` && c.key === `r1`, + ) + expect(optimistic).toBeDefined() + expect(optimistic!.value.$acknowledged).toBe(false) + expect(optimistic!.value.$synced).toBe(false) + expect(tx.acknowledged).toBe(false) + + // Release the ack. + changes.length = 0 + ackGate.release() + await tx.isAcknowledged.promise + await waitForChanges() + + // Transaction is acknowledged but still persisting (handler hasn't returned). + expect(tx.acknowledged).toBe(true) + expect(tx.state).toBe(`persisting`) + expect(tx.isPersisted.isPending()).toBe(true) + + // A virtual-prop-only update was emitted: $acknowledged false -> true, + // value unchanged, $synced still false. + const flip = changes.find((c) => c.type === `update` && c.key === `r1`) + expect(flip).toBeDefined() + expect(flip!.value.$acknowledged).toBe(true) + expect(flip!.previousValue?.$acknowledged).toBe(false) + expect(flip!.value.$synced).toBe(false) + + // Reading the row directly reflects the acknowledged-but-not-synced state. + const row = collection.state.get(`r1`) + expect(row?.$acknowledged).toBe(true) + expect(row?.$synced).toBe(false) + + // Now let the handler return (settle). + settleGate.release() + await tx.isPersisted.promise + expect(tx.state).toBe(`completed`) + + subscription.unsubscribe() + }) + + it(`resolves isAcknowledged together with isPersisted when the collection never calls acknowledge()`, async () => { + const collection = createCollection({ + id: `ack-coincide`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => markReady(), + }, + onInsert: async () => { + // No acknowledge() — a backend without a separate ack signal. + }, + }) + + const tx = collection.insert({ id: `r1`, value: `v` }) + + // isAcknowledged is still safe to await and resolves no later than isPersisted. + await expect(tx.isAcknowledged.promise).resolves.toBe(tx) + await expect(tx.isPersisted.promise).resolves.toBe(tx) + expect(tx.acknowledged).toBe(true) + }) + + it(`rejects isAcknowledged when the transaction fails before acknowledgement`, async () => { + const collection = createCollection({ + id: `ack-fail`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => markReady(), + }, + onInsert: async () => { + throw new Error(`server rejected`) + }, + }) + + const tx = collection.insert({ id: `r1`, value: `v` }) + + await expect(tx.isAcknowledged.promise).rejects.toThrow(`server rejected`) + await expect(tx.isPersisted.promise).rejects.toThrow(`server rejected`) + expect(tx.acknowledged).toBe(false) + }) + + it(`keeps an already-acknowledged ack when a later settle fails`, async () => { + const settleGate = createGate() + const collection = createCollection({ + id: `ack-then-fail`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => markReady(), + }, + onInsert: async ({ transaction }) => { + transaction.acknowledge() + await settleGate.promise + throw new Error(`settle failed`) + }, + }) + + const tx = collection.insert({ id: `r1`, value: `v` }) + await tx.isAcknowledged.promise + expect(tx.acknowledged).toBe(true) + + settleGate.release() + await expect(tx.isPersisted.promise).rejects.toThrow(`settle failed`) + // The ack already resolved; it is not retroactively rejected. + await expect(tx.isAcknowledged.promise).resolves.toBe(tx) + }) + + it(`treats synced rows as acknowledged`, async () => { + const collection = createCollection({ + id: `ack-synced`, + getKey: (item) => item.id, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ type: `insert`, value: { id: `r1`, value: `v` } }) + commit() + markReady() + }, + }, + }) + + // Subscribing starts sync for this lazily-synced collection. + const subscription = collection.subscribeChanges(() => {}) + await waitForChanges() + const row = collection.state.get(`r1`) + expect(row?.$synced).toBe(true) + expect(row?.$acknowledged).toBe(true) + subscription.unsubscribe() + }) + + it(`acknowledge() is idempotent and a no-op after completion`, async () => { + const collection = createCollection({ + id: `ack-idempotent`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => markReady(), + }, + onInsert: async ({ transaction }) => { + transaction.acknowledge() + transaction.acknowledge() // second call is a no-op + }, + }) + + const tx = collection.insert({ id: `r1`, value: `v` }) + await tx.isPersisted.promise + + // No-op after completion, returns the transaction for chaining. + expect(tx.acknowledge()).toBe(tx) + expect(tx.acknowledged).toBe(true) + }) + + // The emission contract: a single state transition that flips both + // $acknowledged and $synced is delivered as ONE update event carrying both; + // two flips at two different times produce two events. + it(`emits a single combined update when ack and settle coincide (pending -> synced)`, async () => { + const changes: Array = [] + const settleGate = createGate() + let syncOps: Parameters[`sync`]>[0] | undefined + + const collection = createCollection({ + id: `ack-emit-coincide`, + getKey: (item) => item.id, + sync: { + sync: (cfg) => { + syncOps = cfg + cfg.markReady() + }, + }, + onInsert: async ({ transaction }) => { + // No acknowledge(): the ack coincides with settle. When released, echo + // the write back through sync so the overlay drops and $synced flips. + await settleGate.promise + syncOps!.begin() + for (const mutation of transaction.mutations) { + syncOps!.write({ type: `insert`, value: mutation.modified }) + } + syncOps!.commit() + }, + }) + + const subscription = collection.subscribeChanges( + (events) => changes.push(...events), + { includeInitialState: false }, + ) + + const tx = collection.insert({ id: `r1`, value: `v` }) + await waitForChanges() + + // Optimistic insert: neither acknowledged nor synced. + expect(changes.length).toBe(1) + expect(changes[0]!.type).toBe(`insert`) + expect(changes[0]!.value.$acknowledged).toBe(false) + expect(changes[0]!.value.$synced).toBe(false) + + // Settle: $acknowledged and $synced flip together in ONE update — not two. + settleGate.release() + await tx.isPersisted.promise + await waitForChanges() + + const updates = changes.filter((c) => c.type === `update`) + expect(updates.length).toBe(1) + expect(updates[0]!.value.$acknowledged).toBe(true) + expect(updates[0]!.value.$synced).toBe(true) + expect(changes.length).toBe(2) + // previousValue is coherent with what was actually emitted before: the row + // was never acknowledged separately, so its previous $acknowledged is false. + expect(updates[0]!.previousValue?.$acknowledged).toBe(false) + expect(updates[0]!.previousValue?.$synced).toBe(false) + + subscription.unsubscribe() + }) + + it(`emits two distinct updates when ack precedes settle (pending -> acked -> synced)`, async () => { + const changes: Array = [] + const ackGate = createGate() + const settleGate = createGate() + let syncOps: Parameters[`sync`]>[0] | undefined + + const collection = createCollection({ + id: `ack-emit-separate`, + getKey: (item) => item.id, + sync: { + sync: (cfg) => { + syncOps = cfg + cfg.markReady() + }, + }, + onInsert: async ({ transaction }) => { + await ackGate.promise + transaction.acknowledge() + await settleGate.promise + syncOps!.begin() + for (const mutation of transaction.mutations) { + syncOps!.write({ type: `insert`, value: mutation.modified }) + } + syncOps!.commit() + }, + }) + + const subscription = collection.subscribeChanges( + (events) => changes.push(...events), + { includeInitialState: false }, + ) + + const tx = collection.insert({ id: `r1`, value: `v` }) + await waitForChanges() + + // 1) Optimistic insert. + expect(changes.length).toBe(1) + expect(changes[0]!.type).toBe(`insert`) + expect(changes[0]!.value.$acknowledged).toBe(false) + expect(changes[0]!.value.$synced).toBe(false) + + // 2) Ack: a virtual-prop-only update — $acknowledged flips, $synced does not. + ackGate.release() + await tx.isAcknowledged.promise + await waitForChanges() + + expect(changes.length).toBe(2) + const ackUpdate = changes[1]! + expect(ackUpdate.type).toBe(`update`) + expect(ackUpdate.value.$acknowledged).toBe(true) + expect(ackUpdate.value.$synced).toBe(false) + expect(ackUpdate.previousValue?.$acknowledged).toBe(false) + + // 3) Settle: a second, separate update — now $synced flips too. + settleGate.release() + await tx.isPersisted.promise + await waitForChanges() + + expect(changes.length).toBe(3) + const settleUpdate = changes[2]! + expect(settleUpdate.type).toBe(`update`) + expect(settleUpdate.value.$acknowledged).toBe(true) + expect(settleUpdate.value.$synced).toBe(true) + // The row was acknowledged before it settled, so previousValue reflects + // that: $acknowledged was already true, only $synced flips here. + expect(settleUpdate.previousValue?.$acknowledged).toBe(true) + expect(settleUpdate.previousValue?.$synced).toBe(false) + + subscription.unsubscribe() + }) +}) diff --git a/packages/db/tests/utils.ts b/packages/db/tests/utils.ts index 41fc65a9d8..6443c1f895 100644 --- a/packages/db/tests/utils.ts +++ b/packages/db/tests/utils.ts @@ -20,6 +20,7 @@ export const stripVirtualProps = | undefined>( if (!value || typeof value !== `object`) return value const { $synced: _synced, + $acknowledged: _acknowledged, $origin: _origin, $key: _key, $collectionId: _collectionId, @@ -30,9 +31,13 @@ export const stripVirtualProps = | undefined>( export const omitVirtualProps = >( value: T, -): Omit => { +): Omit< + T, + '$synced' | '$acknowledged' | '$origin' | '$key' | '$collectionId' +> => { const { $synced: _synced, + $acknowledged: _acknowledged, $origin: _origin, $key: _key, $collectionId: _collectionId,