diff --git a/.changeset/recover-interleaved-electric-source-commits.md b/.changeset/recover-interleaved-electric-source-commits.md new file mode 100644 index 0000000000..567a350ab0 --- /dev/null +++ b/.changeset/recover-interleaved-electric-source-commits.md @@ -0,0 +1,7 @@ +--- +'@tanstack/db': patch +'@tanstack/db-sqlite-persistence-core': patch +'@tanstack/electric-db-collection': patch +--- + +Recover eager Electric collections from a fresh snapshot when interleaved source commits leave persisted rows inconsistent with the resume cursor. Keep cached rows visible while recovery restores the source state. diff --git a/docs/collections/convex-collection.md b/docs/collections/convex-collection.md index 36aca89604..da2740d7eb 100644 --- a/docs/collections/convex-collection.md +++ b/docs/collections/convex-collection.md @@ -81,6 +81,7 @@ const tasks = createCollection( }), ) +const taskId: string = selectedTask._id const transaction = tasks.update(taskId, (task) => { task.completed = true }) diff --git a/docs/collections/electric-collection.md b/docs/collections/electric-collection.md index a206107d60..a7bc97cc5a 100644 --- a/docs/collections/electric-collection.md +++ b/docs/collections/electric-collection.md @@ -377,6 +377,13 @@ state, or incomplete key-set evidence also invalidates the saved cursor. Untagged shapes with complete resume evidence can still resume from their saved offset. +An eager persisted resume also needs compatible source progress when its adapter +supplies that capability. Missing or incompatible progress requires a fresh +snapshot, even with an explicit offset or handle. Those options do not certify +the cached rows. +An interrupted eager recovery keeps its durable reset marker. The next start +also requires a fresh snapshot, even with an explicit offset or handle. + Eager and progressive recovery fetch a full snapshot. Cached rows remain visible until that replacement completes; a partial batch or subset completion cannot publish it early. An on-demand collection with the current SQLite persistence diff --git a/docs/contributing/glossary.md b/docs/contributing/glossary.md index e64d8c7214..61eb31cd5c 100644 --- a/docs/contributing/glossary.md +++ b/docs/contributing/glossary.md @@ -98,6 +98,7 @@ until cleanup ends the run. | durable anchor | The row version and reset epoch observed before one source sync transaction is sent to a coordinator. If both still match and the exact transaction ID is absent, the replacement leader may apply that transaction once. | | exact-ID reconciliation | The replacement leader checks one immutable source transaction ID under the SQLite writer lock after its RPC result becomes indeterminate. A present ID acknowledges the original commit; an absent ID needs an unchanged durable anchor. | | reconciliation reload | A peer notification after exact-ID reconciliation that reloads persisted rows even when the receiver has already observed the same stream position. Equal public rows do not publish a second change batch. | +| source progress token | A sync adapter's opaque identity for one complete materialization at a provider position. Re-delivering that position is idempotent. Its predecessor expectation lets persistence detect a discontinuity between interleaved source commits without ordering provider handles. A discontinuity invalidates source progress evidence until an authoritative replacement. It is separate from a durable row version and a sync-run generation. | ## Offline outbox terms diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 6425c2cebe..51d375dcf6 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -2705,6 +2705,18 @@ serialization, unavailable key evidence after raw loss, concurrent schema migration, WAL across processes, and arbitrary longer transaction histories remain with their existing owners or need explicit future witnesses. +The same owner checks the optional source-progress bridge with an independent +path-of-edges model. A discontinuity stays incompatible across matching and +empty commits; a repeated endpoint is idempotent; a full replacement repairs +continuity. Physical key integrity is checked separately after every transaction. +The fixed/random `sqlite-resume.source-progress` campaigns use 80 runs each, +seed 2281 and direct seed/path replay. Histories have 3–16 actions plus an initial +replacement, opaque positions, stale predecessors, duplicates and optional old +schema migration. Declared reaches require gaps, later matching commits, +replacement after a gap, duplicate endpoints and migration. The primary Browser +owner and real OPFS receiver supply source-head convergence evidence; the storage +model alone does not establish provider framing or cross-host scheduling. + The SQLite resume-snapshot owner also checks a cold, unique-key, non-delete full replacement with 205 rows. It compares exact durable rows, row metadata, key-set evidence, resume metadata, and applied position at transaction return; @@ -2926,8 +2938,11 @@ lagging stream applies v1 at `20_0`, and tab B closes before it reaches v2. Tab A then shows v1, and the durable `electric:resume` offset is back at `20_0`. Nothing redelivers v2 to tab A while it stays open, because its stream is already past it. A tab that opens later resumes from `20_0` and -repairs the cache. The convergence above therefore holds only while the -lagging tab keeps running or another tab restarts its stream. The +repairs the cache in that history. Restart formerly failed when a later commit advanced the resume position +over stale rows (#2281). The eager SQLite bridge now records source +continuity atomically and requires a fresh snapshot on reopen after a gap; +see the [repair evidence](oracle-reviews/issue-2281-investigation.md). +This preserves temporary arrival-order regressions and leaves #2209 open. The multi-tab Electric oracle in `browser-db-sqlite-persistence` pins this history with `it.fails` beside a passing control in which the lagging tab catches up. Ordering the copies would need a generic @@ -3970,10 +3985,17 @@ a hostile wrong-answer control, and an explicit statement of remaining limits. Generated histories in the same file interleave server changes, deploys, partial and complete deliveries, every stop except resume invalidation, restarts, and reloads across two or three eager tabs - (fixed seed 952051961 and a random seed, 60 runs each; the shared + (fixed seed 952051961 and a random seed, 80 runs each; the shared `TANSTACK_DB_ORACLE_SEED`, `_PATH`, and `_PROPERTY= browser.multi-tab-reserved-history` replay one). The - model folds committed batches into the expected durable rows. After + grammar constructs a 24-step prefix before its unchanged 8–24 arbitrary + steps. Every stop kind occurs in a generated order: a source opens a + transaction, stops, lets a distinct peer commit, then recovers and catches + up. Source tab, initially absent key, and recovery method vary. Four + membership toggles restore the original rows before the arbitrary tail. + This makes stop/peer reach structural; an earlier CI random campaign + missed the SDK-give-up boundary with the unconstrained shuffle. + The model folds committed batches into the expected durable rows. After every step it judges the durable rows, every Collection status, and the rows of each tab with no open transaction; after every tab catches up it judges them again. A Collection status changes only through its @@ -3987,6 +4009,10 @@ a hostile wrong-answer control, and an explicit statement of remaining limits. generated histories and pinned with `it.fails`: a resumed tab that lags behind a peer can receive an update for a row the peer already deleted, and Electric's resume check puts it in `error`. + The published Electric resume-safety contract requires rejecting an + unknown partial row. Whether composition must avoid that condition + without failing the source remains a separate contract question; + this repair does not remove the guard or resolve the pinned expectation. The histories also post a peer `collection:reset` on the tabs' coordinator channel; no shipped host sends one yet, so this premise is controlled. The reset abandons every open reserved transaction, and @@ -4002,17 +4028,95 @@ a hostile wrong-answer control, and an explicit statement of remaining limits. A delivery may also commit only a prefix of what a stream owes, as a server answering at an earlier moment does, so lagging commits are generated. The final checkpoint requires the server head (Law 2). - Known gap #2209 is the one history that breaks it: a lagging tab + Known gap #2209 is one history that breaks it: a lagging tab commits an older row and closes for good. Generated histories never close a tab for good, and an `it.fails` replay through the same driver and checks pins that history. - A random CI campaign (seed `-1660880085`, path - `55:0:1:4:4:4:1:1:1:1:3:4:4:4:4`) found a second Law 2 history that - reproduces on `main`: row 1 reaches v2 and both tabs catch up, then - tab B stops, tab A catches up, and tab A stops, and the durable rows - lose row 1. The maintainer chose to leave it open with #2281; an - `it.fails` replay through the same driver and checks pins it, and the - random campaign can still draw it. Owner: this oracle. + The [#2281 repair record](oracle-reviews/issue-2281-investigation.md) + retains the row-loss and resurrection counterexamples on `82c5f999b`. + A lagging source changed rows; another source advanced the shared resume + position over the missing change. The eager Electric/SQLite bridge now + detects a discontinuous predecessor atomically with rows and metadata. + Source continuity remains incompatible until an authoritative replacement; + the physical key ledger stays independent, so cached rows remain visible. + Reopening a missing, foreign, or discontinuous source token requests a + fresh snapshot. The original row-loss replay and both six-step membership + replays now pass. The generated `browser.electric-source-prefix` property + runs 80 histories with seed 2281 and 80 without a seed, with direct + seed/path replay. It crosses initial membership, one to three toggle + pairs, swapped source roles, empty/disjoint advancement, drain/no-drain, + tab reload or source failure followed by same-Collection restart, and + current/missing/foreign progress. Caller positions vary between omitted, + initial `-1`, handle-only, numeric offset with handle, and `now`. + Declared reach + requires all four divergent membership/advancement cells, a draining + control, missing/foreign provenance, and divergent recovery through both + actual lifecycle paths. It also requires divergent recovery with explicit + numeric and `now` positions, plus initial-offset and handle-only controls. + Both Browser random registrations and the SQLite + source-progress registration stand aside during another property's path + replay. The SQLite owner separately + checks opaque predecessor paths, sticky gaps, duplicate endpoints, + replacement repair, and additive schema migration (80 fixed/random + histories each). Disabling gap retention fails both owners; rejecting + duplicate endpoints fails the SQLite law. Bypassing startup provenance + validation fails the Browser owner. + The real installed SDK/HTTP/PostgreSQL/Chromium OPFS receiving driver + `e2e/electric-source-prefix-oracle.opfs.spec.ts` shares the source-head + judgment. The original ten cells cover both membership directions, empty/disjoint + advancement and drain/no-drain, plus missing legacy provenance with + disjoint advancement. Each declares held earlier response, stale commit, + later advancement, and reopened source comparison. All ten pass; gap and + legacy-validation mutants fail the matching real disjoint cells at that + comparison. Ten more cells cross numeric and `now` caller positions with + both membership directions and current/legacy provenance, plus initial + offset and handle-only controls. Eight wrong-row comparisons were RED on + `883b565ab` before the eager-only repair; all ten explicit cells now pass. + Each explicit cell declares its caller-position reach. A mutant that + requests recovery but preserves the caller position fails the final Node + source-head comparison (seed 2281, path `1:0:0`). Explicit positions cannot + bypass eager source-progress recovery. Other sync modes and explicit + malformed/tag/shape regimes retain their earlier behavior and are + outside this added evidence. Real foreign-token provenance and arbitrary + changes of caller form still need separate receiving witnesses. + Each of the source-prefix property's 80 generated inputs now runs both + its original history and an interrupted-recovery neighbor (160 histories + per campaign). The latter starts recovery without a caller cursor, stops + before any source delivery, and reopens with the generated caller form. + Actual persisted reset markers before divergent numeric/`now` recovery + are required reaches: fixed/random counts are 10/6 and 7/8. Both campaigns + fail on `623c59955` before the eager-only reset fix, and pass afterward. + The real matrix adds ten interrupted histories across both membership + directions and all five caller forms. A held first SDK response prevents + replacement publication; the driver observes the durable reset before + closing that run. Numeric/`now` give four wrong-row RED comparisons; + six neighboring controls pass. All thirty real cells pass after the fix. + The reset marker now requires eager recovery and clears the effective + caller position. No expectation, model rule, or comparison is relaxed. + This is bounded same-version eager Browser evidence, not + arbitrary scheduling, mixed client versions, native-device, Electron, + on-demand, optimistic, or monotonic-live-write coverage. + The final checkpoint now observes a reset-abandoned commit's source + failure, restarts it, and only then judges catch-up. The original reset + trace (seed `-1115798760`, path `1:4:3:3:3:3:3:3:3:3:3`) and the later + no-deploy reset trace (seed `-1371495859`, path + `75:0:0:0:0:0:0:1:1:0:0`) pass on unchanged production with that repaired + checkpoint. Both remain fixed inputs to the shared driver. A generated + reach separately requires recovery first discovered during that final + delivery, so an earlier reset failure cannot satisfy this premise. + It reached 2 of 80 fixed-seed histories and 6 of 80 random histories. + Removing the final restart makes both reset traces fail the final row + comparison. The broader Node campaign can also log a leadership term + write against a closed test database during teardown (pilot `PILOT/S02`, + fixed seed `952051961`). This is distinct from injected stream errors; + its user-facing effect is unverified. Owners: broadcast coordination and + this host driver. Needed witness: hold the election term write across + coordinator disposal and database close, then observe its settlement + and resource release on controlled and real hosts. Passing convergence + comparisons do not establish clean teardown. + A generated cached-row regression also preserves intermediate observations during + source-continuity repair. #2209's permanent-close case and #2267's + resumed-update gap remain pinned expected failures under this owner. A claimed leftover-replay/reset failure still needs a concrete schedule; the bounded two-replay-then-reset probe drains replay before reset. Live Electric/browser OPFS delivery needs the matching schedule in diff --git a/docs/contributing/oracle-reviews/issue-2281-investigation.md b/docs/contributing/oracle-reviews/issue-2281-investigation.md new file mode 100644 index 0000000000..1b0bc305ac --- /dev/null +++ b/docs/contributing/oracle-reviews/issue-2281-investigation.md @@ -0,0 +1,239 @@ +# Electric/SQLite source-progress recovery: issue #2281 + +This record summarizes the repair in [PR #2298](https://github.com/TanStack/db/pull/2298), +verified through implementation commit `0ec58057de52162da0eb4598616c2fc2eed2129b`. +The original investigation used production commit `82c5f999b32c9db827dd74e3f172604dfda28013`. +The [full investigation chronology](https://github.com/TanStack/db/blob/0ec58057de52162da0eb4598616c2fc2eed2129b/docs/contributing/oracle-reviews/issue-2281-investigation.md) +retains superseded results, intermediate manifests, and review corrections. +This summary changes no production code, model, history grammar, comparison, +exclusion, or run budget. + +## Law and failure + +After successful source recovery and complete delivery, durable rows and every +open eager Collection must equal the source head. Authority: Law 2 in the +[Browser multi-tab owner](../../../packages/browser-db-sqlite-persistence/tests/multi-tab-electric-oracle.test.ts), +the [coverage map](../oracle-coverage.md), and [#2209](https://github.com/TanStack/db/issues/2209). +Temporary arrival-order regressions remain allowed. The +[Electric recovery contract](../../collections/electric-collection.md#cleanup-and-resume-safety) +requires a fresh baseline after reset. Under the maintainer decision on #2265, +a reset-abandoned commit fails its source; successful catch-up requires restart. + +The row-loss and resurrection traces share this mechanism: + +1. Tab A commits two source changes to one key, reaching position 2. +2. Lagging tab B commits only the first change, restoring an older row state. +3. A advances shared resume metadata to position 2 with an empty batch or a + batch affecting another key. The affected row stays stale. +4. Both tabs restart beyond the missing corrective change, so catch-up cannot + repair the row. + +Atomic rows and metadata within each commit do not establish that commits from +separate streams describe one source history. Keeping the maximum offset still +permits stale row writes. Suppressing empty commits misses disjoint-key batches. +Physical key integrity also cannot certify source progress. + +The two reset traces exposed a separate oracle checkpoint defect. Final delivery +could first fail a reset-abandoned transaction after the initial restart sweep. +The driver now compares that failed state, restarts the newly failed source, +and completes delivery before judging convergence. This preserves the failure +contract; it does not weaken the source-head comparison. + +All four reported traces pass through the shared model, driver, and comparison: + +| Trace | Seed | Shrink path | +| --- | --- | --- | +| Original reset/deploy | `-1115798760` | `1:4:3:3:3:3:3:3:3:3:3` | +| Missing row | `-1660880085` | `55:0:1:4:4:4:1:1:1:1:3:4:4:4:4` | +| Resurrected row | `-699129038` | `78:2:0:3:9:12:12:13:12:13:13:13:13:14:14:14` | +| Reset without deploy | `-1371495859` | `75:0:0:0:0:0:0:1:1:0:0` | + +## Repair + +Each eager Electric commit supplies an opaque source progress token and its +expected predecessor through an optional persistence capability. The token +contains shape identity, handle, and offset. SQLite compares the predecessor +under its existing writer lock, atomically with rows and resume metadata. +A discontinuity makes source progress evidence incompatible until an +authoritative full replacement. A repeated complete endpoint preserves existing +evidence: it neither invents nor clears a gap. Provider handles need no ordering. + +Startup uses existing full-snapshot recovery when progress is missing, foreign, +or incompatible. This recovery overrides explicit caller positions because an +offset cannot supply missing baseline rows. An interrupted eager replacement's +durable reset requires the same recovery on the next start. Both recovery +admission and effective source position honor that obligation. Other sync modes +retain their explicit-position behavior. + +Source progress evidence has its own SQLite column and remains separate from +physical key integrity. A cached-row counterexample rejected combining them: +cached rows must remain visible during recovery. The additive migration also +covers existing databases. Arrival-order writes and their receipts remain. +Custom adapters advertise the optional capability only with an atomic check. + +## Evidence + +Each behavioral extension ran against unchanged production before its repair. +Setup failures and surviving mutants are excluded from the RED evidence. + +| Experiment | RED before its corresponding repair | Repaired GREEN / calibration | +| --- | --- | --- | +| Source continuity | Browser seed `2281`, path `0:0:0:0`, loses row 1 on `82c5f999b`; SQLite accepts missing gap evidence. Real receiver: four wrong-row failures and four passing draining controls. | Full SQLite owner: 61 passes. Browser restart and tab reopen converge. Ignoring durable gaps fails SQLite, Browser, and real disjoint-advancement comparisons. | +| Startup provenance | Missing/foreign tokens cannot justify the cached baseline. | Bypassing validation fails Browser and both real legacy-token cells. Rejecting duplicate endpoints separately fails SQLite at seed `2281`, path `0:7:1:2:2:2:2:2:2`. | +| Explicit caller positions | Extended Node campaigns fail; eight real numeric/`now` cells fail while initial-offset and handle-only controls pass. | Requesting recovery while preserving the raw caller position still fails the shared final comparison. Correcting both decisions passes. | +| Interrupted replacement | Paired Node campaigns fail at seed `2281`, path `1:0:0`, and random seed `-636405297`, path `14:0:0:0:0`. Four real numeric/`now` cells fail; six none/initial/handle controls pass. | All 30 real receiving cells pass. Keeping reset admission but removing its effective-position override fails the same Node replay; exact restoration passes. | +| Final reset checkpoint | Original production and driver reproduce both reported reset mismatches. | Corrected checkpoint passes on original production. Removing only the final restart/delivery block fails both reset replays. | +| Stop/peer coverage | CI reached SDK-give-up followed by peer commit zero times. | Constructed stop blocks establish every required stop boundary. Removing SDK-give-up abandonment fails a public-row assertion at seed `952051961`, path `1:0:0:0:0:0:0:0:1:1:0:0`. | + +Final implementation verification: complete Browser owner, **312 passes and two +tracked expected failures**; all **30 real receiving cells** pass. Electric +checks report 336 and 97 passes across two commands, without type errors. +Browser TypeScript, capability validation, Electron IPC, replay selection, +`pnpm lint:changed`, and whitespace checks pass. Documentation checking reports +zero new failures across 806 examples. Every temporary mutant was restored; +the final Electric source SHA-256 is +`46a417f2d65ccc21adcc72f56d80388d6f2f9fc32a172ab6e3fa1e4702508047`. + +### Independent judgments and exercised paths + +| Owner / driver | Independent judgment | Production path and observation cut | +| --- | --- | --- | +| Browser multi-tab oracle | Server snapshot log, shared source-head comparison | Real Collections, Browser coordinator with Node BroadcastChannel/Web Locks, native SQLite, controlled Electric stream; durable/public rows and health after each step and complete recovery/delivery. | +| SQLite resume-snapshot oracle | Path of opaque predecessor edges since the last authoritative replacement | Runtime capability, SQLite writer transaction, migration and resume snapshot; compatibility evidence after every durable transaction. | +| Real receiver for the Browser law | PostgreSQL source head, same shared comparison | PostgreSQL → Electric HTTP → installed SDK → Chromium Collections → OPFS SQLite; actual earlier responses held before delivery and reopened rows compared after catch-up. | + +The real matrix retains ten original/legacy, ten explicit-position, and ten +interrupted-recovery cells. Every cell requires held earlier response, stale +commit, later advancement, and reopened comparison. Interrupted cells observe +an actual durable reset before stopping replacement. No fabricated SDK messages +or offsets enter this driver. These are finite receiving witnesses. + +The source-prefix property uses 80 fixed-seed inputs and 80 random inputs. Each +input runs uninterrupted and interrupted neighbors: 160 histories per campaign. +Axes include membership, 1–3 toggle pairs, swapped source roles, empty/disjoint +advancement, draining, restart/reopen, token provenance, and five caller forms. +Numeric noninitial offsets without handles are SDK-invalid and excluded. +SQLite uses 80/80 inputs, 3–16 actions plus replacement, with stale/matching/empty +commits, duplicate endpoints, replacements, and old-schema migration. + +Final measured source-prefix and SQLite reaches: + +| Declared state | Fixed / random histories | +| --- | ---: | +| Present row / empty advancement | 28 / 26 | +| Present row / disjoint advancement | 22 / 14 | +| Absent row / empty advancement | 12 / 34 | +| Absent row / disjoint advancement | 14 / 18 | +| Drained before recovery | 84 / 68 | +| Missing progress token | 52 / 64 | +| Foreign progress token | 42 / 56 | +| Divergent same-Collection restart | 50 / 52 | +| Divergent tab reopen | 26 / 40 | +| Divergent explicit numeric resume | 20 / 12 | +| Divergent explicit `now` | 14 / 16 | +| Explicit initial offset | 34 / 42 | +| Explicit handle | 42 / 32 | +| Interrupted reset before numeric resume | 10 / 6 | +| Interrupted reset before `now` | 7 / 8 | +| SQLite discontinuity | 61 / 53 | +| SQLite matching commit after discontinuity | 37 / 42 | +| SQLite replacement after discontinuity | 44 / 36 | +| SQLite duplicate endpoint | 64 / 61 | +| SQLite legacy migration | 34 / 39 | + +The broader lifecycle property retains 80 accepted inputs per campaign. Four +constructed stop blocks precede 8–24 arbitrary actions, for 32–48 actions. +All four stop kinds cross a distinct peer commit and recovery. Final fixed/random +stop-boundary reaches are 81/80, including one fixed history later discarded +under the existing #2267 gap. Final-catch-up reset recovery reaches 4/2. +All other declared reaches pass. No comparison, exclusion, minimum reach, or +run budget was relaxed. + +### Review corrections + +Four independent reviewers found an unreachable focused restart branch and +replay registrations that ran unrelated campaigns. Actual-path instrumentation +first measured restart 0/reopen 38 and failed. Adding a stop made restart +reachable; gap and provenance mutants then failed that path. The original +row-loss trace already supplied a pre-production same-Collection restart RED; +the focused campaign's restart repair occurred later. Replay probes in both +Browser directions, unrelated-property selection, and SQLite selection verify +that only the requested campaign runs. + +Later CodeRabbit reviews identified the explicit-position and interrupted-reset +gaps above. The latest completed external review covers `124ce667c`; automatic +review is paused. Subsequent eager reset semantics have independent local +source review plus the generated, real receiving, and mutant evidence above. +The lossless task-local audit retains 30 items: 12 fixed, five refuted, one +accepted design, and 12 duplicates. No known in-scope source-progress finding +remains open. CodeRabbit supplied useful defect and maintenance findings; +its one-clause recovery suggestion was incomplete, as the mutant demonstrates. + +The rubric evidence is concentrated in the contract/model/path distinctions, +measured legal histories, calibrated comparisons, isolated replay, and retained +failure diagnostics. Cleanup attempts every release and preserves the primary +violation alongside cleanup failures; injected combined failures verify this. +Fixed examples use the shared oracle without separate expected answers. + +## Cost and limits + +The production delta is 153 net source lines across five files. Against +merge-base `672684846`, independently compressed package bundles grow by +DB +37, SQLite core +368, and Electric +168 gzip bytes: **573 bytes total**. +The standard application fixture adds 45 bytes but does not use Electric or +SQLite, so it does not measure their application cost. The repair adds one +metadata read per qualifying incremental source-progress transaction, stored +continuity evidence, and token processing. Runtime overhead was not benchmarked. +Incompatible progress can require a full source snapshot. + +The evidence covers bounded, same-version eager sources sharing SQLite. It +does not establish arbitrary schedules, mixed versions, monotonic intermediate +writes, on-demand subsets, optimism, native/Electron receiving, individual +receipt fulfillment, or clean teardown. Explicit changed-shape/malformed/tagged +resumes, arbitrary cursor selection over certified rows, other modes' reset +policy, changing caller forms within one history, and real foreign-token +receiving remain outside the claim. The coverage map names remaining witnesses. + +The permanently closed lagging-source history in #2209 remains a tracked +expected failure. The #2267 partial-update guard preserves the safety contract; +its no-error composition expectation remains a separate contract question. +The first #2281 comment's seed-row-metadata defect, #2272 metadata receiving, +and the pilot's replay/reset scheduling and teardown gaps retain separate owners. +The independent cluster report confirmed this repair's mechanism and led to a +separate reach for reset recovery first discovered during final catch-up. +It did not combine those other boundaries with source-progress recovery. + +CI on implementation `0ec58057d` passed 25 checks after one failed-job rerun. +Its original observer cleanup/restart defect also reproduces on clean current +main `bea84cd45`; the aggregate ordering reach gap is tracked in #2289. The +maintainer chose to keep #2298 focused. Observer evidence was sent to the +“Survey open issues and PRs” chat; both issues remain open. A passing retry +closes neither. CodeRabbit's skipped check reflects its paused review. + +## Reproduction + +Run from the repository root with two workers: + +```sh +pnpm --filter @tanstack/browser-db-sqlite-persistence exec vitest --run \ + tests/multi-tab-electric-oracle.test.ts --maxWorkers=2 --coverage.enabled=false +pnpm --filter @tanstack/db-sqlite-persistence-core exec vitest --run \ + tests/sqlite-resume-snapshot-oracle.test.ts --maxWorkers=2 --coverage.enabled=false +pnpm --filter @tanstack/browser-db-sqlite-persistence exec playwright test \ + --config playwright.opfs-electric.config.ts \ + electric-source-prefix-oracle.opfs.spec.ts electric-resume-two-tab.opfs.spec.ts \ + --workers=2 +``` + +For a reported trace, set `TANSTACK_DB_ORACLE_PROPERTY=browser.multi-tab-reserved-history` +and its seed/path from the first table; select `-t 'random or replayed'`. +For the final source-prefix replay, use property `browser.electric-source-prefix`, +seed `2281`, and path `1:0:0` with the same selector. Fixed campaigns stand aside +during path replay. Normal campaigns require no replay variables. + +Final local runs used `pnpm_config_verify_deps_before_run=false` after upstream +release-specifier changes triggered automatic reinstall. Installed external +dependencies were unchanged. No override or lockfile workaround accompanies +this PR. Raw logs and the complete review ledger remain task-local under +`notes/issue-2281/`; essential evidence is summarized here and preserved in the +linked historical record. diff --git a/docs/framework/preact/overview.md b/docs/framework/preact/overview.md index 9520c714d0..ff3963438b 100644 --- a/docs/framework/preact/overview.md +++ b/docs/framework/preact/overview.md @@ -26,7 +26,21 @@ For comprehensive documentation on writing queries (filtering, joins, aggregatio The `useLiveQuery` hook creates a live query that automatically updates your component when data changes: ```tsx -import { eq, useLiveQuery } from '@tanstack/preact-db' +import { createCollection, eq, localOnlyCollectionOptions } from '@tanstack/db' +import { useLiveQuery } from '@tanstack/preact-db' + +type Todo = { + id: string + text: string + completed: boolean +} + +const todoCollection = createCollection( + localOnlyCollectionOptions({ + getKey: (todo) => todo.id, + initialData: [], + }), +) function TodoList() { const { data, isLoading } = useLiveQuery((q) => diff --git a/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix-oracle.opfs.spec.ts b/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix-oracle.opfs.spec.ts new file mode 100644 index 0000000000..1fa30afbf0 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix-oracle.opfs.spec.ts @@ -0,0 +1,356 @@ +/** + * Real receiving path for the source-prefix grammar in the multi-tab owner. + * A real Electric HTTP response containing the first membership change waits + * in tab B while A applies a later change. B then commits that older response. + * A advances with an empty batch or disjoint change, and B reopens. PostgreSQL + * supplies the head to the shared judgment; no expected answer is restated. + * The drain branch is the neighboring control. These finite real-host cells + * receive the generated Node law; they do not cover arbitrary browser schedules. + */ +import { expect, test } from '@playwright/test' +import { makePgClient } from '../../db-collection-e2e/support/global-setup' +import { sourceHeadViolation } from '../tests/electric-convergence-oracle' +import { createOracleReach } from '../../db/tests/oracle-reach' +import { declareOracle } from '../../db/tests/oracle-declaration' +import type { Page } from '@playwright/test' +import type {} from './electric-source-prefix.opfs' + +type Cell = { + present: boolean + advance: `empty` | `disjoint` + drain: boolean + legacy: boolean + callerCursor: `none` | `initial-offset` | `handle` | `resume` | `now` + interrupted?: boolean +} +const cells: Array = [true, false] + .flatMap((present) => + ([`empty`, `disjoint`] as const).flatMap((advance) => + [false, true].map((drain) => ({ + present, + advance, + drain, + legacy: false, + callerCursor: `none` as const, + })), + ), + ) + .concat( + [true, false].map((present) => ({ + present, + advance: `disjoint` as const, + drain: false, + legacy: true, + callerCursor: `none` as const, + })), + ) +// Repeat the legal explicit resumed position over both membership failures +// and both current/legacy evidence. Initial-offset and handle-only options +// request a complete snapshot and supply the distinguishing neighbors. +cells.push( + ...([`resume`, `now`] as const).flatMap((callerCursor) => + [true, false].flatMap((present) => + [false, true].map((legacy): Cell => ({ + present, + advance: `disjoint`, + drain: false, + legacy, + callerCursor, + })), + ), + ), + ...([`initial-offset`, `handle`] as const).map((callerCursor): Cell => ({ + present: true, + advance: `disjoint`, + drain: false, + legacy: false, + callerCursor, + })), +) +// Interrupt recovery before the first SDK response can publish its snapshot. +// The real eager adapter durably records reset; numeric/now positions cannot +// turn that marker into a certified baseline. The other forms are controls. +cells.push( + ...([`none`, `initial-offset`, `handle`, `resume`, `now`] as const).flatMap( + (callerCursor) => + [true, false].map((present): Cell => ({ + present, + advance: `disjoint`, + drain: false, + legacy: false, + callerCursor, + interrupted: true, + })), + ), +) +const observe = (page: Page) => + page.evaluate(() => window.__electricSourcePrefix!.observe()) +const rowIds = async (page: Page) => + (await observe(page)).rows.map(({ id }) => id) + +declareOracle({ check: [sourceHeadViolation], grammar: [cells] }) +for (const cell of cells) { + test(`recovers ${cell.present ? `lost` : `resurrected`} rows after ${cell.advance} advancement${cell.drain ? ` with peer drain` : ``}${cell.legacy ? ` from legacy progress` : ``}${cell.callerCursor !== `none` ? ` with explicit ${cell.callerCursor}` : ``}${cell.interrupted ? ` after interrupted recovery` : ``}`, async ({ + context, + }) => { + const reach = createOracleReach( + `browser.electric-source-prefix.opfs`, + [ + `held earlier SDK response`, + `stale commit applied`, + `later advancement applied`, + `reopened source compared`, + ...(cell.interrupted ? [`interrupted recovery reset persisted`] : []), + ...(cell.callerCursor === `none` + ? [] + : [`explicit ${cell.callerCursor} supplied at recovery`]), + ], + { replaying: false }, + ) + reach.beginHistory() + const token = crypto.randomUUID().replaceAll(`-`, ``).slice(0, 12) + const table = `electric_prefix_${token}` + const collectionId = `prefix-${token}` + const parameters = new URLSearchParams({ + table, + collectionId, + databaseId: collectionId, + }) + const url = `/e2e/electric-source-prefix.opfs.html?${parameters}` + const pg = makePgClient() + const pages: Array = [] + const diagnostics: Array = [] + const close = async (page: Page) => { + if (page.isClosed()) return + diagnostics.push( + ...(await page.evaluate( + () => window.__electricSourcePrefix?.cleanup() ?? [], + )), + ) + await page.close() + } + let primary: unknown + let connected = false + try { + await pg.connect() + connected = true + await pg.query( + `CREATE TABLE public.${table} (id TEXT PRIMARY KEY, label TEXT NOT NULL)`, + ) + await pg.query(`INSERT INTO public.${table} VALUES ('anchor', 'anchor')`) + if (cell.present) + await pg.query(`INSERT INTO public.${table} VALUES ('target', 'old')`) + const open = async ( + callerPosition: Record = {}, + holdInitial = false, + ) => { + const page = await context.newPage() + pages.push(page) + page.on(`pageerror`, (error) => diagnostics.push(error.message)) + await page.goto( + `${url}&${new URLSearchParams({ + ...callerPosition, + ...(holdInitial ? { holdInitial: `true` } : {}), + })}`, + ) + await page.waitForFunction( + () => window.__electricSourcePrefix !== undefined, + ) + if (holdInitial) return page + await expect + .poll(async () => (await observe(page)).status, { timeout: 30_000 }) + .toBe(`ready`) + await expect + .poll(async () => (await observe(page)).commits, { timeout: 30_000 }) + .toBeGreaterThan(0) + return page + } + const a = await open() + const b = await open() + await b.evaluate(() => window.__electricSourcePrefix!.arm()) + const bCommits = (await observe(b)).commits + // The first change reverses membership; the second restores it. A has + // already applied the first before the second is issued to PostgreSQL. + if (cell.present) + await pg.query(`DELETE FROM public.${table} WHERE id = 'target'`) + else + await pg.query(`INSERT INTO public.${table} VALUES ('target', 'old')`) + await expect + .poll(() => rowIds(a), { timeout: 30_000 }) + .toEqual(cell.present ? [`anchor`] : [`anchor`, `target`]) + // Empty long-poll responses can precede the row response. Deliver those + // without releasing the first actual changed-row response. + await expect + .poll( + async () => { + const state = await observe(b) + if (state.pending[0] === 0) + await b.evaluate(() => + window.__electricSourcePrefix!.releaseOne(), + ) + return (await observe(b)).pending[0] ?? 0 + }, + { timeout: 30_000 }, + ) + .toBeGreaterThan(0) + reach.reach(`held earlier SDK response`) + if (cell.present) + await pg.query(`INSERT INTO public.${table} VALUES ('target', 'new')`) + else await pg.query(`DELETE FROM public.${table} WHERE id = 'target'`) + await expect + .poll(() => rowIds(a), { timeout: 30_000 }) + .toEqual(cell.present ? [`anchor`, `target`] : [`anchor`]) + await b.evaluate(() => window.__electricSourcePrefix!.releaseOne()) + await expect + .poll(async () => (await observe(b)).commits) + .toBeGreaterThan(bCommits) + await expect + .poll(() => rowIds(a)) + .toEqual(cell.present ? [`anchor`] : [`anchor`, `target`]) + reach.reach(`stale commit applied`) + const beforeAdvance = await observe(a) + if (cell.advance === `disjoint`) { + await pg.query( + `UPDATE public.${table} SET label = 'advanced' WHERE id = 'anchor'`, + ) + await expect + .poll( + async () => + (await observe(a)).rows.find(({ id }) => id === `anchor`)?.label, + { timeout: 30_000 }, + ) + .toBe(`advanced`) + } else { + // Ask the real server for a non-live catch-up response on A's next + // SDK request. The current live request may first time out. + await a.evaluate(() => window.__electricSourcePrefix!.refresh()) + await expect + .poll(async () => (await observe(a)).emptyCommits, { + timeout: 70_000, + }) + .toBeGreaterThan(beforeAdvance.emptyCommits) + } + reach.reach(`later advancement applied`) + if (cell.drain) { + await expect + .poll(async () => (await observe(b)).pending.length, { + timeout: 30_000, + }) + .toBeGreaterThan(0) + await b.evaluate(() => window.__electricSourcePrefix!.releaseOne()) + await expect + .poll(() => rowIds(b), { timeout: 30_000 }) + .toEqual(cell.present ? [`anchor`, `target`] : [`anchor`]) + } + await close(b) + if (cell.legacy) + await a.evaluate(() => window.__electricSourcePrefix!.legacy()) + const callerPosition: Record = {} + if (cell.callerCursor !== `none`) { + const { resume } = await observe(a) + if ( + !resume || + typeof resume !== `object` || + !(`offset` in resume) || + typeof resume.offset !== `string` || + !(`handle` in resume) || + typeof resume.handle !== `string` + ) + throw new Error(`No complete real Electric position at recovery`) + if ( + cell.callerCursor !== `initial-offset` && + cell.callerCursor !== `now` + ) + callerPosition.handle = resume.handle + if (cell.callerCursor === `resume`) + callerPosition.offset = resume.offset + if (cell.callerCursor === `initial-offset`) callerPosition.offset = `-1` + if (cell.callerCursor === `now`) callerPosition.offset = `now` + reach.reach(`explicit ${cell.callerCursor} supplied at recovery`) + } + if (cell.interrupted) { + // Preserve the real position above before recovery replaces its + // resume record with reset. Hold A too, so an empty peer response + // cannot overwrite that marker while this boundary is observed. + await a.evaluate(() => window.__electricSourcePrefix!.arm()) + const pendingRecovery = await open({}, true) + await expect + .poll(async () => (await observe(pendingRecovery)).pending.length, { + timeout: 30_000, + }) + .toBeGreaterThan(0) + await expect + .poll( + async () => { + const { resume } = await observe(pendingRecovery) + return resume && typeof resume === `object` && `kind` in resume + ? resume.kind + : undefined + }, + { timeout: 30_000 }, + ) + .toBe(`reset`) + reach.reach(`interrupted recovery reset persisted`) + await close(pendingRecovery) + } + const reopened = await open(callerPosition) + // Source readiness follows an applied SDK receipt. The wrapper may + // still propagate its notice to A, so poll the shared settled judgment. + const source = await pg.query<{ id: string; label: string }>( + `SELECT id, label FROM public.${table} ORDER BY id`, + ) + reach.reach(`reopened source compared`) + await expect + .poll( + async () => { + const states = await Promise.all([observe(a), observe(reopened)]) + return sourceHeadViolation(source.rows, { + durableRows: states[1].durableRows, + tabs: states, + }) + }, + { timeout: 10_000 }, + ) + .toBeUndefined() + console.info( + `real prefix reach`, + cell, + Object.fromEntries(reach.counts()), + ) + reach.assertReached() + } catch (error) { + primary = error + for (const page of pages) { + if (!page.isClosed()) { + try { + diagnostics.push(JSON.stringify(await observe(page))) + } catch (diagnosticError) { + diagnostics.push(String(diagnosticError)) + } + } + } + } finally { + for (const page of [...pages].reverse()) { + try { + await close(page) + } catch (error) { + diagnostics.push(String(error)) + } + } + if (connected) { + try { + await pg.query(`DROP TABLE IF EXISTS public.${table}`) + } catch (error) { + diagnostics.push(String(error)) + } + await pg.end() + } + } + if (primary !== undefined) + throw new Error( + `real prefix comparison failed; diagnostics=${JSON.stringify(diagnostics)}`, + { cause: primary }, + ) + expect(diagnostics).toEqual([]) + }) +} diff --git a/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix.opfs.html b/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix.opfs.html new file mode 100644 index 0000000000..93b5b9475b --- /dev/null +++ b/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix.opfs.html @@ -0,0 +1,10 @@ + + + + + Electric source prefix receiving driver + + + + + diff --git a/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix.opfs.ts b/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix.opfs.ts new file mode 100644 index 0000000000..1912479240 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/e2e/electric-source-prefix.opfs.ts @@ -0,0 +1,187 @@ +/** Real SDK/OPFS receiving driver for the shared source-head convergence law. + * HTTP responses are acquired from Electric before delivery is held. The gate + * never synthesizes messages or positions; it only delays a tab's next response. + */ +import { createCollection } from '@tanstack/db' +import { electricCollectionOptions } from '../../electric-db-collection/src/electric' +import { + BrowserCollectionCoordinator, + createBrowserWASQLitePersistence, + openBrowserWASQLiteOPFSDatabase, + persistedCollectionOptions, +} from '../src/index' +import type { ElectricCollectionConfig } from '../../electric-db-collection/src/electric' + +type Offset = NonNullable< + ElectricCollectionConfig[`shapeOptions`][`offset`] +> + +type Item = { id: string; label: string } +const parameters = new URL(location.href).searchParams +const databaseId = parameters.get(`databaseId`)! +const collectionId = parameters.get(`collectionId`)! +const table = parameters.get(`table`)! +if (!databaseId || !collectionId || !table) + throw new Error(`Missing prefix driver parameters`) +let hold = parameters.get(`holdInitial`) === `true` +let refresh = false +const pending: Array<{ release: () => void; rows: number }> = [] +const requests: Array<{ offset: string | null; handle: string | null }> = [] +let commits = 0 +let emptyCommits = 0 +let openWrites = 0 +const database = await openBrowserWASQLiteOPFSDatabase({ + databaseName: `${databaseId}.sqlite`, +}) +const coordinator = new BrowserCollectionCoordinator({ dbName: databaseId }) +const persistence = createBrowserWASQLitePersistence({ database, coordinator }) +const source = electricCollectionOptions({ + id: collectionId, + syncMode: `eager`, + getKey: (row) => row.id, + shapeOptions: { + offset: (parameters.get(`offset`) ?? undefined) as Offset | undefined, + handle: parameters.get(`handle`) ?? undefined, + url: `${location.origin}/electric/v1/shape`, + params: { table: `public.${table}` }, + liveSse: false, + fetchClient: async (input, init) => { + const url = new URL( + typeof input === `string` || input instanceof URL ? input : input.url, + ) + requests.push({ + offset: url.searchParams.get(`offset`), + handle: url.searchParams.get(`handle`), + }) + // A non-live catch-up request is answered by Electric with its actual + // up-to-date control, including when no row changed. + if (refresh) { + url.searchParams.delete(`live`) + refresh = false + } + const response = await fetch(url, init) + if (hold && response.ok) { + const messages = (await response.clone().json()) as Array<{ + headers?: { operation?: string } + }> + await new Promise((release) => + pending.push({ + release, + rows: messages.filter( + (message) => message.headers?.operation !== undefined, + ).length, + }), + ) + } + return response + }, + }, +}) +const sourceSync = source.sync +const options = persistedCollectionOptions({ + ...source, + sync: { + ...sourceSync, + sync: (params: Parameters[0]) => + sourceSync.sync({ + ...params, + begin: (...args) => { + openWrites = 0 + return params.begin(...args) + }, + write: (...args) => { + openWrites++ + return params.write(...args) + }, + commit: (...args) => { + const receipt = params.commit(...args) + const writes = openWrites + void Promise.resolve(receipt) + .then(() => { + commits++ + if (writes === 0) emptyCommits++ + }) + .catch(() => undefined) + return receipt + }, + }), + }, + persistence, +}) +const collection = createCollection(options) +const rows = (values: Iterable) => + Array.from(values, ({ id, label }) => ({ id, label })).sort((a, b) => + a.id.localeCompare(b.id), + ) +let cleaned = false +const probe = { + arm: () => { + hold = true + }, + refresh: () => { + refresh = true + }, + legacy: async () => { + // The old format's key ledger followed arrival-order writes even when + // its rows no longer matched the prefix named by the resume position. + await database.execute( + `DELETE FROM collection_metadata WHERE collection_id = ? AND key = ?`, + [collectionId, `electric:source-progress`], + ) + await database.execute( + `UPDATE collection_version SET source_progress_incompatible = 0 WHERE collection_id = ?`, + [collectionId], + ) + }, + releaseOne: () => { + const next = pending.shift() + if (!next) throw new Error(`No held response`) + next.release() + }, + observe: async () => { + const snapshot = await options.persistence.adapter.loadResumeSnapshot({ + kind: `eager`, + collectionId, + }) + return { + status: collection.status, + rows: rows(collection.values()), + durableRows: rows(snapshot.rows.map(({ value }) => value as Item)), + resume: snapshot.collectionMetadata.find( + ({ key }) => key === `electric:resume`, + )?.value, + commits, + emptyCommits, + pending: pending.map(({ rows: count }) => count), + requests: [...requests], + error: collection.health.error, + } + }, + cleanup: async () => { + if (cleaned) return [] + cleaned = true + hold = false + const cleanup = collection.cleanup() + for (const entry of pending.splice(0)) entry.release() + const failures: Array = [] + for (const action of [ + () => cleanup, + () => coordinator.dispose(), + () => database.close?.(), + ]) { + try { + await action() + } catch (error) { + failures.push(String(error)) + } + } + return failures + }, +} +declare global { + interface Window { + __electricSourcePrefix?: typeof probe + } +} +window.__electricSourcePrefix = probe +void collection.preload().catch(() => undefined) diff --git a/packages/browser-db-sqlite-persistence/playwright.opfs-electric.config.ts b/packages/browser-db-sqlite-persistence/playwright.opfs-electric.config.ts index 1b3a300d8d..d5ff75a8cb 100644 --- a/packages/browser-db-sqlite-persistence/playwright.opfs-electric.config.ts +++ b/packages/browser-db-sqlite-persistence/playwright.opfs-electric.config.ts @@ -4,6 +4,7 @@ import opfsConfig from './playwright.opfs.config' export default defineConfig({ ...opfsConfig, testMatch: [ + `electric-source-prefix-oracle.opfs.spec.ts`, `electric-resume-two-tab.opfs.spec.ts`, `electric-hydration-straddle.opfs.spec.ts`, `electric-immediate-reload.opfs.spec.ts`, diff --git a/packages/browser-db-sqlite-persistence/tests/electric-convergence-oracle.ts b/packages/browser-db-sqlite-persistence/tests/electric-convergence-oracle.ts new file mode 100644 index 0000000000..a0498e34b8 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/tests/electric-convergence-oracle.ts @@ -0,0 +1,58 @@ +/** + * Law 2 of the multi-tab Electric owner: after successful recovery and source + * catch-up, durable rows and every open eager Collection equal the source head. + * The authority is the persisted multi-tab eventual-consistency contract in + * oracle-coverage.md (#2209). Arrival-order regressions before this cut remain + * permitted. This judgment knows no cursor, coordinator, or storage algorithm. + * + * The Node driver supplies its independent server log; the Chromium/real SDK + * driver supplies PostgreSQL rows. Both use this same settled-row judgment. + * It does not judge intermediate publication, on-demand subsets, or optimism. + * + * This module owns the independent model and refinement check: the source head + * predicts both durable and public rows, ignoring row order but retaining every + * row occurrence and requiring each tab to be ready. The legal history grammar + * and Node production driver live in multi-tab-electric-oracle.test.ts. The + * receiving grammar and Chromium driver live in + * ../e2e/electric-source-prefix-oracle.opfs.spec.ts and its page module. + * Each driver calls this comparison after recovery and complete source delivery. + */ +import { declareOracle } from '../../db/tests/oracle-declaration' + +type SourceRow = { id: string | number; label: string } + +function canonicalRows(rows: ReadonlyArray): Array { + return rows + .map(({ id, label }) => ({ id, label })) + .sort((a, b) => String(a.id).localeCompare(String(b.id))) +} + +/** Compare durable rows and ready tabs with the independently supplied head. */ +export function sourceHeadViolation( + sourceHead: ReadonlyArray, + observation: { + durableRows: ReadonlyArray + tabs: ReadonlyArray<{ status: string; rows: ReadonlyArray }> + }, +): string | undefined { + // Source-head equality is independent of how storage orders or repairs + // interleaved commits. Readiness is supplied by each tab's own source. + const expected = canonicalRows(sourceHead) + if ( + JSON.stringify(canonicalRows(observation.durableRows)) !== + JSON.stringify(expected) + ) { + return `durable rows differ from source head: ${JSON.stringify({ expected, actual: observation.durableRows })}` + } + for (const [index, tab] of observation.tabs.entries()) { + if ( + tab.status !== `ready` || + JSON.stringify(canonicalRows(tab.rows)) !== JSON.stringify(expected) + ) { + return `tab ${index} differs from ready source head: ${JSON.stringify({ expected, actual: tab })}` + } + } + return undefined +} + +declareOracle({ check: [sourceHeadViolation] }) diff --git a/packages/browser-db-sqlite-persistence/tests/multi-tab-electric-oracle.test.ts b/packages/browser-db-sqlite-persistence/tests/multi-tab-electric-oracle.test.ts index fcdad376c0..f109c45c08 100644 --- a/packages/browser-db-sqlite-persistence/tests/multi-tab-electric-oracle.test.ts +++ b/packages/browser-db-sqlite-persistence/tests/multi-tab-electric-oracle.test.ts @@ -6,7 +6,7 @@ * runs its own Electric stream, and all of them store through one SQLite * database and one elected leader. Tabs coordinate only through durable * commits and the notices those commits broadcast. This file is the Browser - * host owner for three laws about that arrangement. + * host owner for five laws about that arrangement. * * Law 1, reserved commits survive a refetch storm. Authority: * SyncPersistenceCapabilityV1 `reserveCommitTurn` (packages/db/src/types.ts), @@ -103,6 +103,7 @@ import { declareOracle } from '../../db/tests/oracle-declaration' import { oraclePropertyOptions, oracleReplayRequested, + oracleReplayTest, oracleRuns, } from '../../db/tests/oracle-config' import { persistedCollectionOptions } from '../../db-sqlite-persistence-core/src' @@ -110,6 +111,7 @@ import { electricCollectionOptions } from '../../electric-db-collection/src/elec import { BrowserCollectionCoordinator } from '../src/browser-coordinator' import { createBrowserWASQLitePersistence } from '../src/browser-persistence' import { createWASQLiteTestDatabase } from './helpers/wa-sqlite-test-db' +import { sourceHeadViolation } from './electric-convergence-oracle' import type { Message, Row, @@ -117,6 +119,11 @@ import type { import type { ElectricCollectionUtils } from '../../electric-db-collection/src/electric' type Item = Row & { id: number; label: string } +type CallerCursor = `none` | `initial-offset` | `handle` | `resume` | `now` +type ExplicitPosition = { + handle?: string + offset?: `${number}_${number}` | `-1` | `now` +} // --------------------------------------------------------------------------- // Controlled Electric SDK boundary. Each stream records the position Electric @@ -214,6 +221,19 @@ const update = (row: Item): Message => ({ const nextTask = () => new Promise((resolve) => setTimeout(resolve, 0)) +// Attempt every cleanup, preserving failures after all resources are released. +async function closeAll(actions: Array<() => unknown>) { + const failures: Array = [] + for (const action of actions) { + try { + await action() + } catch (error) { + failures.push(error) + } + } + if (failures.length) throw new AggregateError(failures, `host cleanup failed`) +} + // --------------------------------------------------------------------------- // Browser host: one database file, one coordinator and persistence per tab. // --------------------------------------------------------------------------- @@ -252,7 +272,11 @@ function openHost() { } } - const openTab = (tab: string, syncMode: SyncMode) => { + const openTab = ( + tab: string, + syncMode: SyncMode, + callerPosition: ExplicitPosition = {}, + ) => { const database = createWASQLiteTestDatabase({ filename, timeout: 0 }) const execute = database.execute.bind(database) database.execute = async ( @@ -273,6 +297,7 @@ function openHost() { // The application's signal, which can stop the stream from outside. let external = new AbortController() const shapeOptions = { + ...callerPosition, url: `http://electric.test/v1/shape`, params: { table: `items` }, headers: { 'x-tab': tab }, @@ -303,9 +328,11 @@ function openHost() { const close = async () => { if (!open) return open = false - await collection.cleanup().catch(() => undefined) - coordinator.dispose() - await database.close?.() + await closeAll([ + () => collection.cleanup(), + () => coordinator.dispose(), + () => database.close?.(), + ]) } closers.push(close) const stream = () => { @@ -351,10 +378,11 @@ function openHost() { // The application's recovery: clean up and start the Collection again. // A signal that already aborted would stop the new stream too, so the // restarted session gets a fresh signal. - restart: async () => { + restart: async (nextPosition: ExplicitPosition = {}) => { await collection.cleanup() external = new AbortController() shapeOptions.signal = external.signal + Object.assign(shapeOptions, nextPosition) void collection.preload().catch(() => undefined) }, publicRows: () => @@ -386,6 +414,48 @@ function openHost() { openTab, quiesce, commitNotices: () => commitNotices, + hasResetMarker: async () => { + observer ??= createWASQLiteTestDatabase({ filename, timeout: 1000 }) + const records = await observer.execute<{ value: string }>( + `SELECT value FROM collection_metadata WHERE collection_id = ? AND key = ?`, + [`items`, `electric:resume`], + ) + const value: unknown = JSON.parse(records[0]?.value ?? `null`) + return ( + value !== null && + typeof value === `object` && + `kind` in value && + value.kind === `reset` + ) + }, + // A pre-protocol cache or mismatched provenance record still has rows + // and an Electric cursor. Reopen must not trust that cursor blindly. + rewriteSourceProgress: async (format: `missing` | `foreign`) => { + observer ??= createWASQLiteTestDatabase({ filename, timeout: 1000 }) + // The old format's key ledger followed arrival-order row writes and + // therefore certified these wrong rows. Reconstruct that baseline; + // merely deleting a token from an already incompatible cache would + // not distinguish the startup provenance check. + await observer.execute( + `UPDATE collection_version SET source_progress_incompatible = 0 WHERE collection_id = ?`, + [`items`], + ) + if (format === `missing`) { + await observer.execute( + `DELETE FROM collection_metadata WHERE collection_id = ? AND key = ?`, + [`items`, `electric:source-progress`], + ) + } else { + await observer.execute( + `UPDATE collection_metadata SET value = ? WHERE collection_id = ? AND key = ?`, + [ + JSON.stringify(`foreign-position`), + `items`, + `electric:source-progress`, + ], + ) + } + }, // A peer tab's collection reset, posted on the tabs' coordinator channel // as a peer would post it. No shipped host sends this notice yet, so // the premise is controlled; the tabs, their coordinators, and Electric @@ -426,10 +496,12 @@ function openHost() { .sort((left, right) => left.id - right.id) }, close: async () => { - await observer?.close?.() - for (const close of closers.reverse()) await close() - channel.close() - rmSync(directory, { recursive: true, force: true }) + await closeAll([ + () => observer?.close?.(), + ...closers.reverse(), + () => channel.close(), + () => rmSync(directory, { recursive: true, force: true }), + ]) }, } } @@ -505,11 +577,28 @@ const afterDeploy: Array = [ describe(`multi-tab Electric over one persisted database`, () => { let host: Host | undefined - afterEach(async () => { - await host?.close() - host = undefined - streams.length = 0 - }) + // Shrinking must retain the violated law even when releasing the host fails. + // AggregateError keeps the primary mismatch as cause and cleanup separately. + const finishHistory = async (primary?: { error: unknown }) => { + const failures: Array = [] + if (primary) failures.push(primary.error) + try { + await host?.close() + } catch (error) { + failures.push(error) + } finally { + host = undefined + streams.length = 0 + } + if (primary && failures.length > 1) + throw new AggregateError( + failures, + `${String(primary.error)}; secondary host cleanup failure`, + { cause: primary.error }, + ) + if (failures.length) throw failures[0] + } + afterEach(() => finishHistory()) // Law 1 grammar. Dimensions: two or three tabs; eager or on-demand // Collections; which tab finishes its refetch first; whether every tab's @@ -820,7 +909,12 @@ describe(`multi-tab Electric over one persisted database`, () => { // only the pinned #2209 replay does (see below). | { kind: `close`; tab: number } | { kind: `stop`; tab: number; pick: number } - | { kind: `recover`; tab: number; pick: number } + | { + kind: `recover` + tab: number + pick: number + callerCursor?: CallerCursor + } const historyStep: fc.Arbitrary = fc.oneof( { weight: 3, @@ -873,10 +967,46 @@ describe(`multi-tab Electric over one persisted database`, () => { }), }, ) - const generatedHistory = fc.record({ - tabCount: fc.integer({ min: 2, max: 3 }), - steps: fc.array(historyStep, { minLength: 8, maxLength: 24 }), - }) + // Construct stop/peer boundaries before the arbitrary interleavings. A + // partial delivery opens the source's reservation; its stop must abandon + // that reservation so the peer's next commit can apply (Laws 1 and 3). + // Every stop kind occurs in a generated order, with a distinct running + // peer. Recovery and catch-up separate the blocks. Four membership toggles + // of an initially absent key restore the original rows before the tail. + // This adds 24 steps to the unchanged 8–24-step arbitrary tail. Removing + // the open delivery, stop, or peer commit loses the law's premise; recovering + // before the peer commit would test a different boundary. Existing pinned + // histories still use the same driver without this prefix. + const generatedHistory = fc + .record({ + tabCount: fc.integer({ min: 2, max: 3 }), + source: fc.nat(2), + key: fc.constantFrom(3, 4), + recovery: fc.nat(1), + stopOrder: fc.shuffledSubarray([0, 1, 2, 3], { + minLength: 4, + maxLength: 4, + }), + steps: fc.array(historyStep, { minLength: 8, maxLength: 24 }), + }) + .map(({ tabCount, source, key, recovery, stopOrder, steps }) => { + const tab = source % tabCount + const peer = (tab + 1) % tabCount + return { + tabCount, + steps: [ + ...stopOrder.flatMap((pick): Array => [ + { kind: `server change`, key, pick: 0 }, + { kind: `advance`, tab, pick: 0 }, + { kind: `stop`, tab, pick }, + { kind: `catch up`, tab: peer, pick: 0 }, + { kind: `recover`, tab, pick: recovery }, + { kind: `catch up`, tab, pick: 0 }, + ]), + ...steps, + ], + } + }) type GeneratedHistory = { tabCount: number; steps: Array } type ServerLog = { @@ -935,6 +1065,11 @@ describe(`multi-tab Electric over one persisted database`, () => { startedAt: { handle: string | undefined; offset: string }, log: ServerLog, ): Pick => { + // Electric's legal `now` request begins at the current source head and + // supplies no baseline rows, even when no handle was specified. + if (startedAt.offset === `now`) { + return { handle: log.handle, version: log.changes.at(-1)?.version ?? 0 } + } if (startedAt.handle === undefined || startedAt.offset === `-1`) { return { handle: undefined, version: -1 } } @@ -948,6 +1083,14 @@ describe(`multi-tab Electric over one persisted database`, () => { const runGeneratedHistory = async ( { tabCount, steps }: GeneratedHistory, onKnownGap: () => void, + onRecovery?: ( + divergentBeforeRecovery: boolean, + how: `restart` | `reload`, + cursor: CallerCursor, + hadResetMarker: boolean, + ) => void, + progressFormat: `current` | `missing` | `foreign` = `current`, + callerCursor: CallerCursor = `none`, ) => { host = openHost() const activeHost = host @@ -1079,15 +1222,43 @@ describe(`multi-tab Electric over one persisted database`, () => { } await activeHost.quiesce(`tab ${tab.tab} delivered`) } - const recover = async (index: number, how: `restart` | `reload`) => { + const recover = async ( + index: number, + how: `restart` | `reload`, + cursor = callerCursor, + ) => { + if (progressFormat !== `current`) { + await activeHost.rewriteSourceProgress(progressFormat) + progressFormat = `current` + } + const divergentBeforeRecovery = + JSON.stringify(rowsOf(durable)) !== JSON.stringify(rowsOf(log.rows)) + const hadResetMarker = await activeHost.hasResetMarker() const stopped = tabs[index]! const streamsBefore = stopped.streamCount() + // A caller can repeat the current source position, but that position + // supplies no missing rows. Initial-offset and handle-only requests + // are legal full-snapshot neighbors of the explicit resumed request. + const head = position(log, serverVersion) + const callerPosition: ExplicitPosition = + cursor === `resume` + ? { + handle: head.handle, + offset: head.offset as `${number}_${number}`, + } + : cursor === `handle` + ? { handle: head.handle } + : cursor === `initial-offset` + ? { offset: `-1` } + : cursor === `now` + ? { offset: `now` } + : {} let recovered = stopped - if (how === `restart`) await stopped.restart() + if (how === `restart`) await stopped.restart(callerPosition) else { await stopped.close() await activeHost.quiesce(`tab ${stopped.tab} closed`) - recovered = activeHost.openTab(stopped.tab, `eager`) + recovered = activeHost.openTab(stopped.tab, `eager`, callerPosition) tabs[index] = recovered } abandon(models[index]!) @@ -1105,6 +1276,7 @@ describe(`multi-tab Electric over one persisted database`, () => { ...requested, }) await activeHost.quiesce(`tab ${recovered.tab} recovered`) + onRecovery?.(divergentBeforeRecovery, how, cursor, hadResetMarker) } const check = async (label: string, head?: Array) => { @@ -1254,7 +1426,11 @@ describe(`multi-tab Electric over one persisted database`, () => { (step.kind === `recover` && step.pick === 1) ? `reload` : `restart` - await recover(index, how) + await recover( + index, + how, + step.kind === `recover` ? step.callerCursor : undefined, + ) } else if (step.kind === `stop`) { const how = stops[step.pick % stops.length]! if (model.open !== undefined) { @@ -1324,10 +1500,192 @@ describe(`multi-tab Electric over one persisted database`, () => { await recover(index, `restart`) } await deliver(index, `all`, true) + // A reset-abandoned transaction fails only when delivery tries to + // commit it. Judge Law 5 at that cut before the restart establishes + // Law 2's successful catch-up premise. + await check(`after final delivery to tab ${tabs[index]!.tab}`) + if (models[index]!.failedByReset) { + await recover(index, `restart`) + await deliver(index, `all`, true) + await check(`after reset recovery of tab ${tabs[index]!.tab}`) + generatedReach.reach( + `reset-abandoned source recovered during final catch-up`, + ) + } } await check(`after every tab caught up`, rowsOf(log.rows)) + if (models.every((model) => !model.closed)) { + expect( + sourceHeadViolation(rowsOf(log.rows), { + durableRows: await activeHost.durableRows(), + tabs: tabs.map((tab) => ({ + status: tab.collection.status, + rows: tab.publicRows(), + })), + }), + ).toBeUndefined() + } } + // Construct the missing interaction instead of waiting for an unconstrained + // shuffle to draw it. One stream commits an even number of membership + // changes, its peer commits only the first, then the faster stream commits + // either no rows or a disjoint update. Reopening the lagging peer must still + // reach the source head. Swapping tabs, both initial memberships, repeated + // toggles, and draining before recovery distinguish nearby legal histories. + // Recovery 0 stops the source and restarts the same Collection; recovery 1 + // reloads the tab, closing its Collection and opening a new one. The reach + // callback records the driver's actual recovery branch, not the input pick. + // Removing the lagging commit or the later advancement loses the defect; + // draining is the neighboring non-divergent control. A prefix longer than + // the source log is excluded by construction. + // Explicit offset/handle options select a provider position, not a cache + // certificate. The source-head law is unchanged across caller cursor forms. + const sourcePrefixHistory = fc.record({ + key: fc.constantFrom(1, 3), + pairs: fc.integer({ min: 1, max: 3 }), + fastTab: fc.integer({ min: 0, max: 1 }), + advancingBatch: fc.constantFrom(`empty` as const, `disjoint` as const), + drain: fc.boolean(), + recovery: fc.integer({ min: 0, max: 1 }), + progressFormat: fc.constantFrom( + `current` as const, + `missing` as const, + `foreign` as const, + ), + callerCursor: fc.constantFrom( + `none`, + `initial-offset`, + `handle`, + `resume`, + `now`, + ), + }) + const runSourcePrefixCampaign = async ( + parameters: ReturnType, + ) => { + const states = [ + `present/empty`, + `present/disjoint`, + `absent/empty`, + `absent/disjoint`, + `drained before recovery`, + `missing progress at recovery`, + `foreign progress at recovery`, + `divergent source restart`, + `divergent source reload`, + `explicit resume at divergent recovery`, + `explicit now at divergent recovery`, + `explicit initial offset at recovery`, + `explicit handle at recovery`, + `interrupted reset before explicit resume recovery`, + `interrupted reset before explicit now recovery`, + ] as const + const reach = createOracleReach(`browser.electric-source-prefix`, states) + await fc.assert( + fc.asyncProperty(sourcePrefixHistory, async (history) => { + const slowTab = 1 - history.fastTab + // Keep every original history, then repeat it with a recovery that + // stops before its first source delivery. The first start writes its + // reset marker through production; the second must still recover. + // Pairing both paths preserves the uninterrupted domain and gives + // every generated cursor form an interrupted neighbor. + for (const interrupted of [false, true]) { + reach.beginHistory() + const steps: Array = [ + ...Array.from({ length: history.pairs * 2 }, (): HistoryStep => ({ + kind: `server change`, + key: history.key, + pick: 0, + })), + { kind: `catch up`, tab: history.fastTab, pick: 0 }, + { kind: `catch up`, tab: slowTab, pick: 1 }, + ...(history.advancingBatch === `disjoint` + ? [{ kind: `server change` as const, key: 2, pick: 1 }] + : []), + { kind: `catch up`, tab: history.fastTab, pick: 0 }, + ...(history.drain + ? [{ kind: `catch up` as const, tab: slowTab, pick: 0 }] + : []), + ...(history.recovery === 0 + ? [{ kind: `stop` as const, tab: slowTab, pick: 0 }] + : []), + ...(interrupted + ? [ + { + kind: `recover` as const, + tab: slowTab, + pick: history.recovery, + callerCursor: `none` as const, + }, + { kind: `stop` as const, tab: slowTab, pick: 0 }, + ] + : []), + { kind: `recover`, tab: slowTab, pick: history.recovery }, + ] + let primary: { error: unknown } | undefined + try { + await runGeneratedHistory( + { tabCount: 2, steps }, + () => fc.pre(false), + (divergent, how, cursor, hadResetMarker) => { + if (history.drain && !divergent) + reach.reach(`drained before recovery`) + if (!history.drain && divergent) { + if (cursor === `resume`) + reach.reach(`explicit resume at divergent recovery`) + if (cursor === `now`) + reach.reach(`explicit now at divergent recovery`) + if ( + interrupted && + hadResetMarker && + (cursor === `resume` || cursor === `now`) + ) + reach.reach( + `interrupted reset before explicit ${cursor} recovery`, + ) + reach.reach(`divergent source ${how}`) + reach.reach( + `${history.key === 1 ? `present` : `absent`}/${history.advancingBatch}`, + ) + } + if (history.progressFormat !== `current`) { + reach.reach(`${history.progressFormat} progress at recovery`) + } + if (cursor === `initial-offset`) + reach.reach(`explicit initial offset at recovery`) + if (cursor === `handle`) + reach.reach(`explicit handle at recovery`) + }, + history.progressFormat, + history.callerCursor, + ) + } catch (error) { + primary = { error } + } finally { + await finishHistory(primary) + } + } + }), + parameters, + ) + console.info(`source prefix reach`, Object.fromEntries(reach.counts())) + reach.assertReached() + } + it.skipIf(oracleReplayRequested())( + `converges after interleaved source prefixes (fixed seed)`, + () => runSourcePrefixCampaign({ numRuns: oracleRuns(80), seed: 2281 }), + 600_000, + ) + oracleReplayTest(it, `browser.electric-source-prefix`)( + `converges after interleaved source prefixes (random or replayed)`, + () => + runSourcePrefixCampaign( + oraclePropertyOptions(80, `browser.electric-source-prefix`), + ), + 600_000, + ) + type GeneratedReach = | `must-refetch delivered` | `peer commit during an open reserved transaction` @@ -1340,12 +1698,13 @@ describe(`multi-tab Electric over one persisted database`, () => { | `peer commit after signal aborted mid-transaction` | `peer reset abandoned an open transaction` | `reset-abandoned commit put the Collection in error` + | `reset-abandoned source recovered during final catch-up` // The storm driver and the generated-history driver are the checks; the // generated driver folds committed batches into its own model of durable // rows. The storm cells and the generated-history arbitrary are the grammar. declareOracle({ check: [runStormHistory, runGeneratedHistory], - grammar: [stormHistories, generatedHistory], + grammar: [stormHistories, generatedHistory, sourcePrefixHistory], }) const createGeneratedReach = () => @@ -1361,6 +1720,7 @@ describe(`multi-tab Electric over one persisted database`, () => { `peer commit after signal aborted mid-transaction`, `peer reset abandoned an open transaction`, `reset-abandoned commit put the Collection in error`, + `reset-abandoned source recovered during final catch-up`, ]) let generatedReach = createGeneratedReach() @@ -1370,13 +1730,14 @@ describe(`multi-tab Electric over one persisted database`, () => { generatedReach = createGeneratedReach() await fc.assert( fc.asyncProperty(generatedHistory, async (history) => { + let primary: { error: unknown } | undefined try { // fc.pre discards a history that reaches the known gap. await runGeneratedHistory(history, () => fc.pre(false)) + } catch (error) { + primary = { error } } finally { - await host?.close() - host = undefined - streams.length = 0 + await finishHistory(primary) } }), parameters, @@ -1396,7 +1757,7 @@ describe(`multi-tab Electric over one persisted database`, () => { }), 600_000, ) - it( + oracleReplayTest(it, GENERATED_PROPERTY)( `keeps reserved commits lawful across generated multi-tab histories (random or replayed)`, () => runGeneratedCampaign( @@ -1430,6 +1791,88 @@ describe(`multi-tab Electric over one persisted database`, () => { 60_000, ) + // The final delivery can be the first attempt to commit a transaction + // abandoned by the reset. The driver must observe its failure, recover, + // and only then judge convergence. This replays the original issue trace. + it( + `recovers a reset-abandoned source that fails during final catch-up`, + () => + runGeneratedHistory( + { + tabCount: 2, + steps: [ + { kind: `server change`, key: 1, pick: 0 }, + { kind: `advance`, tab: 1, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `advance`, tab: 0, pick: 0 }, + { kind: `peer reset` }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `deploy` }, + { kind: `deploy` }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + ], + }, + () => undefined, + ), + 60_000, + ) + + // The later no-deploy reset trace reaches the same final-delivery failure + // from a reloaded tab. Recovery must happen after that failed commit too. + // Seed -1371495859, path 75:0:0:0:0:0:0:1:1:0:0. + it( + `recovers a reloaded source whose final commit was abandoned by a peer reset`, + () => + runGeneratedHistory( + { + tabCount: 2, + steps: [ + { kind: `recover`, tab: 0, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `advance`, tab: 0, pick: 0 }, + { kind: `advance`, tab: 1, pick: 0 }, + { kind: `advance`, tab: 0, pick: 0 }, + { kind: `peer reset` }, + { kind: `advance`, tab: 0, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + ], + }, + () => undefined, + ), + 60_000, + ) + + // A discontinuous source prefix does not corrupt the cached key set. Keep + // durable rows visible during reload while a fresh source snapshot repairs + // continuation. This is a reduced random history, replayed by the same law. + it( + `keeps cached rows visible while recovering a discontinuous source`, + () => + runGeneratedHistory( + { + tabCount: 2, + steps: [ + { kind: `server change`, key: 1, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `catch up`, tab: 1, pick: 1 }, + { kind: `catch up`, tab: 0, pick: 0 }, + { kind: `recover`, tab: 0, pick: 0 }, + { kind: `recover`, tab: 0, pick: 0 }, + { kind: `stop`, tab: 0, pick: 0 }, + { kind: `peer reset` }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `catch up`, tab: 0, pick: 0 }, + { kind: `server change`, key: 1, pick: 0 }, + { kind: `advance`, tab: 0, pick: 0 }, + ], + }, + () => undefined, + ), + 60_000, + ) + // Known gap #2209, pinned: tab A commits v1 and v2; tab B's lagging stream // then commits only v1, so the durable row falls back to v1, and B closes // for good. Nothing redelivers the head to A, so Law 2's final check fails. @@ -1456,14 +1899,14 @@ describe(`multi-tab Electric over one persisted database`, () => { 60_000, ) - // Known gap, pinned from the random campaign (seed -1660880085, path + // Regression from the random campaign (seed -1660880085, path // 55:0:1:4:4:4:1:1:1:1:3:4:4:4:4): row 1 reaches v2 on the server and both // tabs catch up, but after tab B stops, tab A catches up, and tab A stops, // the durable rows lose row 1. Law 2 expects every open tab and the durable - // rows to show the head. This reproduces on main and is tracked with #2281. + // rows to show the head. The source-progress repair forces a fresh snapshot. // It replays through the same driver and checks with the known-gap detector - // off. Remove `fails` when the durable row survives the stops. - it.fails( + // off. The generated source-prefix campaign extends this same law. + it( `keeps a durable row after both tabs catch up to it and then stop`, () => runGeneratedHistory( @@ -1485,4 +1928,71 @@ describe(`multi-tab Electric over one persisted database`, () => { ), 60_000, ) + + // Issue #2281 reduces to the same six-step history for both membership + // directions. A stores two server changes; B stores only the first. A's + // next up-to-date stores its newer cursor without restoring the second + // change. B then reopens from that cursor and never receives the missing + // change. Law 2 still requires the server head at the final checkpoint. + // These regressions replay the same model and comparison as the generated + // source-prefix campaign above. + for (const { key, behavior } of [ + { key: 1, behavior: `keeps a reinserted row` }, + { key: 3, behavior: `keeps a deleted row absent` }, + ]) { + const stalePrefix: Array = [ + { kind: `server change`, key, pick: 0 }, + { kind: `server change`, key, pick: 0 }, + { kind: `catch up`, tab: 0, pick: 0 }, + { kind: `catch up`, tab: 1, pick: 1 }, + ] + const publishCursor: HistoryStep = { kind: `catch up`, tab: 0, pick: 0 } + const reopenPeer: HistoryStep = { kind: `recover`, tab: 1, pick: 0 } + + // Keeping only the maximum cursor does not repair these rows. + it( + `${behavior} after a cursor-only commit and a lagging tab reload`, + () => + runGeneratedHistory( + { + tabCount: 2, + steps: [...stalePrefix, publishCursor, reopenPeer], + }, + () => undefined, + ), + 60_000, + ) + + // A peer that drains its backlog supplies the missing change itself. + it( + `${behavior} when the lagging stream drains before reloading`, + () => + runGeneratedHistory( + { + tabCount: 2, + steps: [ + ...stalePrefix, + publishCursor, + { kind: `catch up`, tab: 1, pick: 0 }, + reopenPeer, + ], + }, + () => undefined, + ), + 60_000, + ) + + // Without A's cursor-only commit, B resumes from its older position and + // receives the missing change. This distinguishes cursor advancement + // from reload or lag alone as the cause of the permanent mismatch. + it( + `${behavior} when the lagging tab reloads before the cursor advances`, + () => + runGeneratedHistory( + { tabCount: 2, steps: [...stalePrefix, reopenPeer] }, + () => undefined, + ), + 60_000, + ) + } }) diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 666fbf6a1a..666513c54a 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -479,6 +479,7 @@ export type PersistedTx< /** Claim authorizing this transaction's persisted cache generation. */ cacheGenerationClaimId?: string truncate?: boolean + sourceProgress?: { key: string; previous: string | undefined } mutations: Array< | { type: `insert` @@ -516,6 +517,8 @@ export const SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY = Symbol.for( ) export interface PersistenceAdapter { + /** Implements PersistedTx.sourceProgress under the durable writer lock. */ + readonly supportsSourceProgress?: boolean loadSubset: ( target: PersistedStorageTarget, options: LoadSubsetOptions, @@ -544,6 +547,7 @@ export interface PersistenceAdapter { metadata?: unknown }> keySet?: PersistedKeySetEvidence + sourceProgressCompatible?: boolean collectionMetadata: Array<{ key: string; value: unknown }> latestTerm: number latestSeq: number @@ -1304,6 +1308,7 @@ type BufferedSyncTransaction = { lifecycleGeneration: number admittedStorageCollectionId?: string expectedResumeGenerationOwner?: symbol + sourceProgress?: PersistedTx[`sourceProgress`] signal?: AbortSignal prependHydrationRows: ( rows: Array<{ key: TKey; value: T; metadata?: unknown }>, @@ -1668,6 +1673,7 @@ class PersistedCollectionRuntime< private resumeBaselinePromise: Promise | null = null private resumeCertificationPromise: Promise | null = null private persistedKeySetEvidence: PersistedKeySetEvidence | undefined + private sourceProgressCompatible: boolean | undefined private persistedResumeGeneration: PersistedResumeGeneration | undefined private resumeGenerationOwner = Symbol(`persisted resume generation owner`) private lifecycleGeneration = 0 @@ -2631,6 +2637,14 @@ class PersistedCollectionRuntime< return this.persistedKeySetEvidence } + supportsSourceProgress(): boolean { + return this.persistence.adapter.supportsSourceProgress === true + } + + getSourceProgressCompatible(): boolean | undefined { + return this.sourceProgressCompatible + } + hasManagedCacheGeneration(): boolean { return this.cacheGenerationClaim !== undefined } @@ -2913,6 +2927,7 @@ class PersistedCollectionRuntime< if (storageCollectionId !== this.storageCollectionId) return false this.persistedResumeGeneration = this.getResumeSnapshotGeneration(snapshot) this.persistedKeySetEvidence = snapshot.keySet + this.sourceProgressCompatible = snapshot.sourceProgressCompatible this.observeStreamPosition( snapshot.latestTerm, snapshot.latestSeq, @@ -3620,6 +3635,7 @@ class PersistedCollectionRuntime< this.terminalFailure = undefined this.resumeCertificationPromise = null this.persistedKeySetEvidence = undefined + this.sourceProgressCompatible = undefined this.persistedResumeGeneration = undefined this.resumeGenerationOwner = Symbol(`persisted resume generation owner`) } @@ -4705,6 +4721,7 @@ class PersistedCollectionRuntime< rowVersion: streamPosition.rowVersion, cacheGenerationClaimId: this.cacheGenerationClaim?.claimId, truncate: transaction.truncate, + sourceProgress: transaction.sourceProgress, mutations, rowMetadataMutations: toMetadataMutations(transaction.rowMetadataWrites), collectionMetadataMutations: toMetadataMutations( @@ -6729,6 +6746,24 @@ function createWrappedSyncConfig< openTransaction.expectedResumeGenerationOwner = runtime.getResumeGenerationOwner() }, + ...(runtime.supportsSourceProgress() + ? { + expectSourceProgress: ( + progress: NonNullable, + ) => { + if (startupState.cleanedUp) return + const openTransaction = getOpenTransaction() + if (!openTransaction) { + throw new InvalidPersistedCollectionConfigError( + `resumeSnapshot.expectSourceProgress must be called within an open sync transaction`, + ) + } + openTransaction.sourceProgress = progress + }, + getSourceProgressCompatible: () => + runtime.getSourceProgressCompatible(), + } + : {}), }, } @@ -7083,6 +7118,7 @@ function createWrappedSyncConfig< lifecycleGeneration: openTransaction.lifecycleGeneration, expectedResumeGenerationOwner: openTransaction.expectedResumeGenerationOwner, + sourceProgress: openTransaction.sourceProgress, signal, prependHydrationRows: ( rows: Array<{ key: TKey; value: T; metadata?: unknown }>, @@ -7263,6 +7299,22 @@ function createWrappedSyncConfig< if (active()) metadata.persistence.resumeSnapshot.expectCurrentCommit() }, + ...(metadata.persistence.resumeSnapshot.expectSourceProgress + ? { + expectSourceProgress: ( + progress: NonNullable, + ) => { + if (active()) + metadata.persistence.resumeSnapshot.expectSourceProgress?.( + progress, + ) + }, + getSourceProgressCompatible: () => + active() + ? metadata.persistence.resumeSnapshot.getSourceProgressCompatible?.() + : undefined, + } + : {}), }, }, }, diff --git a/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts b/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts index 61ef52fcdb..9a624ddd1b 100644 --- a/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts +++ b/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts @@ -1653,6 +1653,7 @@ function buildIndexName(collectionId: string, signature: string): string { } export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { + readonly supportsSourceProgress = true private readonly driver: SQLiteDriver private readonly schedulingIdentitySource: SQLiteDriver private scheduler: SharedPersistenceScheduler | undefined @@ -2346,6 +2347,7 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { metadata?: unknown }> keySet: PersistedKeySetEvidence + sourceProgressCompatible: boolean collectionMetadata: Array<{ key: string; value: unknown }> latestTerm: number latestSeq: number @@ -2393,10 +2395,8 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { const rows = includeRows ? await this.loadSubsetInternal(tableMapping, {}, transactionDriver) : [] - const { latestRowVersion, keySet } = await this.readKeySetEvidence( - collectionId, - transactionDriver, - ) + const { latestRowVersion, keySet, sourceProgressCompatible } = + await this.readResumeEvidence(collectionId, transactionDriver) const collectionMetadataRows = await transactionDriver.query<{ key: string value: string @@ -2425,6 +2425,7 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { metadata: row.metadata, })), keySet, + sourceProgressCompatible, collectionMetadata: collectionMetadataRows.map((row) => ({ key: row.key, value: deserializePersistedRowValue(row.value), @@ -2599,6 +2600,33 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { const currentRowVersion = version.latest_row_version const nextRowVersion = Math.max(currentRowVersion + 1, tx.rowVersion) const replacesPersistedBaseline = tx.truncate === true + const collectionMetadataMutations = lastMetadataByKey( + tx.collectionMetadataMutations ?? [], + (mutation) => mutation.key, + ) + // A source may commit from a different prefix than the shared baseline. + // Keep accepting its rows, but never certify the resulting baseline for a + // resume past changes that this composition can have lost. Only a full + // replacement can establish complete evidence again. + let sourceProgressDiscontinuous = false + if (tx.sourceProgress && !replacesPersistedBaseline) { + const progress = await transactionDriver.query<{ value: string }>( + `SELECT value FROM collection_metadata WHERE collection_id = ? AND key = ?`, + [collectionId, tx.sourceProgress.key], + ) + const next = collectionMetadataMutations.find( + (mutation) => mutation.key === tx.sourceProgress!.key, + ) + sourceProgressDiscontinuous = + progress[0]?.value !== + (tx.sourceProgress.previous === undefined + ? undefined + : serializePersistedRowValue(tx.sourceProgress.previous)) && + // Delivering a complete source position twice is idempotent. It + // neither breaks a coherent baseline nor repairs a previous gap. + (next?.type !== `set` || + progress[0]?.value !== serializePersistedRowValue(next.value)) + } const tracksPersistedKeySet = version.key_set_evidence_available === 1 || replacesPersistedBaseline const replayDelta: ReplayableTxDelta | null = replacesPersistedBaseline @@ -2816,10 +2844,7 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { } for (const batch of batches( - lastMetadataByKey( - tx.collectionMetadataMutations ?? [], - (mutation) => mutation.key, - ), + collectionMetadataMutations, this.replacementBatchSize, )) { const deletes = batch.filter((mutation) => mutation.type === `delete`) @@ -2857,12 +2882,19 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { key_set_evidence_incompatible = CASE WHEN ? = 1 THEN 0 ELSE key_set_evidence_incompatible + END, + source_progress_incompatible = CASE + WHEN ? = 1 THEN 0 + WHEN ? = 1 THEN 1 + ELSE source_progress_incompatible END WHERE collection_id = ?`, [ nextRowVersion, replacesPersistedBaseline ? 1 : 0, replacesPersistedBaseline ? 1 : 0, + replacesPersistedBaseline ? 1 : 0, + sourceProgressDiscontinuous ? 1 : 0, collectionId, ], ) @@ -3383,22 +3415,25 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { } } - private async readKeySetEvidence( + private async readResumeEvidence( collectionId: string, driver: SQLiteDriver, ): Promise<{ latestRowVersion: number keySet: PersistedKeySetEvidence + sourceProgressCompatible: boolean }> { const versionRows = await driver.query<{ latest_row_version: number key_set_evidence_available: number key_set_incompatible: number + source_progress_incompatible: number }>( `SELECT latest_row_version, key_set_evidence_available, - key_set_evidence_incompatible AS key_set_incompatible + key_set_evidence_incompatible AS key_set_incompatible, + source_progress_incompatible FROM collection_version WHERE collection_id = ? LIMIT 1`, @@ -3408,6 +3443,7 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { return { latestRowVersion: version?.latest_row_version ?? 0, + sourceProgressCompatible: version?.source_progress_incompatible === 0, keySet: { status: version?.key_set_evidence_available !== 1 @@ -4179,7 +4215,8 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { ON CONFLICT(collection_id) DO UPDATE SET latest_row_version = 0, key_set_evidence_available = 1, - key_set_evidence_incompatible = 0`, + key_set_evidence_incompatible = 0, + source_progress_incompatible = 0`, [collectionId], ) await transactionDriver.run( @@ -4230,7 +4267,8 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { AND EXISTS (SELECT 1 FROM sqlite_master WHERE name = 'collection_version' AND sql LIKE '%key_set_evidence_available%' - AND sql LIKE '%key_set_evidence_incompatible%') + AND sql LIKE '%key_set_evidence_incompatible%' + AND sql LIKE '%source_progress_incompatible%') AS complete`, ) return rows[0]?.complete === 1 @@ -4343,17 +4381,19 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { collection_id TEXT PRIMARY KEY, latest_row_version INTEGER NOT NULL, key_set_evidence_available INTEGER NOT NULL DEFAULT 0, - key_set_evidence_incompatible INTEGER NOT NULL DEFAULT 0 + key_set_evidence_incompatible INTEGER NOT NULL DEFAULT 0, + source_progress_incompatible INTEGER NOT NULL DEFAULT 0 )`, ) const collectionVersionColumns = await this.driver.query<{ name: string }>( `PRAGMA table_info(collection_version)`, ) - const keyEvidenceColumns = [ + const versionEvidenceColumns = [ `key_set_evidence_available`, `key_set_evidence_incompatible`, + `source_progress_incompatible`, ] as const - for (const columnName of keyEvidenceColumns) { + for (const columnName of versionEvidenceColumns) { if (collectionVersionColumns.some(({ name }) => name === columnName)) { continue } diff --git a/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts b/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts index 56884b1d2d..61eb42f87c 100644 --- a/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts +++ b/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts @@ -2,8 +2,14 @@ import { DatabaseSync } from 'node:sqlite' import { fc, test as fcTest } from '@fast-check/vitest' import { describe, expect, it } from 'vitest' import { createCollection } from '@tanstack/db' -import { oraclePropertyOptions, oracleRuns } from '../../db/tests/oracle-config' +import { + oraclePropertyOptions, + oracleReplayRequested, + oracleReplayTest, + oracleRuns, +} from '../../db/tests/oracle-config' import { declareOracle } from '../../db/tests/oracle-declaration' +import { createOracleReach } from '../../db/tests/oracle-reach' import { SQLiteCorePersistenceAdapter, createPersistedTableName, @@ -799,14 +805,183 @@ describe(`SQLite resume snapshots`, () => { ) }) + // A resume token may continue only the baseline that established its + // previous token. This is the storage obligation of the optional source + // progress protocol: a discontinuity invalidates resume evidence until an + // authoritative replacement. It does not reject committed rows or order + // opaque tokens. The independent model judges the edges since the last + // replacement; ordinary metadata commits cannot erase a missing edge. + type ProgressStep = { + kind: `advance` | `stale` | `empty` | `snapshot` | `duplicate` + pick: number + } + const progressSteps = fc.array( + fc.record({ + kind: fc.constantFrom( + `advance`, + `stale`, + `empty`, + `snapshot`, + `duplicate`, + ), + pick: fc.nat(3), + }), + { minLength: 3, maxLength: 16 }, + ) + const progressHistory = fc.record({ + legacySchema: fc.boolean(), + steps: progressSteps, + }) + const runProgressCampaign = async ( + parameters: ReturnType, + ) => { + const reach = createOracleReach(`sqlite-resume.source-progress`, [ + `discontinuous commit`, + `matching commit after discontinuity`, + `replacement after discontinuity`, + `duplicate source position`, + `legacy source progress column added`, + ]) + await fc.assert( + fc.asyncProperty( + progressHistory, + async ({ legacySchema, steps: history }) => { + reach.beginHistory() + const database = new DatabaseSync(`:memory:`) + let primary: unknown + try { + let adapter = new SQLiteCorePersistenceAdapter({ + driver: createDriver(database), + }) + const target = { + kind: `eager` as const, + collectionId: `source-progress`, + } + if (legacySchema) { + await adapter.loadResumeSnapshot(target) + database.exec( + `ALTER TABLE collection_version DROP COLUMN source_progress_incompatible`, + ) + adapter = new SQLiteCorePersistenceAdapter({ + driver: createDriver(database), + }) + await adapter.loadResumeSnapshot(target) + reach.reach(`legacy source progress column added`) + } + const tokens = [`initial`] + let current = `initial` + // Model-only path: each edge must begin where its predecessor ended. + let edges: Array<{ from: string; to: string }> = [] + const coherent = () => + edges.every( + (edge, index) => + index === 0 || edge.from === edges[index - 1]!.to, + ) + const steps: Array = [ + { kind: `snapshot`, pick: 0 }, + ...history, + ] + for (const [index, step] of steps.entries()) { + const wasCoherent = coherent() + const previous = + step.kind === `stale` || step.kind === `duplicate` + ? tokens[step.pick % tokens.length]! + : current + const next = + step.kind === `empty` || step.kind === `duplicate` + ? current + : `position-${index}` + if (step.kind === `duplicate`) + reach.reach(`duplicate source position`) + if (step.kind === `snapshot`) { + if (!wasCoherent) reach.reach(`replacement after discontinuity`) + edges = [{ from: previous, to: next }] + } else { + if (previous !== current && next !== current) + reach.reach(`discontinuous commit`) + if (!wasCoherent && previous === current) + reach.reach(`matching commit after discontinuity`) + // Re-delivering the same source position is idempotent; it + // contributes no new edge and cannot repair an earlier gap. + if (next !== current) edges.push({ from: previous, to: next }) + } + // The baseline adapter ignores this proposed optional field. Keeping + // it on a structurally assignable transaction lets the pre-edit RED + // reach the semantic comparison instead of failing type/setup. + const tx = { + txId: `progress-${index}`, + term: 1, + seq: index + 1, + rowVersion: index + 1, + truncate: step.kind === `snapshot`, + sourceProgress: { key: `source-progress`, previous }, + mutations: + step.kind === `empty` + ? [] + : [ + { + type: `insert` as const, + key: `row`, + value: { id: `row`, label: next }, + }, + ], + collectionMetadataMutations: [ + { type: `set` as const, key: `source-progress`, value: next }, + ], + } + await adapter.applyCommittedTx(target, tx) + const observation = await adapter.loadResumeSnapshot(target) + expect( + observation.sourceProgressCompatible, + `source progress at step ${index}: ${JSON.stringify(steps.slice(0, index + 1))}`, + ).toBe(coherent()) + // Physical key integrity and source continuity are separate laws. + // A broken source path must not quarantine an intact cached set. + expect(observation.keySet).toEqual({ status: `consistent` }) + expect( + observation.collectionMetadata.find( + ({ key }) => key === `source-progress`, + )?.value, + ).toBe(next) + current = next + tokens.push(next) + } + } catch (error) { + primary = error + } finally { + closeDatabasePreservingPrimary(database, primary) + } + }, + ), + parameters, + ) + console.info(`source progress reach`, Object.fromEntries(reach.counts())) + reach.assertReached() + } + it.skipIf(oracleReplayRequested())( + `preserves source progress evidence (fixed seed)`, + () => runProgressCampaign({ numRuns: oracleRuns(80), seed: 2281 }), + ) + oracleReplayTest(it, `sqlite-resume.source-progress`)( + `preserves source progress evidence (random or replayed)`, + () => + runProgressCampaign( + oraclePropertyOptions(80, `sqlite-resume.source-progress`), + ), + ) + // The retired-storage and managed-claim checks compare production with independent // models of retained generations and claim intervals. The fixed and generated // cases share those checks. The startup grammar is the generator whose coverage // the fixed startup corpus calibrates. declareOracle({ - check: [assertRetiredStorageHistory, assertManagedClaimReadHistory], + check: [ + assertRetiredStorageHistory, + assertManagedClaimReadHistory, + runProgressCampaign, + ], model: [expectedRetainedGenerations], - grammar: [startupHistoryArbitrary], + grammar: [startupHistoryArbitrary, progressHistory], }) fcTest.prop([startupHistoryArbitrary], { diff --git a/packages/db/src/sync-persistence.ts b/packages/db/src/sync-persistence.ts index 299f47d847..211a19b34a 100644 --- a/packages/db/src/sync-persistence.ts +++ b/packages/db/src/sync-persistence.ts @@ -94,6 +94,18 @@ export function validateSyncPersistenceCapability< `expectCurrentCommit`, `resumeSnapshot.expectCurrentCommit`, ) + if (resumeSnapshot.expectSourceProgress !== undefined) { + requireFunction( + resumeSnapshot, + `expectSourceProgress`, + `resumeSnapshot.expectSourceProgress`, + ) + requireFunction( + resumeSnapshot, + `getSourceProgressCompatible`, + `resumeSnapshot.getSourceProgressCompatible`, + ) + } return value as SyncPersistenceCapabilityV1 } diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index ce37781c99..8d17b2ae8e 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -639,6 +639,21 @@ export type SyncPersistenceCapabilityV1< readonly certify: () => Promise readonly getKeySetEvidence: () => SyncPersistenceKeySetEvidence | undefined readonly expectCurrentCommit: () => void + /** + * Atomically compare the durable source token with this transaction's + * predecessor. The source stages its next opaque string token in collection + * metadata at `key`. A discontinuity invalidates source progress evidence + * until an authoritative truncate; rows and receipts still apply normally. + * Re-delivering the same complete source position must be idempotent; it + * preserves existing evidence, including an earlier incompatibility. + * Only adapters that implement the durable comparison expose this method. + */ + readonly expectSourceProgress?: (progress: { + key: string + previous: string | undefined + }) => void + /** Source progress evidence from the atomic persisted resume snapshot. */ + readonly getSourceProgressCompatible?: () => boolean | undefined } } diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index ca9c1a8927..15e4cb262a 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -160,12 +160,14 @@ const staticOracleProperties = [ `electric.sdk-rotation-release`, `electric-recovery.publication-stream-convergence`, `sqlite-resume.startup-generation`, + `sqlite-resume.source-progress`, `sqlite-resume.managed-claim-read`, `sqlite-resume.retired-storage`, `persisted-demand-coverage.history`, `persisted-demand-coverage.required`, `sqlite-reload.class`, `browser.multi-tab-reserved-history`, + `browser.electric-source-prefix`, `sqlite-reload.class.electron`, `persistence.generation-notice`, `persistence.follower-remote-demand`, diff --git a/packages/db/tests/oracle-replay-manifest.ts b/packages/db/tests/oracle-replay-manifest.ts index 5de3ae76b1..c4e9d54970 100644 --- a/packages/db/tests/oracle-replay-manifest.ts +++ b/packages/db/tests/oracle-replay-manifest.ts @@ -230,7 +230,7 @@ const ownerGroups: ReadonlyArray< [ `db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts`, `sqlite-resume`, - `startup-generation`, + `startup-generation source-progress`, ], [ `db/tests/query/derived-delete-reconciliation-oracle.test.ts`, @@ -410,7 +410,7 @@ const ownerGroups: ReadonlyArray< [ `browser-db-sqlite-persistence/tests/multi-tab-electric-oracle.test.ts`, `browser`, - `multi-tab-reserved-history`, + `multi-tab-reserved-history electric-source-prefix`, ], [ `electron-db-sqlite-persistence/tests/reload-class-oracle.test.ts`, @@ -724,6 +724,14 @@ export type OracleReplayManifestEntry = { // Pagination cells carry their property name in the test title so the guard // can select one cell without running the rest of the matrix. const replayTestNames = new Map([ + [ + `sqlite-resume.source-progress`, + `preserves source progress evidence.*random or replayed`, + ], + [ + `browser.electric-source-prefix`, + `converges after interleaved source prefixes.*random or replayed`, + ], [ `source-health.required-history`, `required histories refine the model.*random`, diff --git a/packages/db/tests/sync-persistence.test.ts b/packages/db/tests/sync-persistence.test.ts index 5e67c43842..87d56a9b48 100644 --- a/packages/db/tests/sync-persistence.test.ts +++ b/packages/db/tests/sync-persistence.test.ts @@ -97,4 +97,27 @@ describe(`sync persistence capability`, () => { }), ).toThrow(/startupCacheGenerationRotated must be a boolean/) }) + + it(`accepts source progress only when the optional capability is callable`, () => { + const capability = { + ...baseCapability, + resumeSnapshot: { + certify: () => Promise.resolve(), + getKeySetEvidence: () => ({ status: `consistent` as const }), + expectCurrentCommit: () => {}, + expectSourceProgress: () => {}, + getSourceProgressCompatible: () => true, + }, + } + expect(validateSyncPersistenceCapability(capability)).toBe(capability) + expect(() => + validateSyncPersistenceCapability({ + ...capability, + resumeSnapshot: { + ...capability.resumeSnapshot, + expectSourceProgress: true, + }, + }), + ).toThrow(/resumeSnapshot.expectSourceProgress must be a function/) + }) }) diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index 14ecda8745..6fb538f0f4 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -1683,6 +1683,8 @@ function createElectricSync>( const certifyResumeSnapshot = resumeSnapshot?.certify const getKeySetEvidence = resumeSnapshot?.getKeySetEvidence const expectCurrentCommit = resumeSnapshot?.expectCurrentCommit + const expectSourceProgress = + syncMode === `eager` ? resumeSnapshot?.expectSourceProgress : undefined const persistedKeySetEvidence = getKeySetEvidence?.() const storedResumeRecord = freshScopedGeneration @@ -1699,15 +1701,31 @@ function createElectricSync>( url: shapeOptions.url, params: shapeOptions.params as Record | undefined, }) + // Handles are opaque. Equality of the complete source position is enough + // to detect a broken continuation; wall-clock timestamps are not progress. + const progressToken = (state: ElectricResumeState | undefined) => + state?.kind === `resume` + ? JSON.stringify([state.shapeId, state.handle, state.offset]) + : undefined + const progressKey = `electric:source-progress` + let sourceProgress = progressToken(persistedResumeState) const hasIncompatiblePersistedResume = persistedResumeState?.kind === `resume` && persistedResumeState.shapeId !== shapeIdentity + const hasIncompatibleSourceProgress = + persistedResumeState?.kind === `resume` && + expectSourceProgress !== undefined && + (metadata?.collection.get(progressKey) !== sourceProgress || + resumeSnapshot?.getSourceProgressCompatible?.() !== true) + const hasPersistedReset = + syncMode === `eager` && persistedResumeState?.kind === `reset` // A pre-ledger `unknown` baseline cannot justify a non-initial cursor. // One fresh replacement establishes consistent evidence for later resumes. - const lacksCompletePersistedKeySet = + const lacksCompletePersistedResume = persistedResumeState?.kind === `resume` && - getKeySetEvidence !== undefined && - persistedKeySetEvidence?.status !== `consistent` + ((getKeySetEvidence !== undefined && + persistedKeySetEvidence?.status !== `consistent`) || + hasIncompatibleSourceProgress) // Claim expiry can leave a partial managed cache without a resume record. // Its incompatible key set alone requires a fresh source snapshot. const hasIncompatibleManagedCache = @@ -1716,13 +1734,15 @@ function createElectricSync>( const requiresFreshSourceEvidence = !freshScopedGeneration && (hasIncompatibleManagedCache || + hasIncompatibleSourceProgress || + hasPersistedReset || (shapeOptions.offset === undefined && shapeOptions.handle === undefined && (persistedResumeState !== undefined || malformedPersistedResume) && (malformedPersistedResume || hasIncompatiblePersistedResume || persistedResumeState?.kind === `reset` || - lacksCompletePersistedKeySet || + lacksCompletePersistedResume || (!retainsTagState && persistedResumeState?.kind === `resume` && persistedResumeState.requiresTagState !== false)))) @@ -1731,6 +1751,14 @@ function createElectricSync>( requiresFreshSourceEvidence && persistence?.managedCacheGeneration === true && persistence.startScopedRecovery !== undefined + // A provider position cannot certify incompatible source progress or + // finish an interrupted eager replacement recorded by a reset marker. + // Eager recovery overrides the caller's position and replaces its rows. + // Other sync modes keep their existing explicit-position behavior. + const requestedPosition = + hasIncompatibleSourceProgress || hasPersistedReset + ? {} + : { offset: shapeOptions.offset, handle: shapeOptions.handle } if (scopedRecovery) lifecycle.forgetEvidenceAfterScopedRecovery() const usesChangesOnlyLog = syncMode === `on-demand` && @@ -1762,7 +1790,8 @@ function createElectricSync>( // Only a complete adapter ledger can justify a persisted cursor. !requiresFreshSourceEvidence const hasExplicitResumeOffset = - shapeOptions.offset !== undefined && shapeOptions.offset !== `-1` + requestedPosition.offset !== undefined && + requestedPosition.offset !== `-1` if (!canUsePersistedResume && !hasExplicitResumeOffset) { clearTagTrackingState() } @@ -1886,14 +1915,14 @@ function createElectricSync>( // In on-demand mode, we only need the changes from the point of time the collection was created // so we default to `now` when there is no saved offset. offset: - shapeOptions.offset ?? + requestedPosition.offset ?? (canUsePersistedResume ? (persistedResumeState.offset as Offset) : usesChangesOnlyLog ? `now` : undefined), handle: - shapeOptions.handle ?? + requestedPosition.handle ?? (canUsePersistedResume ? persistedResumeState.handle : undefined), signal: abortController.signal, onError: async (errorParams) => { @@ -2048,6 +2077,11 @@ function createElectricSync>( updatedAt: Date.now(), requiresTagState: tagState.hasTags(), } + if (expectSourceProgress) { + expectSourceProgress({ key: progressKey, previous: sourceProgress }) + sourceProgress = progressToken(resumeState) + metadata?.collection.set(progressKey, sourceProgress) + } lifecycle.resumeState = resumeState metadata?.collection.set(`electric:resume`, resumeState) } diff --git a/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts b/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts index 674ea6d7a0..70a7407e63 100644 --- a/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts +++ b/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts @@ -27,6 +27,7 @@ import { withFixtureCleanup, } from './electric-recovery-support' import type { Message } from '@electric-sql/client' +import type { PersistedTx } from '../../db-sqlite-persistence-core/src' import type { ElectricCollectionUtils } from '../src/electric' import type { Item, @@ -186,7 +187,7 @@ async function observePersistedRestart( scenario, modelCanonicalRows, ) - const resumeState = + const resumeState: NonNullable = scenario.resumeKind === `none` ? [] : [ @@ -210,6 +211,20 @@ async function observePersistedRestart( driver, schemaVersion: 1, }) + // A compatible warm baseline includes the source's complete-position token. + // Legacy/missing tokens are a separate full-snapshot recovery regime in the + // Browser source-prefix owner, not a compatible-reopen premise here. + if (scenario.resumeKind === `resume`) { + resumeState.push({ + type: `set`, + key: `electric:source-progress`, + value: JSON.stringify([ + `{"params":{"table":"test_table"},"url":"http://test-url"}`, + `shape-old`, + `10_0`, + ]), + }) + } await originalAdapter.applyCommittedTx( { kind: `eager`, collectionId: collectionId }, { diff --git a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts index 925990c2cb..f306d6b511 100644 --- a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts +++ b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts @@ -198,6 +198,19 @@ async function runRace( { type: `insert`, key: 2, value: { id: 2, name: `two` } }, ], collectionMetadataMutations: [ + // Seed the current eager format so this driver still exercises + // certification of a resumed baseline, not legacy-token migration. + { + type: `set`, + key: `electric:source-progress`, + value: JSON.stringify([ + startupReset === `shape-identity` + ? `{"params":{"table":"other_table"},"url":"http://test-url"}` + : `{"params":{"table":"test_table"},"url":"http://test-url"}`, + `shape-old`, + `10_0`, + ]), + }, { type: `set`, key: `electric:resume`, diff --git a/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts b/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts index 38bc6924cc..ee4077ed8d 100644 --- a/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts +++ b/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts @@ -491,6 +491,7 @@ describe(`electron sqlite persistence bridge`, () => { }, ], keySet: { status: `consistent` }, + sourceProgressCompatible: true, collectionMetadata: [ { key: `resume:test`, diff --git a/scripts/doc-snippets-baseline.json b/scripts/doc-snippets-baseline.json index 3ecffabb8d..4786a4be77 100644 --- a/scripts/doc-snippets-baseline.json +++ b/scripts/doc-snippets-baseline.json @@ -58,7 +58,6 @@ "a066dfd28d064b54" ], "docs/framework/angular/overview.md": ["55d749380b445d1a"], - "docs/framework/preact/overview.md": ["a3802a251c868f43"], "docs/framework/react/overview.md": ["1c1f2ec0a48394e3", "d89b6c8f8cab169d"], "docs/guides/collection-options-creator.md": [ "305eb0ca15d244e8",