fix(persistence): fence startup and schema generations

Author: KyleAMathews

.changeset/preserve-resume-baseline-integrity.md

@@ -1,6 +1,7 @@
 ---
 '@tanstack/db-sqlite-persistence-core': minor
 '@tanstack/db': minor
+'@tanstack/browser-db-sqlite-persistence': patch
 '@tanstack/electron-db-sqlite-persistence': minor
 '@tanstack/electric-db-collection': patch
 '@tanstack/query-db-collection': patch

docs/contributing/oracle-coverage.md

@@ -240,10 +240,26 @@ a hostile wrong-answer control, and an explicit statement of remaining limits.
   statement, scan, trigger, or queue-cardinality observations where the
   subsystem promises bounded work. Owners: SQLite resume snapshots, Electric
   acquisition work, and shared-driver scheduling suites.
+- [x] **Startup generation interleavings.** Hold persisted startup between its
+  metadata and hydration snapshots, then cross no write, a managed mutation,
+  and hostile raw loss. Compare public rows with the atomic durable snapshot,
+  and require mutation persistence to wait until the prior stream position is
+  known. Owners: SQLite resume snapshots and Electric resume snapshot races.
+- [x] **Schema-generation fences on cached adapters.** Cross a newer-schema
+  reset with snapshot, row, metadata, delta, position, and index-lifecycle
+  operations from the cached older adapter. Every stale operation rejects;
+  the current adapter remains readable and its index registry remains intact.
+  Owners: SQLite resume snapshots plus browser/electron coordinator routing.
 - [x] **Explicit omission records.** Add a short `Known omissions` section to
   each primary executable owner touched above and keep this map synchronized as
   laws land. An omission record narrows evidence; it does not waive a product
   obligation.
+- [ ] **Atomic active-subset full reload.** Load collection metadata and every
+  active subset from one adapter generation, including filtered and paginated
+  on-demand subsets. A sound implementation needs an atomic multi-subset API;
+  loading all rows or accepting a metadata/row torn pair is not equivalent.
+  RED evidence and the deferred executable placeholder live in
+  `packages/db-sqlite-persistence-core/tests/persisted.test.ts` under R5-007.
 
 ## Deferred contracts and evidence
 

packages/browser-db-sqlite-persistence/src/browser-coordinator.ts

@@ -122,6 +122,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
   private readonly nodeId = safeRandomUUID()
   private readonly dbName: string
   private adapter: AdapterWithPullSince | null
+  private readonly collectionAdapters = new Map<string, AdapterWithPullSince>()
   private readonly channel: BroadcastChannel
   private readonly collections = new Map<string, CollectionState>()
   private readonly pendingRPCs = new Map<string, PendingRPC>()
@@ -133,13 +134,14 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     return this.disposed
   }
 
