From 96b93541e5c6f41d7366eaaa75e1aac5ca7b8792 Mon Sep 17 00:00:00 2001 From: Jose Diego Menenendez del Cueto Date: Tue, 20 Oct 2026 11:18:32 +0200 Subject: [PATCH 1/2] feat(sqlite-persistence): allow progressive initial-render fallback --- .../progressive-persisted-initial-render.md | 5 + docs/contributing/oracle-coverage.md | 12 ++ docs/guides/sqlite-persistence.md | 36 ++++- .../src/persisted.ts | 14 +- .../tests/persisted-readiness-oracle.test.ts | 109 +++++++++----- .../electric-resume-snapshot-races.test.ts | 135 +++++++++++++++++- 6 files changed, 268 insertions(+), 43 deletions(-) create mode 100644 .changeset/progressive-persisted-initial-render.md diff --git a/.changeset/progressive-persisted-initial-render.md b/.changeset/progressive-persisted-initial-render.md new file mode 100644 index 0000000000..23d82b0eb2 --- /dev/null +++ b/.changeset/progressive-persisted-initial-render.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db-sqlite-persistence-core': minor +--- + +Allow progressive sources to opt into persisted initial rendering and restore their local baseline before reporting persisted readiness. diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index 2d3753e74e..a3e2e9d606 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -2472,6 +2472,18 @@ network provider; it does not prove browser scheduling or arbitrary histories. The framework identity scenario does not cover every possible rejection value or an error already present before a hook mounts. +The progressive opt-in extends the empty/nonempty controlled restore histories to +on-demand core sources with a held subset acquisition. The same local-restoration +rule applies; this does not certify a previously completed network snapshot. +The receiving cases in `electric-resume-snapshot-races.test.ts` exercise the +Electric adapter with real Node SQLite managed caches and a controlled SDK +stream, both with and without a saved resume marker. They check cached rows +before network delivery and replacement versus resumed updates afterward. +These cases do not establish OPFS scheduling, multi-tab coordination, cache +rotation during initial fallback, or progressive-specific joined/restart histories. +Those remain outside this draft's validation and need receiving coverage before +a broader guarantee can be made. + ## Deferred contracts and evidence The maintainer assigned offline policy work to diff --git a/docs/guides/sqlite-persistence.md b/docs/guides/sqlite-persistence.md index ca0ccf2cc5..afadea097c 100644 --- a/docs/guides/sqlite-persistence.md +++ b/docs/guides/sqlite-persistence.md @@ -236,7 +236,7 @@ An on-demand cache is not yet a certified offline snapshot. The first managed ge ## Render after SQLite restore -`initialRender` requires an eager persisted Collection. It does not support `syncMode: 'on-demand'`, which has no complete startup snapshot. Set it on every source Collection in a live query. If one source does not opt in, the query cannot report persisted readiness or use the restored snapshot as its initial-render fallback. +`initialRender` supports eager persisted Collections and explicitly opted-in progressive sources. Genuine `syncMode: 'on-demand'` sources remain unsupported. Set it on every source Collection in a live query. If one source does not opt in, the query cannot report persisted readiness or use the restored snapshot as its initial-render fallback. Install the React binding to use the following components: @@ -244,6 +244,38 @@ Install the React binding to use the following components: npm install @tanstack/react-db ``` +Electric progressive sources are represented as on-demand Collections internally. To restore their entire local cache before initial rendering, opt in explicitly: + +```ts +persistedCollectionOptions({ + ...electricCollectionOptions({ + id: 'todos', + syncMode: 'progressive', + getKey: (todo: Todo) => todo.id, + shapeOptions: { + url: 'http://localhost:3000/v1/shape', + params: { table: 'todos' }, + }, + }), + persistence: syncedPersistence, + initialRender: { + strategy: 'network-first', + networkTimeoutMs: 0, + progressive: true, + }, +}) +``` + +Import `electricCollectionOptions` from `@tanstack/electric-db-collection`. +Use `progressive: true` only with progressive sources. The wrapper trusts this +assertion; it cannot distinguish them from genuine on-demand sources. +Restoration loads every cached row into memory. Persisted readiness means that +local restoration finished, not that the previous full network sync completed. +An interrupted initial sync can therefore restore only previously cached subsets, +and a new cache can render empty. Network sync continues and replaces that +provisional cache with the full snapshot. Applications requiring a complete +server snapshot should continue to wait for Collection readiness. + A regular React live query exposes both readiness signals: ```tsx @@ -266,7 +298,7 @@ function TodosView() { } ``` -`isReady` reports Collection readiness. `isPersistedReady` reports that every opted-in source finished its current eager restore, even when SQLite had no rows. These flags do not identify where each visible row came from. A regular `useLiveQuery` does not wait for the network-first deadline before it renders rows. +`isReady` reports Collection readiness. `isPersistedReady` reports that every opted-in source finished its current local restore, even when SQLite had no rows. These flags do not identify where each visible row came from. A regular `useLiveQuery` does not wait for the network-first deadline before it renders rows. The hooks also expose `persistedStatus` and `persistedError`. The status is `unavailable` if a source did not opt in, `loading` during restore, `ready` after every restore completes, or `error` if a restore fails. `persistedError` keeps the restore error. An SSR seed does not count as a SQLite restore. diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 4b3b1ece7e..744f1c55e4 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -823,6 +823,12 @@ export type PersistedInitialRenderOptions = { strategy: `network-first` /** Time to prefer network before restored rows may render; defaults to 3s. */ networkTimeoutMs?: number + /** + * Restore the full local cache for a progressive source represented as + * on-demand. Do not enable for genuinely on-demand sources. This certifies + * local restoration only, not completion of a previous network snapshot. + */ + progressive?: boolean } const DEFAULT_NETWORK_TIMEOUT_MS = 3_000 @@ -2604,7 +2610,7 @@ class PersistedCollectionRuntime< onStartupMetadataLoaded() } const appliedCursor = - this.syncMode !== `on-demand` + this.syncMode !== `on-demand` || this.persistedReadiness !== undefined ? await this.hydrateBaseline(lifecycleGeneration, adapter) : undefined return lifecycleGeneration === this.lifecycleGeneration @@ -6830,7 +6836,11 @@ export function persistedCollectionOptions< } const networkTimeoutMs = getNetworkTimeoutMs(options.initialRender) - if (networkTimeoutMs !== undefined && options.syncMode === `on-demand`) { + if ( + networkTimeoutMs !== undefined && + options.syncMode === `on-demand` && + !options.initialRender?.progressive + ) { throw new InvalidPersistedCollectionConfigError( `network-first initial rendering requires eager persistence syncMode`, ) diff --git a/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts b/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts index aa8633468b..02dc07e01b 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts @@ -45,13 +45,18 @@ function deferred() { function recordingAdapter(rows: ReadonlyArray) { const load = deferred() + const readStarted = vi.fn() const adapter: PersistenceAdapter = { loadSubset: async () => { + readStarted() await load.promise return rows.map((value) => ({ key: value.id, value })) }, loadResumeSnapshot: async (_id, options) => { - if (options?.includeRows !== false) await load.promise + if (options?.includeRows !== false) { + readStarted() + await load.promise + } return { rows: options?.includeRows === false @@ -68,7 +73,7 @@ function recordingAdapter(rows: ReadonlyArray) { applyCommittedTx: async () => {}, ensureIndex: async () => {}, } - return { adapter, load } + return { adapter, load, readStarted } } function source( @@ -76,23 +81,29 @@ function source( adapter: PersistenceAdapter, optIn: boolean, networkTimeoutMs = 0, + progressive = false, ) { return createCollection( persistedCollectionOptions({ id, getKey: (row) => row.id, persistence: { adapter }, + ...(progressive ? { syncMode: `on-demand` as const } : {}), ...(optIn ? { initialRender: { strategy: `network-first` as const, networkTimeoutMs, + ...(progressive ? { progressive: true } : {}), }, } : {}), // Upstream never declares Collection readiness. Local restore is a // separately observable boundary even while this stays loading. - sync: { sync: () => ({}) }, + sync: { + sync: () => + progressive ? { loadSubset: () => new Promise(() => {}) } : {}, + }, }), ) } @@ -141,8 +152,11 @@ async function checkWithCleanup( * # When may a query render its restored data? * * The independent status rule is an all-source conjunction: a query is persisted-ready - * only after every eager persisted source opted in and successfully completed - * its restore in the current sync run. A source that opts out makes the + * only after every eligible persisted source opted in and successfully completed + * its restore in the current sync run. This draft extends eligibility to + * explicitly opted-in progressive sources, which core sees as on-demand. + * This means the cached rows were restored, not that a prior full network + * snapshot completed. The empty/nonempty histories cover both modes. A source that opts out makes the * query unavailable for this signal. A failed restore reports that exact * failure. Neither an empty persisted database nor a pending network source * changes the rule. Initial rendering prefers upstream-signaled Collection @@ -183,6 +197,11 @@ async function checkWithCleanup( * another model formulation. ORC-012 records the bounded contract, driver, * wrong-answer checks, and uncovered histories here and in the coverage map. * It does not claim all network histories. + * Progressive calibration: upstream rejects both new source-mode cases at + * configuration. With only that rejection removed, omitting startup baseline + * hydration fails the pre-query read checkpoint for both empty and nonempty + * caches. Query-triggered subset reads previously masked this omission; the + * pre-query checkpoint distinguishes startup restoration from demand loading. * ORC-013: the joined half-ready snapshot rejects any-one-ready; the empty * restore rejects row-count readiness; the 99/100-ms deadline and 50/100-ms * joined deadlines reject early fallback; the restart witness rejects stale @@ -294,40 +313,54 @@ describe(`persisted-readiness oracle`, () => { expect(caseId).toBe(36) }) - for (const rows of [[], [{ id: `one`, value: `restored` }]]) { - it(`waits for ${rows.length ? `nonempty` : `empty`} local restore without waiting for upstream`, async () => { - const fixture = recordingAdapter(rows) - const persisted = source(`local-${rows.length}`, fixture.adapter, true) - const query = createLiveQueryCollection({ - query: (q) => q.from({ row: persisted }), - }) - const observer = createLiveQueryObserver(query) - const observedPersistedStatuses: Array = [] - const unsubscribe = observer.subscribe(() => { - observedPersistedStatuses.push(observer.getSnapshot().persistedStatus) - }) - await checkWithCleanup(async () => { - expect(observer.getSnapshot().persistedStatus).toBe(`loading`) - expect(observer.getSnapshot().isPersistedReady).toBe(false) - - fixture.load.resolve() - await vi.waitFor(() => { - const snapshot = observer.getSnapshot() - expect(snapshot.persistedStatus).toBe(`ready`) - expect(snapshot.isPersistedReady).toBe(true) - expect(snapshot.status).toBe(`loading`) - expect(snapshot.data).toMatchObject(rows) + // Both source modes owe the same restore-before-fallback rule. The row + // count dimension distinguishes a restored empty cache from a pending read. + for (const progressive of [false, true]) { + for (const rows of [[], [{ id: `one`, value: `restored` }]]) { + it(`waits for ${progressive ? `progressive` : `eager`} ${rows.length ? `nonempty` : `empty`} local restore without waiting for upstream`, async () => { + const fixture = recordingAdapter(rows) + const persisted = source( + `local-${progressive}-${rows.length}`, + fixture.adapter, + true, + 0, + progressive, + ) + // Startup itself must restore the baseline, even before query demand. + // Otherwise a subset read can mask a missing startup hydration step. + persisted.startSyncImmediate() + await vi.waitFor(() => expect(fixture.readStarted).toHaveBeenCalled()) + const query = createLiveQueryCollection({ + query: (q) => q.from({ row: persisted }), }) - await observer.preloadForInitialRender() - expect(observer.getSnapshot().data).toMatchObject(rows) - expect(observedPersistedStatuses).toContain(`ready`) - }, [ - unsubscribe, - () => observer.dispose(), - () => query.cleanup(), - () => persisted.cleanup(), - ]) - }) + const observer = createLiveQueryObserver(query) + const observedPersistedStatuses: Array = [] + const unsubscribe = observer.subscribe(() => { + observedPersistedStatuses.push(observer.getSnapshot().persistedStatus) + }) + await checkWithCleanup(async () => { + expect(observer.getSnapshot().persistedStatus).toBe(`loading`) + expect(observer.getSnapshot().isPersistedReady).toBe(false) + + fixture.load.resolve() + await vi.waitFor(() => { + const snapshot = observer.getSnapshot() + expect(snapshot.persistedStatus).toBe(`ready`) + expect(snapshot.isPersistedReady).toBe(true) + expect(snapshot.status).toBe(`loading`) + expect(snapshot.data).toMatchObject(rows) + }) + await observer.preloadForInitialRender() + expect(observer.getSnapshot().data).toMatchObject(rows) + expect(observedPersistedStatuses).toContain(`ready`) + }, [ + unsubscribe, + () => observer.dispose(), + () => query.cleanup(), + () => persisted.cleanup(), + ]) + }) + } } it(`requires every joined source to opt in and finish its own restore`, async () => { diff --git a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts index 4d8174da91..90e3b7cd22 100644 --- a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts +++ b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts @@ -1,6 +1,11 @@ import { DatabaseSync } from 'node:sqlite' import { beforeEach, describe, expect, it, vi } from 'vitest' -import { IR, createCollection } from '@tanstack/db' +import { + IR, + createCollection, + createLiveQueryCollection, + createLiveQueryObserver, +} from '@tanstack/db' import { ShapeStream } from '@electric-sql/client' import { SQLiteCorePersistenceAdapter, @@ -975,6 +980,134 @@ async function observeLegacyUnknownResume(): Promise { + it.each([false, true])( + `renders the progressive SQLite cache before network delivery (resume marker: %s)`, + async (hasResumeMarker) => { + // This receiving witness uses real SQLite and the Electric adapter with a + // controlled ShapeStream. A missing marker deliberately represents cached + // subset rows from an interrupted initial sync, not a complete snapshot. + const database = new DatabaseSync(`:memory:`) + const adapter = new SQLiteCorePersistenceAdapter({ + driver: createDriver(database), + schemaVersion: 1, + }) + const collectionId = `progressive-initial-render` + const cached = { id: 1, name: `cached` } + const claim = await adapter.claimCacheGeneration(collectionId) + await adapter.applyCommittedTx( + { kind: `managed`, ...claim }, + { + txId: `seed`, + term: 1, + seq: 1, + rowVersion: 1, + cacheGenerationClaimId: claim.claimId, + mutations: [{ type: `insert`, key: 1, value: cached }], + collectionMetadataMutations: hasResumeMarker + ? [ + { + type: `set`, + key: `electric:resume`, + value: { + kind: `resume`, + requiresTagState: false, + offset: `10_0`, + handle: `shape-old`, + shapeId: `{"params":{"table":"test_table"},"url":"http://test-url"}`, + updatedAt: 1, + }, + }, + ] + : [], + }, + ) + await adapter.releaseCacheGenerationClaim(claim.claimId) + const collection = createCollection( + persistedCollectionOptions< + Item, + string | number, + never, + ElectricCollectionUtils + >({ + ...electricCollectionOptions({ + id: collectionId, + getKey: (row) => row.id, + syncMode: `progressive`, + shapeOptions: { + url: `http://test-url`, + params: { table: `test_table` }, + }, + }), + persistence: { adapter }, + initialRender: { + strategy: `network-first`, + networkTimeoutMs: 0, + progressive: true, + }, + }), + ) + const query = createLiveQueryCollection({ + query: (q) => q.from({ row: collection }), + }) + const observer = createLiveQueryObserver(query) + const unsubscribe = observer.subscribe(() => {}) + let primaryFailure: unknown + const cleanupFailures: Array = [] + try { + await reachCheckpoint( + observer.preloadForInitialRender(), + `local initial render`, + ) + expect(observer.getSnapshot().isPersistedReady).toBe(true) + expect(observer.getSnapshot().data).toHaveLength(1) + expect(observer.getSnapshot().data).toMatchObject([cached]) + expect(collection.status).toBe(`loading`) + await vi.waitFor(() => expect(subscribers).toHaveLength(1)) + expect(vi.mocked(ShapeStream).mock.calls[0]![0]).toEqual( + expect.objectContaining({ + offset: hasResumeMarker ? `10_0` : undefined, + }), + ) + // The fresh full snapshot replaces provisional cached rows. A resumed + // stream keeps its established baseline and applies only new changes. + subscribers[0]!([ + change(`insert`, { id: 2, name: `network` }), + { headers: { control: `up-to-date` } }, + ]) + await vi.waitFor(() => expect(collection.status).toBe(`ready`)) + const expected = hasResumeMarker + ? [cached, { id: 2, name: `network` }] + : [{ id: 2, name: `network` }] + expect( + [...collection.values()] + .map(({ id, name }) => ({ id, name })) + .sort((a, b) => a.id - b.id), + ).toEqual(expected) + } catch (error) { + primaryFailure = error + } finally { + for (const cleanup of [ + unsubscribe, + () => observer.dispose(), + () => query.cleanup(), + () => collection.cleanup(), + () => database.close(), + ]) { + try { + await cleanup() + } catch (error) { + cleanupFailures.push(error) + } + } + } + reportReceivingOutcome( + primaryFailure, + cleanupFailures, + `progressive initial render`, + ) + }, + ) + beforeEach(() => { subscribers.length = 0 synchronousMessages = undefined From e0bf64abf5a05a0fe630ad97f234509446434f3d Mon Sep 17 00:00:00 2001 From: Jose Diego Menenendez del Cueto Date: Tue, 20 Oct 2026 11:46:23 +0200 Subject: [PATCH 2/2] Require completed progressive snapshots for persisted initial rendering --- .../progressive-persisted-initial-render.md | 4 +- docs/collections/electric-collection.md | 14 + docs/contributing/oracle-coverage.md | 37 ++- docs/guides/sqlite-persistence.md | 32 +- .../src/persisted.ts | 50 +++- .../tests/persisted-readiness-oracle.test.ts | 274 ++++++++++++------ packages/db/src/live-query-observer.ts | 6 +- packages/db/src/persisted-readiness.ts | 2 + packages/db/src/types.ts | 11 + .../electric-db-collection/src/electric.ts | 49 +++- .../tests/electric-oracle.property.test.ts | 3 + .../tests/electric-recovery-oracle.test.ts | 1 + .../electric-resume-snapshot-races.test.ts | 264 ++++++++++++----- .../tests/electric.test.ts | 1 + .../tests/useLiveSuspenseQuery.test.tsx | 21 +- 15 files changed, 559 insertions(+), 210 deletions(-) diff --git a/.changeset/progressive-persisted-initial-render.md b/.changeset/progressive-persisted-initial-render.md index 23d82b0eb2..241eb35f71 100644 --- a/.changeset/progressive-persisted-initial-render.md +++ b/.changeset/progressive-persisted-initial-render.md @@ -1,5 +1,7 @@ --- +'@tanstack/db': minor +'@tanstack/electric-db-collection': minor '@tanstack/db-sqlite-persistence-core': minor --- -Allow progressive sources to opt into persisted initial rendering and restore their local baseline before reporting persisted readiness. +Allow persisted initial rendering from previously completed Electric progressive snapshots. Validate adapter-owned completion metadata and durable row-set evidence before local fallback. Cold, partial, and legacy progressive caches wait for network data; legacy cursors without completion evidence trigger a fresh full sync. Report unavailable snapshots without hanging initial-render waits on network failure. diff --git a/docs/collections/electric-collection.md b/docs/collections/electric-collection.md index 0b56f39712..e50b09be61 100644 --- a/docs/collections/electric-collection.md +++ b/docs/collections/electric-collection.md @@ -339,6 +339,20 @@ await todosCollection.utils.awaitMatch( ) ``` +### Persisted initial rendering with progressive sync + +Progressive sources expose completion evidence to SQLite persistence. Configure +`initialRender: { strategy: 'network-first', networkTimeoutMs: 0 }` on the +persisted options to render a previously completed snapshot while syncing. +Partial caches and cold starts continue to wait for the network. See +[SQLite persistence](../guides/sqlite-persistence.md#render-after-sqlite-restore). + +Electric writes `snapshotComplete` in its durable resume metadata in the same +transaction as a completed progressive snapshot. Progressive caches written by +older versions lack this evidence and conservatively fetch a fresh full snapshot +once; subsequent launches can resume normally. This also prevents a cursor +left by a subset-only source from being treated as a complete progressive cache. + ### Cleanup and resume safety Transaction evidence and pending `awaitTxId`/`awaitMatch` calls belong to one diff --git a/docs/contributing/oracle-coverage.md b/docs/contributing/oracle-coverage.md index a3e2e9d606..a4ee929f2e 100644 --- a/docs/contributing/oracle-coverage.md +++ b/docs/contributing/oracle-coverage.md @@ -2441,8 +2441,8 @@ a hostile wrong-answer control, and an explicit statement of remaining limits. The persisted-readiness oracle in `packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts` owns the opted-in eager restore boundary through the persisted wrapper, -Collection, live query, and observer. A finite two-source model checks all 36 -opt-in/outcome combinations independently of the observer classifier. Its +Collection, live query, and observer. A finite two-source model checks all 64 +opt-in/outcome combinations, including an unavailable completed snapshot independently of the observer classifier. Its controlled persistence histories check nonempty and empty restore while upstream remains pending, all-source conjunction, mixed opt-out, nested live-query sources, exact failure, SSR seed handoff, @@ -2472,17 +2472,28 @@ network provider; it does not prove browser scheduling or arbitrary histories. The framework identity scenario does not cover every possible rejection value or an error already present before a hook mounts. -The progressive opt-in extends the empty/nonempty controlled restore histories to -on-demand core sources with a held subset acquisition. The same local-restoration -rule applies; this does not certify a previously completed network snapshot. -The receiving cases in `electric-resume-snapshot-races.test.ts` exercise the -Electric adapter with real Node SQLite managed caches and a controlled SDK -stream, both with and without a saved resume marker. They check cached rows -before network delivery and replacement versus resumed updates afterward. -These cases do not establish OPFS scheduling, multi-tab coordination, cache -rotation during initial fallback, or progressive-specific joined/restart histories. -Those remain outside this draft's validation and need receiving coverage before -a broader guarantee can be made. +Progressive sources supply an adapter-owned completed-snapshot capability. +The empty/nonempty controlled histories exercise that capability through core's +on-demand representation, with a held subset acquisition. Joined restoration and +stale cleanup/restart histories cover both eager and progressive representations. +A network failure before or after discovering an unavailable snapshot must reject, +not render or hang. The finite conjunction includes unavailable source outcomes. +The React Suspense receiving case holds rendering when restoration is unavailable. + +The cases in `electric-resume-snapshot-races.test.ts` use the Electric adapter, +real Node SQLite managed caches, and controlled SDK delivery. Completed empty and +nonempty snapshots are eligible before delivery. Cold, partial, legacy, reset, +shape-mismatched, tag-dependent, and row-loss caches are ineligible. Explicit +cursor starts cannot certify an unseen full baseline. Network completion saves +the marker and rows together; reopening renders those rows without SDK delivery. +Removing the completion check makes the partial/cold/legacy cases report ready +at their unavailable checkpoint; all three fail under that mutation. Existing +resume/recovery fixtures explicitly seed completion evidence when their contract +starts from a complete snapshot; the separate legacy case retains missing evidence. +These witnesses do not establish browser OPFS scheduling, multi-tab coordination, +or cache rotation during a pending initial fallback. Those host-specific paths +remain with the persistence lifecycle owners; this change makes no new guarantees +about them. ## Deferred contracts and evidence diff --git a/docs/guides/sqlite-persistence.md b/docs/guides/sqlite-persistence.md index afadea097c..0902d97a41 100644 --- a/docs/guides/sqlite-persistence.md +++ b/docs/guides/sqlite-persistence.md @@ -236,7 +236,7 @@ An on-demand cache is not yet a certified offline snapshot. The first managed ge ## Render after SQLite restore -`initialRender` supports eager persisted Collections and explicitly opted-in progressive sources. Genuine `syncMode: 'on-demand'` sources remain unsupported. Set it on every source Collection in a live query. If one source does not opt in, the query cannot report persisted readiness or use the restored snapshot as its initial-render fallback. +`initialRender` supports eager persisted Collections and Electric progressive sources with a certified completed snapshot. Genuine `syncMode: 'on-demand'` sources remain unsupported. Set it on every source Collection in a live query. If one source does not opt in, the query cannot report persisted readiness or use the restored snapshot as its initial-render fallback. Install the React binding to use the following components: @@ -244,7 +244,7 @@ Install the React binding to use the following components: npm install @tanstack/react-db ``` -Electric progressive sources are represented as on-demand Collections internally. To restore their entire local cache before initial rendering, opt in explicitly: +Electric progressive sources are represented as on-demand Collections internally. Electric supplies the completion capability automatically; no additional flag is needed: ```ts persistedCollectionOptions({ @@ -261,20 +261,28 @@ persistedCollectionOptions({ initialRender: { strategy: 'network-first', networkTimeoutMs: 0, - progressive: true, }, }) ``` Import `electricCollectionOptions` from `@tanstack/electric-db-collection`. -Use `progressive: true` only with progressive sources. The wrapper trusts this -assertion; it cannot distinguish them from genuine on-demand sources. -Restoration loads every cached row into memory. Persisted readiness means that -local restoration finished, not that the previous full network sync completed. -An interrupted initial sync can therefore restore only previously cached subsets, -and a new cache can render empty. Network sync continues and replaces that -provisional cache with the full snapshot. Applications requiring a complete -server snapshot should continue to wait for Collection readiness. +For progressive sources, local fallback requires a full snapshot that completed +in a previous sync, for the same Electric shape, with consistent stored row-set +evidence. Electric saves the completion marker atomically with the full snapshot. +A completed empty snapshot is valid; a new empty cache is not. + +If the cache is partial, has no completion marker, belongs to a different shape, +requires unavailable tag state, uses an explicit cursor, or fails row-set +validation, `persistedStatus` +is `unavailable`. Initial rendering waits for network readiness instead. A +network failure rejects the initial-render wait rather than displaying partial +cached data or waiting indefinitely. Legacy progressive caches without this +marker perform a fresh sync once to establish it. + +Restoration still loads the entire completed snapshot into memory. Ordinary +`useLiveQuery` consumers must respect readiness if they want this gating; +`initialRender` does not hide rows from direct Collection reads. The complete +snapshot may be stale; background sync continues after local fallback. A regular React live query exposes both readiness signals: @@ -300,7 +308,7 @@ function TodosView() { `isReady` reports Collection readiness. `isPersistedReady` reports that every opted-in source finished its current local restore, even when SQLite had no rows. These flags do not identify where each visible row came from. A regular `useLiveQuery` does not wait for the network-first deadline before it renders rows. -The hooks also expose `persistedStatus` and `persistedError`. The status is `unavailable` if a source did not opt in, `loading` during restore, `ready` after every restore completes, or `error` if a restore fails. `persistedError` keeps the restore error. An SSR seed does not count as a SQLite restore. +The hooks also expose `persistedStatus` and `persistedError`. The status is `unavailable` if a source did not opt in or lacks an eligible completed snapshot, `loading` during restore, `ready` after every restore completes, or `error` if a restore fails. `persistedError` keeps the restore error. An SSR seed does not count as a SQLite restore. React, Solid, and Svelte read these values as properties. Angular reads signals such as `result.isPersistedReady()`. Vue reads refs such as `result.isPersistedReady.value`. diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 744f1c55e4..df945865ad 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -823,12 +823,6 @@ export type PersistedInitialRenderOptions = { strategy: `network-first` /** Time to prefer network before restored rows may render; defaults to 3s. */ networkTimeoutMs?: number - /** - * Restore the full local cache for a progressive source represented as - * on-demand. Do not enable for genuinely on-demand sources. This certifies - * local restoration only, not completion of a previous network snapshot. - */ - progressive?: boolean } const DEFAULT_NETWORK_TIMEOUT_MS = 3_000 @@ -1619,6 +1613,10 @@ class PersistedCollectionRuntime< private readonly dbName: string, private readonly persistedReadiness?: PersistedReadinessTracker, private readonly optionsClaim?: () => void | (() => void), + private readonly isPersistedSnapshotComplete?: SyncConfig< + T, + TKey + >[`isPersistedSnapshotComplete`], ) {} private get storageCollectionId(): string { @@ -2260,7 +2258,9 @@ class PersistedCollectionRuntime< } if (lifecycleGeneration === this.lifecycleGeneration) { this.startupSettled = true - this.persistedReadiness?.set({ status: `ready` }) + if (this.persistedReadiness?.getSnapshot().status === `loading`) { + this.persistedReadiness.set({ status: `ready` }) + } } resolveStartupMetadata() })().catch((error) => { @@ -2549,6 +2549,7 @@ class PersistedCollectionRuntime< private async hydrateBaseline( lifecycleGeneration: number, adapter: HydrationPersistenceAdapter, + requireCompleteSnapshot = false, ): Promise { if (lifecycleGeneration !== this.lifecycleGeneration) return undefined @@ -2562,6 +2563,7 @@ class PersistedCollectionRuntime< { lifecycleGeneration, bindKeySetEvidence: true, + requireCompleteSnapshot, }, adapter, ) @@ -2611,7 +2613,11 @@ class PersistedCollectionRuntime< } const appliedCursor = this.syncMode !== `on-demand` || this.persistedReadiness !== undefined - ? await this.hydrateBaseline(lifecycleGeneration, adapter) + ? await this.hydrateBaseline( + lifecycleGeneration, + adapter, + this.isPersistedSnapshotComplete !== undefined, + ) : undefined return lifecycleGeneration === this.lifecycleGeneration ? { @@ -3290,6 +3296,7 @@ class PersistedCollectionRuntime< config: { lifecycleGeneration: number bindKeySetEvidence?: boolean + requireCompleteSnapshot?: boolean requestLocalLoadFailure?: boolean rejectBufferedReplayFailure?: boolean }, @@ -3308,6 +3315,7 @@ class PersistedCollectionRuntime< try { let rows: Array<{ key: TKey; value: T; metadata?: unknown }> let bindResumeSnapshot: (() => void) | undefined + let canHydrate = true if (config.bindKeySetEvidence && !this.scopedRecovery) { const snapshot = await adapter.loadResumeSnapshot( this.storageTargetFor(), @@ -3325,6 +3333,20 @@ class PersistedCollectionRuntime< if (config.lifecycleGeneration !== this.lifecycleGeneration) return true if (resetSequence !== this.resetSequence) return true + if (config.requireCompleteSnapshot) { + canHydrate = + snapshot.keySet?.status === `consistent` && + this.isPersistedSnapshotComplete?.( + new Map( + snapshot.collectionMetadata.map(({ key, value }) => [ + key, + value, + ]), + ), + ) === true + if (!canHydrate) + this.persistedReadiness?.set({ status: `unavailable` }) + } bindResumeSnapshot = () => this.bindResumeSnapshotEvidence(snapshot) boundRowVersion = snapshot.latestRowVersion } else { @@ -3356,8 +3378,9 @@ class PersistedCollectionRuntime< bindResumeSnapshot?.() if ( - !config.bindKeySetEvidence || - this.persistedKeySetEvidence?.status !== `incompatible` + canHydrate && + (!config.bindKeySetEvidence || + this.persistedKeySetEvidence?.status !== `incompatible`) ) { for (const row of rows) { hydrationContext.suppliedRowKeys.add(row.key) @@ -6839,10 +6862,10 @@ export function persistedCollectionOptions< if ( networkTimeoutMs !== undefined && options.syncMode === `on-demand` && - !options.initialRender?.progressive + !(hasOwnSyncKey(options) && options.sync.isPersistedSnapshotComplete) ) { throw new InvalidPersistedCollectionConfigError( - `network-first initial rendering requires eager persistence syncMode`, + `network-first initial rendering requires eager persistence syncMode or an adapter that certifies completed persisted snapshots`, ) } const persistedReadinessTracker = @@ -6895,6 +6918,9 @@ export function persistedCollectionOptions< collectionId, persistedReadinessTracker, claimOptions, + persistedReadinessTracker + ? syncOptions.sync.isPersistedSnapshotComplete + : undefined, ) claimedRuntime = runtime diff --git a/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts b/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts index 02dc07e01b..70b2c5cd6d 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted-readiness-oracle.test.ts @@ -12,7 +12,7 @@ import { persistedCollectionOptions } from '../src/index.js' import type { PersistenceAdapter } from '../src/index.js' type Row = { id: string; value: string } -type ModelOutcome = `loading` | `ready` | `error` +type ModelOutcome = `loading` | `ready` | `error` | `unavailable` type ModelSource = { optedIn: boolean outcome: ModelOutcome @@ -20,7 +20,9 @@ type ModelSource = { } function expectedPersistedSnapshot(sources: ReadonlyArray) { - if (sources.some((entry) => !entry.optedIn)) { + if ( + sources.some((entry) => !entry.optedIn || entry.outcome === `unavailable`) + ) { return { status: `unavailable`, error: undefined } } const failed = sources.find((entry) => entry.outcome === `error`) @@ -94,13 +96,13 @@ function source( initialRender: { strategy: `network-first` as const, networkTimeoutMs, - ...(progressive ? { progressive: true } : {}), }, } : {}), // Upstream never declares Collection readiness. Local restore is a // separately observable boundary even while this stays loading. sync: { + ...(progressive ? { isPersistedSnapshotComplete: () => true } : {}), sync: () => progressive ? { loadSubset: () => new Promise(() => {}) } : {}, }, @@ -154,9 +156,9 @@ async function checkWithCleanup( * The independent status rule is an all-source conjunction: a query is persisted-ready * only after every eligible persisted source opted in and successfully completed * its restore in the current sync run. This draft extends eligibility to - * explicitly opted-in progressive sources, which core sees as on-demand. - * This means the cached rows were restored, not that a prior full network - * snapshot completed. The empty/nonempty histories cover both modes. A source that opts out makes the + * progressive sources with adapter-certified complete snapshots, which core + * sees as on-demand. A source with no complete snapshot reports unavailable. + * The empty/nonempty histories cover both modes. A source that opts out makes the * query unavailable for this signal. A failed restore reports that exact * failure. Neither an empty persisted database nor a pending network source * changes the rule. Initial rendering prefers upstream-signaled Collection @@ -179,7 +181,7 @@ async function checkWithCleanup( * Audit: ORC-001/002 use the approved all-source rule and the independent * expectedPersistedSnapshot model. ORC-003/005 are the finite outcome grammar, * controlled persistence driver, and public snapshot checks. ORC-004/007/008 - * do not apply: the 36-case product is enumerated, not generated or stateful. + * do not apply: the 64-case product is enumerated, not generated or stateful. * ORC-006: the pre-implementation observer lacked persistedStatus and * preloadForInitialRender; the deadline test also failed against the immediate * persisted-ready race. Before the early-error repair, the held-read witness @@ -227,10 +229,15 @@ describe(`persisted-readiness oracle`, () => { }) it(`refines the finite two-source readiness model`, async () => { - const outcomes: ReadonlyArray = [`loading`, `ready`, `error`] + const outcomes: ReadonlyArray = [ + `loading`, + `ready`, + `error`, + `unavailable`, + ] let caseId = 0 // Grammar: each source independently opts in or out and has one of the - // three local outcomes. No network transition is needed to judge this + // four local outcomes. No network transition is needed to judge this // observer-level projection; real persistence histories are below. for (const leftOptedIn of [false, true]) { for (const rightOptedIn of [false, true]) { @@ -310,7 +317,7 @@ describe(`persisted-readiness oracle`, () => { } } } - expect(caseId).toBe(36) + expect(caseId).toBe(64) }) // Both source modes owe the same restore-before-fallback rule. The row @@ -363,37 +370,52 @@ describe(`persisted-readiness oracle`, () => { } } - it(`requires every joined source to opt in and finish its own restore`, async () => { - const left = recordingAdapter([{ id: `one`, value: `left` }]) - const right = recordingAdapter([{ id: `one`, value: `right` }]) - const sourceA = source(`local-join-left`, left.adapter, true) - const sourceB = source(`local-join-right`, right.adapter, true) - const query = createLiveQueryCollection({ - query: (q) => - q - .from({ a: sourceA }) - .join({ b: sourceB }, ({ a, b }) => eq(a.id, b.id)), - }) - const observer = createLiveQueryObserver(query) - const unsubscribe = observer.subscribe(() => {}) - await checkWithCleanup(async () => { - const initialRenderWait = observer.preloadForInitialRender() - left.load.resolve() - await vi.waitFor(() => expect(sourceA.has(`one`)).toBe(true)) - expect(observer.getSnapshot().persistedStatus).toBe(`loading`) + it.each([false, true])( + `requires every joined source to opt in and finish its own restore (progressive: %s)`, + async (progressive) => { + const left = recordingAdapter([{ id: `one`, value: `left` }]) + const right = recordingAdapter([{ id: `one`, value: `right` }]) + const sourceA = source( + `local-join-left`, + left.adapter, + true, + 0, + progressive, + ) + const sourceB = source( + `local-join-right`, + right.adapter, + true, + 0, + progressive, + ) + const query = createLiveQueryCollection({ + query: (q) => + q + .from({ a: sourceA }) + .join({ b: sourceB }, ({ a, b }) => eq(a.id, b.id)), + }) + const observer = createLiveQueryObserver(query) + const unsubscribe = observer.subscribe(() => {}) + await checkWithCleanup(async () => { + const initialRenderWait = observer.preloadForInitialRender() + left.load.resolve() + await vi.waitFor(() => expect(sourceA.has(`one`)).toBe(true)) + expect(observer.getSnapshot().persistedStatus).toBe(`loading`) - right.load.resolve() - await initialRenderWait - expect(observer.getSnapshot().persistedStatus).toBe(`ready`) - expect(observer.getSnapshot().data).toHaveLength(1) - }, [ - unsubscribe, - () => observer.dispose(), - () => query.cleanup(), - () => sourceA.cleanup(), - () => sourceB.cleanup(), - ]) - }) + right.load.resolve() + await initialRenderWait + expect(observer.getSnapshot().persistedStatus).toBe(`ready`) + expect(observer.getSnapshot().data).toHaveLength(1) + }, [ + unsubscribe, + () => observer.dispose(), + () => query.cleanup(), + () => sourceA.cleanup(), + () => sourceB.cleanup(), + ]) + }, + ) it(`follows a live-query source to its underlying persisted Collection`, async () => { const fixture = recordingAdapter([{ id: `one`, value: `local` }]) @@ -479,6 +501,57 @@ describe(`persisted-readiness oracle`, () => { ]) }) + it.each([`before-restore`, `after-restore`] as const)( + `rejects a network failure when no completed snapshot exists (%s)`, + async (order) => { + const fixture = recordingAdapter([ + { id: `partial`, value: `provisional` }, + ]) + let fail: ((error: unknown) => void) | undefined + const persisted = createCollection( + persistedCollectionOptions({ + id: `unavailable-${order}`, + getKey: (row) => row.id, + syncMode: `on-demand`, + initialRender: { strategy: `network-first`, networkTimeoutMs: 0 }, + persistence: { adapter: fixture.adapter }, + sync: { + isPersistedSnapshotComplete: () => false, + sync: ({ markError }) => { + fail = markError + return { loadSubset: () => new Promise(() => {}) } + }, + }, + }), + ) + const query = createLiveQueryCollection({ + query: (q) => q.from({ row: persisted }), + }) + const observer = createLiveQueryObserver(query) + const unsubscribe = observer.subscribe(() => {}) + await checkWithCleanup(async () => { + const failure = new Error(`network unavailable`) + const rejected = expect( + observer.preloadForInitialRender(), + ).rejects.toThrow(`entered error state`) + await vi.waitFor(() => expect(fail).toBeDefined()) + if (order === `before-restore`) fail!(failure) + fixture.load.resolve() + await vi.waitFor(() => + expect(observer.getSnapshot().persistedStatus).toBe(`unavailable`), + ) + if (order === `after-restore`) fail!(failure) + await rejected + expect(observer.getSnapshot().isPersistedReady).toBe(false) + }, [ + unsubscribe, + () => observer.dispose(), + () => query.cleanup(), + () => persisted.cleanup(), + ]) + }, + ) + it(`rejects network-first initial rendering for on-demand persistence`, () => { const fixture = recordingAdapter([]) expect(() => @@ -596,65 +669,76 @@ describe(`persisted-readiness oracle`, () => { ]) }) - it(`does not publish a stale restore after cleanup and restart`, async () => { - const fixture = recordingAdapter([{ id: `one`, value: `local` }]) - fixture.load.resolve() - const firstRead = deferred() - const secondRead = deferred() - const loadResumeSnapshot = fixture.adapter.loadResumeSnapshot - let baselineReads = 0 - let firstReadReturned = false - // The obsolete sync run returns a distinct row, so a leaked publication - // remains observable even after the replacement restore completes. - fixture.adapter.loadResumeSnapshot = async (id, options) => { - if (options?.includeRows !== false) { - const read = ++baselineReads - await (read === 1 ? firstRead.promise : secondRead.promise) - const snapshot = await loadResumeSnapshot(id, options) - if (read === 1) { - firstReadReturned = true - return { - ...snapshot, - rows: [{ key: `stale`, value: { id: `stale`, value: `obsolete` } }], + it.each([false, true])( + `does not publish a stale restore after cleanup and restart (progressive: %s)`, + async (progressive) => { + const fixture = recordingAdapter([{ id: `one`, value: `local` }]) + fixture.load.resolve() + const firstRead = deferred() + const secondRead = deferred() + const loadResumeSnapshot = fixture.adapter.loadResumeSnapshot + let baselineReads = 0 + let firstReadReturned = false + // The obsolete sync run returns a distinct row, so a leaked publication + // remains observable even after the replacement restore completes. + fixture.adapter.loadResumeSnapshot = async (id, options) => { + if (options?.includeRows !== false) { + const read = ++baselineReads + await (read === 1 ? firstRead.promise : secondRead.promise) + const snapshot = await loadResumeSnapshot(id, options) + if (read === 1) { + firstReadReturned = true + return { + ...snapshot, + rows: [ + { key: `stale`, value: { id: `stale`, value: `obsolete` } }, + ], + } } + return snapshot } - return snapshot + return loadResumeSnapshot(id, options) } - return loadResumeSnapshot(id, options) - } - const persisted = source(`local-restart`, fixture.adapter, true) - const observer = createLiveQueryObserver(persisted) - const observedData: Array = [] - const unsubscribe = observer.subscribe(() => { - observedData.push(observer.getSnapshot().data) - }) - await checkWithCleanup(async () => { - await vi.waitFor(() => expect(baselineReads).toBe(1)) - await persisted.cleanup() - persisted.startSyncImmediate() - firstRead.resolve() - await vi.waitFor(() => expect(baselineReads).toBe(2)) - await vi.waitFor(() => expect(firstReadReturned).toBe(true)) - await new Promise((resolve) => setTimeout(resolve, 0)) - expect(observer.getSnapshot().persistedStatus).toBe(`loading`) - expect(observer.getSnapshot().data).toEqual([]) - - secondRead.resolve() - await vi.waitFor(() => { - expect(observer.getSnapshot().persistedStatus).toBe(`ready`) - expect(persisted.has(`one`)).toBe(true) - }) - expect(observedData).not.toContainEqual( - expect.arrayContaining([expect.objectContaining({ id: `stale` })]), + const persisted = source( + `local-restart`, + fixture.adapter, + true, + 0, + progressive, ) - }, [ - () => firstRead.resolve(), - () => secondRead.resolve(), - unsubscribe, - () => observer.dispose(), - () => persisted.cleanup(), - ]) - }) + const observer = createLiveQueryObserver(persisted) + const observedData: Array = [] + const unsubscribe = observer.subscribe(() => { + observedData.push(observer.getSnapshot().data) + }) + await checkWithCleanup(async () => { + await vi.waitFor(() => expect(baselineReads).toBe(1)) + await persisted.cleanup() + persisted.startSyncImmediate() + firstRead.resolve() + await vi.waitFor(() => expect(baselineReads).toBe(2)) + await vi.waitFor(() => expect(firstReadReturned).toBe(true)) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(observer.getSnapshot().persistedStatus).toBe(`loading`) + expect(observer.getSnapshot().data).toEqual([]) + + secondRead.resolve() + await vi.waitFor(() => { + expect(observer.getSnapshot().persistedStatus).toBe(`ready`) + expect(persisted.has(`one`)).toBe(true) + }) + expect(observedData).not.toContainEqual( + expect.arrayContaining([expect.objectContaining({ id: `stale` })]), + ) + }, [ + () => firstRead.resolve(), + () => secondRead.resolve(), + unsubscribe, + () => observer.dispose(), + () => persisted.cleanup(), + ]) + }, + ) // The finite conjunction above has no transitions. This legal history gives // source A two sync runs while source B completes once. A's old restore cannot diff --git a/packages/db/src/live-query-observer.ts b/packages/db/src/live-query-observer.ts index 0dab756aa0..e025331831 100644 --- a/packages/db/src/live-query-observer.ts +++ b/packages/db/src/live-query-observer.ts @@ -370,6 +370,7 @@ class LiveQueryObserverImpl< let failed = false for (const source of sources) { const snapshot = source.readiness.getSnapshot() + if (snapshot.status === `unavailable`) return NO_PERSISTED_READINESS if (snapshot.status === `error` && !failed) { error = snapshot.error failed = true @@ -1117,7 +1118,10 @@ class LiveQueryObserverImpl< persisted.status === `ready` ) { finish() - } else if (networkFailed && persisted.status === `error`) { + } else if ( + networkFailed && + (persisted.status === `error` || persisted.status === `unavailable`) + ) { finish({ error: networkError ?? persisted.error }) } else if (!deadlinePassed && !networkFailed) { timer = setTimeout(check, remainingNetworkMs) diff --git a/packages/db/src/persisted-readiness.ts b/packages/db/src/persisted-readiness.ts index b7bd34c2de..793089429b 100644 --- a/packages/db/src/persisted-readiness.ts +++ b/packages/db/src/persisted-readiness.ts @@ -1,6 +1,8 @@ /** The local-restore boundary of one opted-in persisted Collection. */ export type PersistedReadinessSnapshot = | { status: `loading`; error?: never } + /** Restore finished, but the source cannot certify a usable persisted snapshot. */ + | { status: `unavailable`; error?: never } | { status: `ready`; error?: never } | { status: `error`; error: unknown } diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 298543dc42..e9347942fe 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -499,6 +499,17 @@ export interface SyncConfig< */ exportSyncMeta?: () => unknown + /** + * Certify that durable collection metadata describes a completed full snapshot + * for this source. Persistence checks this against the same atomic snapshot + * as the restored rows, and also requires consistent durable key-set evidence. + * Returning false disables local initial-render fallback for that sync run. + * Subset-only adapters must not provide this capability. + */ + isPersistedSnapshotComplete?: ( + metadata: ReadonlyMap, + ) => boolean + /** * Import adapter-specific metadata produced by exportSyncMeta. */ diff --git a/packages/electric-db-collection/src/electric.ts b/packages/electric-db-collection/src/electric.ts index bb5f8b6048..2f955116d5 100644 --- a/packages/electric-db-collection/src/electric.ts +++ b/packages/electric-db-collection/src/electric.ts @@ -103,6 +103,7 @@ type ElectricResumeState = shapeId: string updatedAt: number requiresTagState?: boolean + snapshotComplete?: boolean } | { kind: `reset` @@ -178,6 +179,9 @@ function parseElectricResumeState( handle: record.handle, shapeId: record.shapeId, updatedAt: record.updatedAt, + ...(typeof record.snapshotComplete === `boolean` && { + snapshotComplete: record.snapshotComplete, + }), ...(typeof record.requiresTagState === `boolean` && { requiresTagState: record.requiresTagState, }), @@ -272,7 +276,9 @@ function getNewestElectricResumeState( (incoming.kind === `resume` && current.offset === incoming.offset && current.handle === incoming.handle && - current.shapeId === incoming.shapeId)) + current.shapeId === incoming.shapeId && + current.snapshotComplete === incoming.snapshotComplete && + current.requiresTagState === incoming.requiresTagState)) // Equal timestamps have no causal ordering. Preserve an identical state, // but collapse every conflict to reset so merge order cannot resurrect an @@ -1866,12 +1872,17 @@ function createElectricSync>( shapeOptions.offset === undefined && shapeOptions.handle === undefined && persistedResumeState?.kind === `resume` && + (syncMode !== `progressive` || + persistedResumeState.snapshotComplete === true) && !hasIncompatiblePersistedResume && // Cached rows do not contain authoritative tag/active-condition state. // Only a complete adapter ledger can justify a persisted cursor. !requiresFreshSourceEvidence const hasExplicitResumeOffset = shapeOptions.offset !== undefined && shapeOptions.offset !== `-1` + // An explicitly supplied cursor can begin after rows we have never seen. + // Only a fresh full snapshot (including must-refetch) can certify it. + let canCertifyFullSnapshot = !hasExplicitResumeOffset if (!canUsePersistedResume && !hasExplicitResumeOffset) { clearTagTrackingState() } @@ -2107,7 +2118,7 @@ function createElectricSync>( const syncedKeys = new Set() let resumeInvalid = false - const stageResumeMetadata = () => { + const stageResumeMetadata = (snapshotComplete: boolean) => { if ( !isActiveLifecycle() || resumeInvalid || @@ -2129,6 +2140,9 @@ function createElectricSync>( shapeId: shapeIdentity, updatedAt: Date.now(), requiresTagState: tagState.hasTags(), + ...(syncMode === `progressive` + ? { snapshotComplete: snapshotComplete && canCertifyFullSnapshot } + : {}), } lifecycle.resumeState = resumeState metadata?.collection.set(`electric:resume`, resumeState) @@ -2510,6 +2524,7 @@ function createElectricSync>( syncedKeys.clear() recordPendingTruncate() isResettingSnapshot = true + canCertifyFullSnapshot = true resetGeneration++ // Reset the loadSubset deduplication state since we're starting fresh @@ -2596,7 +2611,7 @@ function createElectricSync>( } // Commit the atomic swap - stageResumeMetadata() + stageResumeMetadata(true) applied = commitSourceTransaction() // Exit buffering phase by marking that we've received up-to-date @@ -2612,13 +2627,16 @@ function createElectricSync>( // Both up-to-date and subset-end trigger a commit if (transactionStarted) { if (!isResettingSnapshot || finishesReset) { - stageResumeMetadata() + stageResumeMetadata( + syncMode === `progressive` && + (commitPoint === `up-to-date` || hasReceivedUpToDate), + ) } applied = commitSourceTransaction() transactionStarted = false } else if (commitPoint === `up-to-date` && metadata) { beginSourceTransaction() - stageResumeMetadata() + stageResumeMetadata(true) applied = commitSourceTransaction() } } @@ -2774,6 +2792,27 @@ function createElectricSync>( } return { + ...(syncMode === `progressive` + ? { + isPersistedSnapshotComplete: ( + metadata: ReadonlyMap, + ) => { + const resume = parseElectricResumeState( + metadata.get(`electric:resume`), + ) + const shapeId = getStableShapeIdentity(shapeOptions) + return ( + shapeOptions.offset === undefined && + shapeOptions.handle === undefined && + resume?.kind === `resume` && + resume.snapshotComplete === true && + resume.requiresTagState === false && + resume.offset !== `-1` && + resume.shapeId === shapeId + ) + }, + } + : {}), getSyncMetadata: () => ({ relation: shapeOptions.params?.table ? [relationSchema || `public`, shapeOptions.params.table] diff --git a/packages/electric-db-collection/tests/electric-oracle.property.test.ts b/packages/electric-db-collection/tests/electric-oracle.property.test.ts index fa62022c93..5fd99a7c4b 100644 --- a/packages/electric-db-collection/tests/electric-oracle.property.test.ts +++ b/packages/electric-db-collection/tests/electric-oracle.property.test.ts @@ -267,6 +267,9 @@ function resumeState(): ReadonlyMap { `electric:resume`, { kind: `resume`, + // These histories start from a completed full snapshot. Legacy and + // interrupted snapshots are covered by the initial-render receiver. + snapshotComplete: true, requiresTagState: false, offset: `10_0`, handle: `shape-1`, diff --git a/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts b/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts index cfeef06173..5ee0443503 100644 --- a/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts +++ b/packages/electric-db-collection/tests/electric-recovery-oracle.test.ts @@ -174,6 +174,7 @@ function fixture( `electric:resume`, { kind: `resume`, + snapshotComplete: true, requiresTagState: false, offset: `10_0`, handle: `shape-old`, diff --git a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts index 90e3b7cd22..8f17790511 100644 --- a/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts +++ b/packages/electric-db-collection/tests/electric-resume-snapshot-races.test.ts @@ -28,6 +28,7 @@ type Subscriber = (messages: Array>) => void const subscribers: Array = [] let synchronousMessages: Array> | undefined let snapshotRequest: (() => Promise) | undefined +let fetchSnapshotRequest: (() => Promise) | undefined const mockSubscribe = vi.fn((subscriber: Subscriber) => { subscribers.push(subscriber) if (synchronousMessages) subscriber(synchronousMessages) @@ -39,7 +40,10 @@ vi.mock(`@electric-sql/client`, async () => ({ ShapeStream: vi.fn(() => ({ subscribe: mockSubscribe, requestSnapshot: vi.fn(() => snapshotRequest?.() ?? Promise.resolve()), - fetchSnapshot: vi.fn().mockResolvedValue({ metadata: {}, data: [] }), + fetchSnapshot: vi.fn(async () => { + await fetchSnapshotRequest?.() + return { metadata: {}, data: [] } + }), forceDisconnectAndRefresh: vi.fn().mockResolvedValue(undefined), isUpToDate: false, shapeHandle: `shape-current`, @@ -60,7 +64,7 @@ function toBinding(value: unknown): string | number | bigint | null { return String(value) } -function createDriver(database: DatabaseSync): SQLiteDriver { +function createDriver(database: DatabaseSync, serialize = false): SQLiteDriver { const driver: SQLiteDriver = { exec: (sql) => { database.exec(sql) @@ -89,7 +93,25 @@ function createDriver(database: DatabaseSync): SQLiteDriver { } }, } - return driver + if (!serialize) return driver + // Match the host driver's transaction isolation when startup hydration and + // cleanup overlap. Transaction callbacks receive the direct scoped driver. + let queue = Promise.resolve() + const enqueue = (task: () => Promise): Promise => { + const result = queue.then(task) + queue = result.then( + () => undefined, + () => undefined, + ) + return result + } + return { + exec: (sql) => enqueue(() => driver.exec(sql)), + run: (sql, params) => enqueue(() => driver.run(sql, params)), + query: (sql: string, params?: ReadonlyArray) => + enqueue(() => driver.query(sql, params)), + transaction: (task) => enqueue(() => driver.transaction(task)), + } } function deferred() { @@ -980,15 +1002,34 @@ async function observeLegacyUnknownResume(): Promise { - it.each([false, true])( - `renders the progressive SQLite cache before network delivery (resume marker: %s)`, - async (hasResumeMarker) => { - // This receiving witness uses real SQLite and the Electric adapter with a - // controlled ShapeStream. A missing marker deliberately represents cached - // subset rows from an interrupted initial sync, not a complete snapshot. + it.each([ + `complete`, + `complete-empty`, + `partial`, + `cold`, + `legacy`, + `reset`, + `other-shape`, + `tags`, + `row-loss`, + `explicit-offset`, + ] as const)( + `only renders a previously completed progressive snapshot (%s)`, + async (cacheState) => { + const complete = + cacheState === `complete` || cacheState === `complete-empty` + const empty = cacheState === `cold` || cacheState === `complete-empty` + const subset = deferred() + fetchSnapshotRequest = () => subset.promise + // Completion evidence must distinguish empty-complete from cold and + // interrupted caches. Real SQLite owns atomic rows/metadata and key-set + // validation; the controlled SDK delivers no network result until released. + // Calibration: omitting the metadata check makes partial/cold/legacy + // cases report ready at this cut; all three fail their unavailable check. + // An explicit cursor cannot certify rows omitted before that cursor. const database = new DatabaseSync(`:memory:`) const adapter = new SQLiteCorePersistenceAdapter({ - driver: createDriver(database), + driver: createDriver(database, true), schemaVersion: 1, }) const collectionId = `progressive-initial-render` @@ -1002,50 +1043,70 @@ describe(`Electric resume snapshot races`, () => { seq: 1, rowVersion: 1, cacheGenerationClaimId: claim.claimId, - mutations: [{ type: `insert`, key: 1, value: cached }], - collectionMetadataMutations: hasResumeMarker - ? [ - { - type: `set`, - key: `electric:resume`, - value: { - kind: `resume`, - requiresTagState: false, - offset: `10_0`, - handle: `shape-old`, - shapeId: `{"params":{"table":"test_table"},"url":"http://test-url"}`, - updatedAt: 1, + mutations: empty ? [] : [{ type: `insert`, key: 1, value: cached }], + collectionMetadataMutations: + cacheState !== `partial` && cacheState !== `cold` + ? [ + { + type: `set`, + key: `electric:resume`, + value: { + kind: cacheState === `reset` ? `reset` : `resume`, + requiresTagState: cacheState === `tags`, + ...(cacheState !== `legacy` + ? { snapshotComplete: true } + : {}), + offset: `10_0`, + handle: `shape-old`, + shapeId: + cacheState === `other-shape` + ? `other-shape` + : `{"params":{"table":"test_table"},"url":"http://test-url"}`, + updatedAt: 1, + }, }, - }, - ] - : [], + ] + : [], }, ) + if (cacheState === `row-loss`) { + const table = createPersistedTableName(claim.storageCollectionId, `c`) + database.prepare(`DELETE FROM "${table}"`).run() + } await adapter.releaseCacheGenerationClaim(claim.claimId) - const collection = createCollection( - persistedCollectionOptions< - Item, - string | number, - never, - ElectricCollectionUtils - >({ - ...electricCollectionOptions({ - id: collectionId, - getKey: (row) => row.id, - syncMode: `progressive`, - shapeOptions: { - url: `http://test-url`, - params: { table: `test_table` }, + const createSource = () => + createCollection( + persistedCollectionOptions< + Item, + string | number, + never, + ElectricCollectionUtils + >({ + ...electricCollectionOptions({ + id: collectionId, + getKey: (row) => row.id, + syncMode: `progressive`, + shapeOptions: { + url: `http://test-url`, + params: { + table: `test_table`, + ...(cacheState === `explicit-offset` + ? { replica: `full` } + : {}), + }, + ...(cacheState === `explicit-offset` + ? { offset: `10_0` as const, handle: `manual` } + : {}), + }, + }), + persistence: { adapter }, + initialRender: { + strategy: `network-first`, + networkTimeoutMs: 0, }, }), - persistence: { adapter }, - initialRender: { - strategy: `network-first`, - networkTimeoutMs: 0, - progressive: true, - }, - }), - ) + ) + const collection = createSource() const query = createLiveQueryCollection({ query: (q) => q.from({ row: collection }), }) @@ -1054,38 +1115,108 @@ describe(`Electric resume snapshot races`, () => { let primaryFailure: unknown const cleanupFailures: Array = [] try { - await reachCheckpoint( - observer.preloadForInitialRender(), - `local initial render`, + let rendered = false + const initialRender = observer.preloadForInitialRender().then(() => { + rendered = true + }) + void initialRender.catch(() => undefined) + await vi.waitFor(() => + expect(observer.getSnapshot().persistedStatus).toBe( + complete ? `ready` : `unavailable`, + ), ) - expect(observer.getSnapshot().isPersistedReady).toBe(true) - expect(observer.getSnapshot().data).toHaveLength(1) - expect(observer.getSnapshot().data).toMatchObject([cached]) + if (complete) { + await reachCheckpoint(initialRender, `completed local initial render`) + expect(observer.getSnapshot().data).toHaveLength(empty ? 0 : 1) + expect(observer.getSnapshot().data).toMatchObject( + empty ? [] : [cached], + ) + } else { + expect(rendered).toBe(false) + expect(observer.getSnapshot().isPersistedReady).toBe(false) + } expect(collection.status).toBe(`loading`) await vi.waitFor(() => expect(subscribers).toHaveLength(1)) - expect(vi.mocked(ShapeStream).mock.calls[0]![0]).toEqual( - expect.objectContaining({ - offset: hasResumeMarker ? `10_0` : undefined, - }), - ) // The fresh full snapshot replaces provisional cached rows. A resumed // stream keeps its established baseline and applies only new changes. - subscribers[0]!([ - change(`insert`, { id: 2, name: `network` }), - { headers: { control: `up-to-date` } }, - ]) + subscribers[0]!([change(`insert`, { id: 2, name: `network` })]) + if (!complete) { + expect(observer.getSnapshot().isPersistedReady).toBe(false) + expect(rendered).toBe(false) + } + subscribers[0]!([{ headers: { control: `up-to-date` } }]) await vi.waitFor(() => expect(collection.status).toBe(`ready`)) - const expected = hasResumeMarker - ? [cached, { id: 2, name: `network` }] - : [{ id: 2, name: `network` }] + subset.resolve() + await reachCheckpoint(initialRender, `network initial render`) + const expected = + complete && !empty + ? [cached, { id: 2, name: `network` }] + : [{ id: 2, name: `network` }] expect( [...collection.values()] .map(({ id, name }) => ({ id, name })) .sort((a, b) => a.id - b.id), ).toEqual(expected) + const verificationClaim = + await adapter.claimCacheGeneration(collectionId) + try { + const snapshot = await adapter.loadResumeSnapshot({ + kind: `managed`, + ...verificationClaim, + }) + expect(snapshot.keySet).toEqual({ status: `consistent` }) + expect( + snapshot.collectionMetadata.find( + ({ key }) => key === `electric:resume`, + )?.value, + ).toMatchObject({ + kind: `resume`, + snapshotComplete: cacheState !== `explicit-offset`, + requiresTagState: false, + }) + expect( + snapshot.rows + .map(({ value }) => value) + .sort((a, b) => Number(a.id) - Number(b.id)), + ).toEqual(expected) + } finally { + await adapter.releaseCacheGenerationClaim(verificationClaim.claimId) + } + if (cacheState !== `explicit-offset`) { + unsubscribe() + observer.dispose() + await query.cleanup() + await collection.cleanup() + // Reopen the snapshot written by the real adapter above. No SDK batch + // is delivered to this new sync run; its first render must be local. + const reopened = createSource() + const reopenedQuery = createLiveQueryCollection({ + query: (q) => q.from({ row: reopened }), + }) + const reopenedObserver = createLiveQueryObserver(reopenedQuery) + const stop = reopenedObserver.subscribe(() => {}) + try { + await reachCheckpoint( + reopenedObserver.preloadForInitialRender(), + `reopened completed snapshot`, + ) + expect(reopenedObserver.getSnapshot().isPersistedReady).toBe(true) + expect(reopenedObserver.getSnapshot().data).toHaveLength( + expected.length, + ) + expect(reopenedObserver.getSnapshot().data).toMatchObject(expected) + expect(reopened.status).toBe(`loading`) + } finally { + stop() + reopenedObserver.dispose() + await reopenedQuery.cleanup() + await reopened.cleanup() + } + } } catch (error) { primaryFailure = error } finally { + subset.resolve() for (const cleanup of [ unsubscribe, () => observer.dispose(), @@ -1112,6 +1243,7 @@ describe(`Electric resume snapshot races`, () => { subscribers.length = 0 synchronousMessages = undefined snapshotRequest = undefined + fetchSnapshotRequest = undefined vi.clearAllMocks() }) diff --git a/packages/electric-db-collection/tests/electric.test.ts b/packages/electric-db-collection/tests/electric.test.ts index 3f0aa80c5f..18439e2784 100644 --- a/packages/electric-db-collection/tests/electric.test.ts +++ b/packages/electric-db-collection/tests/electric.test.ts @@ -4251,6 +4251,7 @@ describe(`Electric Integration`, () => { `electric:resume`, { kind: `resume`, + snapshotComplete: true, requiresTagState: false, offset: `10_0`, handle: `handle-1`, diff --git a/packages/react-db/tests/useLiveSuspenseQuery.test.tsx b/packages/react-db/tests/useLiveSuspenseQuery.test.tsx index 2b4d56aa55..84fc8664cd 100644 --- a/packages/react-db/tests/useLiveSuspenseQuery.test.tsx +++ b/packages/react-db/tests/useLiveSuspenseQuery.test.tsx @@ -375,6 +375,11 @@ describe(`useLiveSuspenseQuery`, () => { outcome: `timeout`, }, { name: `network readiness`, networkTimeoutMs: 60_000, outcome: `ready` }, + { + name: `no completed snapshot`, + networkTimeoutMs: 0, + outcome: `missing-cache`, + }, { name: `early network failure`, networkTimeoutMs: 60_000, @@ -410,7 +415,7 @@ describe(`useLiveSuspenseQuery`, () => { }, }, }) - let persistedStatus: `loading` | `ready` = `loading` + let persistedStatus: `loading` | `ready` | `unavailable` = `loading` let networkDeadlineAt: number | undefined const listeners = new Set<() => void>() Object.defineProperty( @@ -450,11 +455,14 @@ describe(`useLiveSuspenseQuery`, () => { sync!.begin() sync!.write(initialPersons[0]!) sync!.commit() - persistedStatus = `ready` + persistedStatus = + outcome === `missing-cache` ? `unavailable` : `ready` for (const listener of listeners) listener() }) await waitFor(() => - expect(statusView.result.current.persistedStatus).toBe(`ready`), + expect(statusView.result.current.persistedStatus).toBe( + outcome === `missing-cache` ? `unavailable` : `ready`, + ), ) if (outcome === `pending` || outcome === `timeout`) { await waitFor(() => expect(view.getByText(`John Doe`)).toBeTruthy()) @@ -462,11 +470,14 @@ describe(`useLiveSuspenseQuery`, () => { } else { expect(view.getByText(`Waiting for network`)).toBeTruthy() await act(() => { - if (outcome === `ready`) sync!.markReady() + if (outcome === `ready` || outcome === `missing-cache`) + sync!.markReady() else sync!.markError(new Error(`network failed`)) }) await waitFor(() => expect(view.getByText(`John Doe`)).toBeTruthy()) - expect(source.status).toBe(outcome) + expect(source.status).toBe( + outcome === `missing-cache` ? `ready` : outcome, + ) } } finally { view.unmount()