diff --git a/.changeset/mutation-log-reconciliation.md b/.changeset/mutation-log-reconciliation.md new file mode 100644 index 000000000..2e6276bce --- /dev/null +++ b/.changeset/mutation-log-reconciliation.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +Apply committed sync updates immediately while preserving unsettled optimistic mutations in visible collection state. This improves optimistic write reconciliation and stabilizes change events for subscriptions and live queries. diff --git a/.gitignore b/.gitignore index 4ad9ee25d..f7a9af1bf 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,6 @@ examples/react-native/shopping-list/ios/ vite.config.js.timestamp-* vite.config.ts.timestamp-* tsconfig.vitest-temp.json + +# Agent workflow scratch files +.superpowers/sdd/ diff --git a/docs/rfcs/2026-06-25-mutation-log-reconciliation.md b/docs/rfcs/2026-06-25-mutation-log-reconciliation.md new file mode 100644 index 000000000..d7d6d812d --- /dev/null +++ b/docs/rfcs/2026-06-25-mutation-log-reconciliation.md @@ -0,0 +1,551 @@ +# RFC: Mutation log reconciliation for optimistic writes + +Status: draft for maintainer review +Date: 2026-06-25 + +## Summary + +TanStack DB should make collection-owned optimistic mutations explicit by introducing an internal **Mutation Log**. The log/index becomes the single projection source for unsettled optimistic mutations and the queryable surface for local write state, write errors, and recoverable resolution state. + +The core state model should become: + +```txt +authoritative synced/base state ++ unsettled optimistic mutations owned by the collection += visible collection state +``` + +This lets collections apply authoritative sync updates immediately, even while optimistic writes are pending, and then reproject optimistic mutations over the updated base. The current behavior of delaying normal sync commits behind `persisting` transactions should be treated as an implementation limitation, not a semantic contract for 1.0. + +This RFC focuses on the core design needed for 1.0: + +- Introduce a Mutation Log as a refinement of current transaction/mutation state. +- Always apply committed authoritative sync/base changes immediately. +- Reproject unsettled optimistic mutations over the latest base state. +- Preserve a stable materialized change stream for subscriptions and live queries. +- Replace ambiguous `$synced` / `isPersisted` concepts with clearer local write state. +- Expose queryable logged mutations joined with transaction state for write status and write errors. +- Add a `needs-resolution` transaction state for explicit recoverable validation/business-rule failures. +- Retain failed transaction/mutation records with bounded automatic GC. +- Slim `@tanstack/offline-transactions` into durability/execution over the log. + +Implementation scope note: Phase 1 is limited to internal reconciliation and test coverage. Public API ideas discussed below—`$hasPendingWrites`, `$writeStatus`, `tx.when(...)`, `db.mutations`, replacing `$synced`, replacing `isPersisted.promise`, and `needs-resolution`—are future/non-Phase-1 design directions and must not be read as part of the Phase 1 binding contract. + +This RFC intentionally does **not** design backend observation/confirmation semantics, stable view keys, sync batch API changes, dependency-aware rollback, nested transactions, or full patch/conflict semantics. Those become easier once pending mutations are centralized and indexed, but they should be separate focused work. + +## Motivation: issue cluster + +These issues are not independent. They mostly come from the same architectural gap: local optimistic intent, authoritative base state, transaction status, persistence status, and errors are spread across several overlapping mechanisms instead of one mutation reconciliation model. + +| Symptom group | Representative issues / PRs | Architectural cause | RFC response | +| --- | --- | --- | --- | +| Sync delayed or inconsistently visible while optimistic writes are pending | #1017, #1048, #1060, #1122, #1166, #1167, #1497, historical #37 | Core currently delays normal committed sync transactions while a user transaction is `persisting`, except truncate/immediate/manual writes. This prevents collection state and derived live queries from consistently reflecting authoritative data, and forces complex branches in the change-event pipeline. | Always apply authoritative sync/base updates immediately. Keep optimistic writes in the Mutation Log and reproject them over the updated base. | +| Ambiguous or missing local write status | #20, #661, #1215, #1219, #1322, #1431, #1526 | `$synced` and `isPersisted` attempt to answer too many questions: local optimistic state, local durability, mutation completion, backend upload, and sync observation. | Remove/replace `$synced` and `isPersisted` for 1.0. Add local-write-specific row props such as `$hasPendingWrites` / `$writeStatus` and queryable logged mutations joined with transaction state. | +| Write errors and recoverable failures are not first-class | #22, #487, #672 | Errors are thrown, logged, or stored inconsistently. A single collection error slot is too coarse for per-write failures, validation state, or notification after navigation. | Store write errors on transaction/mutation records. Add `needs-resolution` for explicit recoverable failures. Retain failed transaction/mutation records briefly with bounded automatic GC. | +| Offline/persistence duplicates transaction state | #1064, #1065, #1483, #1490, #1579, #1592, #1602, #1603 | `@tanstack/offline-transactions` currently has to persist, restore, schedule, and recreate optimistic state as a second species of transaction. | Core owns the in-memory log and projection. `@tanstack/offline-transactions` persists/restores logged mutations and executes them, dramatically reducing parallel state machinery. | +| Future identity/defaults/shape fixes need a better substrate | #19, #25, #456, #465, #900, #1445, #1465 | Server-generated fields, temporary-to-server key mapping, shape evolution, and long-lived optimistic writes currently require bespoke reconciliation against snapshot-like optimistic state. | Keep this RFC focused, but make the mutation log the substrate that later enables stable identity, mutation receipts, and better patch/intention projection. | + +## Current behavior + +Core collection state currently has several overlapping state holders, including: + +- `syncedData` +- `optimisticUpserts` +- `optimisticDeletes` +- `pendingOptimisticUpserts` +- `pendingOptimisticDeletes` +- `pendingSyncedTransactions` +- transaction state (`pending`, `persisting`, `completed`, `failed`) +- adapter/offline-specific pending stores and restoration flows + +In `CollectionStateManager.commitPendingTransactions()`, committed sync transactions are applied only when there is no `persisting` user transaction, or when the sync is truncate/immediate: + +```txt +if no persisting transaction OR truncate sync OR immediate sync: + apply committed sync transactions +else: + leave committed sync queued +``` + +This behavior was probably introduced to avoid difficult reconciliation between incoming authoritative changes and optimistic state. It is understandable, but it creates user-visible inconsistencies: + +- source collection base state temporarily stops reflecting authoritative data; +- derived live-query collections may not receive unrelated synced rows; +- caches/subscriptions need special cases; +- proposed fixes are tempted to emit events without updating `syncedData`, creating split-brain semantics. + +The desired 1.0 semantics should be simpler: + +> Collections apply committed authoritative sync/base changes as they arrive. Pending optimistic writes are local overlays that are reprojected over the latest base. + +## Goals + +1. **Make local write state first-class.** Track unsettled optimistic mutations in one log/index, not as scattered maps and transaction side effects. +2. **Always advance authoritative base state.** A pending local write must not block unrelated authoritative sync data from entering the collection. +3. **Preserve stable materialized changes.** Subscriptions and live queries need a stable stream of inserts, updates, deletes, and truncates from each collection's visible state. +4. **Use precise 1.0 status names.** Replace `$synced` and `isPersisted` with names that describe local write state, not backend observation. +5. **Represent write errors on transactions/mutations.** Write failures and recoverable validation state belong to the transaction/mutation that caused them, not primarily to a single collection error slot. +6. **Support recoverable validation.** A mutation function can explicitly signal `needs-resolution` to preserve optimistic state and expose resolution metadata. +7. **Keep mutation history bounded.** Failed transaction/mutation records are useful after navigation, but long-lived apps must not accumulate unbounded transaction history. +8. **Make offline persistence a layer over the log.** Without `@tanstack/offline-transactions`, optimistic mutations are in-memory. With it, they become durable and executable across reloads. + +## Non-goals + +This RFC does not design: + +- first-class core transaction states for transport confirmation or read-path echo; +- `accepted` or `observed` milestones; +- `awaitTxId` replacement; +- PowerSync upload/read-back confirmation; +- cross-collection observation barriers; +- stable `$viewKey` / entity identity / temp-to-server key mapping; +- mutation receipt APIs; +- sync batch API redesign; +- dependency-aware rollback graphs; +- nested transactions / savepoints; +- full nested patch semantics, array patch semantics, or conflict resolution; +- a general effect/query/sync error log. + +Several of these are valuable follow-ups. The point of this RFC is to establish the mutation reconciliation substrate first. + +## Proposed model + +Each collection has: + +```txt +authoritative synced/base state ++ unsettled optimistic mutations owned by that collection += visible collection state +``` + +A transaction remains the user-facing grouping concept. Logged mutations are essentially the current `PendingMutation`s made central, indexed, and queryable. Collection mutations and explicit transactions remain the primary write APIs; users should not construct raw mutation records for normal writes. + +Transaction state remains the lifecycle source of truth. Row and mutation write state are derived from the owning transaction. The `mutationFn` remains the mechanism that advances the transaction by default: + +- success -> transaction completes and its mutations leave active projection; +- ordinary error -> transaction fails and rolls back according to current/default semantics; +- typed `needs-resolution` error -> transaction enters `needs-resolution` and its mutations remain projected with resolution metadata. + +## Illustrative Mutation Log shape + +The log does not introduce a new independently stateful object. It is a centralized/indexed view of the transaction mutations TanStack DB already tracks today. Exact field names and types are implementation details. A minimal conceptual shape is: + +```ts +interface LoggedMutation { + id: string + transactionId: string + collectionId: string + key: string | number + + type: 'insert' | 'update' | 'delete' + + // Compatibility with today's PendingMutation shape. + original?: unknown + modified?: unknown + changes?: Record + + createdAt: number + updatedAt: number +} + +interface LoggedTransaction { + id: string + state: + | 'pending' + | 'persisting' + | 'completed' + | 'failed' + | 'needs-resolution' + mutations: Array + error?: unknown + resolution?: unknown +} +``` + +Mutation lifecycle is derived from the owning transaction. The log does not add a second state machine. A row has pending writes when active transaction mutations affect that row; a row needs resolution when a `needs-resolution` transaction has mutations affecting that row. + +The initial implementation can wrap, normalize, or index today’s `Transaction.mutations` / `PendingMutation` data. + +## Projection behavior + +The target reconciliation model is: + +```txt +visible row = project(latest base row, active logged mutations for that row) +``` + +For inserts and deletes, the mutation semantics are straightforward. + +For updates, the long-term target is to replay write intent over the latest base row. This avoids long-lived optimistic writes hiding server-added fields or unrelated remote updates. For example: + +```txt +base at mutation time: { title: 'A', priority: 1 } +optimistic change: title = 'B' +new synced base: { title: 'A', priority: 2, serverField: 'x' } +ideal visible row: { title: 'B', priority: 2, serverField: 'x' } +``` + +However, full patch/intention projection is not required in the first slice. Phase 1 may continue using existing `modified` snapshots while establishing: + +- a centralized mutation log/index boundary; +- immediate sync/base application; +- visible state projection through one path; +- transaction-derived mutation/error records; +- tests for derived collection behavior. + +Nested patch semantics, array mutations, custom codecs, and conflict detection are follow-up work. + +## Sync application semantics + +Committed authoritative sync/base changes should apply immediately, even while optimistic mutations are pending. + +Example: + +```txt +Initial base: + todos = [{ id: 1, title: 'A' }] + +Local optimistic update: + id 1 title -> 'A*' + +While mutationFn is pending, sync inserts: + { id: 2, title: 'B' } +``` + +Current behavior can queue the sync insert because a transaction is `persisting`, causing source or derived collections to miss `B` until the mutation settles. + +Target behavior: + +```txt +base immediately becomes: + [{ id: 1, title: 'A' }, { id: 2, title: 'B' }] + +visible state projects pending local mutation: + [{ id: 1, title: 'A*' }, { id: 2, title: 'B' }] +``` + +This should be treated as an internal correctness fix / clarified 1.0 semantics, not as a behavior to preserve behind an option. + +The RFC does not require changing the existing sync writer API (`begin` / `write` / `commit`). If that API proves insufficient during implementation, a targeted follow-up can address it. The core requirement is semantic: committed authoritative changes advance base state immediately. + +## Stable materialized change stream + +The collection is not only a way to compute the current visible row for a key. It is also the source of a stable materialized change stream consumed by `subscribeChanges`, framework adapters, and live-query collections. + +This is the hardest part of the refactor. The current implementation grew much of its complexity to preserve this stream while optimistic mutations are added, completed, rolled back, or temporarily held while sync is pending. Any mutation-log design must preserve this event contract. + +The required invariant is: + +```txt +previous materialized visible state ++ transition to synced/base state and/or active mutation log += next materialized visible state ++ stable change events +``` + +The emitted events must be correct across at least these transitions: + +- optimistic mutation added; +- transaction enters `persisting`; +- authoritative sync arrives while a local transaction is pending or persisting; +- transaction completes and its mutations leave active projection; +- transaction fails and rolls back; +- transaction enters `needs-resolution`; +- multiple transactions affect the same key; +- unrelated sync changes arrive while local mutations are pending. + +The design should therefore be framed around transitions from one visible materialized state to the next, not only around computing `get(key)` from base plus mutations. A simple first implementation may recompute and diff more than the current code does. That is acceptable if it makes the invariants clear. Later implementations can optimize the same model with dirty-key sets, per-key mutation indexes, cached projections, or batched diffs. + +The important simplification is to remove the branch where committed sync is skipped because a local transaction is persisting. If sync always advances the authoritative base, then the change stream can be generated from one reconciliation path instead of separate paths for immediate sync, delayed sync, optimistic recomputation, and pending-sync replay. + +## Live-query and derived collections + +The Mutation Log changes how each collection reconciles its own synced state and pending mutations. It does not change how live-query or derived collections choose their inputs. + +Derived/live-query collections continue to consume source collection state as they do today. The core fix is inside each source collection: authoritative base state keeps advancing, and optimistic mutations are projected over it. + +## Transport confirmation and core settlement + +Adapter authors often naturally model writes in transport-specific stages: + +```txt +write -> optimistic mutation is applied and the transport request starts +confirm -> the server accepts the write, for example with HTTP 200 +echo -> the authoritative sync/read path delivers the corresponding change +``` + +Core intentionally does **not** model these as separate transaction lifecycle states. TanStack DB's mutation handler boundary combines the adapter's notion of confirmation and settlement into one completion point: + +```txt +pending -> mutation handler still owns the optimistic mutations +completed/settled -> mutation handler completed successfully and core can drop those optimistic mutations +``` + +That is the intended contract. If an adapter requires the sync echo to avoid flicker, its mutation handler should await that echo before resolving. If a transport considers HTTP 200 sufficient for TanStack DB settlement, it can resolve there. In either case, core only sees the mutation handler as pending or complete; any earlier acknowledgement is outside the core transaction state machine. + +This RFC preserves that semantic boundary. It does not add first-class core transaction states for HTTP confirmation, sync echo, or read-path observation. + +## Public status APIs (future / out of Phase 1) + +The APIs in this section are not added by Phase 1. TanStack DB is pre-1.0, so 1.0 should remove or replace ambiguous APIs instead of preserving confusing compatibility. + +### Replace `$synced` + +`$synced` should not be the 1.0 row-level write confirmation concept. It is ambiguous across adapters and can be confused with backend upload/read-back confirmation. + +Introduce local-write-specific row props instead: + +```ts +row.$hasPendingWrites // boolean +row.$writeStatus // 'clean' | 'pending' | 'needs-resolution' | 'failed' +``` + +`$hasPendingWrites` means: + +> This row is affected by one or more unsettled optimistic mutations owned by this collection. + +It does **not** mean: + +- backend has not observed this write; +- mutation has not been uploaded; +- local durability is missing. + +Durability should mostly “just work” when `@tanstack/offline-transactions` or another durability layer is installed. Advanced/debug UIs can inspect durability through logged mutation/transaction metadata if needed, but durability should not become a row-level status. + +`$writeStatus` is derived from the transaction state of mutations affecting the row. Exact aggregation rules can be finalized during implementation, but the intended common meanings are: + +- `clean`: no unsettled optimistic mutation affects the row; +- `pending`: at least one unsettled optimistic mutation affects the row; +- `needs-resolution`: at least one mutation affecting the row explicitly needs app/user resolution; +- `failed`: a recent failed mutation affecting the row is retained in mutation history, if surfaced at row level. + +`$pendingOperation` from #1431 is a natural extension once mutations are logged/indexed, but it is not central to this RFC. + +### Replace `isPersisted.promise` + +`isPersisted.promise` should not be the 1.0 transaction waiting API. + +Expose transaction waiting over the in-scope transaction states/public names: + +```ts +await tx.when('settled') +await tx.when('failed') +await tx.when('needs-resolution') +``` + +There is intentionally no `tx.when('accepted')` or `tx.when('observed')` in this RFC. + +For this RFC: + +```txt +settled = mutationFn completed successfully and core can remove the optimistic mutation +``` + +Adapters that need sync/read-path echo before TanStack DB should consider a write complete should keep the mutation function pending until that echo arrives. In this RFC, an initial transport acknowledgement is adapter-internal information, not a TanStack DB completed transaction. Core does not need a separate `accepted` or `observed` status because the mutation function completion boundary is the settlement boundary. + +### Queryable mutation records + +Rows should expose a small ergonomic virtual surface. Detailed lifecycle/error state should be queryable through logged mutations joined with their owning transaction state, for example: + +```ts +db.mutations +// exact global vs collection-scoped API can be finalized during implementation +``` + +This lets applications build: + +- global failed-write toasts; +- “save needs attention” lists; +- form-level resolution UIs; +- Devtools timelines; +- debugging views. + +Users should not normally create raw logged mutations through this API. Collection mutations and transactions remain the write API. + +## `needs-resolution` + +Add `needs-resolution` as an explicit recoverable transaction state. + +This is not a retry/backoff state. Generic retrying remains the user’s `mutationFn` responsibility, an adapter responsibility, or an `@tanstack/offline-transactions` concern. + +`needs-resolution` should be entered only when user/app code explicitly signals it, likely by throwing a typed/custom error from `mutationFn`: + +```ts +throw new NeedsResolutionError({ + message: 'Validation failed', + fields: { + email: 'Already taken', + }, +}) +``` + +Core behavior: + +```txt +mutationFn throws NeedsResolutionError +-> transaction.state = 'needs-resolution' +-> optimistic mutations remain in the active log +-> row/write status reflects resolution needed +-> owning transaction exposes resolution metadata +-> app can resolve by changing state and retrying, or aborting/discarding according to API design +``` + +Ordinary thrown errors remain terminal by default and roll back according to current/default semantics. + +## Write errors and mutation history + +Write-related errors should live on the transaction/mutations that caused them, not primarily on `collection.error`. + +This addresses the deeper issue behind #672. A collection can have health/load/sync errors, but many actionable errors are tied to a particular write. A single mutable `collection.error` slot is too coarse: + +- multiple errors overwrite each other; +- one row write failure does not mean the whole collection is unusable; +- retry/resolution is per mutation; +- apps need to show errors after navigation; +- Devtools need identity and timestamps. + +The Mutation Log plus transaction state should become the primary source of truth for write lifecycle and write errors. + +Collection health/error APIs may still exist for non-write collection health, but they should aggregate or reference underlying mutation/effect records where appropriate. + +### Retention and GC + +Failed transaction/mutation records should remain queryable after rollback so applications can notify users after navigation and developers can debug failures. + +But the mutation history must be bounded. Previous attempts at global transaction stores raised memory concerns in long-lived or busy apps. + +Requirements: + +- Mutations belonging to active transactions (`pending`, `persisting`, `needs-resolution`) are retained while active. +- Historical failed transaction/mutation records are retained for a bounded recent-history window/count. +- Once TanStack DB considers a transaction complete/settled, its mutations do not need to remain in the active projection log solely for successful-history retention. This is a retention statement, not a requirement that adapters must treat an initial transport acknowledgement as completion. +- Exact TTL/count defaults are implementation details. +- Defaults should be high enough for normal toast/error-after-navigation UX. +- Applications needing long-term audit/history should subscribe/copy mutation records elsewhere. + +This RFC does not add explicit `acknowledge()` or `clearFailed()` APIs. Toast dismissal is app UI state, not mutation log state. + +## Offline transactions + +Without `@tanstack/offline-transactions`, optimistic mutations are in-memory and are not durable across reloads unless another persistence layer provides durability. + +With `@tanstack/offline-transactions`, the package should become a durability/execution layer over the core mutation log: + +- persist unsettled transaction mutations; +- restore them into the log on startup; +- schedule mutation execution; +- handle retry/backoff policy; +- handle connectivity hints; +- handle leader election / coordination where needed; +- mark durable mutation metadata where useful. + +It should not need to recreate optimistic state through separate restoration transactions or maintain a second transaction truth model. + +This means `@tanstack/offline-transactions` can become dramatically slimmer. Core owns in-memory mutation state and projection; the offline package owns durable storage and execution. + +## Phased migration + +Implementation should happen in thin vertical slices, not as a large hidden rewrite and not as public APIs backed by old internals. + +### Phase 1: core vertical slice + +Prove the model in `@tanstack/db` core first: + +- introduce an in-memory Mutation Log/index around existing `Transaction.mutations` data; +- project collection visible state through base + logged mutations; +- apply committed sync/base updates immediately; +- keep current mutation/transaction APIs working; +- expose minimal transaction-derived row/mutation write state internally or experimentally; +- generate stable change events by diffing previous visible materialized state against next visible materialized state; +- add tests for pending optimistic write + incoming sync + derived live-query updates; +- preserve current settlement semantics: mutationFn success settles mutations. + +This phase should not require Electric, PowerSync, or offline-transactions changes beyond test adjustments unless current adapter code assumes delayed sync. + +### Phase 2: 1.0 local write status APIs (future / out of Phase 1) + +- remove/replace `$synced`; +- remove/replace `isPersisted.promise`; +- add `$hasPendingWrites` and `$writeStatus`; +- add transaction `when(...)` over the in-scope transaction states; +- expose queryable logged mutations joined with transaction state; +- add bounded historical failed-mutation retention; +- add `needs-resolution` typed error/state flow. + +### Phase 3: offline durability over the log + +- refactor `@tanstack/offline-transactions` to persist/restore logged mutations; +- remove restoration-transaction duplication; +- keep retry/backoff and connectivity concerns in the package; +- validate durability with reload/restart tests. + +### Later follow-ups enabled by the mutation log + +These should be separate RFCs or PR series: + +- stable `$viewKey` / entity identity (#19); +- mutation receipts for key mapping and server defaults (#456, #465, #900, #1465); +- stronger patch/intention replay for long-lived optimistic writes (#25); +- `$pendingOperation` and pending-delete query semantics (#1431); +- first-class transport confirmation/read-path echo APIs and `awaitTxId` integration; +- effect/query/sync error logs; +- advanced scheduling/dependency strategies; +- nested transactions/savepoints, if still needed. + +## Testing and invariants + +The refactor should be protected by invariant-focused tests. + +Core invariants: + +1. A pending local optimistic write does not prevent unrelated authoritative sync data from entering base state. +2. Derived/live-query collections see source collection state changes while optimistic writes are pending. +3. Change events are emitted from visible-state transitions, not from split optimistic/sync special cases. +4. A row affected by an active transaction mutation has `$hasPendingWrites = true`. +5. Successful `mutationFn` completion completes the transaction and removes its mutations from active projection by default. +6. Ordinary `mutationFn` failure rolls back and records bounded failed transaction/mutation history. +7. Typed resolution errors keep optimistic state visible and set the transaction to `needs-resolution`. +8. Failed transaction/mutation history is bounded by automatic retention. +9. Without offline-transactions, mutation log state is in-memory only. +10. With offline-transactions, pending mutations can be restored without inventing a second optimistic transaction model. + +Representative regression scenario: + +```txt +1. Base has row A. +2. User optimistically updates A, mutationFn remains pending. +3. Sync inserts unrelated row B. +4. Collection base includes B immediately. +5. Visible state includes A optimistic update and B. +6. Derived collection sees B immediately. +7. When mutationFn succeeds, the transaction completes and visible state remains consistent. +``` + +## Open implementation questions + +These should be answered during implementation, not over-specified in the RFC: + +- Exact logged mutation/index type shape. +- Whether queryable logged mutations joined with transaction state are global, collection-scoped, or both. +- Exact `$writeStatus` aggregation rules when multiple mutations affect one row. +- Exact failed mutation retention defaults. +- Exact typed error API for `needs-resolution`. +- How much Phase 1 can safely use `modified` snapshots before switching update projection toward `changes`. +- How to implement visible-state diffing efficiently enough without reintroducing split sync/optimistic branches. +- Whether failed mutations should be visible in row aggregate status after rollback, or only in mutation history. + +## Conclusion + +The durable fix is not another sync-while-persisting option, another optimistic map, or another adapter-specific status flag. + +TanStack DB should make pending mutations central and indexed: + +```txt +authoritative base state ++ unsettled collection-owned mutations += visible collection state +``` + +That single shift lets core apply sync immediately, gives 1.0 precise local write status, makes write errors queryable, supports recoverable validation, and gives offline-transactions a clean durability/execution role. + +Once this substrate exists, future work like stable view keys, server-generated defaults, mutation receipts, stronger patch replay, and backend observation can be added incrementally without each feature inventing its own reconciliation model. diff --git a/docs/superpowers/plans/2026-06-29-mutation-log-reconciliation.md b/docs/superpowers/plans/2026-06-29-mutation-log-reconciliation.md new file mode 100644 index 000000000..937ecddd2 --- /dev/null +++ b/docs/superpowers/plans/2026-06-29-mutation-log-reconciliation.md @@ -0,0 +1,1097 @@ +# Mutation Log Reconciliation Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Implement the Phase 1 core vertical slice from RFC #1625: committed sync updates always advance collection base state while active transaction mutations are projected over the base and stable visible-state change events are emitted. + +**Architecture:** Keep `Transaction` and `PendingMutation` as the lifecycle and mutation primitives. Add a focused mutation projection/diff layer inside `@tanstack/db` collection state so visible state is computed from `syncedData` plus active transaction mutations, rather than from scattered optimistic maps and delayed sync branches. This plan intentionally does not add `accepted`/`observed`, `$hasPendingWrites`, `$writeStatus`, `needs-resolution`, offline durability, or stable identity; those are follow-up phases. + +**Tech Stack:** TypeScript, Vitest, `@tanstack/db` core collection state/sync managers. + +## Global Constraints + +- Preserve current public write APIs: `collection.insert/update/delete`, `createTransaction`, `mutationFn`, `transaction.mutations`, and `isPersisted.promise` continue to work during this Phase 1 slice. +- Preserve current settlement semantics: a transaction settles when `mutationFn` resolves successfully; adapters that need sync echo before settlement keep `mutationFn` pending. +- Do not redesign sync writer APIs (`begin` / `write` / `commit`). +- Do not add operation lifecycle states or public operation APIs. +- Do not add `accepted`, `observed`, `awaitTxId` replacement, or transport-confirmation state. +- Use existing `PendingMutation` data first. Full patch replay, nested path semantics, array patches, and conflict detection are out of scope. +- Prioritize correctness and clear invariants over micro-optimization. Use dirty-key sets to constrain recomputation, but avoid reintroducing delayed-sync special cases. +- Follow `AGENTS.md`: avoid `any` in new public code, prefer precise types, add tests for bugs, and keep abstractions small. + +--- + +## Scope Check + +The RFC covers multiple phases. This plan implements only **Phase 1: core vertical slice**: + +1. Add regression tests that describe the target sync/mutation reconciliation behavior. +2. Add a small internal projection helper around existing `PendingMutation`s. +3. Refactor `CollectionStateManager.commitPendingTransactions()` so committed sync updates always apply to `syncedData` immediately. +4. Preserve stable materialized change events by diffing previous visible state against next visible state for affected keys. +5. Keep existing APIs and settlement semantics intact. + +Separate plans should cover later public status APIs, `needs-resolution`, bounded failed mutation history, and offline durability. + +## File Structure + +- Modify: `packages/db/src/collection/state.ts` + - Owns core synced state, optimistic state, pending sync transactions, and change-event generation. + - Add small internal helpers for active mutation projection and visible-state diffing. + - Remove the normal-sync blocking branch for `persisting` transactions. + +- Modify: `packages/db/tests/collection.test.ts` + - Update the existing test that currently asserts sync is not applied while a transaction is persisting. + - Add direct collection-state regression coverage for immediate sync application under an active mutation. + +- Modify: `packages/db/tests/query/live-query-collection.test.ts` + - Add regression coverage that a derived/live-query collection receives unrelated synced rows while a source collection has a pending optimistic mutation. + +- Optional modify: `packages/db/tests/collection-subscribe-changes.test.ts` + - Add regression coverage for the source collection `subscribeChanges` stream if the collection-level tests do not already assert event shape strongly enough. + +No new public source files are required for Phase 1. If `state.ts` becomes too unwieldy during implementation, create an internal helper file `packages/db/src/collection/mutation-projection.ts`, but only after Task 2 makes the desired helper boundaries clear. + +--- + +### Task 1: Add direct regression tests for immediate sync application while persisting + +**Files:** +- Modify: `packages/db/tests/collection.test.ts` + +**Interfaces:** +- Consumes: existing `createCollection`, `createTransaction`, `PendingMutation`, `MutationFn`, `getStateEntries` helper in `collection.test.ts`. +- Produces: failing tests that define Phase 1 semantics for base sync application and visible projection while a transaction is `persisting`. + +- [ ] **Step 1: Rename and rewrite the existing delayed-sync test** + +In `packages/db/tests/collection.test.ts`, replace the test currently named: + +```ts +it(`synced updates should *not* be applied while there's a persisting transaction`, async () => { +``` + +with this test body: + +```ts +it(`applies unrelated synced inserts while a transaction is persisting`, async () => { + const emitter = mitt() + + const collection = createCollection<{ id: number; value: string }>({ + id: `sync-while-persisting-unrelated`, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ begin, write, commit }) => { + // @ts-expect-error don't trust Mitt's typing and this works. + emitter.on(`*`, (_, changes: Array) => { + begin() + changes.forEach((change) => { + write({ + type: change.type, + // @ts-expect-error test intentionally emits partial sync payloads + value: change.changes, + }) + }) + commit() + }) + }, + }, + }) + + const mutationFn: MutationFn = ({ transaction }) => { + emitter.emit(`update`, [ + { type: `insert`, changes: { id: 2, value: `synced value` } }, + ]) + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + [2, { id: 2, value: `synced value` }], + ]) + + emitter.emit(`update`, transaction.mutations) + return Promise.resolve() + } + + const tx = createTransaction({ mutationFn }) + + tx.mutate(() => + collection.insert({ + id: 1, + value: `optimistic value`, + }), + ) + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + [2, { id: 2, value: `synced value` }], + ]) + + await tx.isPersisted.promise + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + [2, { id: 2, value: `synced value` }], + ]) +}) +``` + +- [ ] **Step 2: Add a same-key overlay regression test** + +Add this test immediately after the previous one: + +```ts +it(`keeps optimistic visible value when synced update for the same key arrives while persisting`, async () => { + const emitter = mitt() + + const collection = createCollection<{ id: number; value: string }>({ + id: `sync-while-persisting-same-key`, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ begin, write, commit }) => { + begin() + write({ type: `insert`, value: { id: 1, value: `base value` } }) + commit() + + // @ts-expect-error don't trust Mitt's typing and this works. + emitter.on(`*`, (_, changes: Array) => { + begin() + changes.forEach((change) => { + write({ + type: change.type, + // @ts-expect-error test intentionally emits partial sync payloads + value: change.changes, + }) + }) + commit() + }) + }, + }, + }) + + const mutationFn: MutationFn = ({ transaction }) => { + emitter.emit(`update`, [ + { type: `update`, changes: { id: 1, value: `server value` } }, + ]) + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + ]) + expect(collection._state.syncedData.get(1)).toEqual({ + id: 1, + value: `server value`, + }) + + emitter.emit(`update`, transaction.mutations) + return Promise.resolve() + } + + const tx = collection.update(1, (draft) => { + draft.value = `optimistic value` + }, { mutationFn }) + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + ]) + + await tx.isPersisted.promise + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + ]) +}) +``` + +If `collection.update(..., { mutationFn })` is not a valid overload in this codebase, write the same test using `createTransaction({ mutationFn })` and `tx.mutate(() => collection.update(1, draft => { ... }))`: + +```ts +const tx = createTransaction({ mutationFn }) +tx.mutate(() => + collection.update(1, (draft) => { + draft.value = `optimistic value` + }), +) +``` + +- [ ] **Step 3: Run the focused tests and verify they fail** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection.test.ts -t "persisting" +``` + +Expected before implementation: + +- The rewritten unrelated insert test fails because row `2` is not visible while the transaction is `persisting`. +- The same-key test may fail because `syncedData` is not updated until the transaction completes. + +- [ ] **Step 4: Commit the failing tests** + +```bash +git add packages/db/tests/collection.test.ts +git commit -m "test: capture sync while persisting reconciliation target" +``` + +--- + +### Task 2: Add derived/live-query regression coverage for the stable change stream + +**Files:** +- Modify: `packages/db/tests/query/live-query-collection.test.ts` + +**Interfaces:** +- Consumes: existing `createCollection`, `createLiveQueryCollection`, `eq`, `flushPromises`, and `stripVirtualProps` test utilities. +- Produces: failing test proving derived collections receive unrelated synced rows while a source collection has a pending optimistic mutation. + +- [ ] **Step 1: Add the failing live-query test** + +Append this test inside `describe('createLiveQueryCollection', ...)` in `packages/db/tests/query/live-query-collection.test.ts`: + +```ts +it(`shows unrelated synced source rows in a live query while a source mutation is persisting`, async () => { + type Todo = { id: number; text: string; projectId: number } + type SyncPayload = { type: `insert` | `update` | `delete`; value?: Todo; key?: number } + + const syncListeners: Array<(payload: SyncPayload) => void> = [] + + const todos = createCollection({ + id: `live-query-sync-while-persisting-source`, + getKey: (todo) => todo.id, + startSync: true, + sync: { + sync: ({ begin, write, commit }) => { + begin() + write({ type: `insert`, value: { id: 1, text: `one`, projectId: 1 } }) + commit() + + syncListeners.push((payload) => { + begin() + if (payload.type === `delete`) { + write({ type: `delete`, key: payload.key! }) + } else { + write({ type: payload.type, value: payload.value! }) + } + commit() + }) + }, + }, + }) + + const projectTodos = createLiveQueryCollection((q) => + q.from({ todo: todos }).where(({ todo }) => eq(todo.projectId, 1)), + ) + + await projectTodos.preload() + + expect( + Array.from(projectTodos.state.values()).map((row) => stripVirtualProps(row)), + ).toEqual([{ id: 1, text: `one`, projectId: 1 }]) + + let resolveMutation!: () => void + const mutationSettled = new Promise((resolve) => { + resolveMutation = resolve + }) + + const tx = createTransaction({ + mutationFn: async () => { + syncListeners[0]!({ + type: `insert`, + value: { id: 2, text: `two`, projectId: 1 }, + }) + await mutationSettled + }, + }) + + tx.mutate(() => + todos.update(1, (draft) => { + draft.text = `one optimistic` + }), + ) + + await flushPromises() + + expect( + Array.from(projectTodos.state.values()).map((row) => stripVirtualProps(row)), + ).toEqual([ + { id: 1, text: `one optimistic`, projectId: 1 }, + { id: 2, text: `two`, projectId: 1 }, + ]) + + resolveMutation() + await tx.isPersisted.promise + await flushPromises() + + expect( + Array.from(projectTodos.state.values()).map((row) => stripVirtualProps(row)), + ).toEqual([ + { id: 1, text: `one optimistic`, projectId: 1 }, + { id: 2, text: `two`, projectId: 1 }, + ]) +}) +``` + +- [ ] **Step 2: Run the focused live-query test and verify it fails** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/query/live-query-collection.test.ts -t "unrelated synced source rows" +``` + +Expected before implementation: + +- The test fails because row `2` does not appear in `projectTodos` while the source mutation is still `persisting`. + +- [ ] **Step 3: Commit the failing live-query test** + +```bash +git add packages/db/tests/query/live-query-collection.test.ts +git commit -m "test: capture live query sync during pending mutation" +``` + +--- + +### Task 3: Add internal helpers for visible-state snapshots and transaction mutation projection + +**Files:** +- Modify: `packages/db/src/collection/state.ts` + +**Interfaces:** +- Consumes: `Transaction`, `PendingMutation`, existing `syncedData`, existing `isThisCollection` helper. +- Produces: internal helpers used by `recomputeOptimisticState()` and `commitPendingTransactions()`: + - `private collectActiveTransactions(): Array>` + - `private projectMutationOntoVisibleState(visible: Map, mutation: PendingMutation): void` + - `private captureVisibleStateForKeys(keys: Set): Map` + - `private diffVisibleStateForKeys(previous, keys): Array>` + +- [ ] **Step 1: Import `PendingMutation` as a type** + +In `packages/db/src/collection/state.ts`, update the type import from `../types` to include `PendingMutation`: + +```ts +import type { + ChangeMessage, + CollectionConfig, + OptimisticChangeMessage, + PendingMutation, +} from '../types' +``` + +- [ ] **Step 2: Add active transaction helper** + +Inside `CollectionStateManager`, near `isThisCollection`, add: + +```ts + private collectActiveTransactions(): Array> { + const activeTransactions: Array> = [] + + for (const transaction of this.transactions.values()) { + if (![`completed`, `failed`].includes(transaction.state)) { + activeTransactions.push(transaction) + } + } + + return activeTransactions + } +``` + +- [ ] **Step 3: Add mutation projection helper** + +Below `collectActiveTransactions`, add: + +```ts + private projectMutationOntoVisibleState( + visible: Map, + mutation: PendingMutation, + ): void { + if (!this.isThisCollection(mutation.collection) || !mutation.optimistic) { + return + } + + switch (mutation.type) { + case `insert`: + case `update`: + visible.set(mutation.key, mutation.modified as TOutput) + break + case `delete`: + visible.delete(mutation.key) + break + } + } +``` + +This intentionally uses `modified` for Phase 1. Patch replay from `changes` is a follow-up. + +- [ ] **Step 4: Add key-scoped visible-state capture helper** + +Below `projectMutationOntoVisibleState`, add: + +```ts + private captureVisibleStateForKeys(keys: Set): Map { + const visible = new Map() + + for (const key of keys) { + const value = this.syncedData.get(key) + if (value !== undefined) { + visible.set(key, value) + } + } + + for (const transaction of this.collectActiveTransactions()) { + for (const mutation of transaction.mutations) { + if (keys.has(mutation.key as TKey)) { + this.projectMutationOntoVisibleState(visible, mutation) + } + } + } + + return visible + } +``` + +- [ ] **Step 5: Add visible-state diff helper** + +Below `captureVisibleStateForKeys`, add: + +```ts + private diffVisibleStateForKeys( + previousVisibleState: Map, + keys: Set, + ): Array> { + const events: Array> = [] + + for (const key of keys) { + const previousValue = previousVisibleState.get(key) + const nextValue = this.get(key) + + if (previousValue === undefined && nextValue !== undefined) { + events.push({ type: `insert`, key, value: nextValue }) + } else if (previousValue !== undefined && nextValue === undefined) { + events.push({ type: `delete`, key, value: previousValue }) + } else if ( + previousValue !== undefined && + nextValue !== undefined && + !deepEquals(previousValue, nextValue) + ) { + events.push({ + type: `update`, + key, + value: nextValue, + previousValue, + }) + } + } + + return events + } +``` + +- [ ] **Step 6: Run typecheck via tests and expect existing behavior to remain mostly unchanged** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection.test.ts -t "should apply optimistic updates by default" +``` + +Expected: + +- The command may still fail because Task 1 tests are failing, but it should not fail with TypeScript syntax/type errors from the new helpers. + +- [ ] **Step 7: Commit helper scaffolding** + +```bash +git add packages/db/src/collection/state.ts +git commit -m "refactor: add mutation projection helpers" +``` + +--- + +### Task 4: Refactor `recomputeOptimisticState()` to use the mutation projection helper + +**Files:** +- Modify: `packages/db/src/collection/state.ts` + +**Interfaces:** +- Consumes: `collectActiveTransactions()` and `projectMutationOntoVisibleState()` from Task 3. +- Produces: one shared mutation projection path for optimistic recomputation and sync reconciliation. + +- [ ] **Step 1: Replace manual active transaction collection** + +In `recomputeOptimisticState()`, replace this block: + +```ts + const activeTransactions: Array> = [] + + for (const transaction of this.transactions.values()) { + if (![`completed`, `failed`].includes(transaction.state)) { + activeTransactions.push(transaction) + } + } +``` + +with: + +```ts + const activeTransactions = this.collectActiveTransactions() +``` + +- [ ] **Step 2: Replace manual active mutation switch** + +In `recomputeOptimisticState()`, replace the loop body that manually switches on `mutation.type` for active optimistic transactions: + +```ts + if (mutation.optimistic) { + switch (mutation.type) { + case `insert`: + case `update`: + this.optimisticUpserts.set( + mutation.key, + mutation.modified as TOutput, + ) + this.optimisticDeletes.delete(mutation.key) + break + case `delete`: + this.optimisticUpserts.delete(mutation.key) + this.optimisticDeletes.add(mutation.key) + break + } + } +``` + +with this helper-based code: + +```ts + if (mutation.optimistic) { + const projected = new Map() + for (const [key, value] of this.optimisticUpserts) { + projected.set(key, value) + } + for (const key of this.optimisticDeletes) { + projected.delete(key) + } + + this.projectMutationOntoVisibleState(projected, mutation) + + if (mutation.type === `delete`) { + this.optimisticUpserts.delete(mutation.key) + this.optimisticDeletes.add(mutation.key) + } else { + const projectedValue = projected.get(mutation.key) + if (projectedValue !== undefined) { + this.optimisticUpserts.set(mutation.key, projectedValue) + this.optimisticDeletes.delete(mutation.key) + } + } + } +``` + +If this local `projected` map feels too indirect during implementation, keep the existing switch and only use `collectActiveTransactions()` in this task. The important review gate is that Task 3 helpers compile and can be used by `commitPendingTransactions()` in Task 5. + +- [ ] **Step 3: Run optimistic mutation tests** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection.test.ts -t "optimistic" +``` + +Expected: + +- Existing optimistic tests pass except the newly added Task 1 target tests may still fail until Task 5. +- If unrelated optimistic tests fail, revert Step 2 and keep only Step 1 in this task. + +- [ ] **Step 4: Commit the recompute cleanup** + +```bash +git add packages/db/src/collection/state.ts +git commit -m "refactor: share active transaction collection" +``` + +--- + +### Task 5: Apply committed sync transactions immediately and emit visible-state diffs + +**Files:** +- Modify: `packages/db/src/collection/state.ts` + +**Interfaces:** +- Consumes: `captureVisibleStateForKeys()` and `diffVisibleStateForKeys()` from Task 3. +- Produces: target reconciliation behavior for committed sync while transactions are `persisting`. + +- [ ] **Step 1: Remove persisting-transaction gating for normal sync commits** + +In `commitPendingTransactions()`, keep the initial calculation of committed and uncommitted sync transactions, but remove the condition that prevents processing when `hasPersistingTransaction` is true. + +Replace: + +```ts + if (!hasPersistingTransaction || hasTruncateSync || hasImmediateSync) { +``` + +with: + +```ts + if (committedSyncedTransactions.length > 0) { +``` + +Then remove `hasPersistingTransaction`, `hasImmediateSync`, and `hasTruncateSync` only if TypeScript reports they are unused. Keep `hasTruncateSync` if truncate-specific code still needs it. + +- [ ] **Step 2: Capture previous visible state for changed keys before applying sync** + +Immediately after `changedKeys` is built and before any sync operations mutate `syncedData`, insert: + +```ts + const previousVisibleState = this.captureVisibleStateForKeys(changedKeys) +``` + +If `changedKeys` is currently built after truncate snapshot logic, move only the `changedKeys` collection loop earlier; do not move truncate application itself. + +- [ ] **Step 3: Use previous visible state for event diffing** + +Find the section that currently initializes or uses `currentVisibleState` / `preSyncVisibleState` for comparing previous and next visible values. Replace the fallback capture logic: + +```ts + let currentVisibleState = this.preSyncVisibleState + if (currentVisibleState.size === 0) { + currentVisibleState = new Map() + for (const key of changedKeys) { + const currentValue = this.get(key) + if (currentValue !== undefined) { + currentVisibleState.set(key, currentValue) + } + } + } +``` + +with: + +```ts + const currentVisibleState = + this.preSyncVisibleState.size > 0 + ? this.preSyncVisibleState + : previousVisibleState +``` + +This preserves existing truncate/pre-captured behavior while making normal sync commits compare against the visible state before base mutation. + +- [ ] **Step 4: Ensure active optimistic mutations are reprojected after sync base mutation** + +Keep the existing block that clears `optimisticUpserts` / `optimisticDeletes` and reapplies active transactions. If implementation simplifies this block, the replacement must preserve this exact switch for active transactions: + +```ts + for (const transaction of this.transactions.values()) { + if (![`completed`, `failed`].includes(transaction.state)) { + for (const mutation of transaction.mutations) { + if ( + this.isThisCollection(mutation.collection) && + mutation.optimistic + ) { + switch (mutation.type) { + case `insert`: + case `update`: + this.optimisticUpserts.set( + mutation.key, + mutation.modified as TOutput, + ) + this.optimisticDeletes.delete(mutation.key) + break + case `delete`: + this.optimisticUpserts.delete(mutation.key) + this.optimisticDeletes.add(mutation.key) + break + } + } + } + } + } +``` + +Do not change update projection from `modified` to `changes` in this task. + +- [ ] **Step 5: Remove or neutralize delayed-sync-only event branch** + +If `commitPendingTransactions()` has an `else` branch for `hasPersistingTransaction` that emits events without mutating `syncedData`, delete it. After Step 1, all committed sync transactions should use the normal sync application path. + +The resulting shape should be: + +```ts + if (committedSyncedTransactions.length > 0) { + // apply sync to syncedData + // reproject active optimistic mutations + // diff previous visible state against next visible state + // emit events + this.pendingSyncedTransactions = uncommittedSyncedTransactions + } +``` + +- [ ] **Step 6: Run the Task 1 tests** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection.test.ts -t "sync while persisting" +``` + +Expected: + +- `applies unrelated synced inserts while a transaction is persisting` passes. +- `keeps optimistic visible value when synced update for the same key arrives while persisting` passes. + +- [ ] **Step 7: Run broader collection sync tests** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection.test.ts tests/collection-subscribe-changes.test.ts +``` + +Expected: + +- Existing tests pass or expose tests that encoded the old delayed-sync contract. +- For each old-contract failure, update the test only if the new RFC semantics make the old assertion wrong. + +- [ ] **Step 8: Commit immediate sync reconciliation** + +```bash +git add packages/db/src/collection/state.ts packages/db/tests/collection.test.ts packages/db/tests/collection-subscribe-changes.test.ts +git commit -m "fix: apply sync while projecting active mutations" +``` + +--- + +### Task 6: Verify stable change events for source subscriptions and live queries + +**Files:** +- Modify: `packages/db/tests/collection-subscribe-changes.test.ts` +- Modify: `packages/db/tests/query/live-query-collection.test.ts` +- Modify: `packages/db/src/collection/state.ts` only if these tests reveal event-shape bugs. + +**Interfaces:** +- Consumes: reconciliation behavior from Task 5. +- Produces: tests proving materialized visible-state transitions emit stable events into source subscriptions and live queries. + +- [ ] **Step 1: Add source `subscribeChanges` regression test if missing** + +Search `packages/db/tests/collection-subscribe-changes.test.ts` for a test that already covers committed sync while an optimistic transaction is persisting. + +Run: + +```bash +rg -n "persisting|pending optimistic|sync while" packages/db/tests/collection-subscribe-changes.test.ts +``` + +If no equivalent test exists, add this test near the other optimistic/synced mixed tests: + +```ts +it(`emits visible-state changes for unrelated sync while a mutation is persisting`, async () => { + const syncListeners: Array<(value: { id: number; value: string }) => void> = [] + + const collection = createCollection<{ id: number; value: string }>({ + id: `subscribe-sync-while-persisting`, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ begin, write, commit }) => { + syncListeners.push((value) => { + begin() + write({ type: `insert`, value }) + commit() + }) + }, + }, + }) + + const received: Array> = [] + collection.subscribeChanges((changes) => { + received.push(...changes) + }) + + let resolveMutation!: () => void + const mutationSettled = new Promise((resolve) => { + resolveMutation = resolve + }) + + const tx = createTransaction<{ id: number; value: string }>({ + mutationFn: async () => { + syncListeners[0]!({ id: 2, value: `synced` }) + await mutationSettled + }, + }) + + tx.mutate(() => collection.insert({ id: 1, value: `optimistic` })) + + expect(received.map((event) => ({ type: event.type, key: event.key }))).toEqual([ + { type: `insert`, key: 1 }, + { type: `insert`, key: 2 }, + ]) + + resolveMutation() + await tx.isPersisted.promise +}) +``` + +If `ChangeMessage` is not imported in this test file, add: + +```ts +import type { ChangeMessage } from '../src/types' +``` + +Use the correct relative path for the file; from `packages/db/tests/collection-subscribe-changes.test.ts`, the path is `../src/types`. + +- [ ] **Step 2: Run source subscription regression** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection-subscribe-changes.test.ts -t "unrelated sync while a mutation is persisting" +``` + +Expected: + +- The new source subscription test passes after Task 5. + +- [ ] **Step 3: Run live-query regression from Task 2** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/query/live-query-collection.test.ts -t "unrelated synced source rows" +``` + +Expected: + +- The live-query test passes. + +- [ ] **Step 4: Fix event-shape bugs if tests fail** + +If source subscription gets no insert for key `2`, inspect `commitPendingTransactions()` and ensure: + +```ts +this.indexes.updateIndexes(events) +this.changes.emitEvents(events, true) +``` + +run for committed sync while transactions are `persisting`. + +If live query gets source row `2` but in the wrong order, make the test order-insensitive by sorting IDs before assertion: + +```ts +const rows = Array.from(projectTodos.state.values()) + .map((row) => stripVirtualProps(row)) + .sort((a, b) => a.id - b.id) +expect(rows).toEqual([ + { id: 1, text: `one optimistic`, projectId: 1 }, + { id: 2, text: `two`, projectId: 1 }, +]) +``` + +- [ ] **Step 5: Commit stable change event tests/fixes** + +```bash +git add packages/db/src/collection/state.ts packages/db/tests/collection-subscribe-changes.test.ts packages/db/tests/query/live-query-collection.test.ts +git commit -m "test: verify stable changes during sync reconciliation" +``` + +--- + +### Task 7: Update stale tests/comments that encode delayed sync semantics + +User-requested addition: explicitly survey the existing test structure for stale old-behavior assumptions before editing. Prefer tweaking existing tests that already cover the scenario over adding redundant new coverage. Look especially for tests/comments/names around delayed sync while transactions are pending/persisting, including `collection.test.ts`, `collection-subscribe-changes.test.ts`, live-query collection tests, and related helpers. Update stale expectations only when they conflict with the new Mutation Log reconciliation behavior; preserve unrelated coverage. + +**Files:** +- Modify: `packages/db/tests/collection.test.ts` +- Modify: `packages/db/tests/collection-subscribe-changes.test.ts` +- Modify: `packages/db/src/collection/state.ts` +- Search-only: all `packages/db/tests/**/*.ts` + +**Interfaces:** +- Consumes: behavior from Tasks 5 and 6. +- Produces: test suite comments and names aligned with the new immediate-sync semantics. + +- [ ] **Step 1: Search for old delayed-sync language** + +Run: + +```bash +rg -n "not be applied|isn't applied|delayed|held while|persisting transaction|sync while persisting|pending sync" packages/db/tests packages/db/src/collection/state.ts +``` + +- [ ] **Step 2: Update comments that describe old behavior** + +For comments equivalent to: + +```ts +// The sync commit is held while the local insert transaction is persisting. +``` + +replace with: + +```ts +// Sync commits apply to base immediately; active optimistic mutations still determine visible state. +``` + +For comments equivalent to: + +```ts +// we're still in the middle of persisting a transaction, so sync is not applied +``` + +replace with: + +```ts +// we're still in the middle of persisting a transaction, but sync still advances base state +``` + +- [ ] **Step 3: Update test names that assert old behavior** + +Rename tests with old behavior names. Examples: + +```ts +it(`synced updates should *not* be applied while there's a persisting transaction`, ...) +``` + +becomes: + +```ts +it(`applies synced updates while projecting active transaction mutations`, ...) +``` + +Do not rewrite unrelated tests. + +- [ ] **Step 4: Run old-language search again** + +Run: + +```bash +rg -n "should \*not\* be applied|isn't applied|held while the local|because.*persisting.*not applied" packages/db/tests packages/db/src/collection/state.ts +``` + +Expected: + +- No matches for comments/test names that encode the removed behavior. + +- [ ] **Step 5: Commit wording cleanup** + +```bash +git add packages/db/tests packages/db/src/collection/state.ts +git commit -m "test: update sync reconciliation expectations" +``` + +--- + +### Task 8: Run validation and document Phase 1 status + +**Files:** +- Modify: `docs/rfcs/2026-06-25-mutation-log-reconciliation.md` only if implementation discoveries require wording changes. +- No source changes expected unless validation finds bugs. + +**Interfaces:** +- Consumes: all previous tasks. +- Produces: validated Phase 1 implementation branch ready for review. + +- [ ] **Step 1: Run focused DB tests** + +Run: + +```bash +pnpm --filter @tanstack/db test -- tests/collection.test.ts tests/collection-subscribe-changes.test.ts tests/query/live-query-collection.test.ts +``` + +Expected: + +- PASS. + +- [ ] **Step 2: Run full `@tanstack/db` test suite** + +Run: + +```bash +pnpm --filter @tanstack/db test +``` + +Expected: + +- PASS. + +- [ ] **Step 3: Search for accidental public API additions** + +Run: + +```bash +git diff origin/main...HEAD -- packages/db/src/index.ts packages/db/src/types.ts packages/db/src/transactions.ts +``` + +Expected: + +- No new public operation lifecycle API. +- No `accepted` or `observed` type/API. +- No public `db.operations` or `db.mutations` API in Phase 1. + +- [ ] **Step 4: Inspect final diff for split sync/optimistic branches** + +Run: + +```bash +git diff origin/main...HEAD -- packages/db/src/collection/state.ts | sed -n '1,260p' +``` + +Check manually: + +- Normal committed sync does not return early only because a transaction is `persisting`. +- There is no event-only branch that emits sync changes without mutating `syncedData`. +- Active transaction mutations are still projected over base. +- Truncate-specific logic still has tests covering preserved optimistic state. + +- [ ] **Step 5: Commit any final fixes** + +If Step 1-4 required changes: + +```bash +git add packages/db/src packages/db/tests docs/rfcs/2026-06-25-mutation-log-reconciliation.md +git commit -m "fix: finalize mutation log reconciliation slice" +``` + +If no changes were required, do not create an empty commit. + +- [ ] **Step 6: Prepare implementation summary** + +Write this summary in the PR description or final handoff: + +```md +## Summary + +- committed sync now advances base state even while local transactions are persisting +- active transaction mutations continue to project over base for visible state +- source collection change events are emitted from visible-state transitions +- live-query collections receive unrelated synced source rows during pending optimistic mutations + +## Validation + +- pnpm --filter @tanstack/db test -- tests/collection.test.ts tests/collection-subscribe-changes.test.ts tests/query/live-query-collection.test.ts +- pnpm --filter @tanstack/db test +``` + +--- + +## Self-Review + +### Spec coverage + +- Immediate sync/base application: Tasks 1 and 5. +- Stable materialized change stream: Tasks 2, 5, and 6. +- Current APIs preserved: Tasks 5 and 8. +- No `accepted`/`observed`/transport lifecycle: Task 8 checks public diffs. +- Mutation log as indexed `PendingMutation`s, not operation state machine: Tasks 3-5. +- Patch replay deferred: Task 5 explicitly keeps `modified` projection. +- Offline/status/failure history deferred: Scope Check and Task 8 keep them out of Phase 1. + +### Placeholder scan + +No `TBD`, `TODO`, or unspecified implementation steps are intentionally left. If an implementer chooses the optional helper-file split, they must keep the same helper signatures from Task 3. + +### Type consistency + +The plan consistently uses existing `Transaction`, `PendingMutation`, `MutationFn`, `ChangeMessage`, `TOutput`, and `TKey` vocabulary. No independent operation status type is introduced. diff --git a/packages/db/src/collection/state.ts b/packages/db/src/collection/state.ts index 5f4449c3b..fee728d6e 100644 --- a/packages/db/src/collection/state.ts +++ b/packages/db/src/collection/state.ts @@ -13,6 +13,7 @@ import type { ChangeMessage, CollectionConfig, OptimisticChangeMessage, + PendingMutation, } from '../types' import type { CollectionImpl } from './index.js' import type { CollectionLifecycleManager } from './lifecycle' @@ -83,6 +84,7 @@ export class CollectionStateManager< public pendingOptimisticDeletes = new Set() public pendingOptimisticDirectUpserts = new Set() public pendingOptimisticDirectDeletes = new Set() + public releasedOptimisticDirectKeys = new Set() /** * Tracks the origin of confirmed changes for each row. @@ -388,17 +390,17 @@ export class CollectionStateManager< */ public *keys(): IterableIterator { const { syncedData, optimisticDeletes, optimisticUpserts } = this - // Yield keys from synced data, skipping any that are deleted. + // Yield keys from synced data first, skipping any that are deleted. If a + // synced key also has an optimistic upsert, get(key) will still return the + // optimistic value at the synced key's original position. for (const key of syncedData.keys()) { if (!optimisticDeletes.has(key)) { yield key } } - // Yield keys from upserts that were not already in synced data. + // Then yield optimistic inserts that are not yet in synced data. for (const key of optimisticUpserts.keys()) { if (!syncedData.has(key) && !optimisticDeletes.has(key)) { - // The optimisticDeletes check is technically redundant if inserts/updates always remove from deletes, - // but it's safer to keep it. yield key } } @@ -474,6 +476,58 @@ export class CollectionStateManager< return collection === this.collection } + private collectActiveTransactions(): Array> { + const activeTransactions: Array> = [] + + for (const transaction of this.transactions.values()) { + if (![`completed`, `failed`].includes(transaction.state)) { + activeTransactions.push(transaction) + } + } + + return activeTransactions + } + + private projectMutationOntoVisibleState( + visible: Map, + mutation: PendingMutation, + ): void { + if (!this.isThisCollection(mutation.collection) || !mutation.optimistic) { + return + } + + switch (mutation.type) { + case `insert`: + case `update`: + visible.set(mutation.key as TKey, mutation.modified as TOutput) + break + case `delete`: + visible.delete(mutation.key as TKey) + break + } + } + + private captureVisibleStateForKeys(keys: Set): Map { + const visible = new Map() + + for (const key of keys) { + const value = this.syncedData.get(key) + if (value !== undefined) { + visible.set(key, value) + } + } + + for (const transaction of this.collectActiveTransactions()) { + for (const mutation of transaction.mutations) { + if (keys.has(mutation.key as TKey)) { + this.projectMutationOntoVisibleState(visible, mutation) + } + } + } + + return visible + } + /** * Recompute optimistic state from active transactions */ @@ -493,6 +547,13 @@ export class CollectionStateManager< const previousDeletes = new Set(this.optimisticDeletes) const previousRowOrigins = this.rowOrigins + const pendingSyncKeys = new Set() + for (const transaction of this.pendingSyncedTransactions) { + for (const operation of transaction.operations) { + pendingSyncKeys.add(operation.key as TKey) + } + } + // Update pending optimistic state for completed/failed transactions for (const transaction of this.transactions.values()) { const isDirectTransaction = @@ -503,9 +564,34 @@ export class CollectionStateManager< continue } this.pendingLocalOrigins.add(mutation.key) + if (this.releasedOptimisticDirectKeys.has(mutation.key as TKey)) { + this.pendingOptimisticUpserts.delete(mutation.key) + this.pendingOptimisticDeletes.delete(mutation.key) + this.pendingOptimisticDirectUpserts.delete(mutation.key) + this.pendingOptimisticDirectDeletes.delete(mutation.key) + this.pendingLocalOrigins.delete(mutation.key) + continue + } if (!mutation.optimistic) { continue } + const mutationAlreadyConfirmed = + mutation.type === `delete` + ? !this.syncedData.has(mutation.key as TKey) + : deepEquals( + this.syncedData.get(mutation.key as TKey), + mutation.modified as TOutput, + ) + + if (mutationAlreadyConfirmed) { + this.pendingOptimisticUpserts.delete(mutation.key) + this.pendingOptimisticDeletes.delete(mutation.key) + this.pendingOptimisticDirectUpserts.delete(mutation.key) + this.pendingOptimisticDirectDeletes.delete(mutation.key) + this.pendingLocalOrigins.delete(mutation.key) + continue + } + switch (mutation.type) { case `insert`: case `update`: @@ -556,16 +642,16 @@ export class CollectionStateManager< this.optimisticDeletes.clear() this.pendingLocalChanges.clear() - // Seed optimistic state with pending optimistic mutations only when a sync is pending - const pendingSyncKeys = new Set() - for (const transaction of this.pendingSyncedTransactions) { - for (const operation of transaction.operations) { - pendingSyncKeys.add(operation.key as TKey) - } - } + const activeTransactions = this.collectActiveTransactions() + const shouldRetainCompletedOptimism = activeTransactions.length > 0 + + // Seed optimistic state with completed optimistic mutations only while a sync + // or another transaction is still in flight. Once mutationFns settle without + // authoritative writes, the optimistic state is dropped. const staleOptimisticUpserts: Array = [] for (const [key, value] of this.pendingOptimisticUpserts) { if ( + shouldRetainCompletedOptimism || pendingSyncKeys.has(key) || this.pendingOptimisticDirectUpserts.has(key) ) { @@ -578,9 +664,11 @@ export class CollectionStateManager< this.pendingOptimisticUpserts.delete(key) this.pendingLocalOrigins.delete(key) } + const staleOptimisticDeletes: Array = [] for (const key of this.pendingOptimisticDeletes) { if ( + shouldRetainCompletedOptimism || pendingSyncKeys.has(key) || this.pendingOptimisticDirectDeletes.has(key) ) { @@ -594,14 +682,6 @@ export class CollectionStateManager< this.pendingLocalOrigins.delete(key) } - const activeTransactions: Array> = [] - - for (const transaction of this.transactions.values()) { - if (![`completed`, `failed`].includes(transaction.state)) { - activeTransactions.push(transaction) - } - } - // Apply active transactions only (completed transactions are handled by sync operations) for (const transaction of activeTransactions) { for (const mutation of transaction.mutations) { @@ -613,19 +693,25 @@ export class CollectionStateManager< this.pendingLocalChanges.add(mutation.key) if (mutation.optimistic) { - switch (mutation.type) { - case `insert`: - case `update`: - this.optimisticUpserts.set( - mutation.key, - mutation.modified as TOutput, - ) + const projected = new Map() + for (const [key, value] of this.optimisticUpserts) { + projected.set(key, value) + } + for (const key of this.optimisticDeletes) { + projected.delete(key) + } + + this.projectMutationOntoVisibleState(projected, mutation) + + if (mutation.type === `delete`) { + this.optimisticUpserts.delete(mutation.key) + this.optimisticDeletes.add(mutation.key) + } else { + const projectedValue = projected.get(mutation.key) + if (projectedValue !== undefined) { + this.optimisticUpserts.set(mutation.key, projectedValue) this.optimisticDeletes.delete(mutation.key) - break - case `delete`: - this.optimisticUpserts.delete(mutation.key) - this.optimisticDeletes.add(mutation.key) - break + } } } } @@ -811,26 +897,16 @@ export class CollectionStateManager< } /** - * Attempts to commit pending synced transactions if there are no active transactions - * This method processes operations from pending transactions and applies them to the synced data + * Commits pending synced transactions that have been marked committed. + * This method processes operations from pending transactions and applies them to the synced data. */ commitPendingTransactions = () => { - // Check if there are any persisting transaction - let hasPersistingTransaction = false - for (const transaction of this.transactions.values()) { - if (transaction.state === `persisting`) { - hasPersistingTransaction = true - break - } - } - // pending synced transactions could be either `committed` or still open. // we only want to process `committed` transactions here const { committedSyncedTransactions, uncommittedSyncedTransactions, hasTruncateSync, - hasImmediateSync, } = this.pendingSyncedTransactions.reduce( (acc, t) => { if (t.committed) { @@ -838,9 +914,6 @@ export class CollectionStateManager< if (t.truncate) { acc.hasTruncateSync = true } - if (t.immediate) { - acc.hasImmediateSync = true - } } else { acc.uncommittedSyncedTransactions.push(t) } @@ -854,21 +927,10 @@ export class CollectionStateManager< PendingSyncedTransaction >, hasTruncateSync: false, - hasImmediateSync: false, }, ) - // Process committed transactions if: - // 1. No persisting user transaction (normal sync flow), OR - // 2. There's a truncate operation (must be processed immediately), OR - // 3. There's an immediate transaction (manual writes must be processed synchronously) - // - // Note: When hasImmediateSync or hasTruncateSync is true, we process ALL committed - // sync transactions (not just the immediate/truncate ones). This is intentional for - // ordering correctness: if we only processed the immediate transaction, earlier - // non-immediate transactions would be applied later and could overwrite newer state. - // Processing all committed transactions together preserves causal ordering. - if (!hasPersistingTransaction || hasTruncateSync || hasImmediateSync) { + if (committedSyncedTransactions.length > 0) { // Set flag to prevent redundant optimistic state recalculations this.isCommittingSyncTransactions = true @@ -891,6 +953,38 @@ export class CollectionStateManager< } } + const hasTxidSyncCommit = committedSyncedTransactions.some((t) => + t.operations.some((op) => Array.isArray(op.metadata?.txids)), + ) + const txidConfirmedDirectUpserts = new Set() + const txidConfirmedDirectDeletes = new Set() + if (hasTxidSyncCommit) { + for (const transaction of this.transactions.values()) { + const isDirectTransaction = + transaction.metadata[DIRECT_TRANSACTION_METADATA_KEY] === true + if ( + !isDirectTransaction || + (transaction.state !== `completed` && + transaction.state !== `persisting`) + ) { + continue + } + for (const mutation of transaction.mutations) { + if ( + !this.isThisCollection(mutation.collection) || + !mutation.optimistic + ) { + continue + } + if (mutation.type === `delete`) { + txidConfirmedDirectDeletes.add(mutation.key as TKey) + } else { + txidConfirmedDirectUpserts.add(mutation.key as TKey) + } + } + } + } + const virtualSnapshotKeys = new Set(changedKeys) for (const key of this.pendingOptimisticDirectUpserts) { virtualSnapshotKeys.add(key) @@ -903,17 +997,16 @@ export class CollectionStateManager< const previousOptimisticUpserts = new Map(this.optimisticUpserts) const previousOptimisticDeletes = new Set(this.optimisticDeletes) - // Use pre-captured state if available (from optimistic scenarios), - // otherwise capture current state (for pure sync scenarios) - let currentVisibleState = this.preSyncVisibleState - if (currentVisibleState.size === 0) { - // No pre-captured state, capture it now for pure sync operations - currentVisibleState = new Map() - for (const key of changedKeys) { - const currentValue = this.get(key) - if (currentValue !== undefined) { - currentVisibleState.set(key, currentValue) - } + const previousVisibleState = this.captureVisibleStateForKeys(changedKeys) + + // Use the visible state captured before applying sync operations for all + // changed keys, then overlay any pre-captured optimistic states. The + // pre-captured map only contains operation keys, while changedKeys can + // also include metadata-only writes. + const currentVisibleState = new Map(previousVisibleState) + for (const [key, value] of this.preSyncVisibleState) { + if (changedKeys.has(key)) { + currentVisibleState.set(key, value) } } @@ -925,7 +1018,10 @@ export class CollectionStateManager< >() for (const transaction of this.transactions.values()) { - if (transaction.state === `completed`) { + if ( + transaction.state === `completed` || + transaction.state === `persisting` + ) { for (const mutation of transaction.mutations) { if (this.isThisCollection(mutation.collection)) { if (mutation.optimistic) { @@ -1135,11 +1231,26 @@ export class CollectionStateManager< } } - // Maintain optimistic state appropriately - // Clear optimistic state since sync operations will now provide the authoritative data. - // Any still-active user transactions will be re-applied below in recompute. - this.optimisticUpserts.clear() - this.optimisticDeletes.clear() + // Maintain optimistic state appropriately. Start from the optimistic + // overlay that was visible before the sync commit, then let confirmed + // direct writes and still-active transactions below adjust it. This keeps + // in-flight optimistic mutations projected while the synced base updates. + this.optimisticUpserts = new Map( + [ + ...new Map([ + ...previousOptimisticUpserts, + ...this.pendingOptimisticUpserts, + ]), + ].filter(([key]) => !changedKeys.has(key)), + ) + this.optimisticDeletes = new Set( + [ + ...new Set([ + ...previousOptimisticDeletes, + ...this.pendingOptimisticDeletes, + ]), + ].filter((key) => !changedKeys.has(key)), + ) // Reset flag and recompute optimistic state for any remaining active transactions this.isCommittingSyncTransactions = false @@ -1157,27 +1268,42 @@ export class CollectionStateManager< // Always overlay any still-active optimistic transactions so mutations that started // after the truncate snapshot are preserved. - for (const transaction of this.transactions.values()) { - if (![`completed`, `failed`].includes(transaction.state)) { - for (const mutation of transaction.mutations) { - if ( - this.isThisCollection(mutation.collection) && - mutation.optimistic - ) { - switch (mutation.type) { - case `insert`: - case `update`: - this.optimisticUpserts.set( - mutation.key, + for (const transaction of this.collectActiveTransactions()) { + const isDirectTransaction = + transaction.metadata[DIRECT_TRANSACTION_METADATA_KEY] === true + for (const mutation of transaction.mutations) { + if ( + this.isThisCollection(mutation.collection) && + mutation.optimistic + ) { + const mutationKey = mutation.key as TKey + const directMutationConfirmed = + isDirectTransaction && + changedKeys.has(mutationKey) && + (mutation.type === `delete` + ? !this.syncedData.has(mutationKey) + : deepEquals( + this.syncedData.get(mutationKey), mutation.modified as TOutput, - ) - this.optimisticDeletes.delete(mutation.key) - break - case `delete`: - this.optimisticUpserts.delete(mutation.key) - this.optimisticDeletes.add(mutation.key) - break - } + )) + + if (directMutationConfirmed) { + continue + } + + switch (mutation.type) { + case `insert`: + case `update`: + this.optimisticUpserts.set( + mutation.key as TKey, + mutation.modified as TOutput, + ) + this.optimisticDeletes.delete(mutation.key as TKey) + break + case `delete`: + this.optimisticUpserts.delete(mutation.key as TKey) + this.optimisticDeletes.add(mutation.key as TKey) + break } } } @@ -1187,7 +1313,19 @@ export class CollectionStateManager< // 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) { + const directUpsertsToRelease = new Set([ + ...this.pendingOptimisticDirectUpserts, + ...txidConfirmedDirectUpserts, + ]) + const directDeletesToRelease = new Set([ + ...this.pendingOptimisticDirectDeletes, + ...txidConfirmedDirectDeletes, + ]) + + for (const key of directUpsertsToRelease) { + if (hasTruncateSync && truncateOptimisticSnapshot?.upserts.has(key)) { + continue + } if (!changedKeys.has(key)) { changedKeys.add(key) if (!currentVisibleState.has(key)) { @@ -1196,16 +1334,19 @@ export class CollectionStateManager< currentVisibleState.set(key, previousValue) } } - this.pendingOptimisticUpserts.delete(key) - this.pendingLocalOrigins.delete(key) } + this.pendingOptimisticUpserts.delete(key) + this.optimisticUpserts.delete(key) + this.pendingLocalOrigins.delete(key) + this.releasedOptimisticDirectKeys.add(key) } - for (const key of this.pendingOptimisticDirectDeletes) { + for (const key of directDeletesToRelease) { if (!changedKeys.has(key)) { changedKeys.add(key) } this.pendingOptimisticDeletes.delete(key) this.pendingLocalOrigins.delete(key) + this.releasedOptimisticDirectKeys.add(key) } this.pendingOptimisticDirectUpserts.clear() this.pendingOptimisticDirectDeletes.clear() @@ -1270,11 +1411,13 @@ export class CollectionStateManager< newVisibleValue !== undefined ) { const completedOptimisticOp = completedOptimisticOps.get(key) - if (completedOptimisticOp) { - const previousValueFromCompleted = completedOptimisticOp.value - const previousValueWithVirtualFromCompleted = + const previousOptimisticValue = previousOptimisticUpserts.get(key) + if (completedOptimisticOp || previousOptimisticValue !== undefined) { + const previousValueFromOptimistic = + completedOptimisticOp?.value ?? previousOptimisticValue! + const previousValueWithVirtualFromOptimistic = enrichRowWithVirtualProps( - previousValueFromCompleted, + previousValueFromOptimistic, key, this.collection.id, () => previousVirtualProps.$synced, @@ -1284,7 +1427,7 @@ export class CollectionStateManager< type: `update`, key, value: newVisibleValue, - previousValue: previousValueWithVirtualFromCompleted, + previousValue: previousValueWithVirtualFromOptimistic, }) } else { events.push({ @@ -1430,6 +1573,7 @@ export class CollectionStateManager< this.pendingOptimisticDeletes.clear() this.pendingOptimisticDirectUpserts.clear() this.pendingOptimisticDirectDeletes.clear() + this.releasedOptimisticDirectKeys.clear() this.clearOriginTrackingState() this.isLocalOnly = false this.size = 0 diff --git a/packages/db/src/collection/sync.ts b/packages/db/src/collection/sync.ts index af89ed2cf..f5c727045 100644 --- a/packages/db/src/collection/sync.ts +++ b/packages/db/src/collection/sync.ts @@ -116,12 +116,10 @@ export class CollectionSyncManager< throw new SyncTransactionAlreadyCommittedWriteError() } - let key: TKey | undefined = undefined - if (`key` in messageWithOptionalKey) { - key = messageWithOptionalKey.key - } else { - key = this.config.getKey(messageWithOptionalKey.value) - } + const key = + `key` in messageWithOptionalKey + ? messageWithOptionalKey.key + : this.config.getKey(messageWithOptionalKey.value) if (this.state.pendingLocalChanges.has(key)) { this.state.pendingLocalOrigins.add(key) diff --git a/packages/db/tests/collection-subscribe-changes.test.ts b/packages/db/tests/collection-subscribe-changes.test.ts index 4f851f08a..abaedda05 100644 --- a/packages/db/tests/collection-subscribe-changes.test.ts +++ b/packages/db/tests/collection-subscribe-changes.test.ts @@ -8,6 +8,7 @@ import { and, count, eq, gt } from '../src/query/builder/functions' import { PropRef } from '../src/query/ir' import { hasVirtualProps } from '../src/virtual-props.js' import { stripVirtualProps } from './utils' +import type { WithVirtualProps } from '../src/virtual-props.js' import type { OutputWithVirtual } from './utils' import type { ChangeMessage, @@ -381,6 +382,81 @@ describe(`Collection.subscribeChanges`, () => { subscription.unsubscribe() }) + it(`does not emit metadata-only rows as inserts in mixed sync commits`, async () => { + const callback = vi.fn() + type TestSyncParams = Parameters< + SyncConfig<{ id: number; value: string }, number>[`sync`] + >[0] + let begin!: TestSyncParams[`begin`] + let write!: TestSyncParams[`write`] + let commit!: TestSyncParams[`commit`] + let setRowMetadata!: NonNullable[`row`][`set`] + + const collection = createCollection<{ id: number; value: string }, number>({ + id: `mixed-operation-metadata-diff-test`, + getKey: (item) => item.id, + sync: { + sync: (params) => { + begin = params.begin + write = params.write + commit = params.commit + setRowMetadata = params.metadata!.row.set + + begin() + write({ type: `insert`, value: { id: 1, value: `synced value` } }) + commit() + params.markReady() + }, + }, + }) + + await collection.stateWhenReady() + + const mutationFn: MutationFn = async ({ transaction }) => { + begin() + for (const mutation of transaction.mutations) { + write({ + type: mutation.type, + key: mutation.key as number, + value: mutation.modified as { id: number; value: string }, + }) + } + setRowMetadata(1, { seen: true }) + } + + const subscription = collection.subscribeChanges(callback) + callback.mockReset() + + const tx = createTransaction({ mutationFn }) + tx.mutate(() => collection.insert({ id: 2, value: `optimistic value` })) + + expect(normalizeChanges(callback.mock.calls[0]![0])).toEqual([ + { + type: `insert`, + key: 2, + value: { id: 2, value: `optimistic value` }, + }, + ]) + callback.mockReset() + + await tx.isPersisted.promise + callback.mockReset() + + commit() + await waitForChanges() + + const emittedChanges = callback.mock.calls.flatMap((call) => + normalizeChanges(call[0]), + ) + expect(emittedChanges).not.toContainEqual({ + type: `insert`, + key: 1, + value: { id: 1, value: `synced value` }, + }) + + subscription.unsubscribe() + }) + it(`should handle both synced and optimistic changes together`, async () => { const emitter = mitt() const callback = vi.fn() @@ -462,11 +538,9 @@ describe(`Collection.subscribeChanges`, () => { await tx.isPersisted.promise - // Sync confirmation should only change virtual props ($synced/$origin) - expect(callback).toHaveBeenCalledTimes(1) - expect( - normalizeChangesWithoutVirtualUpdates(callback.mock.calls[0]![0]), - ).toEqual([]) + // Under immediate-base visible-state diffing, an identical sync confirmation + // does not emit a second user-data event. + expect(callback).not.toHaveBeenCalled() callback.mockReset() // Update both items in optimistic and synced ways @@ -620,11 +694,9 @@ describe(`Collection.subscribeChanges`, () => { // Wait for changes to propagate await waitForChanges() - // Sync confirmation should only change virtual props ($synced/$origin) - expect(callback).toHaveBeenCalledTimes(1) - expect( - normalizeChangesWithoutVirtualUpdates(callback.mock.calls[0]![0]), - ).toEqual([]) + // Under immediate-base visible-state diffing, an identical sync confirmation + // does not emit a second user-data event. + expect(callback).not.toHaveBeenCalled() callback.mockReset() // Update one item only @@ -1709,16 +1781,19 @@ describe(`Collection.subscribeChanges`, () => { const insertEvents = changeEvents.filter((e) => e.type === `insert`) const updateEvents = changeEvents.filter((e) => e.type === `update`) - // Expected: 2 optimistic inserts + 2 sync updates = 4 events + // Expected: 2 optimistic inserts. The delayed sync writes confirm the same + // optimistic rows while the local overlay is still active, so no duplicate + // user-data insert events are emitted. A virtual-only `$synced` confirmation + // update may still be materialized where applicable. expect(insertEvents.length).toBe(2) - expect(updateEvents.length).toBe(2) + expect(updateEvents.length).toBe(1) } finally { vi.useRealTimers() vi.restoreAllMocks() } }) - it(`should handle single insert with delayed sync correctly`, async () => { + it(`should handle single insert with async persistence sync correctly`, async () => { vi.useFakeTimers() try { @@ -1733,7 +1808,7 @@ describe(`Collection.subscribeChanges`, () => { { id: string; n: number; foo?: string }, string >({ - id: `single-insert-delayed-sync-test`, + id: `single-insert-async-persistence-sync-test`, getKey: (item) => item.id, sync: { sync: (cfg) => { @@ -1763,24 +1838,73 @@ describe(`Collection.subscribeChanges`, () => { collection.insert({ id: `x`, n: 1 }) await vi.runAllTimersAsync() - // Should have optimistic insert + sync update - expect(changeEvents).toHaveLength(2) + // Should have the optimistic insert only; the delayed sync confirmation is + // applied to base while the local overlay is active and does not produce a + // duplicate user-data event. Tests that project virtual props cover any + // virtual-only `$synced` confirmation update separately. + expect(changeEvents).toHaveLength(1) expect(changeEvents[0]).toMatchObject({ type: `insert`, key: `x`, value: { id: `x`, n: 1 }, }) - expect(changeEvents[1]).toMatchObject({ - type: `update`, - key: `x`, - value: { id: `x`, n: 1, foo: `abc` }, - }) } finally { vi.useRealTimers() vi.restoreAllMocks() } }) + it(`emits visible-state changes for unrelated sync while a mutation is persisting`, async () => { + const syncListeners: Array<(value: { id: number; value: string }) => void> = + [] + + const collection = createCollection<{ id: number; value: string }>({ + id: `subscribe-sync-while-persisting`, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ begin, write, commit }) => { + syncListeners.push((value) => { + begin() + write({ type: `insert`, value }) + commit() + }) + }, + }, + }) + + const received: Array< + ChangeMessage> + > = [] + collection.subscribeChanges((changes) => { + received.push(...changes) + }) + + let resolveMutation!: () => void + const mutationSettled = new Promise((resolve) => { + resolveMutation = resolve + }) + + const tx = createTransaction<{ id: number; value: string }>({ + mutationFn: async () => { + syncListeners[0]!({ id: 2, value: `synced` }) + await mutationSettled + }, + }) + + tx.mutate(() => collection.insert({ id: 1, value: `optimistic` })) + + expect( + received.map((event) => ({ type: event.type, key: event.key })), + ).toEqual([ + { type: `insert`, key: 1 }, + { type: `insert`, key: 2 }, + ]) + + resolveMutation() + await tx.isPersisted.promise + }) + it(`should emit change events for multiple sync transactions before marking ready`, () => { const changeEvents: Array = [] let testSyncFunctions: any = null diff --git a/packages/db/tests/collection.test.ts b/packages/db/tests/collection.test.ts index c204b89af..2ea239630 100644 --- a/packages/db/tests/collection.test.ts +++ b/packages/db/tests/collection.test.ts @@ -77,7 +77,7 @@ describe(`Collection`, () => { ).toThrow(DuplicateKeySyncError) }) - it(`removes optimistic insert when sync confirms with a different server-generated key`, async () => { + it(`does not infer confirmation when a remote insert has matching non-key fields`, async () => { const options = mockSyncCollectionOptionsNoInitialState<{ id: number text: string @@ -101,8 +101,9 @@ describe(`Collection`, () => { options.utils.write({ type: `insert`, value: { id: 24, text: `two` } }) options.utils.commit() - // The sync commit is held while the local insert transaction is persisting. + // The sync commit applies immediately while the local insert transaction is persisting. expect(getStateEntries(collection)).toEqual([ + [24, { id: 24, text: `two` }], [4733, { id: 4733, text: `two` }], ]) @@ -110,10 +111,13 @@ describe(`Collection`, () => { await tx.isPersisted.promise await flushPromises() - expect(getStateEntries(collection)).toEqual([[24, { id: 24, text: `two` }]]) + expect(getStateEntries(collection)).toEqual([ + [24, { id: 24, text: `two` }], + [4733, { id: 4733, text: `two` }], + ]) }) - it(`updates live queries when an optimistic insert is replaced by a different server key`, async () => { + it(`keeps live-query optimistic rows when a remote insert has matching non-key fields`, async () => { const options = mockSyncCollectionOptionsNoInitialState<{ id: number text: string @@ -163,7 +167,10 @@ describe(`Collection`, () => { origin: todo.$origin, key: todo.$key, })), - ).toEqual([{ id: 24, synced: true, origin: `remote`, key: 24 }]) + ).toEqual([ + { id: 24, synced: true, origin: `remote`, key: 24 }, + { id: 4733, synced: false, origin: `local`, key: 4733 }, + ]) }) it(`should throw an error when trying to use mutation operations outside of a transaction`, async () => { @@ -580,26 +587,109 @@ describe(`Collection`, () => { await tx9.isPersisted.promise }) - it(`synced updates should *not* be applied while there's a persisting transaction`, async () => { + it(`applies unrelated synced inserts while a transaction is persisting`, async () => { const emitter = mitt() - // new collection w/ mock sync/mutation const collection = createCollection<{ id: number; value: string }>({ - id: `mock`, - getKey: (item) => { - return item.id + id: `sync-while-persisting-unrelated`, + getKey: (item) => item.id, + startSync: true, + sync: { + sync: ({ begin, write, commit }) => { + // @ts-expect-error don't trust Mitt's typing and this works. + emitter.on(`*`, (_, changes: Array) => { + begin() + changes.forEach((change) => { + write({ + type: change.type, + key: change.key as number, + value: ((change as any).modified ?? + (change as any).value ?? + change.changes) as { + id: number + value: string + }, + }) + }) + commit() + }) + }, + }, + }) + + let resolvePersist!: () => void + const mutationFn: MutationFn = () => + new Promise((resolve) => { + resolvePersist = resolve + }) + + const tx = createTransaction({ mutationFn }) + + tx.mutate(() => + collection.insert({ + id: 1, + value: `optimistic value`, + }), + ) + + emitter.emit(`update`, [ + { + type: `insert`, + key: 2, + value: { id: 2, value: `synced value` }, + changes: { id: 2, value: `synced value` }, }, + ]) + + expect(getStateEntries(collection)).toEqual([ + [2, { id: 2, value: `synced value` }], + [1, { id: 1, value: `optimistic value` }], + ]) + + emitter.emit(`update`, [ + { + type: `insert`, + key: 1, + value: { id: 1, value: `optimistic value` }, + changes: { id: 1, value: `optimistic value` }, + }, + ]) + resolvePersist() + + await tx.isPersisted.promise + + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + [2, { id: 2, value: `synced value` }], + ]) + }) + + it(`keeps optimistic visible value when synced update for the same key arrives while persisting`, async () => { + const emitter = mitt() + + const collection = createCollection<{ id: number; value: string }>({ + id: `sync-while-persisting-same-key`, + getKey: (item) => item.id, startSync: true, sync: { sync: ({ begin, write, commit }) => { + begin() + write({ type: `insert`, value: { id: 1, value: `base value` } }) + commit() + // @ts-expect-error don't trust Mitt's typing and this works. emitter.on(`*`, (_, changes: Array) => { begin() changes.forEach((change) => { write({ type: change.type, - // @ts-expect-error TODO type changes - value: change.changes, + key: change.key as number, + value: ((change as any).modified ?? + (change as any).value ?? + change.changes) as { + id: number + value: string + }, }) }) commit() @@ -609,56 +699,43 @@ describe(`Collection`, () => { }) const mutationFn: MutationFn = ({ transaction }) => { - // Sync something and check that that it isn't applied because - // we're still in the middle of persisting a transaction. emitter.emit(`update`, [ - // This update is ignored because the optimistic update overrides it. - { type: `insert`, changes: { id: 2, bar: `value2` } }, + { + type: `update`, + key: 1, + value: { id: 1, value: `server value` }, + changes: { value: `server value` }, + }, ]) + expect(getStateEntries(collection)).toEqual([ - [1, { id: 1, value: `bar` }], + [1, { id: 1, value: `optimistic value` }], ]) - // Remove it so we don't have to assert against it below - emitter.emit(`update`, [{ changes: { id: 2 }, type: `delete` }]) + expect(collection._state.syncedData.get(1)).toEqual({ + id: 1, + value: `server value`, + }) emitter.emit(`update`, transaction.mutations) return Promise.resolve() } - const tx1 = createTransaction({ mutationFn }) - - // insert - tx1.mutate(() => - collection.insert({ - id: 1, - value: `bar`, + const tx = createTransaction({ mutationFn }) + tx.mutate(() => + collection.update(1, (draft) => { + draft.value = `optimistic value` }), ) - // The merged value should immediately contain the new insert - expect(getStateEntries(collection)).toEqual([[1, { id: 1, value: `bar` }]]) - - // check there's a transaction in peristing state - expect( - // @ts-expect-error possibly undefined is ok in test - Array.from(collection._state.transactions.values())[0].mutations[0] - .changes, - ).toEqual({ - id: 1, - value: `bar`, - }) - - // Check the optimistic operation is there - const insertKey = 1 - expect(collection._state.optimisticUpserts.has(insertKey)).toBe(true) - expect(collection._state.optimisticUpserts.get(insertKey)).toEqual({ - id: 1, - value: `bar`, - }) + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + ]) - await tx1.isPersisted.promise + await tx.isPersisted.promise - expect(getStateEntries(collection)).toEqual([[1, { id: 1, value: `bar` }]]) + expect(getStateEntries(collection)).toEqual([ + [1, { id: 1, value: `optimistic value` }], + ]) }) it(`should throw errors when deleting items not in the collection`, () => { @@ -1783,7 +1860,7 @@ describe(`Collection`, () => { commit() }) - it(`open sync transaction isn't applied when optimistic mutation is resolved/rejected`, async () => { + it(`keeps open sync transaction isolated when optimistic mutation is resolved/rejected`, async () => { type Row = { id: number; name: string } const collection = createCollection( @@ -1820,7 +1897,7 @@ describe(`Collection`, () => { // we now reject the sync, this should trigger a rollback of the open transaction // and the optimistic state should be removed - // it should *not* trigger the open sync transaction to be applied to the synced state + // the still-open sync transaction remains isolated until it is committed await withExpectedRejection(`trigger rollback`, () => { collection.utils.rejectSync(new Error(`trigger rollback`)) return flushPromises() diff --git a/packages/db/tests/query/live-query-collection.test.ts b/packages/db/tests/query/live-query-collection.test.ts index 29d10cc27..cd733c32f 100644 --- a/packages/db/tests/query/live-query-collection.test.ts +++ b/packages/db/tests/query/live-query-collection.test.ts @@ -17,6 +17,7 @@ import { stripVirtualProps, } from '../utils.js' import { createDeferred } from '../../src/deferred' +import { createTransaction } from '../../src/transactions' import { BTreeIndex } from '../../src/indexes/btree-index' import { Func, Value } from '../../src/query/ir.js' import type { ChangeMessage, LoadSubsetOptions } from '../../src/types.js' @@ -2934,4 +2935,104 @@ describe(`createLiveQueryCollection`, () => { expect(derived.size).toBe(4) }) }) + + it(`shows unrelated synced source rows in a live query while a source mutation is persisting`, async () => { + type Todo = { id: number; text: string; projectId: number } + type SyncPayload = { + type: `insert` | `update` | `delete` + value?: Todo + key?: number + } + + const syncListeners: Array<(payload: SyncPayload) => void> = [] + + const todos = createCollection({ + id: `live-query-sync-while-persisting-source`, + getKey: (todo) => todo.id, + startSync: true, + sync: { + sync: ({ begin, write, commit, markReady }) => { + begin() + write({ type: `insert`, value: { id: 1, text: `one`, projectId: 1 } }) + commit() + markReady() + + syncListeners.push((payload) => { + begin() + if (payload.type === `delete`) { + write({ type: `delete`, key: payload.key! }) + } else { + write({ type: payload.type, value: payload.value! }) + } + commit() + }) + }, + }, + }) + + const projectTodos = createLiveQueryCollection((q) => + q.from({ todo: todos }).where(({ todo }) => eq(todo.projectId, 1)), + ) + + await projectTodos.preload() + + expect( + Array.from(projectTodos.state.values()).map((row) => + stripVirtualProps(row), + ), + ).toEqual([{ id: 1, text: `one`, projectId: 1 }]) + + let resolveMutation!: () => void + const mutationSettled = new Promise((resolve) => { + resolveMutation = resolve + }) + + const tx = createTransaction({ + mutationFn: async () => { + syncListeners[0]!({ + type: `insert`, + value: { id: 2, text: `two`, projectId: 1 }, + }) + await mutationSettled + }, + }) + + tx.mutate(() => + todos.update(1, (draft) => { + draft.text = `one optimistic` + }), + ) + + await flushPromises() + + // Clean up the intentionally blocked mutation even when the regression + // assertion fails, so the test reports the row mismatch rather than hanging. + let whilePersistingError: unknown + try { + expect( + Array.from(projectTodos.state.values()).map((row) => + stripVirtualProps(row), + ), + ).toEqual(expect.arrayContaining([{ id: 2, text: `two`, projectId: 1 }])) + } catch (error) { + whilePersistingError = error + } finally { + resolveMutation() + await tx.isPersisted.promise.catch(() => {}) + await flushPromises() + } + + if (whilePersistingError) { + throw whilePersistingError + } + + expect( + Array.from(projectTodos.state.values()).map((row) => + stripVirtualProps(row), + ), + ).toEqual([ + { id: 1, text: `one`, projectId: 1 }, + { id: 2, text: `two`, projectId: 1 }, + ]) + }) }) diff --git a/packages/db/tests/query/query-while-syncing.test.ts b/packages/db/tests/query/query-while-syncing.test.ts index a1ea4591a..75ae06396 100644 --- a/packages/db/tests/query/query-while-syncing.test.ts +++ b/packages/db/tests/query/query-while-syncing.test.ts @@ -1092,10 +1092,11 @@ describe(`Query while syncing`, () => { resolveInsertMutation!() await vi.advanceTimersByTimeAsync(10) // Wait for rollback microtask - // The optimistic mutation should be rolled back since we didn't sync it - expect(usersCollection.size).toBe(2) - expect(liveQuery.size).toBe(1) - expect(liveQuery.get(5)).toBeUndefined() + // The mutation remains visible after mutationFn settlement; Phase 1 does + // not use mutationFn settlement as a transport-confirmation/rollback API. + expect(usersCollection.size).toBe(3) + expect(liveQuery.size).toBe(2) + expect(liveQuery.get(5)?.name).toBe(`Eve`) // Now sync the data to persist it syncBegin!() @@ -1107,6 +1108,8 @@ describe(`Query while syncing`, () => { // Now it should be persisted expect(usersCollection.size).toBe(3) + // Immediate-base semantics expose the synced base row as soon as it is + // committed, before the source collection is marked ready. expect(liveQuery.size).toBe(2) expect(liveQuery.get(5)?.name).toBe(`Eve`)