diff --git a/.changeset/persisted-follower-applies-before-ready.md b/.changeset/persisted-follower-applies-before-ready.md new file mode 100644 index 0000000000..fa1edf15e9 --- /dev/null +++ b/.changeset/persisted-follower-applies-before-ready.md @@ -0,0 +1,9 @@ +--- +'@tanstack/db-sqlite-persistence-core': patch +--- + +In a follower tab, an on-demand load that the leader fetches now settles only after the follower applies the leader's committed rows. Before, the leader's reply could settle the load while the follower was still applying the broadcast commit, so a live query in the follower tab could first be ready without those rows. A follower load now rejects, and releases its acquisition, when the leader's broadcast commit fails to apply in the follower, instead of settling without rows. + +A follower now replaces its rows in place when it applies the leader's commits, instead of replaying its demand to a leader that commits again on every load. It also requests its demand again after the leader resets the collection, instead of showing an empty ready result. If its cache claim renewal fails transiently while a load waits for the leader's rows, the load rejects and the demand is retried, as after a failed request to the leader. + +Any persisted collection, not only a follower, now keeps commit notices that arrive while its cache claim is expired and applies them in arrival order once the claim is renewed or rotated; before, they were dropped. A commit that fails to apply from the queue now makes the collection fail, as a commit applied when it arrives already did. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index f2f2c7f873..4d96d490d9 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -435,7 +435,7 @@ generic visitor behavior outside the named collectors. | Experimental no-peek continuation | [test-only live-window probe](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination-oracle.no-peek.integration.test.ts) | The candidate uses an authoritative fact only for the same complete output prefix and otherwise requests the extra row. It remains test-only; the shipped opaque-pagination owner keeps peek-ahead. See [experiment limits](https://github.com/TanStack/db/blob/main/packages/query-db-collection/tests/cursor-pagination/NO-PEEK-RESULTS.md). | | Electric and TrailBase | [Electric histories](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-oracle.property.test.ts), [recovery histories](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts), [held resume snapshots](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts), [PostgreSQL semantics](https://github.com/TanStack/db/blob/main/packages/electric-db-collection/e2e/sql-predicate-semantics-oracle.e2e.test.ts), [TrailBase contract](https://github.com/TanStack/db/blob/main/packages/trailbase-db-collection/tests/ORACLE.md) | Installed SDK delivery/framing, independent predicates, exact subscription arguments, restart/reset lineage, held certification and durability races, source-order publication before durability, and late errors. The queued-presence property runs identical fixed/random generators plus isolated seed-and-path replay across insert, update, delete, and truncate callbacks. The recovery fixtures use a mocked ShapeStream; they do not establish live Electric-service framing or native persistence-host behavior. | | PowerSync | [tests](https://github.com/TanStack/db/tree/main/packages/powersync-db-collection/tests), `tests/correctness-oracle.test.ts` | Applied receipt positions crossed with held peers, native SQLite/SDK and cleanup evidence. A metadata-bearing `Collection.update` passes through an observed SDK watcher callback to the public and SQLite rows and an exact queued CRUD patch. Held callback ordering and synthetic zero-field cases remain controlled-only, and remote upload has no receiving fixture here. Run the focused owner with the package's `test:oracles` command. A timeout mutant proves a progress failure, not every value assertion. | -| SQLite persistence and native hosts | [persisted histories](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts), [reset/resume histories](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/sqlite-core-adapter-oracle.test.ts), [dual-adapter resume snapshots](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts), [Browser composed-owner histories](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts), [Browser coordinator RPC](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/browser-coordinator-oracle.test.ts), [shared-driver fairness](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.test.ts), [driver contracts](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/contracts/sqlite-driver-contract.ts), [Node shared-handle scheduling](https://github.com/TanStack/db/blob/main/packages/node-db-sqlite-persistence/tests/node-driver.test.ts), [OP-SQLite shared-handle scheduling](https://github.com/TanStack/db/blob/main/packages/react-native-db-sqlite-persistence/tests/op-sqlite-driver-oracle.test.ts), [browser OPFS lifecycle](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-page-lifecycle-oracle.test.ts), [worker diagnostics](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-worker-diagnostics-oracle.test.ts), [Electron IPC and composed owner](https://github.com/TanStack/db/blob/main/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts), [113-law manifest](https://github.com/TanStack/db/blob/main/packages/db-collection-e2e/src/fixtures/persisted-conformance-manifest.ts) | Core cache/remote rejection/peer/reopen histories, atomic reset/resume lineage, key-set evidence, dual-adapter races, queued coordinator reload/gap notifications during startup, unscheduled startup/reset overlap, and exact driver results. The shared core-adapter contract checks custom local comparator dispatch for ascending, descending, and comparator-equal public-key ordering at `loadSubset` completion. Host packages register this contract, but the local sqlite3 CLI run does not prove native host execution. Browser composes public source commits with per-collection elected-owner routing and covers the complete committed-transaction wire partition through deterministic Node transport seams. Remote-subset histories distinguish logical demand, physical acquisitions, exact acquisition leases, and released replay tombstones. Electron composes source commits with a per-collection renderer owner, IPC adapter, real SQLite, and reopen checks. The shared-driver fairness owner checks K=1 complete-logical-hydrate scheduling with identical fixed, random, and seed-plus-path campaigns. It records public rows, raw dequeue reach, and a persist-first FIFO hostile control. Its Chromium OPFS fixture refines the provider boundary but does not establish a browser matrix, elapsed-time latency, unbounded eventuality, or multi-process coordination. Same-handle Node and OP-SQLite tests cover transaction admission. Controlled OPFS page/worker histories cover ownership and diagnostic-cause retention. The reset/resume owners use sqlite3 CLI and in-memory node:sqlite seams; they do not prove multi-process WAL, mobile/Tauri, or other native-device execution. Distinct database handles rely on SQLite lock admission rather than one in-process queue. React Native hosts without async-context propagation must use the transaction driver supplied to the callback for nested work. Fake workers and synthetic page events do not prove native handle release or real bfcache admission. The Browser composed seams are not real multi-context/OPFS-worker execution; the Electron harness is not an actual Electron process unless its explicit runtime-bridge mode runs. An ownerless elected node suppresses core routing, while a follower may route demand to the elected owner; host coordinators retry only classified transport or admission failures while demand remains retained. The manifest excludes progressive and move suites; registration and shim runs are not device execution. | +| SQLite persistence and native hosts | [persisted histories](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts), [reset/resume histories](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/sqlite-core-adapter-oracle.test.ts), [dual-adapter resume snapshots](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/sqlite-resume-snapshot-oracle.test.ts), [Browser composed-owner histories](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts), [Browser coordinator RPC](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/browser-coordinator-oracle.test.ts), [shared-driver fairness](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.test.ts), [driver contracts](https://github.com/TanStack/db/blob/main/packages/db-sqlite-persistence-core/tests/contracts/sqlite-driver-contract.ts), [Node shared-handle scheduling](https://github.com/TanStack/db/blob/main/packages/node-db-sqlite-persistence/tests/node-driver.test.ts), [OP-SQLite shared-handle scheduling](https://github.com/TanStack/db/blob/main/packages/react-native-db-sqlite-persistence/tests/op-sqlite-driver-oracle.test.ts), [browser OPFS lifecycle](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-page-lifecycle-oracle.test.ts), [worker diagnostics](https://github.com/TanStack/db/blob/main/packages/browser-db-sqlite-persistence/tests/opfs-worker-diagnostics-oracle.test.ts), [Electron IPC and composed owner](https://github.com/TanStack/db/blob/main/packages/electron-db-sqlite-persistence/tests/electron-ipc.test.ts), [113-law manifest](https://github.com/TanStack/db/blob/main/packages/db-collection-e2e/src/fixtures/persisted-conformance-manifest.ts) | Core cache/remote rejection/peer/reopen histories, atomic reset/resume lineage, key-set evidence, dual-adapter races, queued coordinator reload/gap notifications during startup, unscheduled startup/reset overlap, and exact driver results. The shared core-adapter contract checks custom local comparator dispatch for ascending, descending, and comparator-equal public-key ordering at `loadSubset` completion. Host packages register this contract, but the local sqlite3 CLI run does not prove native host execution. Browser composes public source commits with per-collection elected-owner routing and covers the complete committed-transaction wire partition through deterministic Node transport seams. Remote-subset histories distinguish logical demand, physical acquisitions, exact acquisition leases, and released replay tombstones. Electron composes source commits with a per-collection renderer owner, IPC adapter, real SQLite, and reopen checks. The shared-driver fairness owner checks K=1 complete-logical-hydrate scheduling with identical fixed, random, and seed-plus-path campaigns. It records public rows, raw dequeue reach, and a persist-first FIFO hostile control. Its Chromium OPFS fixture refines the provider boundary but does not establish a browser matrix, elapsed-time latency, unbounded eventuality, or multi-process coordination. Same-handle Node and OP-SQLite tests cover transaction admission. Controlled OPFS page/worker histories cover ownership and diagnostic-cause retention. The reset/resume owners use sqlite3 CLI and in-memory node:sqlite seams; they do not prove multi-process WAL, mobile/Tauri, or other native-device execution. Distinct database handles rely on SQLite lock admission rather than one in-process queue. React Native hosts without async-context propagation must use the transaction driver supplied to the callback for nested work. Fake workers and synthetic page events do not prove native handle release or real bfcache admission. The Browser composed seams are not real multi-context/OPFS-worker execution; the Electron harness is not an actual Electron process unless its explicit runtime-bridge mode runs. An ownerless elected node suppresses core routing, while a follower may route demand to the elected owner. The follower generation-notice and rotated-reacquisition lanes state the rotation law: a rotated follower holds the new generation's commit notices until the owner has reacquired every active demand, then ends scoped recovery and applies them. Production does not yet end that recovery, so their cells that require it are pinned as failing under the known gap [#2139](https://github.com/TanStack/db/issues/2139); the random generation-notice variant is skipped, because some of its histories pass today. Cells that still pass cover failed reacquisitions (scoped recovery stays and the failure is reported), a superseding rotation as leader, and a scoped recovery without rotation, with or without an active demand (untrusted durable rows stay hidden). Other cells cover a follower whose claim expires while its ensure is in flight and an owner reset (the follower requests its demand again). The follower remote-demand lane checks that a follower's load, and a live query over its demand, settle only after the owner's broadcast applies here, for a broadcast that applies at once, one queued behind another subset's hydration, and one while the cache claim is expired; the claim-rotation placement is pinned under #2139. A failed apply rejects the load, errors the query, and fails the source. A renewal that fails transiently at the reply rejects the load but keeps and retries the demand, without releasing the owner's acquisition; a rotation during the reply wait is tracked under #2139. The flush-side expiry check and arrival-order queueing have no killing witness yet. Removing the follower's wait on either ensure path, its terminal check, or the queued flush's terminal marking each fails it. The Electron IPC readiness oracle checks a fresh or reopened renderer, a successful or failed IPC hydration, and an upstream load that succeeds, fails, or is released during hydration, against a model of the three readiness checkpoints; a renderer that replies before the main-process read or treats an IPC failure as empty fails it. The owner is a controlled coordinator double, not two real coordinators; host coordinators retry only classified transport or admission failures while demand remains retained. The manifest excludes progressive and move suites; registration and shim runs are not device execution. The per-collection coordinator oracle's real-tab lane runs a follower and an owner tab with real Browser coordinators over Node's BroadcastChannel and Web Locks and one real SQLite file: 16 histories over the owner's source (custom or Query Collection), the broadcast's placement, the follower's apply outcome, and the consumer, plus a rotation cell. Main fails all 16 histories; an owner tab for a rotated follower generation is not reachable there. The Electron IPC readiness oracle adds a follower and an owner renderer with Electron coordinators: 4 histories over the apply outcome and the consumer, in process and through a real Electron main process; main fails all 4. | | SQLite expression-index planning | [Node expression-index oracle](https://github.com/TanStack/db/blob/main/packages/node-db-sqlite-persistence/tests/expression-index-oracle.test.ts) | RFC #1659 invariant 8 owns identical persisted-index and runtime-expression shapes. Independent expected keys are checked against direct captured SQL, adapter results, and named-index plans. Generated BigInts use SQLite's signed range; one fixed case checks legacy oversized-value reads. Other limits: bounded unqualified JSON paths/scalars, Node BetterSQLite, and no null, arbitrary raw SQL, or native-host planning. Run the package's `test:oracles` campaign. | | Offline execution | [scheduler](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/KeyScheduler-oracle.property.test.ts), [leadership](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/leadership-replay-oracle.property.test.ts), [settlement](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/transaction-settlement-oracle.property.test.ts), [serialization](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/transaction-serializer-oracle.property.test.ts), [web connectivity replay](https://github.com/TanStack/db/blob/main/packages/offline-transactions/tests/connectivity-replay-oracle.test.ts), [review record](oracle-reviews/pr-1879-visible-replay.md) | Declarative FIFO eligibility, per-transaction outcomes, durable state and typed wire trees. The web connectivity owner checks a persisted write under a false browser online hint, then visible replay by event or explicit retry, at provider-call, outbox, and caller-settlement cuts. It uses controlled browser globals and fake storage; it does not prove actual browser network detection, genuine offline retry timing, multi-tab leadership transfer, or React Native behavior. Issued work may finish after ownership loss, but new work must not start. Exactly-once network execution is not promised. | | Frameworks | [React conformance](https://github.com/TanStack/db/blob/main/packages/react-db/tests/conformance.test.tsx), [React pagination](https://github.com/TanStack/db/blob/main/packages/react-db/tests/infinite-query-conformance.test.tsx), [shared framework contracts](https://github.com/TanStack/db/tree/main/packages/db/tests/conformance), [Vue synchronous publication](https://github.com/TanStack/db/blob/main/packages/vue-db/tests/useLiveQuery-publication-oracle.test.ts) | Exact exposed rows/pages and each framework's own lifecycle cuts. Vue's synchronous watcher checks insert, update, and nonterminal delete through supplied Collection and identity query inputs; it does not cover an empty final result, multiple changes in one transaction, or query recompilation. A React witness does not prove Vue/Solid/Angular/Svelte scheduling. Preserve their receiving registrations. The shared suite also checks initial-query readiness over a lagging on-demand source, for a list and a `findOne` query and for the infinite query (React, Vue, and Svelte): not ready before the subset settles, ready at settlement while the source is still `loading`, and still ready after the source becomes ready. Solid's Suspense over that source has its own witness. | diff --git a/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.spec.ts b/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.spec.ts index 8515aa8d85..779df5fba1 100644 --- a/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.spec.ts +++ b/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.spec.ts @@ -23,6 +23,48 @@ async function observe(page: Page): Promise { return page.evaluate(() => window.__remoteSubsetProbe!.observe()) } +/** + * A follower's load settles only after the leader's rows show in the follower + * (ARCHITECTURE.md law 7). Here the follower requests an ID that the leader has + * not loaded, so the follower's rows can arrive only through the leader's + * committed broadcast. The check reads the rows at the follower live query's + * first-ready transition. + */ +test(`a follower is first ready with rows the leader fetched for it`, async ({ + context, +}) => { + const databaseId = `rs-${crypto.randomUUID()}` + const pages: Array = [] + try { + const leader = await context.newPage() + pages.push(leader) + await leader.goto( + `/e2e/remote-subset-two-tab.opfs.html?databaseId=${databaseId}&itemId=leader-item`, + ) + const first = await observe(leader) + if (first.phase === `failed`) throw new Error(first.failure) + expect(first.isLeader).toBe(true) + + const follower = await context.newPage() + pages.push(follower) + await follower.goto( + `/e2e/remote-subset-two-tab.opfs.html?databaseId=${databaseId}&itemId=follower-item`, + ) + const second = await observe(follower) + if (second.phase === `failed`) throw new Error(second.failure) + expect(second.isLeader).toBe(false) + expect(second.upstreamLoads).toBe(0) + expect(second.rowsAtFirstReady).toEqual([ + { id: `follower-item`, label: `Item follower-item` }, + ]) + } finally { + for (const page of pages.reverse()) { + await page.evaluate(() => window.__remoteSubsetProbe?.cleanup()) + await page.close() + } + } +}) + test(`two real OPFS tabs transport a filtered subset without a clone retry`, async ({ context, }) => { diff --git a/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.ts b/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.ts index 5705e05748..c419b8a3ce 100644 --- a/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.ts +++ b/packages/browser-db-sqlite-persistence/e2e/remote-subset-two-tab.opfs.ts @@ -32,6 +32,8 @@ export type RemoteSubsetObservation = { isLeader: boolean storageCollectionId?: string rows: Array + // The rows visible at the live query's first-ready transition. + rowsAtFirstReady?: Array upstreamLoads: number ensureRequests: Array<{ collectionId: string @@ -68,6 +70,7 @@ if (!databaseId || !itemId) throw new Error(`Missing oracle parameters`) let phase: RemoteSubsetObservation[`phase`] = `starting` let failure: string | undefined let upstreamLoads = 0 +let rowsAtFirstReady: Array | undefined let remoteSubsetPosts = 0 let storageCollectionId: string | undefined const remoteSubsetPostFailures: Array = [] @@ -87,6 +90,7 @@ window.__remoteSubsetProbe = { phase, ...(failure ? { failure } : {}), ...readRuntime(), + ...(rowsAtFirstReady ? { rowsAtFirstReady } : {}), upstreamLoads, ensureRequests: [...ensureRequests], remoteSubsetPosts, @@ -219,6 +223,12 @@ try { .select(({ item }) => ({ id: item.id, label: item.label })), ) cleanupTasks.unshift(() => live.cleanup()) + live.onFirstReady(() => { + rowsAtFirstReady = [...live.values()].map(({ id, label }) => ({ + id, + label, + })) + }) readRuntime = () => ({ isLeader: storageCollectionId ? coordinator.isLeader(storageCollectionId) diff --git a/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts b/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts index 2e37a3f8a4..d94319f941 100644 --- a/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts +++ b/packages/browser-db-sqlite-persistence/tests/per-collection-coordinator-oracle.test.ts @@ -3,7 +3,15 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { setImmediate as nextTurn } from 'node:timers/promises' import { fc } from '@fast-check/vitest' -import { IR, createCollection, whenSyncAccepted } from '@tanstack/db' +import { + IR, + createCollection, + createLiveQueryCollection, + eq, + whenSyncAccepted, +} from '@tanstack/db' +import { QueryClient } from '@tanstack/query-core' +import { queryCollectionOptions } from '@tanstack/query-db-collection' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { BrowserCollectionCoordinator } from '../src/browser-coordinator' import { @@ -4984,6 +4992,607 @@ describe(`sync-ingested write ownership oracle`, () => { }) }) +/* +Follower remote-demand settlement across real tabs +================================================== + +Law and source: the applied-settlement law (law 7 in +packages/db/src/query/live/ARCHITECTURE.md) for a follower tab. A follower +routes on-demand demand it cannot serve to the elected owner. Its load, and so +its live query's first ready, settles only when the rows the owner fetched for +it are visible in the follower; if applying the owner's commit fails here, the +load rejects and the follower's source enters error. The single-runtime lanes +`persistence.follower-remote-demand` and `persistence.follower-generation-notice` +in db-sqlite-persistence-core/tests/persisted-oracle.test.ts own the runtime +mechanism with a coordinator double. This lane is their receiving witness +across the real boundary: two tabs, each with its own real +BrowserCollectionCoordinator over Node's real BroadcastChannel and Web Locks, +and one real SQLite file opened by both tabs. + +Independent model: it knows only whether the owner's commit applies in the +follower. While that apply is outstanding, the consumer has not settled. If it +applies, the consumer settles successfully with the owner's row. If it fails, +the consumer rejects without the row and the follower source is in error. + +Grammar: the owner's source is a receipt-awaiting custom source or a real +Query Collection source, crossed with the placement of the owner's broadcast +in the follower (direct, or queued while another subset's hydration holds the +apply mutex), crossed with whether the follower's durable read of that commit +succeeds, crossed with the consumer (a live query, or the follower's load). +The driver holds or fails the first follower read whose rows contain the +owner's row, and for the queued placement it holds the other subset's +hydration read. It does not reorder the channel, which stays FIFO as in a +browser. The custom source delivers each row once; a source that re-inserts +an existing row on every ensure made the follower release and request its +demand again without end, on main as well, and is outside this lane. + +Checkpoint and observations: once the owner's reply has reached the follower +and the apply is still outstanding, whether the consumer has settled. After +release: its outcome, the demanded rows it saw when it settled (first-ready +rows, or rows at load settlement), and the follower source's status. A live +query cannot show a load that succeeded after a failed apply, because the +failure errors the source; the load consumer observes it directly. + +Calibration: main's persisted.ts (load settles at the reply) fails all 16 +histories at the checkpoint. Removing the follower's wait fails all 16. +Removing the terminal check after the wait fails the 4 load histories whose +apply fails. The Query Collection histories pass, so query-db's +`whenSyncAccepted` wait resolves after the owner's broadcast under real +persistence. + +Rotation: when only the follower's claim expires, it rotates into a +generation no other tab joins and becomes that generation's owner. The +rotation cell pins that its rows stay visible and new demand settles from its +own source. Another tab never owns the rotated generation in this topology, +so the elected-owner notice path has no receiving witness here. +*/ + +// Node's own BroadcastChannel and Web Locks, captured before any test installs +// the controlled seams. This lane swaps them in for its body only. +const nativeBroadcastChannel = globalThis.BroadcastChannel +const nativeNavigator = globalThis.navigator + +function withNativeBrowserTransport(body: () => Promise): Promise { + const controlledChannel = globalThis.BroadcastChannel + const controlledNavigator = globalThis.navigator + const define = (channel: unknown, navigator: unknown) => { + Object.defineProperty(globalThis, `BroadcastChannel`, { + value: channel, + writable: true, + configurable: true, + }) + Object.defineProperty(globalThis, `navigator`, { + value: navigator, + writable: true, + configurable: true, + }) + } + define(nativeBroadcastChannel, nativeNavigator) + return body().finally(() => define(controlledChannel, controlledNavigator)) +} + +type RealTabSource = `receipt` | `query-db` +type RealTabPlacement = `direct` | `queued` +type RealTabHistory = { + source: RealTabSource + placement: RealTabPlacement + apply: `ok` | `fails` + // A live query over the demand, or the follower Collection's load itself. + consumer: `live-query` | `load` +} +type RealTabRow = { id: string; group: string } + +function expectedRealTabFollower(history: RealTabHistory) { + const applied = history.apply === `ok` + return { + settledWhileApplyHeld: false, + settled: { + outcome: applied ? `fulfilled` : `rejected`, + rows: applied ? [`remote`] : undefined, + source: applied ? `ready` : `error`, + }, + } +} + +function createDeferredVoid() { + let resolve!: () => void + let reject!: (error: unknown) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + void promise.catch(() => undefined) + return { promise, resolve, reject } +} + +// One tab: a real SQLite connection to the shared file, a real coordinator, +// browser persistence, and the persisted on-demand Collection every tab +// defines. Only the leader's source runs; a follower routes demand. +function createRealTab(options: { + dbName: string + filename: string + source: RealTabSource + upstream: Map + holdUpstream?: () => Promise + readGate?: (rows: ReadonlyArray) => Promise + now?: () => number + // Commit on every load: a newer version of a delivered row (`update`), or + // the same row inserted again (`reinsert`, an equal-value insert core + // reconciles as an update). + commitEachLoad?: `update` | `reinsert` +}) { + const database = createWASQLiteTestDatabase({ filename: options.filename }) + const gatedDatabase: BrowserWASQLiteDatabase = { + execute: async (sql, params) => { + const rows = await database.execute(sql, params) + if (options.readGate && /^\s*select/i.test(sql)) { + await options.readGate(rows) + } + return rows as never + }, + close: () => database.close?.(), + } + const coordinator = new BrowserCollectionCoordinator({ + dbName: options.dbName, + }) + liveCoordinators.add(coordinator) + const persistence = createBrowserWASQLitePersistence({ + database: gatedDatabase, + coordinator, + ...(options.now + ? { now: options.now, cacheGenerationClaimTtlMs: 10_000 } + : {}), + }) + const upstreamRows = (where: unknown) => { + const wanted = JSON.stringify(where ?? null) + return Array.from(options.upstream.values()).filter((row) => + wanted.includes(`"${row.id}"`), + ) + } + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const trace: Array = [] + let storageCollectionId: string | undefined + const setAdapter = coordinator.setAdapterForCollection.bind(coordinator) + coordinator.setAdapterForCollection = (collectionId, adapter, claimId) => { + if (claimId) storageCollectionId = collectionId + setAdapter(collectionId, adapter, claimId) + } + const ensure = coordinator.requestEnsureRemoteSubset.bind(coordinator) + coordinator.requestEnsureRemoteSubset = (collectionId, subset) => { + trace.push(`ensure`) + return ensure(collectionId, subset).finally(() => + trace.push(`ensure-settled`), + ) + } + const loadUpstream = async (where: unknown) => { + trace.push(`upstream`) + await options.holdUpstream?.() + return upstreamRows(where) + } + const delivered = new Set() + let version = 0 + const receiptSource: SyncConfig = { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: async (subset) => { + if (options.commitEachLoad) { + const fresh = await loadUpstream(subset.where) + if (fresh.length === 0) return + begin() + for (const row of fresh) { + const delivered_ = delivered.has(row.id) + delivered.add(row.id) + if (options.commitEachLoad === `update` && delivered_) { + const next = { ...row, group: `v${++version}` } + options.upstream.set(row.id, next) + write({ type: `update`, value: next }) + } else write({ type: `insert`, value: row }) + } + await whenSyncAccepted(commit()) + return + } + // Like a real source, deliver each upstream row version once. + const rows = (await loadUpstream(subset.where)).filter( + (row) => !delivered.has(row.id), + ) + for (const row of rows) delivered.add(row.id) + if (rows.length === 0) return + begin() + for (const row of rows) write({ type: `insert`, value: row }) + await whenSyncAccepted(commit()) + }, + } + }, + } + const collection = + options.source === `receipt` + ? createCollection( + persistedCollectionOptions({ + id: `real-tab-items`, + schemaVersion: 1, + persistence, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: receiptSource, + }), + ) + : createCollection( + persistedCollectionOptions({ + id: `real-tab-items`, + schemaVersion: 1, + persistence, + ...queryCollectionOptions< + RealTabRow, + unknown, + Array, + string + >({ + queryKey: [`real-tab-items`], + queryClient, + syncMode: `on-demand`, + getKey: (row) => row.id, + queryFn: (context) => + loadUpstream(context.meta?.loadSubsetOptions?.where), + }), + }), + ) + return { + coordinator, + collection, + trace, + storageCollectionId: () => storageCollectionId, + cleanup: async () => { + await collection.cleanup() + queryClient.clear() + coordinator.dispose() + liveCoordinators.delete(coordinator) + database.close?.() + }, + } +} + +function assertRealTabFollowerHistory(history: RealTabHistory): Promise { + return withNativeBrowserTransport(() => runRealTabFollowerHistory(history)) +} + +async function runRealTabFollowerHistory( + history: RealTabHistory, +): Promise { + const directory = mkdtempSync(join(tmpdir(), `db-real-tab-follower-`)) + const filename = join(directory, `tabs.sqlite`) + const dbName = `real-tab-follower-${Math.random().toString(36).slice(2)}` + const upstream = new Map([ + [`remote`, { id: `remote`, group: `a` }], + ]) + const applyGate = createDeferredVoid() + const hydrationGate = createDeferredVoid() + const upstreamGate = createDeferredVoid() + let gateApply = false + let applyReached = false + let hydrationReached = false + let gateHydration = false + let holdLeaderUpstream = false + const applyFailure = new Error(`follower apply read failed`) + const tabs: Array> = [] + const queries: Array<{ cleanup: () => Promise }> = [] + try { + const leader = createRealTab({ + dbName, + filename, + source: history.source, + upstream, + holdUpstream: () => + holdLeaderUpstream ? upstreamGate.promise : Promise.resolve(), + }) + tabs.push(leader) + leader.collection.startSyncImmediate() + await leader.collection.stateWhenReady() + const follower = createRealTab({ + dbName, + filename, + source: history.source, + upstream, + readGate: async (rows) => { + if (gateHydration) { + gateHydration = false + hydrationReached = true + await hydrationGate.promise + return + } + if (gateApply && JSON.stringify(rows).includes(`remote`)) { + gateApply = false + applyReached = true + await applyGate.promise + if (history.apply === `fails`) throw applyFailure + } + }, + }) + tabs.push(follower) + follower.collection.startSyncImmediate() + + await vi.waitFor(() => expect(follower.storageCollectionId()).toBeDefined()) + gateApply = true + if (history.placement === `queued`) holdLeaderUpstream = true + // The consumer settles at first ready (live query) or when the load + // settles; `rows` are the demanded rows it sees at that moment. + let settledAt: { rows: Array } | undefined + let outcome: Promise<`fulfilled` | `rejected`> + if (history.consumer === `live-query`) { + const query = createLiveQueryCollection((q) => + q + .from({ item: follower.collection }) + .where(({ item }) => eq(item.id, `remote`)), + ) + queries.push(query) + query.onFirstReady(() => { + settledAt = { rows: query.toArray.map((row) => row.id).sort() } + }) + outcome = query.preload().then( + () => `fulfilled` as const, + () => `rejected` as const, + ) + } else { + outcome = Promise.resolve( + follower.collection._sync.loadSubset({ + where: new IR.Func(`eq`, [ + new IR.PropRef([`id`]), + new IR.Value(`remote`), + ]), + }), + ).then( + () => { + settledAt = { + rows: follower.collection.has(`remote`) ? [`remote`] : [], + } + return `fulfilled` as const + }, + () => `rejected` as const, + ) + } + if (history.placement === `queued`) { + // While the owner's upstream load is held, another subset's hydration + // takes the follower's apply mutex. The owner's broadcast then arrives + // queued behind it. + await vi.waitFor( + () => { + expect(follower.trace).toContain(`ensure`) + expect(leader.trace).toContain(`upstream`) + }, + { timeout: 15_000 }, + ) + gateHydration = true + const other = createLiveQueryCollection((q) => + q + .from({ item: follower.collection }) + .where(({ item }) => eq(item.id, `other`)), + ) + queries.push(other) + void other.preload().catch(() => undefined) + await vi.waitFor(() => expect(hydrationReached).toBe(true), { + timeout: 15_000, + }) + holdLeaderUpstream = false + upstreamGate.resolve() + } + // Checkpoint: the owner's reply has reached the follower, and the apply + // of the owner's commit is still outstanding here: a direct broadcast's + // durable read is held, or a queued broadcast waits behind the held + // hydration. The model says the query has not reached first ready. + await vi.waitFor( + () => { + expect( + history.placement === `direct` ? applyReached : hydrationReached, + ).toBe(true) + expect(follower.trace).toContain(`ensure-settled`) + }, + { timeout: 15_000 }, + ) + for (let turn = 0; turn < 20; turn++) await nextTurn() + const expected = expectedRealTabFollower(history) + expect(settledAt !== undefined).toBe(expected.settledWhileApplyHeld) + hydrationGate.resolve() + applyGate.resolve() + const settled = await outcome + await vi.waitFor(() => + expect(follower.collection.status).toBe(expected.settled.source), + ) + expect({ + outcome: settled, + rows: settledAt?.rows, + source: follower.collection.status, + }).toEqual(expected.settled) + } finally { + hydrationGate.resolve() + applyGate.resolve() + upstreamGate.resolve() + for (const query of queries) await query.cleanup().catch(() => undefined) + for (const tab of tabs.reverse()) await tab.cleanup().catch(() => undefined) + rmSync(directory, { recursive: true, force: true }) + } +} + +describe(`follower remote-demand settlement across real tabs`, () => { + const histories: Array = [] + for (const source of [`receipt`, `query-db`] as const) + for (const placement of [`direct`, `queued`] as const) + for (const apply of [`ok`, `fails`] as const) + for (const consumer of [`live-query`, `load`] as const) + histories.push({ source, placement, apply, consumer }) + it.each(histories)( + `settles a follower $consumer with the owner's rows: $source source, $placement, apply $apply`, + (history) => assertRealTabFollowerHistory(history), + 20_000, + ) + + // Rotation across real tabs. When only the follower's cache claim expires, + // it rotates into a new generation that no other tab participates in, so + // it becomes that generation's owner and serves new demand from its own + // source. The model owes the same applied settlement there: the rows it + // already showed stay visible, and a new demand settles with its rows. + // This topology never routes to another tab's owner, so the + // elected-owner notice path stays covered by the single-runtime + // `persistence.follower-generation-notice` lane. + it( + `keeps a rotated follower's rows and serves its new demand from its own generation`, + () => + withNativeBrowserTransport(async () => { + const directory = mkdtempSync(join(tmpdir(), `db-real-tab-rotation-`)) + const filename = join(directory, `tabs.sqlite`) + const dbName = `real-tab-rotation-${Math.random().toString(36).slice(2)}` + const upstream = new Map([ + [`remote`, { id: `remote`, group: `a` }], + [`other`, { id: `other`, group: `b` }], + ]) + let followerNow = Date.now() + const tabs: Array> = [] + const where = (id: string): LoadSubsetOptions => ({ + where: new IR.Func(`eq`, [new IR.PropRef([`id`]), new IR.Value(id)]), + }) + try { + const leader = createRealTab({ + dbName, + filename, + source: `receipt`, + upstream, + }) + tabs.push(leader) + leader.collection.startSyncImmediate() + await leader.collection.stateWhenReady() + const follower = createRealTab({ + dbName, + filename, + source: `receipt`, + upstream, + now: () => followerNow, + }) + tabs.push(follower) + follower.collection.startSyncImmediate() + await vi.waitFor(() => + expect(follower.storageCollectionId()).toBeDefined(), + ) + await follower.collection._sync.loadSubset(where(`remote`)) + const original = follower.storageCollectionId() + expect(follower.coordinator.isLeader(original!)).toBe(false) + expect(follower.collection.has(`remote`)).toBe(true) + + followerNow += 20_000 + await follower.collection._sync.loadSubset(where(`other`)) + const rotated = follower.storageCollectionId() + expect(rotated).not.toBe(original) + expect(follower.coordinator.isLeader(rotated!)).toBe(true) + expect({ + remote: follower.collection.has(`remote`), + other: follower.collection.has(`other`), + status: follower.collection.status, + }).toEqual({ remote: true, other: true, status: `ready` }) + } finally { + for (const tab of tabs.reverse()) + await tab.cleanup().catch(() => undefined) + rmSync(directory, { recursive: true, force: true }) + } + }), + 30_000, + ) +}) + +/* +A leader source may commit on every load: a legal update with a new value, or +an equal-value insert of a row it already delivered, which core reconciles as +an update. A follower applies the leader's commit notice from durable rows. +That apply must not release and request the follower's demand again: a new +request makes the leader's source load and commit again, and the follower +would never settle. The model: the follower's query is first ready with the +leader's row, the follower sends at most two ensures (its demand, and at most +one replay), and a later leader commit still reaches the follower with the +leader's current value. +*/ +describe(`follower stays bounded when the leader commits on every load`, () => { + it.each([`update`, `reinsert`] as const)( + `settles and stays bounded: commit on every load as $0`, + (commitEachLoad) => + withNativeBrowserTransport(async () => { + const directory = mkdtempSync(join(tmpdir(), `db-real-tab-each-load-`)) + const filename = join(directory, `tabs.sqlite`) + const dbName = `real-tab-each-load-${Math.random().toString(36).slice(2)}` + const upstream = new Map([ + [`remote`, { id: `remote`, group: `a` }], + ]) + const tabs: Array> = [] + const queries: Array<{ cleanup: () => Promise }> = [] + const remote: LoadSubsetOptions = { + where: new IR.Func(`eq`, [ + new IR.PropRef([`id`]), + new IR.Value(`remote`), + ]), + } + try { + const leader = createRealTab({ + dbName, + filename, + source: `receipt`, + upstream, + commitEachLoad, + }) + tabs.push(leader) + leader.collection.startSyncImmediate() + await leader.collection.stateWhenReady() + const follower = createRealTab({ + dbName, + filename, + source: `receipt`, + upstream, + commitEachLoad, + }) + tabs.push(follower) + follower.collection.startSyncImmediate() + await vi.waitFor(() => + expect(follower.storageCollectionId()).toBeDefined(), + ) + const items: Collection = follower.collection + const query = createLiveQueryCollection((q) => + q.from({ item: items }).where(({ item }) => eq(item.id, `remote`)), + ) + queries.push(query) + let rowsAtFirstReady: Array | undefined + query.onFirstReady(() => { + rowsAtFirstReady = query.toArray.map((row) => row.id) + }) + void query.preload().catch(() => undefined) + await vi.waitFor(() => expect(query.isReady()).toBe(true), { + timeout: 8_000, + }) + expect(rowsAtFirstReady).toEqual([`remote`]) + for (let turn = 0; turn < 200; turn++) await nextTurn() + const ensures = () => + follower.trace.filter((entry) => entry === `ensure`).length + expect(ensures()).toBeLessThanOrEqual(2) + + // A later leader commit still reaches the follower. + await leader.collection._sync.loadSubset(remote) + await vi.waitFor( + () => + expect(follower.collection.get(`remote`)?.group).toBe( + upstream.get(`remote`)!.group, + ), + { timeout: 8_000 }, + ) + for (let turn = 0; turn < 200; turn++) await nextTurn() + expect(ensures()).toBeLessThanOrEqual(2) + } finally { + for (const query of queries) + await query.cleanup().catch(() => undefined) + for (const tab of tabs.reverse()) + await tab.cleanup().catch(() => undefined) + rmSync(directory, { recursive: true, force: true }) + } + }), + 30_000, + ) +}) + describe(`coordinator fixture lifecycle`, () => { it(`releases delayed transport and leadership state before a fresh lifecycle`, async () => { let first: BrowserCollectionCoordinator | undefined diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index ae681969ef..30d9aa8861 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -1747,9 +1747,13 @@ class PersistedCollectionRuntime< // release its turn before checking the claim and rotating the cache. queueMicrotask(() => { if (lifecycleGeneration !== this.lifecycleGeneration) return - void this.ensureCacheClaimActive().catch((error) => { - this.markTerminalFailure(error, lifecycleGeneration) - }) + void this.ensureCacheClaimActive() + .then(() => + this.applyMutex.run(() => this.flushQueuedTxCommittedUnsafe()), + ) + .catch((error) => { + this.markTerminalFailure(error, lifecycleGeneration) + }) }) } @@ -1915,6 +1919,31 @@ class PersistedCollectionRuntime< ) } + // An ensure reply settles a follower's load (law 7). A source's loadSubset + // must await every commit receipt that establishes its result, so the + // leader broadcasts those commits before it replies on the same channel. + // They are already queued on the apply mutex; settle after they show here. + private async applyLeaderCommitsBeforeReply( + options: LoadSubsetOptions, + ): Promise { + try { + // Commits kept while the claim was expired apply after it is active. + if (this.hasExpiredCacheClaim() || this.cacheClaimRecovery) { + await this.ensureCacheClaimActive() + } + await this.applyMutex.run(() => this.flushQueuedTxCommittedUnsafe()) + // A broadcast that failed to apply made the runtime terminal; its rows + // are not visible, so the load must not settle successfully. + this.throwIfTerminal() + } catch (error) { + // A terminal runtime releases the acquisition the owner granted the + // load. Otherwise, as after a failed ensure, the demand retries. + if (this.getCurrentTerminalFailure()) this.unloadSubset(options) + else this.queueRemoteSubsetEnsure(options) + throw error + } + } + private clearSyncControls(): void { this.syncControls = { begin: null, @@ -2740,6 +2769,7 @@ class PersistedCollectionRuntime< this.storageCollectionId, rotatedOptions, )) + await this.applyLeaderCommitsBeforeReply(options) } return } @@ -2793,6 +2823,7 @@ class PersistedCollectionRuntime< this.queueRemoteSubsetEnsure(options) throw error } + await this.applyLeaderCommitsBeforeReply(options) return } @@ -3692,6 +3723,7 @@ class PersistedCollectionRuntime< private replaceCollectionSnapshot( rows: Array<{ key: TKey; value: T; metadata?: unknown }>, collectionMetadata: Array<{ key: string; value: unknown }>, + replayDemand: boolean, ): SyncAppliedReceipt { if ( !this.syncControls.begin || @@ -3709,10 +3741,18 @@ class PersistedCollectionRuntime< ), ) const rowKeys = new Set(rows.map((row) => row.key)) + // A truncate replays the Collection's demand. A follower routes that + // demand to its leader, whose source loads and commits again, so a + // follower replaces its rows in place, except after a reset, which owes + // the replay. + const followsLeader = + this.canRouteRemoteDemandThroughCoordinator() && + !this.persistence.coordinator.isLeader(this.storageCollectionId) + const truncate = hydratedKeys.size === 0 && (replayDemand || !followsLeader) return this.withInternalApply(() => { this.syncControls.begin?.() - if (hydratedKeys.size === 0) { + if (truncate) { this.syncControls.truncate?.() } else { for (const key of base?.keys() ?? []) { @@ -3728,7 +3768,7 @@ class PersistedCollectionRuntime< for (const row of rows) { if (hydratedKeys.has(row.key)) continue - if (hydratedKeys.size > 0 && base?.has(row.key)) { + if (!truncate && base?.has(row.key)) { // Remove stale fields before applying a complete persisted row in // the default partial-update mode, without truncating seed keys. this.syncControls.write?.({ type: `delete`, key: row.key }) @@ -4537,38 +4577,37 @@ class PersistedCollectionRuntime< if (this.getCurrentTerminalFailure()) { return } - if (this.hasExpiredCacheClaim()) { - void this.ensureCacheClaimActive().catch((error) => { - this.markTerminalFailure(error) - }) - return - } - const { payload } = message const lifecycleGeneration = this.lifecycleGeneration const isSelf = message.senderId === this.persistence.coordinator.getNodeId() + const claimExpired = this.hasExpiredCacheClaim() + if (claimExpired && !isTxCommittedPayload(payload)) { + this.scheduleCacheClaimRecovery(lifecycleGeneration) + return + } // Allow tx:committed from self — the coordinator produces these on behalf // of both local and remote mutations. The seq dedup in // processCommittedTxUnsafe prevents double-processing of our own writes. if (isTxCommittedPayload(payload)) { - if (this.isHydratingNow()) { - this.queuedTxCommitted.push({ - txCommitted: payload, - lifecycleGeneration, - storageCollectionId: message.collectionId, - }) - return + // Commits apply in arrival order through the queue. While the claim is + // expired, keep them until renewal or rotation decides this generation; + // a renewed generation still owes its rows. A hydration flushes the + // queue when it ends. + this.queuedTxCommitted.push({ + txCommitted: payload, + lifecycleGeneration, + storageCollectionId: message.collectionId, + }) + if (claimExpired) { + this.scheduleCacheClaimRecovery(lifecycleGeneration) + } else if (!this.isHydratingNow()) { + void this.applyMutex + .run(() => this.flushQueuedTxCommittedUnsafe()) + .catch((error) => { + this.markTerminalFailure(error, lifecycleGeneration) + }) } - - void this.applyMutex - .run(() => { - if (message.collectionId !== this.storageCollectionId) return - return this.processCommittedTxUnsafe(payload, lifecycleGeneration) - }) - .catch((error) => { - this.markTerminalFailure(error, lifecycleGeneration) - }) return } @@ -4606,15 +4645,26 @@ class PersistedCollectionRuntime< private async flushQueuedTxCommittedUnsafe(): Promise { while (this.queuedTxCommitted.length > 0) { + // Claim recovery flushes the queue once the generation is decided. + if (this.hasExpiredCacheClaim()) { + this.scheduleCacheClaimRecovery(this.lifecycleGeneration) + return + } const queued = this.queuedTxCommitted.shift() if (!queued) { continue } if (queued.storageCollectionId !== this.storageCollectionId) continue - await this.processCommittedTxUnsafe( - queued.txCommitted, - queued.lifecycleGeneration, - ) + // A queued commit has left the queue, so a failure must be terminal, + // as it is for a commit applied when it arrives. + try { + await this.processCommittedTxUnsafe( + queued.txCommitted, + queued.lifecycleGeneration, + ) + } catch (error) { + throw this.markTerminalFailure(error, queued.lifecycleGeneration) + } } } @@ -4623,10 +4673,6 @@ class PersistedCollectionRuntime< lifecycleGeneration = this.lifecycleGeneration, ): Promise { if (lifecycleGeneration !== this.lifecycleGeneration) return - if (this.hasExpiredCacheClaim()) { - this.scheduleCacheClaimRecovery(lifecycleGeneration) - return - } if (this.resetTermFence !== undefined) { if (txCommitted.term <= this.resetTermFence) { // This payload could have been committed before the schema reset. @@ -4906,10 +4952,13 @@ class PersistedCollectionRuntime< } if (lifecycleGeneration !== this.lifecycleGeneration) return + // A reset replays demand: the owner's reset rows return only when the + // follower requests its demand again. await this.reloadActiveSubsetsUnsafe( adapter, false, this.syncMode === `eager`, + true, ) } @@ -5064,6 +5113,7 @@ class PersistedCollectionRuntime< adapter: HydrationPersistenceAdapter, skipUnchangedReplace = false, restoreResetBaseline = false, + replayDemand = false, ): Promise { if (this.scopedRecovery) { this.refreshScopedSubsets() @@ -5217,6 +5267,7 @@ class PersistedCollectionRuntime< metadata: row.metadata, })), collectionMetadata, + replayDemand, ) await whenSyncAccepted(applied) } diff --git a/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts b/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts index f49827d4b5..bd5fd592e6 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted-oracle.test.ts @@ -131,7 +131,9 @@ import type { * conformance portfolio have separate owners. This file proves the role * partition for non-single-process remote demand: an ownerless elected node * does not route, while a follower may route to the elected owner's registered - * source. Native SQLite hosts, live Electric service behavior, adapter + * source. During scoped recovery a follower takes a current-generation commit + * notice from the elected owner as source evidence and publishes its demanded + * durable rows; a notice from any other sender still requests a refetch. Native SQLite hosts, live Electric service behavior, adapter * cancellation, and the open B2-failure/E4 schedule remain separate evidence. */ @@ -212,7 +214,9 @@ const requestedOracleReplayProperty = readOracleRunConfig().replayProperty const describeUnlessOracleReplay = requestedOracleReplayProperty === undefined || requestedOracleReplayProperty === `persistence.retained-demand` || - requestedOracleReplayProperty === `persistence.generation-notice` + requestedOracleReplayProperty === `persistence.generation-notice` || + requestedOracleReplayProperty === `persistence.follower-generation-notice` || + requestedOracleReplayProperty === `persistence.follower-remote-demand` ? describe : describe.skip @@ -6345,6 +6349,49 @@ describeUnlessOracleReplay(`persistedCollectionOptions`, () => { ).toThrow(InvalidPersistedCollectionCoordinatorError) }) + // Every coordinator method the runtime calls is part of the contract, so a + // coordinator missing any one of them fails at configuration, naming it, + // rather than later inside message dispatch. + it.each([ + `getNodeId`, + `subscribe`, + `publish`, + `isLeader`, + `ensureLeadership`, + `requestEnsureRemoteSubset`, + `requestReleaseRemoteSubset`, + `registerRemoteSubsetOwner`, + `requestEnsurePersistedIndex`, + `requestApplyCommittedTx`, + ] as const)(`rejects a coordinator without %s`, (method) => { + const complete: Record = { + getNodeId: () => `node-1`, + subscribe: () => () => {}, + publish: () => {}, + isLeader: () => true, + ensureLeadership: async () => {}, + requestEnsureRemoteSubset: async () => {}, + requestReleaseRemoteSubset: async () => {}, + registerRemoteSubsetOwner: () => () => {}, + requestEnsurePersistedIndex: async () => {}, + requestApplyCommittedTx: async () => ({ term: 1, seq: 1 }), + } + delete complete[method] + + expect(() => + persistedCollectionOptions({ + id: `coordinator-without-${method}`, + getKey: (item) => item.id, + persistence: { + adapter: createNoopAdapter(), + coordinator: complete as unknown as PersistedCollectionCoordinator, + }, + }), + ).toThrow( + `Invalid persisted collection coordinator: missing required "${method}" method`, + ) + }) + it(`rejects an incomplete coordinator before starting a rich sync source`, async () => { type SourceParams = Parameters[`sync`]>[0] @@ -25509,6 +25556,1639 @@ describeUnlessOracleReplay(`persistedCollectionOptions`, () => { assertGenerationNoticeHistory, ) + /** + * A follower has no local source. It routes demand to the elected owner, + * and the owner's commits reach it only as durable rows plus a commit + * notice. After a cache rotation the follower is in scoped recovery, where + * durable rows cannot replace source rows. Rotation gives the follower a + * fresh cache generation, and the follower asks the owner to reacquire + * every active demand into it. Once those reacquisitions settle, the new + * generation holds exactly the owner's rows for that demand, so durable + * rows are authoritative again. The law is therefore: while the + * reacquisitions are in flight, the follower holds the generation's commit + * notices; when they settle, scoped recovery ends for that generation and + * the held notices, and every later one, apply through the normal path. + * No sender is special: a peer's notice after recovery is an ordinary + * invalidation that rereads durable rows. + * + * The independent model is `expectedScopedSnapshot` with complete demand + * over the authored durable rows of the current generation. It knows + * nothing about notices or scoped recovery; the driver applies it only at + * checkpoints where the owner's evidence has arrived. The ensure log is a + * separate observation: one acquisition and one reacquisition after + * rotation, and no other ensure. A bounded leader that re-commits on every + * ensure makes the alternative design (answer every notice during scoped + * recovery with a refetch) loop, and the bound turns that loop into an + * assertion failure. + * + * The grammar crosses when the owner first commits the row (before or + * after the rotation) with a later broadcast for the same key: none, an + * owner update, or a peer notice that claims a different value. An owner + * update's durable read either succeeds or fails. Recovery has ended, so a + * failed read of it is a real failure: the follower becomes terminal and + * keeps the rows it showed, instead of silently ignoring the failure as if + * a new recovery had started during the read. + * The + * observations are the follower's rows when its first load settles, after + * the rotation's reacquisition, and after the later broadcast, plus the + * ensure log at quiescence. The coordinator is a controlled double of an + * elected owner; it does not prove the browser transport. + * + * Known gap (#2139): production does not end a rotated follower's scoped + * recovery, so this lane's cells are pinned as failing until it does. + */ + type FollowerGenerationNoticeHistory = { + ownerCommit: `before-rotation` | `after-rotation` + later: `none` | `owner` | `peer` + // Applies only to a later owner update. + ownerRead: `ok` | `fails` + // Control: an owner notice's read is in flight when rotation starts, and + // then fails. Recovery owns that read, so its failure is ignored. + readFailsAsRotationStarts?: boolean + // How the owner's notices in the rotated generation reach the follower: + // in order, after a gap in the stream position (the follower's position + // restarts at zero when it rotates), or as an exact-ID reconciliation + // reload. Each must publish the same rows once recovery ends. + noticeShape?: `plain` | `gap` | `reconciliation` + } + const assertFollowerGenerationNoticeHistory = async ({ + ownerCommit, + later, + ownerRead, + readFailsAsRotationStarts = false, + noticeShape = `plain`, + }: FollowerGenerationNoticeHistory): Promise => { + const readFailure = new Error(`owner update read failed`) + let failNextRead = false + let holdNextRead: + { promise: Promise; reject: (e: unknown) => void } | undefined + const oldStorageId = `follower-notice-old` + const newStorageId = `follower-notice-new` + const ownerId = `elected-owner` + const v1: Todo = { id: `remote`, title: `Owner row` } + const v2: Todo = { id: `remote`, title: `Owner update` } + const durable = new Map>([ + [oldStorageId, new Map()], + [newStorageId, new Map()], + ]) + let storageId = oldStorageId + // The owner's upstream row: present from the start, or only once the + // owner fetched it after the rotation began. + let upstream: Todo | undefined = + ownerCommit === `before-rotation` ? v1 : undefined + const ensures: Array<{ id: string; refetch: boolean }> = [] + const subscribers = new Map< + string, + (message: ProtocolEnvelope) => void + >() + const versions = new Map() + // Enough commits for every legal history; a refetch loop exceeds it. + const commitBudget = 8 + let commits = 0 + const notify = (id: string, row: Todo, senderId: string) => { + // A gap skips one stream position in the rotated generation. + const skip = + noticeShape === `gap` && id === newStorageId && !versions.has(id) + ? 1 + : 0 + const version = (versions.get(id) ?? 0) + 1 + skip + versions.set(id, version) + const reconciliation = + noticeShape === `reconciliation` && id === newStorageId + subscribers.get(id)?.({ + v: 1, + dbName: `test-db`, + collectionId: id, + senderId, + ts: Date.now(), + payload: { + type: `tx:committed`, + term: 1, + seq: version, + txId: `${senderId}-${id}-${version}`, + latestRowVersion: version, + requiresFullReload: reconciliation, + ...(reconciliation ? { reconciliationReload: true } : {}), + changedRows: [{ key: row.id, value: { ...row } }], + deletedKeys: [], + } satisfies TxCommitted, + }) + } + const ownerCommits = (id: string, row: Todo) => { + if (commits++ >= commitBudget) return + durable.get(id)!.set(row.id, structuredClone(row)) + notify(id, row, ownerId) + } + const adapter: PersistenceAdapter = { + loadSubset: async (target) => { + const hold = holdNextRead + if (hold) { + holdNextRead = undefined + await hold.promise + } + if (failNextRead) { + failNextRead = false + throw readFailure + } + return Array.from( + durable.get(recordedStorageId(target))!.values(), + (value) => ({ + key: value.id, + value: structuredClone(value), + }), + ) + }, + claimCacheGeneration: async () => ({ + storageCollectionId: oldStorageId, + claimId: `follower-notice-claim`, + expiresAtMs: Date.now() + 60_000, + }), + renewCacheGenerationClaim: async () => Date.now() + 60_000, + rotateCacheGeneration: async () => { + storageId = newStorageId + if (ownerCommit === `after-rotation`) upstream = v1 + return { + storageCollectionId: newStorageId, + claimId: `follower-notice-claim`, + expiresAtMs: Date.now() + 60_000, + } + }, + releaseCacheGenerationClaim: async () => {}, + loadResumeSnapshot: async () => ({ + rows: [], + keySet: { status: `consistent` }, + collectionMetadata: [], + latestTerm: 0, + latestSeq: 0, + latestRowVersion: 0, + resetEpoch: 0, + }), + applyCommittedTx: async () => {}, + ensureIndex: async () => {}, + } + const coordinator: PersistedCollectionCoordinator = { + getNodeId: () => `follower-node`, + subscribe: (id, onMessage) => { + subscribers.set(id, onMessage) + return () => { + if (subscribers.get(id) === onMessage) subscribers.delete(id) + } + }, + publish: () => {}, + isLeader: () => false, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + // The elected owner loads its upstream and commits it into the + // requested generation before it replies, on every ensure. + requestEnsureRemoteSubset: async (id, options) => { + ensures.push({ id, refetch: options.refetch === true }) + if (upstream) ownerCommits(id, upstream) + }, + requestReleaseRemoteSubset: async () => {}, + registerRemoteSubsetOwner: () => () => {}, + requestApplyCommittedTx: (_id, tx) => + Promise.resolve({ + type: `rpc:applyCommittedTx:res`, + rpcId: tx.txId, + ok: true, + term: tx.term, + seq: tx.seq, + latestRowVersion: tx.rowVersion, + }), + pullSince: () => + Promise.resolve({ + type: `rpc:pullSince:res`, + rpcId: `follower-notice-pull`, + ok: true, + latestTerm: 1, + latestSeq: 0, + latestRowVersion: 0, + requiresFullReload: false, + changedKeys: [], + deletedKeys: [], + }), + } + let source!: TodoSyncParams + const collection = createCollection( + persistedCollectionOptions({ + id: `follower-generation-notice`, + syncMode: `on-demand`, + getKey: (row) => row.id, + sync: { + sync: (params) => { + source = params + params.markReady() + return { + loadSubset: async () => {}, + restartAfterScopedRecovery: async (cacheRotated) => { + await cacheRotated + }, + } + }, + }, + persistence: { adapter, coordinator }, + }), + ) + const demand: LoadSubsetOptions = { limit: 10 } + const publicRows = () => + new Map( + Array.from(collection.state, ([key, row]) => [ + String(key), + stripVirtualProps(row) as Todo, + ]), + ) + // A checkpoint after the follower has processed every delivered notice: + // the scan crosses the same apply mutex the notices use. + const settled = async (label: string) => { + await flushAsyncWork() + await atPersistedOracleCheckpoint( + source.metadata!.persistence!.scanPersistedRows(), + label, + ) + await flushAsyncWork() + } + const expectOwnerEvidence = (previous: ReadonlyMap) => { + const expected = expectedScopedSnapshot( + previous, + durable.get(storageId)!, + new Map(), + true, + ) + expectExactSnapshot(publicRows(), expected) + return expected + } + let hasPrimaryFailure = false + try { + collection.startSyncImmediate() + await collection.stateWhenReady() + await atPersistedOracleCheckpoint( + Promise.resolve(collection._sync.loadSubset(demand)), + `follower demand settled`, + ) + await settled(`first-ready notices processed`) + let visible = expectOwnerEvidence(new Map()) + + const persistence = source.metadata?.persistence + if (!persistence?.startScopedRecovery) { + throw new Error(`Expected scoped recovery capability`) + } + let rotation: Promise + if (readFailsAsRotationStarts) { + let reject!: (error: unknown) => void + const promise = new Promise((_resolve, rejectRead) => { + reject = rejectRead + }) + void promise.catch(() => undefined) + holdNextRead = { promise, reject } + ownerCommits(oldStorageId, { + id: `remote`, + title: `Pre-rotation update`, + }) + await flushAsyncWork() + rotation = persistence.startScopedRecovery() + reject(readFailure) + } else { + rotation = persistence.startScopedRecovery() + } + await atPersistedOracleCheckpoint( + rotation, + `follower cache rotation settled`, + ) + await atPersistedOracleCheckpoint( + vi.waitFor(() => + expect(ensures.map(({ id }) => id)).toContain(newStorageId), + ), + `reacquisition reached the owner`, + ) + await settled(`reacquisition notices processed`) + // Answering the owner's notice with another refetch would make the + // owner commit again, and the follower would refetch forever. + expect.soft(ensures).toHaveLength(2) + visible = expectOwnerEvidence(visible) + + if (later === `owner`) { + upstream = v2 + failNextRead = ownerRead === `fails` + ownerCommits(newStorageId, v2) + if (ownerRead === `fails`) { + // Not settled(): a terminal runtime rejects the checkpoint scan. + await flushAsyncWork() + await vi.waitFor(() => expect(collection.status).toBe(`error`)) + expectExactSnapshot(publicRows(), visible) + } else { + await settled(`owner update processed`) + expectOwnerEvidence(visible) + } + } else if (later === `peer`) { + // A peer's notice after recovery is an ordinary invalidation: the + // follower rereads durable rows, which the peer did not change, so + // its rows stay as they were and it sends no other ensure. + notify(newStorageId, { id: `remote`, title: `Peer claim` }, `peer-node`) + await settled(`peer notice processed`) + expectOwnerEvidence(visible) + } + + expect(ensures).toEqual([ + { id: oldStorageId, refetch: false }, + { id: newStorageId, refetch: true }, + ]) + } catch (error) { + hasPrimaryFailure = true + throw error + } finally { + await cleanupPersistedOracle( + [() => collection.cleanup()], + hasPrimaryFailure, + ) + } + } + + // Known gap: see #2139. A rotated follower stays in scoped recovery, so + // the owner's rows in the new generation never become visible. + it.fails.each([ + { ownerCommit: `before-rotation`, later: `none` }, + { ownerCommit: `after-rotation`, later: `none` }, + { ownerCommit: `after-rotation`, later: `owner` }, + { ownerCommit: `before-rotation`, later: `peer` }, + ] as const)( + `publishes an elected owner's rows to a follower after rotation: commit $ownerCommit, later $later`, + (history) => + assertFollowerGenerationNoticeHistory({ ...history, ownerRead: `ok` }), + ) + + // Known gap: see #2139. A rotated follower stays in scoped recovery, so + // the owner's rows in the new generation never become visible. + it.fails.each( + ([`gap`, `reconciliation`] as const).flatMap((noticeShape) => + ([`before-rotation`, `after-rotation`] as const).flatMap((ownerCommit) => + ([`none`, `owner`] as const).map((later) => ({ + ownerCommit, + later, + noticeShape, + })), + ), + ), + )( + `publishes an elected owner's rows after rotation when its notice is a $noticeShape: commit $ownerCommit, later $later`, + (history) => + assertFollowerGenerationNoticeHistory({ ...history, ownerRead: `ok` }), + ) + + // Known gap: see #2139. A rotated follower stays in scoped recovery, so + // the owner's rows in the new generation never become visible. + it.fails.each([ + { ownerCommit: `before-rotation`, later: `owner` }, + { ownerCommit: `after-rotation`, later: `owner` }, + ] as const)( + `fails a follower when its elected owner's update cannot be read after rotation: commit $ownerCommit`, + (history) => + assertFollowerGenerationNoticeHistory({ ...history, ownerRead: `fails` }), + ) + + // Known gap: see #2139. A rotated follower stays in scoped recovery, so + // the owner's rows in the new generation never become visible. + it.fails( + `ignores a follower read failure when rotation starts while the read is in flight`, + () => + assertFollowerGenerationNoticeHistory({ + ownerCommit: `before-rotation`, + later: `none`, + ownerRead: `ok`, + readFailsAsRotationStarts: true, + }), + ) + + const followerGenerationNoticeHistory = fc.record({ + ownerCommit: fc.constantFrom< + FollowerGenerationNoticeHistory[`ownerCommit`] + >(`before-rotation`, `after-rotation`), + later: fc.constantFrom( + `none`, + `owner`, + `peer`, + ), + ownerRead: fc.constantFrom( + `ok`, + `fails`, + ), + }) + // Known gap: see #2139. A rotated follower stays in scoped recovery, so + // the owner's rows in the new generation never become visible. + it.fails( + `checks generated follower generation-notice histories (fixed)`, + () => + fc.assert( + fc.asyncProperty( + followerGenerationNoticeHistory, + assertFollowerGenerationNoticeHistory, + ), + { seed: 2128, numRuns: oracleRuns(12) }, + ), + ) + // Known gap: see #2139. Some generated histories pass today, so a random + // draw cannot be pinned as failing; the fixed seed above pins the gap. + it.skip(`checks generated follower generation-notice histories (random or replayed)`, () => + fc.assert( + fc.asyncProperty( + followerGenerationNoticeHistory, + assertFollowerGenerationNoticeHistory, + ), + oraclePropertyOptions(12, `persistence.follower-generation-notice`), + )) + + /** + * A follower routes demand it cannot serve to the elected owner. The owner + * loads the rows upstream, commits them durably, and broadcasts the commit + * before it answers the ensure request on the same channel. The follower + * applies that broadcast asynchronously, by reading durable rows under its + * apply mutex. Law 7 therefore owes: the follower's load settles only after + * the broadcasts that preceded the reply have applied here, and it rejects, + * with the runtime terminal, if applying them failed. + * + * The independent model knows only whether the owner's commit applies. + * While that apply is held, no consumer of the demand has settled. If it + * applies, the consumer settles successfully with the owner's row visible. + * If it fails, the consumer rejects without the row, and the source is in + * error. The model does not know how the follower queues commits. + * + * The grammar crosses where the owner's broadcast meets the follower with + * whether its apply succeeds. A direct broadcast applies at once. A queued + * broadcast arrives while another subset's hydration holds the apply mutex, + * so it waits in the queue. A rotated history loses the cache claim after + * the first hydration read, so the demand reacquires through the owner + * under a new generation. An expired history lets the follower's cache + * claim lapse while its ensure is in flight, so the owner's notice arrives + * while the claim is expired; renewal keeps the same generation. The read + * that applies the owner's row is the one + * whose durable snapshot contains it; the driver holds that read to observe + * the not-ready checkpoint, or fails it. The consumer is a live query + * over the demanded rows or the follower's load itself. Observations, while + * the apply is held and after it settles: whether the consumer settled, its + * outcome, the demanded rows it sees when it settles (the query's + * first-ready rows, or the source's rows at load settlement), and the + * query's and source's status. A load that rejects owes the release of the + * remote acquisition the owner granted it (LoadSubsetFn), so for the load + * consumer the model also counts the follower's release requests for the + * current cache generation when it settles. The coordinator + * is a controlled double of an elected owner that broadcasts before it + * replies, as the built-in coordinator does; the two-tab OPFS suite is the + * receiving witness for the browser transport. + * + * A separate cell lets the claim lapse after the owner's row applied, so + * the follower's renewal at the reply fails transiently. That is not claim + * loss: the load rejects as after a failed ensure, and the demand stays + * active and retries instead of releasing the owner's acquisition. + * + * Known gap (#2139): a rotated follower never ends scoped recovery, so the + * owner's row never becomes visible in the rotated placement. Its cells are + * pinned as failing, and the generated histories omit that placement. A + * rotation that starts while a load waits for the owner's commits is also + * outside this lane and tracked there. + */ + type FollowerRemoteDemandHistory = { + placement: `direct` | `queued` | `rotated` | `expired` + apply: `ok` | `fails` + // A live query over the demand, or the follower's load itself. + consumer: `live-query` | `load` + // The owner's row applies, then the claim lapses before the reply and + // the follower's first renewal fails transiently. + renewal?: `fails-at-reply` + } + type FollowerRow = { id: string; group: string } + + const expectedFollowerRemoteDemand = ( + history: FollowerRemoteDemandHistory, + ) => { + const applied = history.apply === `ok` + return { + // While the owner's commit has not applied here, nothing has settled. + whileApplyHeld: { settled: false, rows: undefined }, + settled: + history.consumer === `live-query` + ? applied + ? { outcome: `fulfilled`, rows: [`remote`], status: `ready` } + : { outcome: `rejected`, rows: undefined, status: `error` } + : applied + ? { + outcome: `fulfilled`, + rows: [`remote`], + status: undefined, + releases: 0, + } + : { + outcome: `rejected`, + rows: undefined, + status: `error`, + releases: 1, + }, + } + } + + function createFollowerTab(history: FollowerRemoteDemandHistory) { + const durable: Array = [] + const applyRead = createDeferred() + const applyReadStarted = createDeferred() + const otherHydrationStarted = createDeferred() + const otherHydration = createDeferred() + let reads = 0 + let storageId = `follower-demand-cache-1` + let loseClaim = false + const claim = () => ({ + storageCollectionId: storageId, + claimId: `follower-demand-claim`, + expiresAtMs: Date.now() + 10_000, + }) + const rotated = history.placement === `rotated` + const expiring = history.placement === `expired` + const renewalFailsAtReply = history.renewal === `fails-at-reply` + let failRenewal = false + let now = 1_000_000 + const adapter: PersistenceAdapter = { + loadSubset: async () => { + const read = ++reads + const snapshot = durable.slice() + if (snapshot.some((row) => row.id === `remote`)) { + applyReadStarted.resolve() + await applyRead.promise + if (history.apply === `fails`) throw new Error(`apply read failed`) + } else if (history.placement === `queued` && read === 2) { + otherHydrationStarted.resolve() + await otherHydration.promise + } + if (rotated && read === 1) loseClaim = true + return snapshot.map((value) => ({ key: value.id, value })) + }, + ...(expiring + ? { + getCacheGenerationNow: () => now, + claimCacheGeneration: async () => ({ + storageCollectionId: storageId, + claimId: `follower-demand-claim`, + expiresAtMs: now + 10_000, + }), + // The claim is still held, so renewal keeps this generation. + renewCacheGenerationClaim: async () => now + 10_000, + rotateCacheGeneration: async () => { + throw new Error(`an expired history must not rotate`) + }, + releaseCacheGenerationClaim: async () => {}, + } + : {}), + ...(renewalFailsAtReply + ? { + getCacheGenerationNow: () => now, + claimCacheGeneration: async () => ({ + storageCollectionId: storageId, + claimId: `follower-demand-claim`, + expiresAtMs: now + 10_000, + }), + // A transient storage error, not a lost claim. + renewCacheGenerationClaim: async () => { + if (failRenewal) { + failRenewal = false + throw new Error(`transient renewal failure`) + } + return now + 10_000 + }, + rotateCacheGeneration: async () => { + throw new Error(`a transient renewal failure must not rotate`) + }, + releaseCacheGenerationClaim: async () => {}, + } + : {}), + ...(rotated + ? { + claimCacheGeneration: async () => claim(), + renewCacheGenerationClaim: async () => + loseClaim ? undefined : Date.now() + 10_000, + rotateCacheGeneration: async () => { + loseClaim = false + storageId = `follower-demand-cache-2` + return claim() + }, + releaseCacheGenerationClaim: async () => {}, + } + : {}), + loadResumeSnapshot: () => + Promise.resolve({ + rows: [], + keySet: { status: `consistent` }, + collectionMetadata: [], + latestTerm: 0, + latestSeq: 0, + latestRowVersion: 0, + resetEpoch: 0, + }), + applyCommittedTx: async () => {}, + ensureIndex: async () => {}, + } as PersistenceAdapter + let subscriber: ((message: ProtocolEnvelope) => void) | undefined + let ensures = 0 + let releases = 0 + let resetPending = false + const coordinator: PersistedCollectionCoordinator = { + getNodeId: () => `follower`, + subscribe: (_collectionId, onMessage) => { + subscriber = onMessage + return () => { + subscriber = undefined + } + }, + publish: () => {}, + isLeader: () => false, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + // The owner commits once, broadcasts, then replies; later ensures find + // nothing new. + requestEnsureRemoteSubset: async (collectionId) => { + if (++ensures > 1 && !resetPending) return + resetPending = false + if (history.placement === `queued`) await otherHydrationStarted.promise + // The follower's claim lapses before the owner's notice reaches it. + if (expiring) now += 20_000 + const value = { id: `remote`, group: `a` } + durable.push(value) + const payload: TxCommitted = { + type: `tx:committed`, + term: 1, + seq: ensures, + txId: `owner-tx-${ensures}`, + latestRowVersion: ensures, + requiresFullReload: false, + changedRows: [{ key: value.id, value }], + deletedKeys: [], + } + subscriber?.({ + v: 1, + dbName: `test-db`, + collectionId, + senderId: `owner`, + ts: Date.now(), + payload, + }) + if (renewalFailsAtReply) { + // Let the owner's row apply here before the claim lapses. + for (let i = 0; i < 5; i++) + await new Promise((resolve) => setTimeout(resolve, 0)) + now += 20_000 + failRenewal = true + } + }, + // Rotation legitimately releases the previous generation's + // acquisition, so only the current generation's releases count. + requestReleaseRemoteSubset: async (collectionId) => { + if (!rotated || collectionId === storageId) releases++ + }, + registerRemoteSubsetOwner: () => () => {}, + requestApplyCommittedTx: (_collectionId, tx) => + Promise.resolve({ + type: `rpc:applyCommittedTx:res`, + rpcId: tx.txId, + ok: true, + term: tx.term, + seq: tx.seq, + latestRowVersion: tx.rowVersion, + }), + pullSince: () => + Promise.resolve({ + type: `rpc:pullSince:res`, + rpcId: `follower-demand-pull`, + ok: true, + latestTerm: 1, + latestSeq: 0, + latestRowVersion: 0, + requiresFullReload: false, + changedKeys: [], + deletedKeys: [], + }), + } + const source = createCollection( + persistedCollectionOptions({ + id: `follower-demand-${Math.random().toString(36).slice(2)}`, + getKey: (row) => row.id, + persistence: { adapter, coordinator }, + syncMode: `on-demand`, + sync: { + sync: () => ({ + loadSubset: () => Promise.resolve(), + restartAfterScopedRecovery: () => {}, + }), + }, + }), + ) + source.startSyncImmediate() + return { + source, + applyReadStarted: applyReadStarted.promise, + otherHydrationStarted: otherHydrationStarted.promise, + releaseApply: () => applyRead.resolve(), + releaseOtherHydration: () => otherHydration.resolve(), + releases: () => releases, + ensures: () => ensures, + // The owner resets durable storage and announces it; it loads the + // demanded row again on the next ensure. + resetOwner: () => { + durable.length = 0 + resetPending = true + subscriber?.({ + v: 1, + dbName: `test-db`, + collectionId: source.id, + senderId: `owner`, + ts: Date.now(), + payload: { + type: `collection:reset`, + schemaVersion: 1, + resetEpoch: 1, + }, + }) + }, + } + } + + // Wait for a driver event, but turn a missing event into an observation + // rather than a timeout. + const within = (promise: Promise, ms = 200) => + Promise.race([ + promise.then(() => true), + new Promise((resolve) => setTimeout(() => resolve(false), ms)), + ]) + + const assertFollowerRemoteDemandHistory = async ( + history: FollowerRemoteDemandHistory, + ) => { + const expected = expectedFollowerRemoteDemand(history) + const errors = vi.spyOn(console, `error`).mockImplementation(() => {}) + const warnings = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const tab = createFollowerTab(history) + const groupQuery = (group: string) => + createLiveQueryCollection((q) => + q.from({ row: tab.source }).where(({ row }) => eq(row.group, group)), + ) + const query = groupQuery(`a`) + const other = groupQuery(`b`) + try { + // The consumer's observation: its outcome, and the demanded rows it + // sees when it settles. + const seen: { + outcome: `pending` | `fulfilled` | `rejected` + rows?: Array + } = { outcome: `pending` } + const demandedRows = () => + Array.from(tab.source.values()) + .filter((row) => row.group === `a`) + .map((row) => row.id) + .sort() + let settled: Promise + if (history.consumer === `live-query`) { + query.onFirstReady(() => { + seen.rows = query.toArray.map((row) => (row as FollowerRow).id).sort() + }) + settled = query.preload().then( + () => (seen.outcome = `fulfilled`), + () => (seen.outcome = `rejected`), + ) + } else { + settled = Promise.resolve( + tab.source._sync.loadSubset({ + where: new IR.Func(`eq`, [ + new IR.PropRef([`group`]), + new IR.Value(`a`), + ]), + }), + ).then( + () => { + seen.outcome = `fulfilled` + seen.rows = demandedRows() + }, + () => (seen.outcome = `rejected`), + ) + } + if (history.placement === `queued`) void other.preload().catch(() => {}) + + if (history.placement === `queued`) { + // The owner's broadcast arrives while this hydration holds the apply + // mutex, so it waits in the queue until the hydration ends. + await within(tab.otherHydrationStarted) + await new Promise((resolve) => setTimeout(resolve, 0)) + tab.releaseOtherHydration() + } + const applyStarted = await within(tab.applyReadStarted) + await new Promise((resolve) => setTimeout(resolve, 0)) + // Checkpoint: the owner has replied, but its row has not applied here. + expect({ + applyStarted, + settled: seen.outcome !== `pending`, + rows: seen.rows, + }).toEqual({ applyStarted: true, ...expected.whileApplyHeld }) + + tab.releaseApply() + await within(settled) + await new Promise((resolve) => setTimeout(resolve, 0)) + // Checkpoint: the apply settled. + const releasesAtSettle = tab.releases() + expect({ + outcome: seen.outcome, + rows: seen.rows, + ...(`releases` in expected.settled + ? { releases: releasesAtSettle } + : {}), + status: + expected.settled.status === undefined + ? undefined + : history.consumer === `live-query` && + expected.settled.status === `ready` + ? query.status + : expected.settled.status === `error` + ? history.consumer === `live-query` + ? `${query.status}/${tab.source.status}` + : tab.source.status + : tab.source.status, + }).toEqual({ + outcome: expected.settled.outcome, + rows: expected.settled.rows, + ...(`releases` in expected.settled + ? { releases: expected.settled.releases } + : {}), + status: + expected.settled.status === `error` && + history.consumer === `live-query` + ? `error/error` + : expected.settled.status, + }) + } finally { + await query.cleanup() + await other.cleanup() + await tab.source.cleanup() + errors.mockRestore() + warnings.mockRestore() + } + } + + const followerRemoteDemandHistories: Array = ( + [`direct`, `queued`, `rotated`, `expired`] as const + ).flatMap((placement) => + ([`ok`, `fails`] as const).flatMap((apply) => + ([`live-query`, `load`] as const).map((consumer) => ({ + placement, + apply, + consumer, + })), + ), + ) + it.each( + followerRemoteDemandHistories.filter( + ({ placement }) => placement !== `rotated`, + ), + )( + `settles a follower's remote demand with the owner's applied rows: $placement broadcast, apply $apply, $consumer`, + assertFollowerRemoteDemandHistory, + ) + // Known gap: see #2139. A rotated follower stays in scoped recovery, so + // the owner's rows in the new generation never become visible. + it.fails.each( + followerRemoteDemandHistories.filter( + ({ placement }) => placement === `rotated`, + ), + )( + `settles a follower's remote demand with the owner's applied rows: $placement broadcast, apply $apply, $consumer`, + assertFollowerRemoteDemandHistory, + ) + // A reset clears the owner's durable rows. Its rows return only when the + // follower requests its demand again, so the model predicts one more + // ensure, the owner's row visible again, and no further ensures: replacing + // rows in place must not drop the replay that a reset owes. Only a node + // that follows replaces rows in place. A node elected leader that still + // routes demand through the coordinator truncates and replays as the + // leader does; this follower-only cell does not exercise that path. + // A renewal that fails after the owner replied does not establish claim + // loss. The load cannot confirm its rows, so it rejects, as a failed ensure + // does; the demand stays active and retries, so the owner's acquisition + // is kept. The model predicts the rejected load, no release, one retried + // ensure, the owner's row still visible, and a healthy source. + it(`keeps a follower's demand when a renewal fails transiently at its reply`, async () => { + const errors = vi.spyOn(console, `error`).mockImplementation(() => {}) + const warnings = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const tab = createFollowerTab({ + placement: `direct`, + apply: `ok`, + consumer: `load`, + renewal: `fails-at-reply`, + }) + try { + tab.releaseApply() + let outcome: `pending` | `fulfilled` | `rejected` = `pending` + const settled = Promise.resolve( + tab.source._sync.loadSubset({ + where: new IR.Func(`eq`, [ + new IR.PropRef([`group`]), + new IR.Value(`a`), + ]), + }), + ).then( + () => (outcome = `fulfilled`), + () => (outcome = `rejected`), + ) + await within(settled, 1_000) + for (let i = 0; i < 5; i++) + await new Promise((resolve) => setTimeout(resolve, 0)) + // Checkpoint: the load settled and any retry has run. + expect({ + outcome, + releases: tab.releases(), + ensures: tab.ensures(), + rows: Array.from(tab.source.values()).map((row) => row.id), + status: tab.source.status === `error` ? `error` : `healthy`, + }).toEqual({ + outcome: `rejected`, + releases: 0, + ensures: 2, + rows: [`remote`], + status: `healthy`, + }) + } finally { + await tab.source.cleanup() + errors.mockRestore() + warnings.mockRestore() + } + }) + + it(`a follower requests its demand again after the owner resets, once`, async () => { + const errors = vi.spyOn(console, `error`).mockImplementation(() => {}) + const warnings = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const tab = createFollowerTab({ + placement: `direct`, + apply: `ok`, + consumer: `live-query`, + }) + const query = createLiveQueryCollection((q) => + q.from({ row: tab.source }).where(({ row }) => eq(row.group, `a`)), + ) + try { + void query.preload() + await within(tab.applyReadStarted) + tab.releaseApply() + await vi.waitFor(() => expect(query.size).toBe(1), { timeout: 1_000 }) + expect(tab.ensures()).toBe(1) + + tab.resetOwner() + await new Promise((resolve) => setTimeout(resolve, 100)) + expect({ + ensures: tab.ensures(), + rows: query.toArray.map((row) => (row as FollowerRow).id), + status: query.status, + }).toEqual({ ensures: 2, rows: [`remote`], status: `ready` }) + } finally { + await query.cleanup() + await tab.source.cleanup() + errors.mockRestore() + warnings.mockRestore() + } + }) + + // Scoped recovery means durable rows cannot replace source rows. Owner + // evidence is an exception only in a rotated generation, which holds + // nothing but the owner's writes. Without a managed generation the old + // durable cache stays in place, so the model predicts that an owner notice + // during that recovery republishes none of the untrusted durable rows. + /** + * After a cache rotation, a follower reacquires every active demand into + * the new generation through the owner. The owner commits each demand's + * rows into that generation and broadcasts them before it replies, so the + * follower holds those notices. Scoped recovery may end only once every + * reacquisition has succeeded: before that, the new generation is still + * incomplete and its durable rows have no authority. Starting scoped + * recovery withdraws the old generation's rows. + * + * The independent model knows each demand's row version per generation: + * the owner wrote version 1 into the old generation and writes version 2 + * into the new one when it reacquires the demand. Until every + * reacquisition has settled, neither demand shows a row. If all succeed, + * both show the new generation's rows, including an owner update that + * arrived while the reacquisitions were settling. If any fails, the + * follower stays in scoped recovery, shows no row from the incomplete + * generation, and the source reports the failure as soon as that + * reacquisition settles. + * + * A demand that the caller releases or aborts while it reacquires no longer + * needs the new generation's rows: its aborted ensure is neither a failure + * nor a reported error, and the other demands decide the outcome. Releasing + * it twice changes nothing. A load that fails on its own still counts as a + * failure even though the failed load gives up its demand. + * + * A new demand loaded while the reacquisitions are in flight settles only + * once they all have: with the rows of the new generation if recovery + * ended, or rejected if it did not. Its own ensure answering early does not + * make its held rows visible. + * + * Only a recovery that rotated can end this way. A recovery that keeps the + * old generation, or one in which this node is the leader, leaves scoped + * recovery in place, and a newer rotation supersedes an older one's + * reacquisition. + * + * Known gap (#2139): production does not yet end scoped recovery after a + * rotation, so the histories in which recovery should end, or in which a + * new demand waits for it, are pinned as failing. + * + * Limits. A failed reacquisition is not retried; the follower stays in + * scoped recovery until a later recovery rotates again. + * + * The grammar crosses the order in which two demands' reacquisitions + * settle, whether the owner updates demand a's row (version 3) between + * the first and last settlement, which reacquisitions fail or are released, + * and whether a new demand loads during reacquisition. Separate cells start + * rotation from two reads that find the claim lost, supersede reacquisition + * by a rotation as leader, and recover without rotating. The driver + * observes the rows, the new load's outcome, and the source status after + * every settlement, plus the ensure count, which a refetch loop would + * exceed. + */ + type RotatedReacquisitionHistory = { + order: [`a`, `b`] | [`b`, `a`] + noticeBetween: boolean + // Which reacquisitions fail. `released` means demand b is released + // while it reacquires, so its ensure aborts; that is not a failure. + fails: `none` | `a` | `b` | `both` | `released` + // A new demand, c, loads while the reacquisitions are in flight. + newDemand?: boolean + // While reacquisition is in flight, the node is elected leader and a + // second rotation supersedes the first; it then follows again, and a new + // demand c loads. With nothing reacquiring, the node stays in scoped + // recovery: held notices must still be handled, not queued forever. + supersededAsLeader?: boolean + // How rotation starts: the source asks for scoped recovery after both + // demands loaded, or both demands' first reads are in flight when the + // cache claim is found lost. + trigger?: `source` | `claim-lost-during-reads` + } + type VersionedRow = { id: string; group: string; v: number } + type ReacquisitionObservation = { + rows: Record<`a` | `b` | `c`, number | undefined> + // The new demand's load, when the history has one. + newLoad?: `pending` | `fulfilled` | `rejected` + status: `error` | `healthy` + } + const expectedReacquisition = ( + history: RotatedReacquisitionHistory, + settled: number, + ): ReacquisitionObservation => { + const fails = (group: `a` | `b`) => + history.fails === group || history.fails === `both` + const allSettled = settled === 2 + const ended = allSettled && !fails(`a`) && !fails(`b`) + // A failed reacquisition is reported when it settles. + const failed = history.order.slice(0, settled).some(fails) + return { + rows: ended + ? { + a: history.noticeBetween ? 3 : 2, + // A released demand's held notice still applies once recovery + // ends; the new generation's rows are trusted then. + b: 2, + c: history.newDemand ? 2 : undefined, + } + : { a: undefined, b: undefined, c: undefined }, + ...(history.newDemand + ? { + newLoad: !allSettled + ? (`pending` as const) + : ended + ? (`fulfilled` as const) + : (`rejected` as const), + } + : {}), + status: failed ? `error` : `healthy`, + } + } + + const assertRotatedReacquisition = async ( + history: RotatedReacquisitionHistory, + ) => { + const errors = vi.spyOn(console, `error`).mockImplementation(() => {}) + const warnings = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const oldStorageId = `reacquire-old` + const newStorageId = `reacquire-new` + const durable = new Map>([ + [oldStorageId, new Map()], + [newStorageId, new Map()], + ]) + let subscriber: ((message: ProtocolEnvelope) => void) | undefined + const seqs = new Map() + const ownerCommits = (storageId: string, row: VersionedRow) => { + const seq = (seqs.get(storageId) ?? 0) + 1 + seqs.set(storageId, seq) + durable.get(storageId)!.set(row.id, row) + const payload: TxCommitted = { + type: `tx:committed`, + term: 1, + seq, + txId: `owner-${storageId}-${seq}`, + latestRowVersion: seq, + requiresFullReload: false, + changedRows: [{ key: row.id, value: row }], + deletedKeys: [], + } + subscriber?.({ + v: 1, + dbName: `test-db`, + collectionId: storageId, + senderId: `owner`, + ts: Date.now(), + payload, + }) + } + type Reply = { + resolve: () => void + reject: (error: unknown) => void + } + const replies = new Map() + let ensures = 0 + let cEnsures = 0 + const groupOf = (options: LoadSubsetOptions) => { + const where = options.where as IR.Func | undefined + const value = where?.args[1] as IR.Value | undefined + return value?.value as `a` | `b` | `c` | undefined + } + const duringReads = history.trigger === `claim-lost-during-reads` + const firstReads = createDeferred() + let reads = 0 + let claimLost = false + const adapter = { + loadSubset: async ( + target: PersistedStorageTarget, + ): Promise> => { + if (duringReads && ++reads <= 2) { + await firstReads.promise + claimLost = true + } + return Array.from( + durable.get(recordedStorageId(target))!.values(), + (value) => ({ key: value.id, value: { ...value } }), + ) + }, + claimCacheGeneration: async () => ({ + storageCollectionId: oldStorageId, + claimId: `reacquire-claim`, + expiresAtMs: Date.now() + 60_000, + }), + renewCacheGenerationClaim: async () => + claimLost ? undefined : Date.now() + 60_000, + rotateCacheGeneration: async () => { + claimLost = false + return { + storageCollectionId: newStorageId, + claimId: `reacquire-claim-2`, + expiresAtMs: Date.now() + 60_000, + } + }, + releaseCacheGenerationClaim: async () => {}, + loadResumeSnapshot: () => + Promise.resolve({ + rows: [], + keySet: { status: `consistent` }, + collectionMetadata: [], + latestTerm: 0, + latestSeq: 0, + latestRowVersion: 0, + resetEpoch: 0, + }), + applyCommittedTx: async () => {}, + ensureIndex: async () => {}, + } as unknown as PersistenceAdapter + let leader = false + const coordinator = { + getNodeId: () => `follower`, + subscribe: ( + _collectionId: string, + onMessage: (message: ProtocolEnvelope) => void, + ) => { + subscriber = onMessage + return () => { + subscriber = undefined + } + }, + publish: () => {}, + isLeader: () => leader, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + // The owner commits the demand's row into the requesting generation + // and broadcasts it before it replies. In the new generation the + // driver releases the reply; a later refetch finds nothing new. + requestEnsureRemoteSubset: ( + storageId: string, + options: LoadSubsetOptions, + ) => { + ensures++ + const group = groupOf(options) + if (!group) return Promise.resolve() + if (storageId === oldStorageId) { + ownerCommits(oldStorageId, { id: `${group}-row`, group, v: 1 }) + return Promise.resolve() + } + if (group === `c`) cEnsures++ + if (replies.has(group) || (group === `c` && cEnsures > 1)) + return Promise.resolve() + ownerCommits(newStorageId, { id: `${group}-row`, group, v: 2 }) + // The new demand's ensure answers at once; the follower must still + // hold its rows until reacquisition ends. + if (group === `c`) return Promise.resolve() + return new Promise((resolve, reject) => { + replies.set(group, { resolve, reject }) + }) + }, + requestReleaseRemoteSubset: async () => {}, + registerRemoteSubsetOwner: () => () => {}, + requestApplyCommittedTx: (_collectionId: string, tx: TxCommitted) => + Promise.resolve({ + type: `rpc:applyCommittedTx:res`, + rpcId: tx.txId, + ok: true, + term: tx.term, + seq: tx.seq, + latestRowVersion: tx.seq, + }), + pullSince: (storageId: string) => + Promise.resolve({ + type: `rpc:pullSince:res`, + rpcId: `reacquire-pull`, + ok: true, + latestTerm: 1, + latestSeq: seqs.get(storageId) ?? 0, + latestRowVersion: seqs.get(storageId) ?? 0, + requiresFullReload: true, + changedKeys: [], + deletedKeys: [], + }), + } as unknown as PersistedCollectionCoordinator + let sourceParams: SyncConfig[`sync`] extends ( + params: infer P, + ) => unknown + ? P + : never + const source = createCollection( + persistedCollectionOptions({ + id: `follower-reacquire-${Math.random().toString(36).slice(2)}`, + getKey: (row) => row.id, + persistence: { adapter, coordinator }, + syncMode: `on-demand`, + sync: { + sync: (params) => { + sourceParams = params + return { + loadSubset: () => Promise.resolve(), + restartAfterScopedRecovery: () => {}, + } + }, + }, + }), + ) + source.startSyncImmediate() + let newLoad: `pending` | `fulfilled` | `rejected` = `pending` + const observe = (): ReacquisitionObservation => ({ + rows: { + a: source.get(`a-row`)?.v, + b: source.get(`b-row`)?.v, + c: source.get(`c-row`)?.v, + }, + ...(history.newDemand ? { newLoad } : {}), + status: source.status === `error` ? `error` : `healthy`, + }) + const turn = () => new Promise((resolve) => setTimeout(resolve, 0)) + const settle = async () => { + for (let i = 0; i < 8; i++) await turn() + } + const demand = (group: `a` | `b` | `c`): LoadSubsetOptions => ({ + where: new IR.Func(`eq`, [ + new IR.PropRef([`group`]), + new IR.Value(group), + ]), + }) + const demands = { a: demand(`a`), b: demand(`b`), c: demand(`c`) } + const load = (group: `a` | `b` | `c`) => + source._sync.loadSubset(demands[group]) + try { + if (duringReads) { + // Both first reads are in flight when the claim is found lost. + void Promise.resolve(load(`a`)).catch(() => undefined) + void Promise.resolve(load(`b`)).catch(() => undefined) + await settle() + firstReads.resolve() + } else { + for (const group of [`a`, `b`] as const) await load(group) + await settle() + expect(observe().rows).toEqual({ a: 1, b: 1 }) + void sourceParams!.metadata!.persistence!.startScopedRecovery!() + } + await vi.waitFor(() => expect(replies.size).toBe(2), { timeout: 2_000 }) + await settle() + if (history.newDemand) { + void Promise.resolve(load(`c`)).then( + () => (newLoad = `fulfilled`), + () => (newLoad = `rejected`), + ) + } + if (history.fails === `released`) { + // Releasing the same demand twice must be idempotent. + source._sync.unloadSubset(demands.b) + source._sync.unloadSubset(demands.b) + } + await settle() + // Checkpoint: both reacquisitions are in flight. + expect(observe()).toEqual(expectedReacquisition(history, 0)) + if (history.supersededAsLeader) { + leader = true + await sourceParams!.metadata!.persistence!.startScopedRecovery!() + await settle() + leader = false + let outcome: `pending` | `fulfilled` | `rejected` = `pending` + void Promise.resolve(load(`c`)).then( + () => (outcome = `fulfilled`), + () => (outcome = `rejected`), + ) + await settle() + // Checkpoint: the superseded reacquisition no longer holds notices. + // The node stays in scoped recovery with nothing reacquiring, so the + // owner's notice for c is handled by one refetch instead of waiting + // in the queue forever; the untrusted generation shows no row. + expect({ outcome, cEnsures, c: source.get(`c-row`)?.v }).toEqual({ + outcome: `fulfilled`, + cEnsures: 2, + c: undefined, + }) + for (const reply of replies.values()) reply.resolve() + return + } + for (const [index, group] of history.order.entries()) { + if (index === 1 && history.noticeBetween) { + ownerCommits(newStorageId, { id: `a-row`, group: `a`, v: 3 }) + await settle() + } + if (history.fails === `released` && group === `b`) { + const aborted = new Error(`released`) + aborted.name = `AbortError` + replies.get(group)!.reject(aborted) + } else if (history.fails === group || history.fails === `both`) + replies.get(group)!.reject(new Error(`reacquire ${group} failed`)) + else replies.get(group)!.resolve() + await settle() + // Checkpoint: after each settlement. + expect(observe()).toEqual(expectedReacquisition(history, index + 1)) + } + // Two acquisitions and two reacquisitions, a new demand's ensure, plus + // one refresh per held notice when recovery stays; a refetch loop + // exceeds the bound. + expect(ensures).toBeLessThanOrEqual(16) + } finally { + await source.cleanup() + errors.mockRestore() + warnings.mockRestore() + } + } + + const rotatedReacquisitionHistories: Array = ( + [ + [`a`, `b`], + [`b`, `a`], + ] as const + ).flatMap((order) => + [false, true].flatMap((noticeBetween) => + ([`none`, `a`, `b`, `both`, `released`] as const).flatMap((fails) => + [false, true].map((newDemand) => ({ + order: [...order], + noticeBetween, + fails, + newDemand, + })), + ), + ), + ) + // Known gap: see #2139. Recovery never ends today, so the histories in + // which it should end, or in which a new demand must wait for it, fail. + const reacquisitionGap = ({ + fails, + newDemand, + }: RotatedReacquisitionHistory) => + newDemand === true || fails === `none` || fails === `released` + it.each( + rotatedReacquisitionHistories.filter( + (history) => !reacquisitionGap(history), + ), + )( + `ends a follower's scoped recovery only after every reacquisition succeeds: order $order, notice between $noticeBetween, fails $fails, new demand $newDemand`, + assertRotatedReacquisition, + 10_000, + ) + it.fails.each(rotatedReacquisitionHistories.filter(reacquisitionGap))( + `ends a follower's scoped recovery only after every reacquisition succeeds: order $order, notice between $noticeBetween, fails $fails, new demand $newDemand`, + assertRotatedReacquisition, + 10_000, + ) + it( + `releases held notices when a second rotation supersedes reacquisition as leader`, + () => + assertRotatedReacquisition({ + order: [`a`, `b`], + noticeBetween: false, + fails: `none`, + supersededAsLeader: true, + }), + 10_000, + ) + // Two hydrations that both find the claim lost queue work while the + // follower reacquires. Their flushes must leave the held notices alone; + // otherwise each notice refreshes the scoped demand and the follower spins. + it( + `holds a rotated follower's notices when two reads find the claim lost: fails 'a'`, + () => + assertRotatedReacquisition({ + order: [`a`, `b`], + noticeBetween: false, + fails: `a`, + trigger: `claim-lost-during-reads`, + }), + 10_000, + ) + // Known gap: see #2139. Recovery never ends, so the rows stay hidden. + it.fails( + `holds a rotated follower's notices when two reads find the claim lost: fails 'none'`, + () => + assertRotatedReacquisition({ + order: [`a`, `b`], + noticeBetween: false, + fails: `none`, + trigger: `claim-lost-during-reads`, + }), + 10_000, + ) + + // The same law holds when no demand is active as recovery starts: with + // nothing reacquired and no new generation, a later load must still not + // hydrate the untrusted rows. + it.each([`active demand`, `no active demand`] as const)( + `does not trust untrusted durable rows after a recovery without rotation: %s`, + async (activeAtRecovery) => { + const errors = vi.spyOn(console, `error`).mockImplementation(() => {}) + const warnings = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const durable: Array = [{ id: `stale`, group: `a` }] + let subscriber: ((message: ProtocolEnvelope) => void) | undefined + let seq = 0 + const notice = (collectionId: string, value: FollowerRow) => { + seq++ + if (!durable.some((row) => row.id === value.id)) durable.push(value) + const payload: TxCommitted = { + type: `tx:committed`, + term: 1, + seq, + txId: `owner-tx-${seq}`, + latestRowVersion: seq, + requiresFullReload: false, + changedRows: [{ key: value.id, value }], + deletedKeys: [], + } + subscriber?.({ + v: 1, + dbName: `test-db`, + collectionId, + senderId: `owner`, + ts: Date.now(), + payload, + }) + } + let ensures = 0 + const adapter = { + loadSubset: async () => + durable.map((value) => ({ key: value.id, value })), + loadResumeSnapshot: () => + Promise.resolve({ + rows: [], + keySet: { status: `consistent` }, + collectionMetadata: [], + latestTerm: 0, + latestSeq: 0, + latestRowVersion: 0, + resetEpoch: 0, + }), + applyCommittedTx: async () => {}, + ensureIndex: async () => {}, + } as unknown as PersistenceAdapter + const coordinator = { + getNodeId: () => `follower`, + subscribe: ( + _collectionId: string, + onMessage: (message: ProtocolEnvelope) => void, + ) => { + subscriber = onMessage + return () => { + subscriber = undefined + } + }, + publish: () => {}, + isLeader: () => false, + ensureLeadership: async () => {}, + requestEnsurePersistedIndex: async () => {}, + requestEnsureRemoteSubset: async (collectionId: string) => { + if (++ensures === 1) + notice(collectionId, { id: `remote`, group: `a` }) + }, + requestReleaseRemoteSubset: async () => {}, + registerRemoteSubsetOwner: () => () => {}, + requestApplyCommittedTx: (_collectionId: string, tx: TxCommitted) => + Promise.resolve({ + type: `rpc:applyCommittedTx:res`, + rpcId: tx.txId, + ok: true, + term: tx.term, + seq: tx.seq, + latestRowVersion: seq, + }), + pullSince: () => + Promise.resolve({ + type: `rpc:pullSince:res`, + rpcId: `unrotated-pull`, + ok: true, + latestTerm: 1, + latestSeq: seq, + latestRowVersion: seq, + requiresFullReload: false, + changedKeys: [], + deletedKeys: [], + }), + } as unknown as PersistedCollectionCoordinator + let sourceParams: SyncConfig[`sync`] extends ( + params: infer P, + ) => unknown + ? P + : never + const source = createCollection( + persistedCollectionOptions({ + id: `follower-unrotated-${Math.random().toString(36).slice(2)}`, + getKey: (row) => row.id, + persistence: { adapter, coordinator }, + syncMode: `on-demand`, + sync: { + sync: (params) => { + sourceParams = params + return { + loadSubset: () => Promise.resolve(), + restartAfterScopedRecovery: () => {}, + } + }, + }, + }), + ) + source.startSyncImmediate() + const query = createLiveQueryCollection((q) => + q.from({ row: source }).where(({ row }) => eq(row.group, `a`)), + ) + try { + if (activeAtRecovery === `active demand`) { + await query.preload() + expect(Array.from(source.keys()).sort()).toEqual([`remote`, `stale`]) + } else { + // Start the source without any demand. + void source.preload().catch(() => undefined) + await vi.waitFor(() => expect(sourceParams).toBeDefined()) + } + + // The source reports that the old durable rows are untrusted. + await sourceParams!.metadata!.persistence!.startScopedRecovery!() + await new Promise((resolve) => setTimeout(resolve, 50)) + if (activeAtRecovery === `no active demand`) { + void query.preload().catch(() => undefined) + } + notice(source.id, { id: `fresh`, group: `a` }) + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(Array.from(source.keys())).not.toContain(`stale`) + } finally { + await query.cleanup() + await source.cleanup() + errors.mockRestore() + warnings.mockRestore() + } + }, + ) + + const followerRemoteDemandHistory = fc.record({ + // Known gap: see #2139. The rotated placement is pinned above. + placement: fc.constantFrom( + `direct`, + `queued`, + `expired`, + ), + apply: fc.constantFrom(`ok`, `fails`), + consumer: fc.constantFrom( + `live-query`, + `load`, + ), + }) + fcTest.prop( + [followerRemoteDemandHistory], + oraclePropertyOptions(12, `persistence.follower-remote-demand`), + )( + `checks generated follower remote-demand histories (random or replayed)`, + assertFollowerRemoteDemandHistory, + ) + // The model has one one-shot refresh lease. The grammar crosses cleanup // before and after source settlement. Whichever event wins releases the // lease once. The raw release log retains multiplicity at both checkpoints. diff --git a/packages/db/tests/oracle-config.ts b/packages/db/tests/oracle-config.ts index 5aa174b416..cc9409f295 100644 --- a/packages/db/tests/oracle-config.ts +++ b/packages/db/tests/oracle-config.ts @@ -37,6 +37,8 @@ const staticOracleProperties = [ `sqlite-resume.managed-claim-read`, `sqlite-resume.retired-storage`, `persistence.generation-notice`, + `persistence.follower-generation-notice`, + `persistence.follower-remote-demand`, `sqlite-ordinary.write-history`, `sqlite-ordinary.independent-work`, `collection-sync.reentrant-drain`, diff --git a/packages/electron-db-sqlite-persistence/tests/electron-ipc-readiness-oracle.test.ts b/packages/electron-db-sqlite-persistence/tests/electron-ipc-readiness-oracle.test.ts new file mode 100644 index 0000000000..285669f087 --- /dev/null +++ b/packages/electron-db-sqlite-persistence/tests/electron-ipc-readiness-oracle.test.ts @@ -0,0 +1,623 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { persistedCollectionOptions } from '@tanstack/db-sqlite-persistence-core' +import { createNodeSQLitePersistence } from '@tanstack/node-db-sqlite-persistence' +import { + IR, + createCollection, + createLiveQueryCollection, + eq, + whenSyncAccepted, +} from '../../db/src' +import { createDeferred } from '../../db/src/deferred' +import { BetterSqlite3SQLiteDriver } from '../../node-db-sqlite-persistence/src/node-driver' +import { + createElectronSQLitePersistence, + exposeElectronSQLitePersistence, +} from '../src' +import { ElectronCollectionCoordinator } from '../src/electron-coordinator' +import { + createElectronRuntimeBridgeInvoke, + isElectronFullE2EEnabled, +} from './e2e/electron-process-client' +import type { Collection, LoadSubsetOptions } from '../../db/src' +import type { + ElectronPersistenceInvoke, + ElectronPersistenceRequestEnvelope, + ElectronPersistenceResponseEnvelope, +} from '../src/protocol' + +/** + * # Does the Electron renderer carry initial-query readiness? + * + * ARCHITECTURE.md law 10 makes a settled `loadSubset` the readiness evidence + * for an on-demand source, and law 7 says a successful load settles only after + * its rows are visible. In a renderer, the persisted wrapper hydrates the + * subset through the Electron IPC adapter, which reads SQLite in the main + * process, and then awaits the upstream source's load. The upstream never + * calls `markReady()`, as a progressive source would not during its full + * sync. + * + * The independent model below predicts, for each history, what a live query + * over the demanded rows shows at three checkpoints: while the main process's + * `loadSubset` reply is held, after that reply but before the upstream load + * settles, and after the upstream load settles. It knows only the rows that an + * earlier renderer run persisted and the outcome of each step; it knows + * nothing about the IPC protocol or the wrapper's internals. + * + * The history grammar crosses a fresh or reopened renderer (an earlier run + * persisted one local row) with the IPC hydration outcome (a reply, or a + * failed reply) and what follows a reply: the upstream load succeeds, fails, + * or the query releases its demand while the IPC hydration is in flight. + * + * The driver uses the real renderer adapter and the real main-process + * handler over a real SQLite file. The `ipcMain` double forwards each + * envelope in process and lets the driver hold or fail the `loadSubset` + * reply, which is the only part of the IPC transport that this law depends + * on. With `TANSTACK_DB_ELECTRON_E2E_ALL=1`, each envelope instead crosses a + * real Electron main process through the runtime bridge. The histories cover + * a renderer without a coordinator. A second lane below runs a follower and + * an owner renderer with coordinators. + */ + +type Row = { id: string; group: string } + +// A real Electron process starts for each envelope, so waits are longer. +const settleTimeout = { timeout: isElectronFullE2EEnabled() ? 60_000 : 1_000 } +// Bound each history, so a stalled main process fails one history instead of +// holding the run until the suite's timeout. +const historyTimeout = isElectronFullE2EEnabled() ? 90_000 : 10_000 + +type IpcHandler = ( + event: unknown, + request: ElectronPersistenceRequestEnvelope, +) => Promise + +function mainProcessInvoke(dbPath: string): ElectronPersistenceInvoke { + if (isElectronFullE2EEnabled()) { + return createElectronRuntimeBridgeInvoke({ + dbPath, + collectionId: `electron-readiness`, + allowAnyCollectionId: true, + timeoutMs: 60_000, + }) + } + const driver = new BetterSqlite3SQLiteDriver({ filename: dbPath }) + let handler: IpcHandler | undefined + const dispose = exposeElectronSQLitePersistence({ + ipcMain: { + handle: (_channel: string, listener: IpcHandler) => { + handler = listener + }, + removeHandler: () => {}, + }, + persistence: createNodeSQLitePersistence({ + database: driver.getDatabase(), + }), + }) + onTestFinished(() => { + dispose() + driver.close() + }) + return (_channel, request) => handler!(undefined, request) +} + +function electronMain() { + const directory = mkdtempSync(join(tmpdir(), `electron-readiness-`)) + onTestFinished(() => rmSync(directory, { recursive: true, force: true })) + const invoke = mainProcessInvoke(join(directory, `state.sqlite`)) + + // A held `loadSubset` reply waits for `release`; `failNext` rejects the + // next reply. `arrived` resolves once the main process has answered. + let hold: { arrived: () => void; released: Promise } | undefined + let failNext: Error | undefined + const loadReplies: Array<`sent` | `failed`> = [] + const renderer = createElectronSQLitePersistence({ + timeoutMs: isElectronFullE2EEnabled() ? 90_000 : undefined, + invoke: async (channel, request) => { + const response = await invoke(channel, request) + if (request.method === `loadSubset`) { + if (hold) { + hold.arrived() + await hold.released + } + if (failNext) { + const error = failNext + failNext = undefined + loadReplies.push(`failed`) + throw error + } + loadReplies.push(`sent`) + } + return response + }, + }) + return { + renderer, + loadReplies, + holdLoads: () => { + const arrived = createDeferred() + const released = createDeferred() + hold = { arrived: () => arrived.resolve(), released: released.promise } + return { + arrived: arrived.promise, + release: () => released.resolve(), + } + }, + failNextLoad: (error: Error) => { + failNext = error + }, + } +} + +type Upstream = { + calls: Array + respond: (rows: ReadonlyArray) => void + fail: (error: unknown) => void +} + +function persistedOnDemand( + main: ReturnType, + id = `electron-readiness-${Math.random().toString(36).slice(2)}`, +): [Collection, Upstream] { + let pending = createDeferred() + let write!: (rows: ReadonlyArray) => void + const upstream: Upstream = { + calls: [], + respond: (rows) => { + write(rows) + pending.resolve() + }, + fail: (error) => pending.reject(error), + } + const collection = createCollection( + persistedCollectionOptions({ + id, + getKey: (row) => row.id, + persistence: main.renderer, + syncMode: `on-demand`, + sync: { + sync: (params) => { + write = (rows) => { + params.begin() + for (const value of rows) params.write({ type: `insert`, value }) + params.commit() + } + return { + loadSubset: (options: LoadSubsetOptions) => { + upstream.calls.push(options) + pending = createDeferred() + return pending.promise + }, + } + }, + }, + }), + ) + onTestFinished(() => collection.cleanup()) + return [collection, upstream] +} + +function groupA(source: Collection) { + const query = createLiveQueryCollection((q) => + q.from({ row: source }).where(({ row }) => eq(row.group, `a`)), + ) + onTestFinished(() => query.cleanup()) + return query +} + +function rowsAtFirstReady(query: Collection) { + const seen: { rows?: Array } = {} + query.onFirstReady(() => { + seen.rows = query.toArray.map((row: Row) => row.id).sort() + }) + return seen +} + +function quietLogs() { + const warned = vi.spyOn(console, `warn`).mockImplementation(() => {}) + const loggedError = vi.spyOn(console, `error`).mockImplementation(() => {}) + onTestFinished(() => { + warned.mockRestore() + loggedError.mockRestore() + }) +} + +type IpcReadinessHistory = { + renderer: `fresh` | `reopened` + hydration: `reply` | `fails` + // What follows a hydration reply. + next: `upstream-ok` | `upstream-fails` | `release-during-hydration` +} + +type Checkpoints = { + whileReplyHeld?: { + ready: boolean + rows: Array + upstreamCalls: number + } + afterReply: + | { ready: false; rows: Array; upstreamCalls: 1 } + | { status: `error`; upstreamCalls: 0; firstReadyRows: undefined } + | { releasedQuietly: true } + afterUpstream?: + | { firstReadyRows: Array; sourceReady: false } + | { status: `error`; firstReadyRows: undefined } +} + +// The model: readiness waits for both loads, and each load's rows are visible +// when the query is first ready. A failure of either load errors the query +// before it is ever ready. A demand released while its hydration is in flight +// starts no live upstream load, installs no upstream rows, and causes no +// error. Whether locally hydrated rows stay is retention, which the persisted +// oracle's retained-demand lane owns, so the model does not predict it. +function expectedCheckpoints(history: IpcReadinessHistory): Checkpoints { + const local = history.renderer === `reopened` ? [`local`] : [] + if (history.hydration === `fails`) + return { + afterReply: { + status: `error`, + upstreamCalls: 0, + firstReadyRows: undefined, + }, + } + const whileReplyHeld = { ready: false, rows: [], upstreamCalls: 0 } + if (history.next === `release-during-hydration`) + return { whileReplyHeld, afterReply: { releasedQuietly: true } } + return { + whileReplyHeld, + afterReply: { ready: false, rows: local, upstreamCalls: 1 }, + afterUpstream: + history.next === `upstream-ok` + ? { firstReadyRows: [...local, `remote`].sort(), sourceReady: false } + : { status: `error`, firstReadyRows: undefined }, + } +} + +const ipcReadinessHistories: Array = ( + [`fresh`, `reopened`] as const +).flatMap((renderer) => [ + { renderer, hydration: `fails` as const, next: `upstream-ok` as const }, + ...( + [`upstream-ok`, `upstream-fails`, `release-during-hydration`] as const + ).map((next) => ({ renderer, hydration: `reply` as const, next })), +]) + +async function observeIpcReadinessHistory( + history: IpcReadinessHistory, +): Promise { + quietLogs() + const main = electronMain() + const id = `electron-readiness-${history.renderer}-${Math.random().toString(36).slice(2)}` + if (history.renderer === `reopened`) { + // An earlier renderer run persists one local row. + const [source, upstream] = persistedOnDemand(main, id) + const query = groupA(source) + const preload = query.preload() + await vi.waitFor( + () => expect(upstream.calls).toHaveLength(1), + settleTimeout, + ) + upstream.respond([{ id: `local`, group: `a` }]) + await preload + await query.cleanup() + await source.cleanup() + } + + const observed: Checkpoints = { afterReply: { releasedQuietly: true } } + if (history.hydration === `fails`) { + main.failNextLoad(new Error(`main process unavailable`)) + const [source, upstream] = persistedOnDemand(main, id) + const query = groupA(source) + const seen = rowsAtFirstReady(query) + void query.preload().catch(() => {}) + await vi.waitFor( + () => expect(main.loadReplies).toContain(`failed`), + settleTimeout, + ) + await vi.waitFor(() => expect(query.status).toBe(`error`), settleTimeout) + observed.afterReply = { + status: query.status as `error`, + upstreamCalls: upstream.calls.length as 0, + firstReadyRows: seen.rows as undefined, + } + return observed + } + + const held = main.holdLoads() + const [source, upstream] = persistedOnDemand(main, id) + const query = groupA(source) + const seen = rowsAtFirstReady(query) + const preload = query.preload() + void preload.catch(() => {}) + await held.arrived + observed.whileReplyHeld = { + ready: query.isReady(), + rows: query.toArray.map((row: Row) => row.id).sort(), + upstreamCalls: upstream.calls.length, + } + + if (history.next === `release-during-hydration`) { + await query.cleanup() + held.release() + await vi.waitFor( + () => expect(main.loadReplies).toContain(`sent`), + settleTimeout, + ) + await new Promise((resolve) => setTimeout(resolve, 20)) + observed.afterReply = { + releasedQuietly: (upstream.calls.every( + (options) => options.signal?.aborted, + ) && + !source.has(`remote`) && + source.status !== `error`) as true, + } + return observed + } + + held.release() + await vi.waitFor( + () => expect(upstream.calls.length).toBeGreaterThan(0), + settleTimeout, + ) + observed.afterReply = { + ready: query.isReady() as false, + rows: query.toArray.map((row: Row) => row.id).sort(), + upstreamCalls: upstream.calls.length as 1, + } + + if (history.next === `upstream-ok`) { + upstream.respond([{ id: `remote`, group: `a` }]) + await preload + observed.afterUpstream = { + firstReadyRows: seen.rows as Array, + sourceReady: source.isReady() as false, + } + } else { + upstream.fail(new Error(`upstream failed`)) + await vi.waitFor(() => expect(query.status).toBe(`error`), settleTimeout) + observed.afterUpstream = { + status: query.status as `error`, + firstReadyRows: seen.rows as undefined, + } + } + return observed +} + +describe(`Electron renderer initial readiness follows the composed load`, () => { + it.each(ipcReadinessHistories)( + `matches the model: $renderer renderer, hydration $hydration, then $next`, + async (history) => { + expect(await observeIpcReadinessHistory(history)).toEqual( + expectedCheckpoints(history), + ) + }, + historyTimeout, + ) +}) + +/** + * # Does a follower renderer settle with its owner's rows? + * + * Two renderers share one main process and SQLite file, each with its own + * ElectronCollectionCoordinator over Node's BroadcastChannel and Web Locks. + * The first renderer owns the Collection and runs the source; the second is a + * follower and routes its demand to the owner. Law 7 owes the follower: its + * load, and its query's first ready, settle only once the owner's rows are + * visible in the follower, and they reject, with the follower source in + * error, if applying the owner's commit fails here. + * + * The model knows only whether the owner's commit applies in the follower. + * The grammar crosses that outcome with the consumer: a live query over the + * demanded row, or the follower's load. The driver holds the follower's IPC + * `loadSubset` reply that carries the owner's row, which is the follower's + * read of the owner's commit, and fails it for the `fails` histories. The + * checkpoint is after the owner's reply reached the follower while that read + * is held: the consumer has not settled. The browser per-collection oracle + * owns the queued placement and the Query Collection source across real tabs. + */ +type CoordinatedRendererHistory = { + apply: `ok` | `fails` + consumer: `live-query` | `load` +} + +function expectedCoordinatedRenderer(history: CoordinatedRendererHistory) { + const applied = history.apply === `ok` + return { + settledWhileApplyHeld: false, + outcome: applied ? `fulfilled` : `rejected`, + rows: applied ? [`remote`] : undefined, + source: applied ? `ready` : `error`, + } +} + +function coordinatedRenderer(options: { + invoke: ElectronPersistenceInvoke + dbName: string + collectionId: string + gateRead?: (response: unknown) => Promise +}) { + const coordinator = new ElectronCollectionCoordinator({ + dbName: options.dbName, + }) + onTestFinished(() => coordinator.dispose()) + const trace: Array = [] + const ensure = coordinator.requestEnsureRemoteSubset.bind(coordinator) + coordinator.requestEnsureRemoteSubset = (collectionId, subset) => { + trace.push(`ensure`) + return ensure(collectionId, subset).finally(() => + trace.push(`ensure-settled`), + ) + } + // Record when the owner's commit notice reaches this renderer, so the lane + // can check the premise it relies on: the notice precedes the reply. + const subscribe = coordinator.subscribe.bind(coordinator) + coordinator.subscribe = (collectionId, onMessage) => + subscribe(collectionId, (message) => { + const payload = message.payload as { type?: string } | undefined + if (payload?.type === `tx:committed`) trace.push(`broadcast`) + onMessage(message) + }) + const persistence = createElectronSQLitePersistence({ + timeoutMs: isElectronFullE2EEnabled() ? 90_000 : undefined, + coordinator, + invoke: async (channel, request) => { + const response = await options.invoke(channel, request) + if (request.method === `loadSubset`) await options.gateRead?.(response) + return response + }, + }) + const delivered = new Set() + const collection = createCollection( + persistedCollectionOptions({ + id: options.collectionId, + getKey: (row) => row.id, + persistence, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + markReady() + return { + loadSubset: async () => { + // Deliver the owner's upstream row once, as a real source does. + if (delivered.has(`remote`)) return + delivered.add(`remote`) + trace.push(`upstream`) + begin() + write({ type: `insert`, value: { id: `remote`, group: `a` } }) + await whenSyncAccepted(commit()) + }, + } + }, + }, + }), + ) + onTestFinished(() => collection.cleanup()) + return { coordinator, collection, trace } +} + +async function observeCoordinatedRendererHistory( + history: CoordinatedRendererHistory, +) { + quietLogs() + const directory = mkdtempSync(join(tmpdir(), `electron-follower-`)) + onTestFinished(() => rmSync(directory, { recursive: true, force: true })) + const invoke = mainProcessInvoke(join(directory, `state.sqlite`)) + const dbName = `electron-follower-${Math.random().toString(36).slice(2)}` + const collectionId = `electron-follower-items` + const owner = coordinatedRenderer({ invoke, dbName, collectionId }) + owner.collection.startSyncImmediate() + await owner.collection.stateWhenReady() + + const applyHeld = createDeferred() + let applyReached = false + let gate = false + const follower = coordinatedRenderer({ + invoke, + dbName, + collectionId, + gateRead: async (response) => { + if (!gate || !JSON.stringify(response).includes(`remote`)) return + gate = false + applyReached = true + await applyHeld.promise + if (history.apply === `fails`) throw new Error(`follower read failed`) + }, + }) + onTestFinished(() => applyHeld.resolve()) + follower.collection.startSyncImmediate() + await follower.collection.stateWhenReady() + gate = true + + let settledAt: { rows: Array } | undefined + let outcome: Promise<`fulfilled` | `rejected`> + if (history.consumer === `live-query`) { + const query = createLiveQueryCollection((q) => + q + .from({ row: follower.collection }) + .where(({ row }) => eq(row.id, `remote`)), + ) + onTestFinished(() => query.cleanup()) + query.onFirstReady(() => { + settledAt = { rows: query.toArray.map((row: Row) => row.id).sort() } + }) + outcome = query.preload().then( + () => `fulfilled` as const, + () => `rejected` as const, + ) + } else { + outcome = Promise.resolve( + follower.collection._sync.loadSubset({ + where: new IR.Func(`eq`, [ + new IR.PropRef([`id`]), + new IR.Value(`remote`), + ]), + }), + ).then( + () => { + settledAt = { + rows: follower.collection.has(`remote`) ? [`remote`] : [], + } + return `fulfilled` as const + }, + () => `rejected` as const, + ) + } + + await vi.waitFor( + () => { + expect(applyReached).toBe(true) + expect(follower.trace).toContain(`ensure-settled`) + }, + { timeout: isElectronFullE2EEnabled() ? 60_000 : 8_000 }, + ) + for (let turn = 0; turn < 20; turn++) + await new Promise((resolve) => setImmediate(resolve)) + // Checkpoint: the owner has replied while the follower's apply is held. + // The premise transferred: its notice reached the follower before the reply. + expect(follower.trace.indexOf(`broadcast`)).toBeGreaterThan(-1) + expect(follower.trace.indexOf(`broadcast`)).toBeLessThan( + follower.trace.indexOf(`ensure-settled`), + ) + const settledWhileApplyHeld = settledAt !== undefined + // The topology this lane claims: the owner loads upstream, and the + // follower routes its demand instead of running its own source. + expect(owner.trace).toContain(`upstream`) + expect(follower.trace).not.toContain(`upstream`) + applyHeld.resolve() + const settled = await outcome + await vi.waitFor( + () => + expect(follower.collection.status).toBe( + expectedCoordinatedRenderer(history).source, + ), + settleTimeout, + ) + return { + settledWhileApplyHeld, + outcome: settled, + rows: settledAt?.rows, + source: follower.collection.status, + } +} + +describe(`Electron follower renderer settles with its owner's rows`, () => { + it.each( + ([`ok`, `fails`] as const).flatMap((apply) => + ([`live-query`, `load`] as const).map((consumer) => ({ + apply, + consumer, + })), + ), + )( + `matches the model: apply $apply, consumer $consumer`, + async (history) => { + expect(await observeCoordinatedRendererHistory(history)).toEqual( + expectedCoordinatedRenderer(history), + ) + }, + isElectronFullE2EEnabled() ? 180_000 : 20_000, + ) +})