-  private requireAdapter(): AdapterWithPullSince {
-    if (!this.adapter) {
+  private requireAdapter(collectionId: string): AdapterWithPullSince {
+    const adapter = this.collectionAdapters.get(collectionId) ?? this.adapter
+    if (!adapter) {
       throw new Error(
         `BrowserCollectionCoordinator: adapter not set. Call setAdapter() before using leader-side operations.`,
       )
     }
-    return this.adapter
+    return adapter
   }
 
   constructor(options: BrowserCollectionCoordinatorOptions) {
@@ -160,6 +162,14 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     this.adapter = adapter
   }
 
+  /** Register the schema/mode-specific adapter for one collection. */
+  setCollectionAdapter(
+    collectionId: string,
+    adapter: AdapterWithPullSince,
+  ): void {
+    this.collectionAdapters.set(collectionId, adapter)
+  }
+
   // -----------------------------------------------------------------------
   // PersistedCollectionCoordinator interface
   // -----------------------------------------------------------------------
@@ -223,7 +233,11 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     spec: PersistedIndexSpec,
   ): Promise<void> {
     if (this.isLeader(collectionId)) {
-      await this.requireAdapter().ensureIndex(collectionId, signature, spec)
+      await this.requireAdapter(collectionId).ensureIndex(
+        collectionId,
+        signature,
+        spec,
+      )
       return
     }
 
@@ -305,6 +319,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
 
     this.channel.close()
     this.collections.clear()
+    this.collectionAdapters.clear()
   }
 
   // -----------------------------------------------------------------------
@@ -348,7 +363,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
 
           try {
             // Restore stream position from DB before claiming leadership
-            const adapter = this.requireAdapter()
+            const adapter = this.requireAdapter(collectionId)
             if (adapter.getStreamPosition) {
               const pos = await adapter.getStreamPosition(collectionId)
               state.latestTerm = pos.latestTerm
@@ -610,7 +625,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     },
   ): Promise<RPCResponse> {
     await this.withWriterLock(() =>
-      this.requireAdapter().ensureIndex(
+      this.requireAdapter(collectionId).ensureIndex(
         collectionId,
         request.signature,
         request.spec,
@@ -676,7 +691,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     }
 
     await this.withWriterLock(() =>
-      this.requireAdapter().applyCommittedTx(collectionId, tx),
+      this.requireAdapter(collectionId).applyCommittedTx(collectionId, tx),
     )
 
     // Track envelope for dedup
@@ -736,7 +751,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
   ): Promise<PullSinceResponse> {
     const state = this.collections.get(collectionId)
 
-    const adapter = this.requireAdapter()
+    const adapter = this.requireAdapter(collectionId)
     if (!adapter.pullSince) {
       return {
         type: `rpc:pullSince:res`,

packages/browser-db-sqlite-persistence/src/browser-persistence.ts

@@ -130,34 +130,39 @@ export function createBrowserWASQLitePersistence(
       ...(schemaVersion === undefined ? {} : { schemaVersion }),
     })
     adapterCache.set(cacheKey, adapter)
-
-    // Wire the adapter into the multi-tab coordinator so it can handle
-    // leader-side RPCs (applyCommittedTx, pullSince, ensureIndex, etc.)
-    if (resolvedCoordinator instanceof BrowserCollectionCoordinator) {
-      resolvedCoordinator.setAdapter(adapter)
-    }
-
     return adapter
   }
 
   const createCollectionPersistence = (
+    collectionId: string | undefined,
     mode: PersistedCollectionMode,
     schemaVersion: number | undefined,
-  ): PersistedCollectionPersistence => ({
-    adapter: getAdapterForCollection(mode, schemaVersion),
-    coordinator: resolvedCoordinator,
-  })
+  ): PersistedCollectionPersistence => {
+    const adapter = getAdapterForCollection(mode, schemaVersion)
+    if (resolvedCoordinator instanceof BrowserCollectionCoordinator) {
+      if (collectionId === undefined) {
+        resolvedCoordinator.setAdapter(adapter)
+      } else {
+        resolvedCoordinator.setCollectionAdapter(collectionId, adapter)
+      }
+    }
+    return {
+      adapter,
+      coordinator: resolvedCoordinator,
+    }
+  }
 
   const defaultPersistence = createCollectionPersistence(
+    undefined,
     `sync-absent`,
     undefined,
   )
 
   return {
     ...defaultPersistence,
-    resolvePersistenceForCollection: ({ mode, schemaVersion }) =>
-      createCollectionPersistence(mode, schemaVersion),
+    resolvePersistenceForCollection: ({ collectionId, mode, schemaVersion }) =>
+      createCollectionPersistence(collectionId, mode, schemaVersion),
     resolvePersistenceForMode: (mode) =>
-      createCollectionPersistence(mode, undefined),
+      createCollectionPersistence(undefined, mode, undefined),
   }
 }

packages/browser-db-sqlite-persistence/tests/browser-coordinator.test.ts

@@ -1,5 +1,10 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { BrowserCollectionCoordinator } from '../src/browser-coordinator'
+import {
+  createBrowserWASQLitePersistence,
+  persistedCollectionOptions,
+} from '../src'
+import { createWASQLiteTestDatabase } from './helpers/wa-sqlite-test-db'
 import type { BrowserCollectionCoordinatorOptions } from '../src/browser-coordinator'
 import type { PersistenceAdapter } from '@tanstack/db-sqlite-persistence-core'
 
@@ -295,6 +300,91 @@ describe(`BrowserCollectionCoordinator`, () => {
       coord2.dispose()
     })
 
+    it.each([
+      { schemaV1: 1, schemaV2: 2 },
+      { schemaV1: 2, schemaV2: 4 },
+    ])(
+      `routes each collection through its schema-version adapter: $schemaV1/$schemaV2`,
+      async ({ schemaV1, schemaV2 }) => {
+        const database = createWASQLiteTestDatabase({ filename: `:memory:` })
+        const coordinator = new BrowserCollectionCoordinator({
+          dbName: `schema-routed-db`,
+        })
+        const persistence = createBrowserWASQLitePersistence({
+          database,
+          coordinator,
+        })
+        const collectionV1 = `schema-routed-v1`
+        const collectionV2 = `schema-routed-v2`
+
+        try {
+          const optionsV1 = persistedCollectionOptions<
+            { id: string; title: string },
+            string
+          >({
+            id: collectionV1,
+            schemaVersion: schemaV1,
+            getKey: (row) => row.id,
+            persistence,
+          })
+          await optionsV1.persistence.adapter.loadResumeSnapshot(collectionV1)
+
+          const optionsV2 = persistedCollectionOptions<
+            { id: string; title: string },
+            string
+          >({
+            id: collectionV2,
+            schemaVersion: schemaV2,
+            getKey: (row) => row.id,
+            persistence,
+          })
+          await optionsV2.persistence.adapter.loadResumeSnapshot(collectionV2)
+
+          coordinator.subscribe(collectionV1, () => {})
+          coordinator.subscribe(collectionV2, () => {})
+          await vi.waitFor(() => {
+            expect(coordinator.isLeader(collectionV1)).toBe(true)
+            expect(coordinator.isLeader(collectionV2)).toBe(true)
+          })
+
+          const [resultV1, resultV2] = await Promise.all([
+            coordinator.requestApplyLocalMutations(collectionV1, [
+              {
+                mutationId: `mutation-v1`,
+                type: `insert`,
+                key: `v1`,
+                value: { id: `v1`, title: `schema one` },
+              },
+            ]),
+            coordinator.requestApplyLocalMutations(collectionV2, [
+              {
+                mutationId: `mutation-v2`,
+                type: `insert`,
+                key: `v2`,
+                value: { id: `v2`, title: `schema two` },
+              },
+            ]),
+          ])
+
+          expect(resultV1.ok).toBe(true)
+          expect(resultV2.ok).toBe(true)
+          expect(
+            await optionsV1.persistence.adapter.loadSubset(collectionV1, {}),
+          ).toMatchObject([
+            { key: `v1`, value: { id: `v1`, title: `schema one` } },
+          ])
+          expect(
+            await optionsV2.persistence.adapter.loadSubset(collectionV2, {}),
+          ).toMatchObject([
+            { key: `v2`, value: { id: `v2`, title: `schema two` } },
+          ])
+        } finally {
+          coordinator.dispose()
+          await Promise.resolve(database.close?.())
+        }
+      },
+    )
+
     it(`different collections have independent leaders`, async () => {
       const coord1 = createCoordinator()
       const coord2 = createCoordinator()

packages/db-sqlite-persistence-core/src/persisted.ts

@@ -936,13 +936,14 @@ class PersistedCollectionRuntime<
     }
   }
 
-  async ensureStarted(): Promise<void> {
+  ensureStarted(): Promise<void> {
     if (this.startPromise) {
       return this.startPromise
     }
 
     const lifecycleGeneration = this.lifecycleGeneration
     this.startPromise = this.startInternal(lifecycleGeneration)
+    void this.startPromise.catch(() => undefined)
     return this.startPromise
   }
 
@@ -1247,6 +1248,10 @@ class PersistedCollectionRuntime<
     }
 
     await this.applyMutex.run(async () => {
+      // Startup metadata establishes the durable term/sequence boundary. A
+      // mutation admitted before it resolves must wait rather than allocate a
+      // default position that can collide with an already-applied transaction.
+      await this.ensureStartupMetadataLoaded()
       const acceptedMutationIds =
         await this.persistCollectionMutationsUnsafe(mutations)
       const acceptedMutationIdSet = new Set(acceptedMutationIds)
@@ -1479,18 +1484,27 @@ class PersistedCollectionRuntime<
     resetEpoch: number
   }): void {
     const generation = this.getResumeSnapshotGeneration(snapshot)
-    // Moving between equally uncertified snapshots cannot make either
-    // trustworthy; preserve that status so sync performs a fresh replacement.
-    const remainsUncertified =
-      this.persistedKeySetEvidence?.status === snapshot.keySet?.status &&
-      snapshot.keySet?.status !== `consistent`
+    const previousEvidenceStatus = this.persistedKeySetEvidence?.status
+    // An uncertified sync baseline never authorized a persisted resume cursor,
+    // so a later atomic snapshot may replace its evidence while the source
+    // performs the already-required fresh snapshot. Local-only collections
+    // have no remote cursor to fence; a managed write may advance a still-
+    // consistent SQLite baseline during startup. Incompatible evidence remains
+    // fail-closed in both modes.
+    const mayAcceptUnownedGeneration =
+      (this.mode === `sync-present` &&
+        previousEvidenceStatus !== `consistent` &&
+        previousEvidenceStatus !== `incompatible`) ||
+      (this.mode === `sync-absent` &&
+        previousEvidenceStatus === `consistent` &&
+        snapshot.keySet?.status === `consistent`)
     this.observeStreamPosition(
       snapshot.latestTerm,
       snapshot.latestSeq,
       snapshot.latestRowVersion,
     )
     this.persistedKeySetEvidence =
-      this.isExpectedResumeGeneration(generation) || remainsUncertified
+      this.isExpectedResumeGeneration(generation) || mayAcceptUnownedGeneration
         ? snapshot.keySet
         : { status: `incompatible` }
   }