diff --git a/.changeset/fix-shared-sqlite-hydration-fairness.md b/.changeset/fix-shared-sqlite-hydration-fairness.md new file mode 100644 index 0000000000..d2055d929a --- /dev/null +++ b/.changeset/fix-shared-sqlite-hydration-fairness.md @@ -0,0 +1,7 @@ +--- +'@tanstack/db-sqlite-persistence-core': patch +'@tanstack/browser-db-sqlite-persistence': patch +'@tanstack/electron-db-sqlite-persistence': patch +--- + +Prevent cold SQLite hydrations from being starved by unrelated queued writes when collections share a browser driver. Schedule each complete hydrate fairly while preserving transaction atomicity and leader-local persistence coordination. diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index a340502b46..6b3756ff74 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -74,6 +74,11 @@ jobs: cd examples/react/start-ssr-e2e pnpm exec playwright install --with-deps chromium + - name: Run Browser SQLite OPFS fairness E2E tests + run: | + cd packages/browser-db-sqlite-persistence + pnpm test:opfs-fairness + - name: Run React Start SSR E2E tests run: | cd examples/react/start-ssr-e2e diff --git a/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.html b/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.html new file mode 100644 index 0000000000..215878cfd3 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.html @@ -0,0 +1,13 @@ + + + + + + + Shared driver OPFS fairness oracle + + + running + + + diff --git a/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.spec.ts b/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.spec.ts new file mode 100644 index 0000000000..849de99ccc --- /dev/null +++ b/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.spec.ts @@ -0,0 +1,102 @@ +/** + * Browser checkpoint assertions for the OPFS refinement. Expected public rows + * are rebuilt from the scenario IDs rather than from production output. The + * neutral case proves cold-query reach; the storm case requires no K=1 + * violation. Failure-before-checkpoint, semantic mismatch, driver cleanup, and + * OPFS cleanup remain distinct outcomes so setup or teardown cannot satisfy the + * scheduling law. + */ +import { expect, test } from '@playwright/test' +import type { Page } from '@playwright/test' +import type { OPFSOracleResult } from './shared-driver-fairness.opfs' + +async function readOracleResult( + page: Page, + mode: `neutral` | `storm`, +): Promise { + let rejectPageError!: (error: Error) => void + const pageError = new Promise((_resolve, reject) => { + rejectPageError = reject + }) + void pageError.catch(() => undefined) + const onPageError = (error: Error) => { + rejectPageError( + new Error( + `OPFS fairness page failed before publishing a result: ${error.message}`, + ), + ) + } + page.on(`pageerror`, onPageError) + + try { + await page.goto(`/e2e/shared-driver-fairness.opfs.html?mode=${mode}`) + await Promise.race([ + page.waitForFunction( + () => window.__tanstackDriverFairnessOracle !== undefined, + ), + pageError, + ]) + return page.evaluate(() => window.__tanstackDriverFairnessOracle!) + } finally { + page.off(`pageerror`, onPageError) + } +} + +function expectedHydratedCollections(scenarioId: string, count: number) { + return Array.from({ length: count }, (_, index) => ({ + collectionId: `${scenarioId}-hydrate-${index}`, + rows: [ + { id: `row-${index}-0`, value: index * 10 }, + { id: `row-${index}-1`, value: index * 10 + 1 }, + ], + })) +} + +test(`real Chromium OPFS fixture reaches and cleans up cold hydration`, async ({ + page, +}) => { + const result = await readOracleResult(page, `neutral`) + + if (result.status !== `complete`) throw new Error(result.primaryFailure) + expect(result.provider).toBe(`Chromium OPFSCoopSyncVFS worker`) + expect(result.observation.admittedHydrateIds).toHaveLength(2) + // These are actual public Collection rows captured after preload, compared + // with seed values built independently by this browser assertion. + expect(result.observation.hydratedCollections).toEqual( + expectedHydratedCollections(`opfs-neutral-reach`, 2), + ) + expect( + result.observation.rawDequeues.some((entry) => + entry.sql.startsWith(`SELECT key, value, metadata, row_version FROM`), + ), + ).toBe(true) + expect(result.observation.cleanupFailures).toEqual([]) + expect(result.opfsCleanupFailures).toEqual([]) +}) + +test(`real Chromium OPFS fixture bounds pending cold hydration behind persists`, async ({ + page, +}) => { + const result = await readOracleResult(page, `storm`) + + if (result.status !== `complete`) throw new Error(result.primaryFailure) + expect(result.provider).toBe(`Chromium OPFSCoopSyncVFS worker`) + expect(result.observation.admittedHydrateIds).toHaveLength(4) + expect(result.observation.hydratedCollections).toEqual( + expectedHydratedCollections(`opfs-fixed-persist-storm`, 4), + ) + // This is the semantic RED checkpoint. Setup, wall time, and cleanup are + // reported independently and cannot satisfy this assertion. + if (result.violation !== undefined) { + throw new Error( + `real OPFS fairness mismatch: ${JSON.stringify(result.violation)}; ` + + `logical completion order: ${JSON.stringify(result.observation.logicalCompletionOrder)}; ` + + `driver admissions: ${result.observation.driverAdmissions.length}; ` + + `raw dequeues: ${result.observation.rawDequeues.length}; ` + + `driver cleanup diagnostics: ${JSON.stringify(result.observation.cleanupFailures)}; ` + + `OPFS cleanup diagnostics: ${JSON.stringify(result.opfsCleanupFailures)}`, + ) + } + expect(result.observation.cleanupFailures).toEqual([]) + expect(result.opfsCleanupFailures).toEqual([]) +}) diff --git a/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.ts b/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.ts new file mode 100644 index 0000000000..a443f680f2 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/e2e/shared-driver-fairness.opfs.ts @@ -0,0 +1,160 @@ +/** + * Real-provider refinement of the shared-driver fairness oracle. The page runs + * the same legal neutral and fixed-storm histories through public `preload()`, + * the browser/core adapter boundary, and Chromium's OPFSCoopSyncVFS worker. It + * freezes logical completions, public rows, raw dequeue reach, and the K=1 + * violation result before cleanup, then reports provider cleanup separately. + * This fixture adds real OPFS/worker evidence; it does not claim multi-tab, + * multi-process, non-Chromium, latency, or unbounded-eventuality coverage. + */ +import { openBrowserWASQLiteOPFSDatabase } from '../src/index' +import { + findSharedDriverFairnessViolation, + observeSharedDriverFairness, +} from '../tests/shared-driver-fairness-oracle' +import type { + SharedDriverFairnessObservation, + SharedDriverFairnessScenario, + SharedDriverFairnessViolation, +} from '../tests/shared-driver-fairness-oracle' + +export type OPFSOracleResult = + | { + status: `complete` + provider: `Chromium OPFSCoopSyncVFS worker` + observation: SharedDriverFairnessObservation + violation: SharedDriverFairnessViolation | undefined + opfsCleanupFailures: ReadonlyArray + } + | { + status: `failed-before-checkpoint` + provider: `Chromium OPFSCoopSyncVFS worker` + primaryFailure: string + opfsCleanupFailures: ReadonlyArray + } + +declare global { + interface Window { + __tanstackDriverFairnessOracle?: OPFSOracleResult + } +} + +async function removeOPFSArtifacts( + databaseName: string, +): Promise> { + const failures: Array = [] + const root = await navigator.storage.getDirectory() + for (const suffix of [``, `-journal`, `-wal`]) { + try { + await root.removeEntry(`${databaseName}${suffix}`) + } catch (error) { + if (!(error instanceof DOMException && error.name === `NotFoundError`)) { + failures.push( + `${databaseName}${suffix}: ${error instanceof Error ? error.message : String(error)}`, + ) + } + } + } + + // OPFSCoopSyncVFS creates one private temporary access-handle directory per + // worker. This page owns its isolated origin for the fixture run. + const iterableRoot = root as FileSystemDirectoryHandle & { + entries: () => AsyncIterableIterator<[string, FileSystemHandle]> + } + for await (const [name, handle] of iterableRoot.entries()) { + if (handle.kind !== `directory` || !name.startsWith(`.ahp-`)) continue + try { + await root.removeEntry(name, { recursive: true }) + } catch (error) { + failures.push( + `${name}: ${error instanceof Error ? error.message : String(error)}`, + ) + } + } + return failures +} + +function scenarioFromLocation(): SharedDriverFairnessScenario { + const mode = new URL(location.href).searchParams.get(`mode`) + if (mode === `neutral`) { + return { + id: `opfs-neutral-reach`, + work: [0, 1].map((index) => ({ + kind: `hydrate` as const, + id: `hydrate-${index}`, + seededRows: [ + { id: `row-${index}-0`, value: index * 10 }, + { id: `row-${index}-1`, value: index * 10 + 1 }, + ], + })), + } + } + return { + id: `opfs-fixed-persist-storm`, + work: [ + ...Array.from({ length: 5 }, (_, index) => ({ + kind: `persist` as const, + id: `persist-${index}`, + mutationsPerPersist: 2, + })), + ...Array.from({ length: 4 }, (_, index) => ({ + kind: `hydrate` as const, + id: `hydrate-${index}`, + seededRows: [ + { id: `row-${index}-0`, value: index * 10 }, + { id: `row-${index}-1`, value: index * 10 + 1 }, + ], + })), + ], + } +} + +async function run(): Promise { + const status = document.querySelector(`#oracle-status`) + const databaseName = `ws5b-${crypto.randomUUID()}.sqlite` + let observation: SharedDriverFairnessObservation | undefined + let violation: SharedDriverFairnessViolation | undefined + let primaryFailure: string | undefined + let opfsCleanupFailures: ReadonlyArray = [] + try { + observation = await observeSharedDriverFairness( + () => openBrowserWASQLiteOPFSDatabase({ databaseName }), + scenarioFromLocation(), + ) + // Freeze the reached semantic checkpoint before attempting OPFS cleanup. + violation = findSharedDriverFairnessViolation(observation) + } catch (error) { + primaryFailure = error instanceof Error ? error.message : String(error) + } + + try { + opfsCleanupFailures = await removeOPFSArtifacts(databaseName) + } catch (cleanupError) { + opfsCleanupFailures = [ + cleanupError instanceof Error + ? cleanupError.message + : String(cleanupError), + ] + } + + if (observation) { + window.__tanstackDriverFairnessOracle = { + status: `complete`, + provider: `Chromium OPFSCoopSyncVFS worker`, + observation, + violation, + opfsCleanupFailures, + } + if (status) status.value = `complete` + } else { + window.__tanstackDriverFairnessOracle = { + status: `failed-before-checkpoint`, + provider: `Chromium OPFSCoopSyncVFS worker`, + primaryFailure: primaryFailure ?? `unknown failure before checkpoint`, + opfsCleanupFailures, + } + if (status) status.value = `failed-before-checkpoint` + } +} + +void run() diff --git a/packages/browser-db-sqlite-persistence/package.json b/packages/browser-db-sqlite-persistence/package.json index 9bae7bb83c..26b2b70fef 100644 --- a/packages/browser-db-sqlite-persistence/package.json +++ b/packages/browser-db-sqlite-persistence/package.json @@ -23,6 +23,8 @@ "dev": "vite build --watch", "lint": "eslint . --fix", "test": "vitest --run", + "test:oracles": "vitest --run tests/shared-driver-fairness-oracle.test.ts", + "test:opfs-fairness": "playwright test --config playwright.opfs.config.ts", "test:e2e": "pnpm --filter @tanstack/db-ivm build && pnpm --filter @tanstack/db build && pnpm --filter @tanstack/db-sqlite-persistence-core build && pnpm --filter @tanstack/browser-db-sqlite-persistence build && vitest --config vitest.e2e.config.ts --run" }, "type": "module", @@ -56,6 +58,7 @@ }, "devDependencies": { "@journeyapps/wa-sqlite": "^1.4.1", + "@playwright/test": "^1.60.0", "@types/better-sqlite3": "^7.6.13", "@vitest/coverage-istanbul": "^3.2.4", "better-sqlite3": "^12.6.2" diff --git a/packages/browser-db-sqlite-persistence/playwright.opfs.config.ts b/packages/browser-db-sqlite-persistence/playwright.opfs.config.ts new file mode 100644 index 0000000000..ecfa8af0a6 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/playwright.opfs.config.ts @@ -0,0 +1,25 @@ +import { defineConfig } from '@playwright/test' + +const baseURL = `http://127.0.0.1:4185` +const browserChannel = + process.env.PLAYWRIGHT_CHANNEL ?? (process.env.CI ? undefined : `chrome`) + +export default defineConfig({ + testDir: `./e2e`, + testMatch: `shared-driver-fairness.opfs.spec.ts`, + timeout: 60_000, + fullyParallel: false, + workers: 1, + use: { + baseURL, + ...(browserChannel ? { channel: browserChannel } : {}), + headless: true, + trace: `retain-on-failure`, + }, + webServer: { + command: `vite --config vite.opfs.config.ts --host 127.0.0.1 --port 4185`, + reuseExistingServer: false, + timeout: 120_000, + url: `${baseURL}/e2e/shared-driver-fairness.opfs.html`, + }, +}) diff --git a/packages/browser-db-sqlite-persistence/src/browser-coordinator.ts b/packages/browser-db-sqlite-persistence/src/browser-coordinator.ts index 1babddc5a7..995a63ac94 100644 --- a/packages/browser-db-sqlite-persistence/src/browser-coordinator.ts +++ b/packages/browser-db-sqlite-persistence/src/browser-coordinator.ts @@ -1,6 +1,7 @@ import { safeRandomUUID } from '@tanstack/db-sqlite-persistence-core' import type { ApplyLocalMutationsResponse, + HydrationPersistenceAdapter, PersistedCollectionCoordinator, PersistedIndexSpec, PersistedMutationEnvelope, @@ -221,9 +222,15 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina collectionId: string, signature: string, spec: PersistedIndexSpec, + scopedAdapter?: HydrationPersistenceAdapter, ): Promise { if (this.isLeader(collectionId)) { - await this.requireAdapter().ensureIndex(collectionId, signature, spec) + // A scoped adapter is a leader-local capability and never crosses RPC. + await (scopedAdapter ?? this.requireAdapter()).ensureIndex( + collectionId, + signature, + spec, + ) return } @@ -270,13 +277,19 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina async pullSince( collectionId: string, fromRowVersion: number, + scopedAdapter?: HydrationPersistenceAdapter, ): Promise { if (this.isLeader(collectionId)) { - return this.handlePullSince(collectionId, { - type: `rpc:pullSince:req`, - rpcId: safeRandomUUID(), - fromRowVersion, - }) + // A scoped adapter is a leader-local capability and never crosses RPC. + return this.handlePullSince( + collectionId, + { + type: `rpc:pullSince:req`, + rpcId: safeRandomUUID(), + fromRowVersion, + }, + scopedAdapter, + ) } return this.sendRPC(collectionId, { @@ -733,10 +746,11 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina rpcId: string fromRowVersion: number }, + scopedAdapter?: HydrationPersistenceAdapter, ): Promise { const state = this.collections.get(collectionId) - const adapter = this.requireAdapter() + const adapter = scopedAdapter ?? this.requireAdapter() if (!adapter.pullSince) { return { type: `rpc:pullSince:res`, diff --git a/packages/browser-db-sqlite-persistence/src/wa-sqlite-driver.ts b/packages/browser-db-sqlite-persistence/src/wa-sqlite-driver.ts index c1ae91b9e0..3a6be4c901 100644 --- a/packages/browser-db-sqlite-persistence/src/wa-sqlite-driver.ts +++ b/packages/browser-db-sqlite-persistence/src/wa-sqlite-driver.ts @@ -1,4 +1,7 @@ -import { InvalidPersistedCollectionConfigError } from '@tanstack/db-sqlite-persistence-core' +import { + InvalidPersistedCollectionConfigError, + SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY, +} from '@tanstack/db-sqlite-persistence-core' import type { SQLiteDriver } from '@tanstack/db-sqlite-persistence-core' export type BrowserWASQLiteDatabase = { @@ -36,6 +39,7 @@ function assertDatabaseShape( } export class BrowserWASQLiteDriver implements SQLiteDriver { + readonly [SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY] = {} private readonly database: BrowserWASQLiteDatabase private queue: Promise = Promise.resolve() private nextSavepointId = 1 @@ -46,29 +50,33 @@ export class BrowserWASQLiteDriver implements SQLiteDriver { this.database = options.database } - async exec(sql: string): Promise { - await this.enqueue(async () => { + exec(sql: string): Promise { + return this.enqueue(async () => { await this.database.execute(sql) }) } - async query( + query( sql: string, params: ReadonlyArray = [], ): Promise> { return this.enqueue(() => this.database.execute(sql, params)) } - async run(sql: string, params: ReadonlyArray = []): Promise { - await this.enqueue(async () => { + run(sql: string, params: ReadonlyArray = []): Promise { + return this.enqueue(async () => { await this.database.execute(sql, params) }) } - async transaction( + transaction( fn: (transactionDriver: SQLiteDriver) => Promise, ): Promise { - assertTransactionCallbackHasDriverArg(fn) + try { + assertTransactionCallbackHasDriverArg(fn) + } catch (error) { + return Promise.reject(error) + } return this.enqueue(async () => { await this.database.execute(`BEGIN IMMEDIATE`) @@ -87,7 +95,7 @@ export class BrowserWASQLiteDriver implements SQLiteDriver { }) } - async transactionWithDriver( + transactionWithDriver( fn: (transactionDriver: SQLiteDriver) => Promise, ): Promise { return this.transaction(fn) @@ -144,6 +152,14 @@ export class BrowserWASQLiteDriver implements SQLiteDriver { private enqueue(operation: () => Promise | T): Promise { const queuedOperation = this.queue.then(operation, operation) + // Brand the exact Promise returned to callers. Transparent wrappers may + // preserve this identity for late discovery; wrappers that create a new + // Promise must forward the driver key before adapter construction. + Object.defineProperty( + queuedOperation, + SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY, + { value: this[SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY] }, + ) this.queue = queuedOperation.then( () => undefined, () => undefined, diff --git a/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.test.ts b/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.test.ts new file mode 100644 index 0000000000..7fc27c5d67 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.test.ts @@ -0,0 +1,295 @@ +/** + * Node campaign for the shared-driver K=1 contract documented in + * `shared-driver-fairness-oracle.ts`. Fixed histories prove neutral reach and a + * persist storm; generated legal histories vary both lane sizes, mutation + * width, and tail order. Public rows and logical completion checkpoints are + * checked independently of production scheduling, and an executable + * persist-first FIFO mutant proves the checker rejects the original fault. + * Failures replay through TANSTACK_DB_DRIVER_FAIRNESS_SEED/PATH. + */ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import fc from 'fast-check' +import { describe, expect, it } from 'vitest' +import { + SHARED_DRIVER_FAIRNESS_BOUND, + createPersistFirstFaultObservation, + findSharedDriverFairnessViolation, + observeSharedDriverFairness, +} from './shared-driver-fairness-oracle' +import { createWASQLiteTestDatabase } from './helpers/wa-sqlite-test-db' +import type { + SharedDriverFairnessObservation, + SharedDriverFairnessOptions, + SharedDriverFairnessScenario, + SharedDriverFairnessWork, +} from './shared-driver-fairness-oracle' + +const DEFAULT_SEED = 165_905 +const replaySeed = Number( + process.env.TANSTACK_DB_DRIVER_FAIRNESS_SEED ?? DEFAULT_SEED, +) +const replayPath = process.env.TANSTACK_DB_DRIVER_FAIRNESS_PATH + +type WorkKind = SharedDriverFairnessWork[`kind`] + +function createScenario( + id: string, + orderedKinds: ReadonlyArray, + mutationsPerPersist = 1, +): SharedDriverFairnessScenario { + let hydrateIndex = 0 + let persistIndex = 0 + return { + id, + work: orderedKinds.map((kind) => { + if (kind === `persist`) { + const index = persistIndex++ + return { + kind, + id: `persist-${index}`, + mutationsPerPersist, + } + } + const index = hydrateIndex++ + return { + kind, + id: `hydrate-${index}`, + seededRows: [ + { id: `row-${index}-0`, value: index * 10 }, + { id: `row-${index}-1`, value: index * 10 + 1 }, + ], + } + }), + } +} + +function expectedHydratedCollections(scenario: SharedDriverFairnessScenario) { + return scenario.work.flatMap((work) => + work.kind === `hydrate` + ? [ + { + collectionId: `${scenario.id}-${work.id}`, + rows: work.seededRows.map((row) => ({ ...row })), + }, + ] + : [], + ) +} + +async function withNodeScenario( + scenario: SharedDriverFairnessScenario, + assertion: ( + observation: SharedDriverFairnessObservation, + ) => void | Promise, + options: SharedDriverFairnessOptions = {}, +): Promise { + const directory = mkdtempSync(join(tmpdir(), `db-driver-fairness-`)) + let primaryFailure: unknown + try { + const observation = await observeSharedDriverFairness( + () => + createWASQLiteTestDatabase({ + filename: join(directory, `state.sqlite`), + }), + scenario, + options, + ) + await assertion(observation) + } catch (error) { + primaryFailure = error + } + + let nodeCleanupFailure: unknown + try { + rmSync(directory, { recursive: true, force: true }) + } catch (error) { + nodeCleanupFailure = error + } + + if (primaryFailure !== undefined) { + if (nodeCleanupFailure !== undefined) { + const primaryMessage = + primaryFailure instanceof Error + ? primaryFailure.message + : String(primaryFailure) + const cleanupMessage = + nodeCleanupFailure instanceof Error + ? nodeCleanupFailure.message + : String(nodeCleanupFailure) + throw new Error( + `${primaryMessage}; node cleanup diagnostics: ${cleanupMessage}`, + ) + } + throw primaryFailure + } + if (nodeCleanupFailure !== undefined) throw nodeCleanupFailure +} + +function expectObservationReach( + observation: SharedDriverFairnessObservation, +): void { + const expectedHydrates = expectedHydratedCollections(observation.scenario) + expect(observation.admittedHydrateIds).toEqual( + expectedHydrates.map(({ collectionId }) => collectionId), + ) + expect(observation.hydrationCompletions).toHaveLength(expectedHydrates.length) + // Actual rows come from the public Collection after preload. The expected + // rows above are derived independently from the generated seed history. + expect(observation.hydratedCollections).toEqual(expectedHydrates) +} + +function expectFairObservation( + observation: SharedDriverFairnessObservation, +): void { + expectObservationReach(observation) + const violation = findSharedDriverFairnessViolation( + observation, + SHARED_DRIVER_FAIRNESS_BOUND, + ) + if (violation) { + throw new Error( + `shared-driver fairness mismatch at ${violation.checkpoint.collectionId}: ` + + `${violation.checkpoint.completedPersistIds.length} persists completed ` + + `(maximum ${violation.expectedMaximumCompletedPersists}), ` + + `${violation.checkpoint.pendingPersistCount} remained pending ` + + `(minimum ${violation.expectedMinimumPendingPersists}); ` + + `permitted completed ids: ${JSON.stringify(violation.permittedCompletedPersistIds)}; ` + + `cleanup diagnostics: ${JSON.stringify(observation.cleanupFailures)}`, + ) + } + expect(observation.cleanupFailures).toEqual([]) +} + +const generatedHistoryArbitrary = fc + .record({ + hydrateCount: fc.integer({ min: 2, max: 7 }), + persistCount: fc.integer({ min: 2, max: 7 }), + mutationsPerPersist: fc.integer({ min: 1, max: 3 }), + }) + .chain(({ hydrateCount, persistCount, mutationsPerPersist }) => { + const tail = [ + ...Array.from({ length: persistCount - 1 }, (_, index) => ({ + kind: `persist` as const, + token: `p${index + 1}`, + })), + ...Array.from({ length: hydrateCount }, (_, index) => ({ + kind: `hydrate` as const, + token: `h${index}`, + })), + ] + return fc + .shuffledSubarray(tail, { + minLength: tail.length, + maxLength: tail.length, + }) + .map((shuffledTail) => ({ + hydrateCount, + persistCount, + mutationsPerPersist, + orderedKinds: [ + `persist` as const, + ...shuffledTail.map(({ kind }) => kind), + ], + replayHistory: `p0,${shuffledTail.map(({ token }) => token).join(`,`)}`, + })) + }) + +describe(`shared BrowserWASQLiteDriver fairness oracle`, () => { + it(`reaches cold hydration and preserves independently seeded public rows without queued persists`, async () => { + await withNodeScenario( + createScenario(`neutral-reach`, [`hydrate`, `hydrate`, `hydrate`]), + (observation) => { + expectFairObservation(observation) + expect( + observation.rawDequeues.some((entry) => + entry.sql.startsWith( + `SELECT key, value, metadata, row_version FROM`, + ), + ), + ).toBe(true) + }, + ) + }) + + it(`completes a pending cold hydrate before an unrelated persist backlog drains`, async () => { + await withNodeScenario( + createScenario( + `fixed-persist-storm`, + [`persist`, `persist`, `persist`, `hydrate`], + 2, + ), + expectFairObservation, + ) + }) + + it(`bounds persist completions for generated ordered cold-hydrate/persist histories`, async () => { + await fc.assert( + fc.asyncProperty( + generatedHistoryArbitrary, + async ({ + hydrateCount, + persistCount, + mutationsPerPersist, + orderedKinds, + replayHistory, + }) => { + const observationId = + `generated-h${hydrateCount}-p${persistCount}-m${mutationsPerPersist}-` + + replayHistory.replaceAll(`,`, `-`) + await withNodeScenario( + createScenario(observationId, orderedKinds, mutationsPerPersist), + expectFairObservation, + ) + }, + ), + { + seed: replaySeed, + path: replayPath, + numRuns: 12, + endOnFailure: true, + verbose: 2, + }, + ) + }) + + it(`kills a persist-first FIFO scheduling mutant through the real package path`, async () => { + await withNodeScenario( + createScenario( + `persist-first-fixture-fault`, + [`persist`, `persist`, `persist`, `persist`, `hydrate`], + 1, + ), + (observation) => { + expectObservationReach(observation) + const violation = findSharedDriverFairnessViolation(observation) + if (!violation) { + throw new Error( + `persist-first scheduling mutant escaped the fairness checker; ` + + `cleanup diagnostics: ${JSON.stringify(observation.cleanupFailures)}`, + ) + } + expect(violation.checkpoint.completedPersistIds).toHaveLength(4) + expect(violation.checkpoint.pendingPersistCount).toBe(0) + expect(observation.cleanupFailures).toEqual([]) + }, + { schedulingFault: `persist-first-fifo` }, + ) + }) + + it(`calibrates the checker against a synthetic persist-first observation`, () => { + const scenario = createScenario( + `persist-first-checker-calibration`, + [`persist`, `persist`, `persist`, `persist`, `hydrate`], + 1, + ) + const fault = createPersistFirstFaultObservation(scenario) + + expect(findSharedDriverFairnessViolation(fault)).toMatchObject({ + checkpoint: fault.hydrationCompletions[0], + expectedMaximumCompletedPersists: 1, + expectedMinimumPendingPersists: 3, + }) + }) +}) diff --git a/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.ts b/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.ts new file mode 100644 index 0000000000..a2cb29b5d9 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/tests/shared-driver-fairness-oracle.ts @@ -0,0 +1,745 @@ +/** + * # When does a cold hydrate get a turn on a shared SQLite driver? + * + * Contract and source: RFC #1659 accepts K=1 complete-logical-cold-hydrate + * scheduling. The persist already executing when the storm begins is + * non-preemptible. After that, at most one additional persist may complete + * between consecutive hydrate completions, with FIFO identity preserved + * inside the hydrate and persist lanes. + * + * History grammar and domain: a legal storm has unique work IDs, starts with + * that already-running persist when any persist exists, and then permutes + * complete hydrate and persist requests. Hydrates contain nonempty unique + * seeded rows; persists contain one or more mutations. The generated campaign + * varies 2..7 hydrates, 2..7 persists, 1..3 mutations per persist, and sampled + * tail permutations. Neutral histories contain only hydrates. + * + * Independent model and production boundary: `createFairnessReference` + * computes permitted completed persist IDs from the ordered history and K; it + * does not import or simulate the production scheduler. The driver exercises + * public `Collection.preload()` through persisted collection options, the core + * adapter, and one real `BrowserWASQLiteDriver`. Each preload completion is a + * checkpoint after the complete logical hydrate, not after an individual SQL + * statement. + * + * Observed public facts: admitted and completed logical IDs, completed and + * pending persists at every hydrate checkpoint, independently seeded public + * collection rows, raw SQL dequeue reach, and cleanup diagnostics. Known + * omissions: the oracle does not establish elapsed-time latency, unbounded + * eventuality, multi-process coordination, or a browser matrix; the Chromium + * OPFS fixture separately refines the provider boundary. + * + * Challenge and replay: the executable persist-first FIFO driver must violate + * the same K=1 checker while using the public/core/driver path. Re-run a + * generated failure with TANSTACK_DB_DRIVER_FAIRNESS_SEED and + * TANSTACK_DB_DRIVER_FAIRNESS_PATH. Cleanup preserves the primary failure and + * reports secondary resource-release diagnostics separately. + */ +import { createCollection } from '../../db/src/index' +import { persistedCollectionOptions } from '../src/index' +import { BrowserWASQLiteDriver } from '../src/wa-sqlite-driver' +import { + SingleProcessCoordinator, + createSQLiteCorePersistenceAdapter, +} from '../../db-sqlite-persistence-core/src/index' +import type { Collection } from '../../db/src/index' +import type { + PersistedCollectionPersistence, + PersistedTx, + SQLiteDriver, +} from '../../db-sqlite-persistence-core/src/index' +import type { BrowserWASQLiteDatabase } from '../src/index' + +export const SHARED_DRIVER_FAIRNESS_BOUND = 1 + +export type FairnessRow = { + id: string + value: number +} + +export type SharedDriverFairnessWork = + | { + kind: `hydrate` + id: string + seededRows: ReadonlyArray + } + | { + kind: `persist` + id: string + mutationsPerPersist: number + } + +export type SharedDriverFairnessScenario = { + id: string + /** + * Logical admission history. A storm begins with the one persist that is + * already non-preemptibly running; the remaining hydrate and persist work + * may be interleaved in any order. + */ + work: ReadonlyArray +} + +export type RawDriverDequeue = { + ordinal: number + sql: string + params: ReadonlyArray +} + +export type RawDriverAdmission = RawDriverDequeue & { + kind: `exec` | `query` | `run` | `transaction` +} + +export type HydrationCompletionCheckpoint = { + collectionId: string + completionOrdinal: number + completedPersistIds: ReadonlyArray + pendingPersistCount: number + rawDequeueCount: number +} + +export type HydratedCollectionRows = { + collectionId: string + rows: ReadonlyArray +} + +export type SharedDriverFairnessObservation = { + scenario: SharedDriverFairnessScenario + admittedHydrateIds: ReadonlyArray + logicalCompletionOrder: ReadonlyArray + hydrationCompletions: ReadonlyArray + driverAdmissions: ReadonlyArray + rawDequeues: ReadonlyArray + hydratedCollections: ReadonlyArray + cleanupFailures: ReadonlyArray +} + +export type SharedDriverFairnessViolation = { + checkpoint: HydrationCompletionCheckpoint + expectedCollectionId: string + expectedMaximumCompletedPersists: number + expectedMinimumPendingPersists: number + permittedCompletedPersistIds: ReadonlyArray + unexpectedCompletedPersistIds: ReadonlyArray +} + +export type SharedDriverFairnessOptions = { + /** Test-only hostile control that preserves the current global FIFO fault. */ + schedulingFault?: `persist-first-fifo` +} + +type Deferred = { + promise: Promise + resolve: () => void +} + +type CloseableSQLiteDriver = SQLiteDriver & { + transactionWithDriver: ( + fn: (transactionDriver: SQLiteDriver) => Promise, + ) => Promise + close: () => Promise +} + +type FairnessReferenceCheckpoint = { + collectionId: string + expectedMaximumCompletedPersists: number + expectedMinimumPendingPersists: number + permittedCompletedPersistIds: ReadonlyArray +} + +export type BrowserWASQLiteDatabaseFactory = () => + | BrowserWASQLiteDatabase + | Promise + +function createDeferred(): Deferred { + let resolve!: () => void + const promise = new Promise((settle) => { + resolve = settle + }) + return { promise, resolve } +} + +function normalizeSql(sql: string): string { + return sql.replace(/\s+/g, ` `).trim() +} + +function failureMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error) +} + +function collectionIdFor( + scenario: SharedDriverFairnessScenario, + work: SharedDriverFairnessWork, +): string { + return `${scenario.id}-${work.id}` +} + +function validateScenario(scenario: SharedDriverFairnessScenario): void { + const hydrateWork = scenario.work.filter((work) => work.kind === `hydrate`) + const persistWork = scenario.work.filter((work) => work.kind === `persist`) + if (hydrateWork.length < 1) { + throw new Error(`work must contain at least one hydrate`) + } + if (persistWork.length > 0 && scenario.work[0]?.kind !== `persist`) { + throw new Error( + `a storm history must begin with its already-running persist`, + ) + } + if ( + new Set(scenario.work.map((work) => work.id)).size !== scenario.work.length + ) { + throw new Error(`work ids must be unique`) + } + for (const work of scenario.work) { + if (work.kind === `persist` && work.mutationsPerPersist < 1) { + throw new Error(`mutationsPerPersist must be at least one`) + } + if (work.kind === `hydrate`) { + if (work.seededRows.length < 1) { + throw new Error(`each hydrate must contain at least one seeded row`) + } + if ( + new Set(work.seededRows.map((row) => row.id)).size !== + work.seededRows.length + ) { + throw new Error(`seeded row ids must be unique within each hydrate`) + } + } + } +} + +class ObservedDatabase implements BrowserWASQLiteDatabase { + readonly rawDequeues: Array = [] + private heldBegin: Deferred | undefined + private beginEntered: Deferred | undefined + + constructor(private readonly database: BrowserWASQLiteDatabase) {} + + holdNextTransactionBegin(): { entered: Promise; release: () => void } { + if (this.heldBegin) { + throw new Error(`a transaction begin is already held`) + } + this.heldBegin = createDeferred() + this.beginEntered = createDeferred() + return { + entered: this.beginEntered.promise, + release: () => this.releaseHeldBegin(), + } + } + + clearTrace(): void { + this.rawDequeues.length = 0 + } + + async execute( + sql: string, + params: ReadonlyArray = [], + ): Promise> { + const normalizedSql = normalizeSql(sql) + this.rawDequeues.push({ + ordinal: this.rawDequeues.length, + sql: normalizedSql, + params: [...params], + }) + + if (normalizedSql === `BEGIN IMMEDIATE` && this.heldBegin) { + const heldBegin = this.heldBegin + this.beginEntered?.resolve() + await heldBegin.promise + } + + return this.database.execute(sql, params) + } + + async close(): Promise { + this.releaseHeldBegin() + await this.database.close?.() + } + + private releaseHeldBegin(): void { + this.heldBegin?.resolve() + this.heldBegin = undefined + this.beginEntered = undefined + } +} + +/** + * A test-only hostile scheduler. It admits calls through the same public/core/ + * BrowserWASQLiteDriver path but serializes them in global admission order, so + * a future fair production driver cannot accidentally make this control pass. + */ +class PersistFirstFIFOFaultDriver implements SQLiteDriver { + private tail = Promise.resolve() + + constructor(private readonly driver: CloseableSQLiteDriver) {} + + exec(sql: string): Promise { + return this.enqueue(() => this.driver.exec(sql)) + } + + query( + sql: string, + params: ReadonlyArray = [], + ): Promise> { + return this.enqueue(() => this.driver.query(sql, params)) + } + + run(sql: string, params: ReadonlyArray = []): Promise { + return this.enqueue(() => this.driver.run(sql, params)) + } + + transaction( + fn: (transactionDriver: SQLiteDriver) => Promise, + ): Promise { + return this.enqueue(() => this.driver.transaction(fn)) + } + + transactionWithDriver( + fn: (transactionDriver: SQLiteDriver) => Promise, + ): Promise { + return this.enqueue(() => this.driver.transactionWithDriver(fn)) + } + + close(): Promise { + return this.enqueue(() => this.driver.close()) + } + + private enqueue(operation: () => Promise): Promise { + const result = this.tail.then(operation) + this.tail = result.then( + () => undefined, + () => undefined, + ) + return result + } +} + +class AdmissionObservedDriver implements SQLiteDriver { + readonly admissions: Array = [] + + constructor( + private readonly driver: CloseableSQLiteDriver, + private readonly onAdmission?: (entry: RawDriverAdmission) => void, + ) {} + + exec(sql: string): Promise { + this.record(`exec`, sql, []) + return this.driver.exec(sql) + } + + query( + sql: string, + params: ReadonlyArray = [], + ): Promise> { + this.record(`query`, sql, params) + return this.driver.query(sql, params) + } + + run(sql: string, params: ReadonlyArray = []): Promise { + this.record(`run`, sql, params) + return this.driver.run(sql, params) + } + + transaction( + fn: (transactionDriver: SQLiteDriver) => Promise, + ): Promise { + this.record(`transaction`, `BEGIN IMMEDIATE`, []) + return this.driver.transaction(fn) + } + + transactionWithDriver( + fn: (transactionDriver: SQLiteDriver) => Promise, + ): Promise { + this.record(`transaction`, `BEGIN IMMEDIATE`, []) + return this.driver.transactionWithDriver(fn) + } + + close(): Promise { + return this.driver.close() + } + + clearAdmissions(): void { + this.admissions.length = 0 + } + + private record( + kind: RawDriverAdmission[`kind`], + sql: string, + params: ReadonlyArray, + ): void { + const entry = { + ordinal: this.admissions.length, + kind, + sql: normalizeSql(sql), + params: [...params], + } + this.admissions.push(entry) + this.onAdmission?.(entry) + } +} + +function createSharedPersistence( + database: BrowserWASQLiteDatabase, + onAdmission?: (entry: RawDriverAdmission) => void, + schedulingFault?: SharedDriverFairnessOptions[`schedulingFault`], +): { + driver: AdmissionObservedDriver + persistence: PersistedCollectionPersistence +} { + const productionDriver = new BrowserWASQLiteDriver({ database }) + const scheduledDriver = + schedulingFault === `persist-first-fifo` + ? new PersistFirstFIFOFaultDriver(productionDriver) + : productionDriver + const driver = new AdmissionObservedDriver(scheduledDriver, onAdmission) + const adapter = createSQLiteCorePersistenceAdapter({ + driver, + schemaMismatchPolicy: `sync-absent-error`, + appliedTxPruneMaxRows: 0, + appliedTxPruneMaxAgeSeconds: 0, + }) + return { + driver, + persistence: { + adapter, + coordinator: new SingleProcessCoordinator(), + }, + } +} + +function createPersistedTx( + collectionId: string, + sequence: number, + mutationsPerPersist: number, +): PersistedTx { + return { + txId: `${collectionId}-tx-${sequence}`, + term: 1, + seq: sequence, + rowVersion: sequence, + mutations: Array.from({ length: mutationsPerPersist }, (_, index) => ({ + type: `insert` as const, + key: `${sequence}-${index}`, + value: { + id: `${sequence}-${index}`, + value: sequence * 100 + index, + }, + })), + } +} + +function createFairnessReference( + scenario: SharedDriverFairnessScenario, + fairnessBound: number, +): ReadonlyArray { + const persistIds = scenario.work + .filter((work) => work.kind === `persist`) + .map((work) => collectionIdFor(scenario, work)) + const hydrateIds = scenario.work + .filter((work) => work.kind === `hydrate`) + .map((work) => collectionIdFor(scenario, work)) + + return hydrateIds.map((collectionId, hydrateIndex) => { + // The first persist is the one already executing when hydrate work reaches + // the queue. The ordered reference then permits at most K additional + // persists between successive hydrate completions, preserving FIFO order + // within each logical lane. + const expectedMaximumCompletedPersists = Math.min( + persistIds.length, + persistIds.length === 0 ? 0 : 1 + hydrateIndex * fairnessBound, + ) + return { + collectionId, + expectedMaximumCompletedPersists, + expectedMinimumPendingPersists: + persistIds.length - expectedMaximumCompletedPersists, + permittedCompletedPersistIds: persistIds.slice( + 0, + expectedMaximumCompletedPersists, + ), + } + }) +} + +export function findSharedDriverFairnessViolation( + observation: SharedDriverFairnessObservation, + fairnessBound = SHARED_DRIVER_FAIRNESS_BOUND, +): SharedDriverFairnessViolation | undefined { + const reference = createFairnessReference(observation.scenario, fairnessBound) + for (const [ + hydrateIndex, + checkpoint, + ] of observation.hydrationCompletions.entries()) { + const expected = reference[hydrateIndex] + if (!expected) continue + const permitted = new Set(expected.permittedCompletedPersistIds) + const unexpectedCompletedPersistIds = checkpoint.completedPersistIds.filter( + (id) => !permitted.has(id), + ) + if ( + checkpoint.collectionId !== expected.collectionId || + checkpoint.completedPersistIds.length > + expected.expectedMaximumCompletedPersists || + checkpoint.pendingPersistCount < + expected.expectedMinimumPendingPersists || + unexpectedCompletedPersistIds.length > 0 + ) { + return { + checkpoint, + expectedCollectionId: expected.collectionId, + expectedMaximumCompletedPersists: + expected.expectedMaximumCompletedPersists, + expectedMinimumPendingPersists: expected.expectedMinimumPendingPersists, + permittedCompletedPersistIds: expected.permittedCompletedPersistIds, + unexpectedCompletedPersistIds, + } + } + } + return undefined +} + +/** Checker calibration only; the executable hostile control uses the fixture. */ +export function createPersistFirstFaultObservation( + scenario: SharedDriverFairnessScenario, +): SharedDriverFairnessObservation { + const persistIds = scenario.work + .filter((work) => work.kind === `persist`) + .map((work) => collectionIdFor(scenario, work)) + const firstHydrate = scenario.work.find((work) => work.kind === `hydrate`) + if (!firstHydrate) throw new Error(`fault observation requires a hydrate`) + const firstHydrateId = collectionIdFor(scenario, firstHydrate) + return { + scenario, + admittedHydrateIds: [firstHydrateId], + logicalCompletionOrder: [ + ...persistIds.map((id) => `persist:${id}`), + `hydrate:${firstHydrateId}`, + ], + hydrationCompletions: [ + { + collectionId: firstHydrateId, + completionOrdinal: persistIds.length, + completedPersistIds: persistIds, + pendingPersistCount: 0, + rawDequeueCount: persistIds.length + 1, + }, + ], + rawDequeues: [], + driverAdmissions: [], + hydratedCollections: [], + cleanupFailures: [], + } +} + +/** + * Runs the public persisted-collection startup path over one shared + * BrowserWASQLiteDriver. The expected scheduler is deliberately not imported: + * the oracle observes only logical admissions/completions and raw SQL dequeue. + */ +export async function observeSharedDriverFairness( + openDatabase: BrowserWASQLiteDatabaseFactory, + scenario: SharedDriverFairnessScenario, + options: SharedDriverFairnessOptions = {}, +): Promise { + validateScenario(scenario) + + const hydrateWork = scenario.work.filter((work) => work.kind === `hydrate`) + const persistWork = scenario.work.filter((work) => work.kind === `persist`) + const hydrateIds = hydrateWork.map((work) => collectionIdFor(scenario, work)) + const persistIds = persistWork.map((work) => collectionIdFor(scenario, work)) + const admittedHydrateIds: Array = [] + const allHydratesRequested = createDeferred() + + // A separate connection and adapter own seed/setup. Closing and reopening + // makes every measured hydration cold at the adapter and driver layers. + const seedDatabase = await openDatabase() + const seed = createSharedPersistence(seedDatabase) + let seedPrimaryFailure: unknown + try { + for (const work of hydrateWork) { + const collectionId = collectionIdFor(scenario, work) + await seed.persistence.adapter.applyCommittedTx(collectionId, { + txId: `${collectionId}-seed`, + term: 1, + seq: 1, + rowVersion: 1, + mutations: work.seededRows.map((row) => ({ + type: `insert` as const, + key: row.id, + value: { ...row }, + })), + }) + } + for (const collectionId of persistIds) { + await seed.persistence.adapter.loadSubset(collectionId, {}) + } + } catch (error) { + seedPrimaryFailure = error + } + let seedCleanupFailure: unknown + try { + await seed.driver.close() + } catch (error) { + seedCleanupFailure = error + } + if (seedPrimaryFailure !== undefined) { + if (seedCleanupFailure !== undefined) { + throw new Error( + `${failureMessage(seedPrimaryFailure)}; seed cleanup diagnostics: ${failureMessage(seedCleanupFailure)}`, + ) + } + throw seedPrimaryFailure + } + if (seedCleanupFailure !== undefined) throw seedCleanupFailure + + const observedDatabase = new ObservedDatabase(await openDatabase()) + const { driver, persistence } = createSharedPersistence( + observedDatabase, + undefined, + options.schedulingFault, + ) + + const collections: Array> = [] + const persistPromises: Array> = [] + const preloadPromises: Array> = [] + const logicalCompletionOrder: Array = [] + const completedPersistIds: Array = [] + const hydrationCompletions: Array = [] + const observedRows = new Map>() + const cleanupFailures: Array = [] + let releaseHeldBegin: (() => void) | undefined + let beginEntered: Promise | undefined + let persistSequence = 0 + let primaryFailure: unknown + let observation: SharedDriverFairnessObservation | undefined + + try { + // Cache only the unrelated persist tables. Hydrate tables remain cold. + for (const collectionId of persistIds) { + await persistence.adapter.loadSubset(collectionId, {}) + } + + observedDatabase.clearTrace() + driver.clearAdmissions() + + if (persistIds.length > 0) { + const heldBegin = observedDatabase.holdNextTransactionBegin() + releaseHeldBegin = heldBegin.release + beginEntered = heldBegin.entered + } + + // Admit the explicit legal history while the first persist is held. The + // first item is the non-preemptible transaction; every tail permutation + // is therefore observable at the same deterministic release checkpoint. + for (const work of scenario.work) { + const collectionId = collectionIdFor(scenario, work) + if (work.kind === `persist`) { + persistSequence += 1 + const sequence = persistSequence + const persist = persistence.adapter + .applyCommittedTx( + collectionId, + createPersistedTx(collectionId, sequence, work.mutationsPerPersist), + ) + .then(() => { + completedPersistIds.push(collectionId) + logicalCompletionOrder.push(`persist:${collectionId}`) + }) + persistPromises.push(persist) + continue + } + + const collection = createCollection( + persistedCollectionOptions({ + id: collectionId, + getKey: (row) => row.id, + persistence, + }), + ) + collections.push(collection) + const preloadRequest = collection.preload() + admittedHydrateIds.push(collectionId) + if (admittedHydrateIds.length === hydrateIds.length) { + allHydratesRequested.resolve() + } + const preload = preloadRequest.then(() => { + logicalCompletionOrder.push(`hydrate:${collectionId}`) + observedRows.set( + collectionId, + collection.toArray.map((row) => ({ id: row.id, value: row.value })), + ) + hydrationCompletions.push({ + collectionId, + completionOrdinal: logicalCompletionOrder.length - 1, + completedPersistIds: [...completedPersistIds], + pendingPersistCount: persistIds.length - completedPersistIds.length, + rawDequeueCount: observedDatabase.rawDequeues.length, + }) + }) + preloadPromises.push(preload) + } + + await beginEntered + await allHydratesRequested.promise + // Every public preload request is now pending. Releasing the one + // non-preemptible persist here leaves production responsible for admitting + // and completing each full logical hydrate under the approved scheduler. + releaseHeldBegin?.() + releaseHeldBegin = undefined + + await Promise.all([...persistPromises, ...preloadPromises]) + + observation = { + scenario, + admittedHydrateIds, + logicalCompletionOrder, + hydrationCompletions, + driverAdmissions: driver.admissions.map((entry) => ({ + ...entry, + params: [...entry.params], + })), + rawDequeues: observedDatabase.rawDequeues.map((entry) => ({ + ...entry, + params: [...entry.params], + })), + hydratedCollections: hydrateWork.map((work) => { + const collectionId = collectionIdFor(scenario, work) + return { + collectionId, + rows: + observedRows.get(collectionId)?.map((row) => ({ ...row })) ?? [], + } + }), + cleanupFailures, + } + } catch (error) { + primaryFailure = error + } finally { + releaseHeldBegin?.() + await Promise.allSettled([...persistPromises, ...preloadPromises]) + for (const collection of collections) { + try { + await collection.cleanup() + } catch (error) { + cleanupFailures.push(failureMessage(error)) + } + } + try { + await driver.close() + } catch (error) { + cleanupFailures.push(failureMessage(error)) + } + } + + if (primaryFailure !== undefined) { + if (cleanupFailures.length > 0) { + throw new Error( + `${failureMessage(primaryFailure)}; active cleanup diagnostics: ${JSON.stringify(cleanupFailures)}`, + ) + } + throw primaryFailure + } + if (!observation) { + throw new Error(`shared-driver observation ended without a result`) + } + return observation +} diff --git a/packages/browser-db-sqlite-persistence/tsconfig.json b/packages/browser-db-sqlite-persistence/tsconfig.json index 5b14f299c7..b8bc1c02dc 100644 --- a/packages/browser-db-sqlite-persistence/tsconfig.json +++ b/packages/browser-db-sqlite-persistence/tsconfig.json @@ -19,6 +19,14 @@ ] } }, - "include": ["src", "tests", "e2e", "vite.config.ts", "vitest.e2e.config.ts"], + "include": [ + "src", + "tests", + "e2e", + "vite.config.ts", + "vite.opfs.config.ts", + "vitest.e2e.config.ts", + "playwright.opfs.config.ts" + ], "exclude": ["node_modules", "dist"] } diff --git a/packages/browser-db-sqlite-persistence/vite.opfs.config.ts b/packages/browser-db-sqlite-persistence/vite.opfs.config.ts new file mode 100644 index 0000000000..c66d394740 --- /dev/null +++ b/packages/browser-db-sqlite-persistence/vite.opfs.config.ts @@ -0,0 +1,30 @@ +import { dirname, resolve } from 'node:path' +import { fileURLToPath } from 'node:url' +import { defineConfig } from 'vite' + +const packageDirectory = dirname(fileURLToPath(import.meta.url)) + +export default defineConfig({ + base: `./`, + // wa-sqlite locates its sibling WASM file through import.meta.url. Keeping + // the module out of Vite's dependency prebundle preserves that relationship + // for this real-browser fixture. + optimizeDeps: { + exclude: [`@journeyapps/wa-sqlite`], + }, + resolve: { + alias: { + '@tanstack/db': resolve(packageDirectory, `../db/src`), + '@tanstack/db-ivm': resolve(packageDirectory, `../db-ivm/src`), + '@tanstack/db-sqlite-persistence-core': resolve( + packageDirectory, + `../db-sqlite-persistence-core/src`, + ), + }, + }, + server: { + fs: { + allow: [resolve(packageDirectory, `../..`)], + }, + }, +}) diff --git a/packages/db-sqlite-persistence-core/src/persisted.ts b/packages/db-sqlite-persistence-core/src/persisted.ts index 0f1e4112ac..8cbcf7d811 100644 --- a/packages/db-sqlite-persistence-core/src/persisted.ts +++ b/packages/db-sqlite-persistence-core/src/persisted.ts @@ -219,6 +219,21 @@ export type PersistedRowScanOptions = { metadataOnly?: boolean } +export type PersistencePullSinceResult = + | { + latestRowVersion: number + requiresFullReload: true + } + | { + latestRowVersion: number + requiresFullReload: false + changedKeys: Array + deletedKeys: Array + deltas?: Array< + ReplayableTxDelta, string | number> + > + } + export type PersistedTx< T extends object = Record, TKey extends string | number = string | number, @@ -249,6 +264,21 @@ export type PersistedTx< collectionMetadataMutations?: Array } +/** + * Opaque identity shared by every adapter over the same physical SQLite driver. + * The core adapter uses it to serialize non-preemptible logical operations + * while alternating one regular operation between queued hydrations. + * + * Delegating drivers must forward this property before they are passed to a + * SQLite persistence adapter. A driver may also brand each returned Promise + * with the same key for late capability discovery, but a wrapper must return + * that exact Promise: discovering the key after a call cannot retroactively + * schedule the wrapper's first logical operation. + */ +export const SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY = Symbol.for( + `tanstack-db.sqlite-driver-supports-shared-logical-scheduling`, +) + export interface PersistenceAdapter { loadSubset: ( collectionId: string, @@ -280,9 +310,25 @@ export interface PersistenceAdapter { latestSeq: number latestRowVersion: number }> + /** + * Runs one complete logical hydrate as a non-preemptible scheduler unit. + * The callback must use the supplied unscheduled adapter for all nested + * persistence work and must not retain it after the callback settles. + */ + runInHydrationScope?: ( + task: (adapter: HydrationPersistenceAdapter) => Promise, + ) => Promise +} + +export type HydrationPersistenceAdapter = PersistenceAdapter & { + pullSince?: ( + collectionId: string, + fromRowVersion: number, + ) => Promise } export interface SQLiteDriver { + readonly [SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY]?: object exec: (sql: string) => Promise query: ( sql: string, @@ -297,6 +343,24 @@ export interface SQLiteDriver { ) => Promise } +/** + * Forwards a driver's shared logical scheduling identity to a transparent + * delegating driver before the wrapper is used by a persistence adapter. + */ +export function forwardSQLiteDriverSharedLogicalScheduling< + TDriver extends SQLiteDriver, +>(source: SQLiteDriver, wrapper: TDriver): TDriver { + const key = source[SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY] + if (key) { + Object.defineProperty( + wrapper, + SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY, + { value: key }, + ) + } + return wrapper +} + export interface PersistedCollectionCoordinator { getNodeId: () => string subscribe: ( @@ -310,18 +374,22 @@ export interface PersistedCollectionCoordinator { collectionId: string, options: LoadSubsetOptions, ) => Promise + /** The scoped adapter is leader-local and is never serialized to a follower. */ requestEnsurePersistedIndex: ( collectionId: string, signature: string, spec: PersistedIndexSpec, + scopedAdapter?: HydrationPersistenceAdapter, ) => Promise requestApplyLocalMutations?: ( collectionId: string, mutations: Array, ) => Promise + /** The scoped adapter is leader-local and is never serialized to a follower. */ pullSince?: ( collectionId: string, fromRowVersion: number, + scopedAdapter?: HydrationPersistenceAdapter, ) => Promise } @@ -802,7 +870,6 @@ class PersistedCollectionRuntime< truncate: null, metadata: null, } - private started = false private startupMetadataPromise: Promise | null = null private startPromise: Promise | null = null private resumeBaselinePromise: Promise | null = null @@ -899,13 +966,72 @@ class PersistedCollectionRuntime< } } + private runInHydrationScope( + task: (adapter: HydrationPersistenceAdapter) => Promise, + ): Promise { + if (this.persistence.adapter.runInHydrationScope) { + return this.persistence.adapter.runInHydrationScope(task) + } + return Promise.resolve().then(() => task(this.persistence.adapter)) + } + async ensureStarted(): Promise { if (this.startPromise) { return this.startPromise } const lifecycleGeneration = this.lifecycleGeneration - this.startPromise = this.startInternal(lifecycleGeneration) + let resolveStartupMetadata!: () => void + let rejectStartupMetadata!: (error: unknown) => void + this.startupMetadataPromise = new Promise((resolve, reject) => { + resolveStartupMetadata = resolve + rejectStartupMetadata = reject + }) + void this.startupMetadataPromise.catch(() => undefined) + + this.startPromise = (async () => { + const loadStartupMetadata = async ( + adapter: HydrationPersistenceAdapter, + ) => { + if (lifecycleGeneration !== this.lifecycleGeneration) { + resolveStartupMetadata() + return false + } + + try { + await this.loadStartupMetadataInternal(lifecycleGeneration, adapter) + resolveStartupMetadata() + return lifecycleGeneration === this.lifecycleGeneration + } catch (error) { + rejectStartupMetadata(error) + throw error + } + } + + let appliedCursor: number | undefined + if (this.persistence.adapter.runInHydrationScope) { + appliedCursor = await this.applyMutex.run(() => + this.runInHydrationScope(async (adapter) => { + if (!(await loadStartupMetadata(adapter))) return undefined + return this.startInternal(lifecycleGeneration, adapter) + }), + ) + } else { + // Preserve the existing unscheduled-adapter lifecycle contract: a + // replacement upstream may start while stale hydration is settling. + if (await loadStartupMetadata(this.persistence.adapter)) { + appliedCursor = await this.applyMutex.run(() => + this.startInternal(lifecycleGeneration, this.persistence.adapter), + ) + } + } + if ( + appliedCursor !== undefined && + lifecycleGeneration === this.lifecycleGeneration + ) { + await this.waitForAppliedReceiptsAfter(appliedCursor) + } + })() return this.startPromise } @@ -920,26 +1046,41 @@ class PersistedCollectionRuntime< if (lifecycleGeneration !== this.lifecycleGeneration) return if (this.syncMode !== `on-demand`) return - await this.hydrateBaseline(lifecycleGeneration) + const appliedCursor = await this.applyMutex.run(() => + this.runInHydrationScope((adapter) => + this.hydrateBaseline(lifecycleGeneration, adapter), + ), + ) + if ( + appliedCursor !== undefined && + lifecycleGeneration === this.lifecycleGeneration + ) { + await this.waitForAppliedReceiptsAfter(appliedCursor) + } })() return this.resumeBaselinePromise } - private async hydrateBaseline(lifecycleGeneration: number): Promise { - if (lifecycleGeneration !== this.lifecycleGeneration) return + private async hydrateBaseline( + lifecycleGeneration: number, + adapter: HydrationPersistenceAdapter, + ): Promise { + if (lifecycleGeneration !== this.lifecycleGeneration) return undefined const baseline = {} this.activeSubsets.set(this.getSubsetKey(baseline), baseline) const appliedCursor = this.appliedReceiptSequence - await this.applyMutex.run(async () => { - if (lifecycleGeneration !== this.lifecycleGeneration) return - await this.hydrateSubsetUnsafe(baseline, { + await this.hydrateSubsetUnsafe( + baseline, + { requestRemoteEnsure: false, lifecycleGeneration, - }) - }) - if (lifecycleGeneration !== this.lifecycleGeneration) return - await this.waitForAppliedReceiptsAfter(appliedCursor) + }, + adapter, + ) + return lifecycleGeneration === this.lifecycleGeneration + ? appliedCursor + : undefined } async ensureStartupMetadataLoaded(): Promise { @@ -947,41 +1088,35 @@ class PersistedCollectionRuntime< return this.startupMetadataPromise } - const lifecycleGeneration = this.lifecycleGeneration - this.startupMetadataPromise = - this.loadStartupMetadataInternal(lifecycleGeneration) - return this.startupMetadataPromise + void this.ensureStarted() + return this.startupMetadataPromise! } - private async startInternal(lifecycleGeneration: number): Promise { - if (this.started) { - return - } - - this.started = true - - await this.ensureStartupMetadataLoaded() - if (lifecycleGeneration !== this.lifecycleGeneration) return + private async startInternal( + lifecycleGeneration: number, + adapter: HydrationPersistenceAdapter, + ): Promise { + if (lifecycleGeneration !== this.lifecycleGeneration) return undefined const indexBootstrapSnapshot = this.collection?.getIndexMetadata() ?? [] this.attachIndexLifecycleListeners() - await this.bootstrapPersistedIndexes(indexBootstrapSnapshot) - if (lifecycleGeneration !== this.lifecycleGeneration) return + await this.bootstrapPersistedIndexes(indexBootstrapSnapshot, adapter) + if (lifecycleGeneration !== this.lifecycleGeneration) return undefined if (this.syncMode !== `on-demand`) { - await this.hydrateBaseline(lifecycleGeneration) + return this.hydrateBaseline(lifecycleGeneration, adapter) } + return undefined } private async loadStartupMetadataInternal( lifecycleGeneration: number, + adapter: HydrationPersistenceAdapter, ): Promise { // Restore stream position from the database so that new mutations // don't collide with previously applied transactions. - if (this.persistence.adapter.getStreamPosition) { - const position = await this.persistence.adapter.getStreamPosition( - this.collectionId, - ) + if (adapter.getStreamPosition) { + const position = await adapter.getStreamPosition(this.collectionId) if (lifecycleGeneration !== this.lifecycleGeneration) return this.observeStreamPosition( position.latestTerm, @@ -990,19 +1125,20 @@ class PersistedCollectionRuntime< ) } - const collectionMetadata = await this.loadCollectionMetadataSnapshot() + const collectionMetadata = + await this.loadCollectionMetadataSnapshot(adapter) if (lifecycleGeneration !== this.lifecycleGeneration) return this.replaceCollectionMetadataSnapshot(collectionMetadata) } - private async loadCollectionMetadataSnapshot(): Promise< - Array<{ key: string; value: unknown }> - > { - if (!this.persistence.adapter.loadCollectionMetadata) { + private async loadCollectionMetadataSnapshot( + adapter: HydrationPersistenceAdapter, + ): Promise> { + if (!adapter.loadCollectionMetadata) { return [] } - return this.persistence.adapter.loadCollectionMetadata(this.collectionId) + return adapter.loadCollectionMetadata(this.collectionId) } private replaceCollectionMetadataSnapshot( @@ -1046,13 +1182,19 @@ class PersistedCollectionRuntime< ): Promise { const lifecycleGeneration = this.lifecycleGeneration this.activeSubsets.set(this.getSubsetKey(options), options) - const appliedCursor = this.appliedReceiptSequence + await this.applyMutex.run(() => - this.hydrateSubsetUnsafe(options, { - requestRemoteEnsure: this.mode === `sync-present`, - lifecycleGeneration, - }), + this.runInHydrationScope((adapter) => + this.hydrateSubsetUnsafe( + options, + { + requestRemoteEnsure: this.mode === `sync-present`, + lifecycleGeneration, + }, + adapter, + ), + ), ) if (lifecycleGeneration !== this.lifecycleGeneration) return await this.waitForAppliedReceiptsAfter(appliedCursor) @@ -1092,10 +1234,16 @@ class PersistedCollectionRuntime< const lifecycleGeneration = this.lifecycleGeneration // A one-shot refresh does not acquire an enduring subscription lease. await this.applyMutex.run(() => - this.hydrateSubsetUnsafe(options, { - requestRemoteEnsure: false, - lifecycleGeneration, - }), + this.runInHydrationScope((adapter) => + this.hydrateSubsetUnsafe( + options, + { + requestRemoteEnsure: false, + lifecycleGeneration, + }, + adapter, + ), + ), ) } @@ -1243,7 +1391,6 @@ class PersistedCollectionRuntime< private advanceLifecycle(): void { this.lifecycleGeneration++ - this.started = false this.startupMetadataPromise = null this.startPromise = null this.resumeBaselinePromise = null @@ -1270,8 +1417,9 @@ class PersistedCollectionRuntime< private loadSubsetRowsUnsafe( options: LoadSubsetOptions, + adapter: HydrationPersistenceAdapter, ): Promise> { - return this.persistence.adapter.loadSubset(this.collectionId, options, { + return adapter.loadSubset(this.collectionId, options, { requiredIndexSignatures: this.getRequiredIndexSignatures(), }) as Promise> } @@ -1301,10 +1449,11 @@ class PersistedCollectionRuntime< requestRemoteEnsure: boolean lifecycleGeneration: number }, + adapter: HydrationPersistenceAdapter, ): Promise { this.hydratingGeneration = config.lifecycleGeneration try { - const rows = await this.loadSubsetRowsUnsafe(options) + const rows = await this.loadSubsetRowsUnsafe(options, adapter) if (config.lifecycleGeneration !== this.lifecycleGeneration) return this.applyRowsToCollection(rows) @@ -1314,8 +1463,8 @@ class PersistedCollectionRuntime< } } - await this.flushQueuedHydrationTransactionsUnsafe() - await this.flushQueuedTxCommittedUnsafe() + await this.flushQueuedHydrationTransactionsUnsafe(adapter) + await this.flushQueuedTxCommittedUnsafe(adapter) if (config.requestRemoteEnsure) { this.queueRemoteSubsetEnsure(options) @@ -1397,14 +1546,16 @@ class PersistedCollectionRuntime< }) } - private async flushQueuedHydrationTransactionsUnsafe(): Promise { + private async flushQueuedHydrationTransactionsUnsafe( + adapter: HydrationPersistenceAdapter, + ): Promise { while (this.queuedHydrationTransactions.length > 0) { const transaction = this.queuedHydrationTransactions.shift() if (!transaction) { continue } try { - await this.applyBufferedSyncTransactionUnsafe(transaction) + await this.applyBufferedSyncTransactionUnsafe(transaction, adapter) } catch (error) { transaction.rejectApplied?.(error) for (const abandoned of this.queuedHydrationTransactions) { @@ -1418,6 +1569,7 @@ class PersistedCollectionRuntime< private async applyBufferedSyncTransactionUnsafe( transaction: BufferedSyncTransaction, + adapter: HydrationPersistenceAdapter, ): Promise { if (transaction.signal?.aborted) { transaction.rejectApplied?.(new SyncTransactionAbortedError()) @@ -1431,7 +1583,10 @@ class PersistedCollectionRuntime< } const applyToCollection = (): SyncAppliedReceipt => { - begin() + // Buffered source replay is part of persistence hydration. Apply it + // immediately so it cannot wait for a persisting mutation whose + // persistence is queued behind this hydrate's apply mutex. + begin({ immediate: true }) if (transaction.truncate) { truncate?.() @@ -1480,7 +1635,10 @@ class PersistedCollectionRuntime< } if (!transaction.internal) { - await this.persistAndBroadcastExternalSyncTransactionUnsafe(transaction) + await this.persistAndBroadcastExternalSyncTransactionUnsafe( + transaction, + adapter, + ) } transaction.resolveApplied?.() } catch (error) { @@ -1491,6 +1649,7 @@ class PersistedCollectionRuntime< private async persistAndBroadcastExternalSyncTransactionUnsafe( transaction: BufferedSyncTransaction, + adapter: HydrationPersistenceAdapter = this.persistence.adapter, ): Promise { if (transaction.internal) { return @@ -1520,7 +1679,7 @@ class PersistedCollectionRuntime< const tx = this.createPersistedTxFromOperations(transaction, streamPosition) - await this.persistence.adapter.applyCommittedTx(this.collectionId, tx) + await adapter.applyCommittedTx(this.collectionId, tx) this.publishTxCommittedEvent( this.createTxCommittedPayload({ term: tx.term, @@ -1961,7 +2120,9 @@ class PersistedCollectionRuntime< } void this.applyMutex - .run(() => this.processCommittedTxUnsafe(payload)) + .run(() => + this.processCommittedTxUnsafe(payload, this.persistence.adapter), + ) .catch((error) => { console.warn(`Failed to process tx:committed message:`, error) }) @@ -1975,25 +2136,33 @@ class PersistedCollectionRuntime< if (isCollectionResetPayload(payload)) { void this.applyMutex - .run(() => this.truncateAndReloadUnsafe()) + .run(() => + this.runInHydrationScope((adapter) => + this.truncateAndReloadUnsafe(adapter), + ), + ) .catch((error) => { console.warn(`Failed to process collection reset message:`, error) }) } } - private async flushQueuedTxCommittedUnsafe(): Promise { + private async flushQueuedTxCommittedUnsafe( + adapter: HydrationPersistenceAdapter, + ): Promise { while (this.queuedTxCommitted.length > 0) { const queued = this.queuedTxCommitted.shift() if (!queued) { continue } - await this.processCommittedTxUnsafe(queued) + await this.processCommittedTxUnsafe(queued, adapter, true) } } private async processCommittedTxUnsafe( txCommitted: TxCommitted, + adapter: HydrationPersistenceAdapter, + hydrationScopeAlreadyActive = false, ): Promise { if (txCommitted.term < this.latestTerm) { return @@ -2014,7 +2183,13 @@ class PersistedCollectionRuntime< const hasGap = hasGapInCurrentTerm || hasGapAcrossTerms if (hasGap) { - await this.recoverFromSeqGapUnsafe() + if (hydrationScopeAlreadyActive) { + await this.recoverFromSeqGapUnsafe(adapter) + } else { + await this.runInHydrationScope((scopedAdapter) => + this.recoverFromSeqGapUnsafe(scopedAdapter), + ) + } if ( txCommitted.term < this.latestTerm || (txCommitted.term === this.latestTerm && @@ -2030,15 +2205,22 @@ class PersistedCollectionRuntime< txCommitted.latestRowVersion, ) - await this.invalidateFromCommittedTxUnsafe(txCommitted) + await this.invalidateFromCommittedTxUnsafe( + txCommitted, + adapter, + hydrationScopeAlreadyActive, + ) } - private async recoverFromSeqGapUnsafe(): Promise { + private async recoverFromSeqGapUnsafe( + adapter: HydrationPersistenceAdapter, + ): Promise { if (this.persistence.coordinator.pullSince && this.latestRowVersion >= 0) { try { const pullResponse = await this.persistence.coordinator.pullSince( this.collectionId, this.latestRowVersion, + adapter, ) if (pullResponse.ok) { @@ -2048,23 +2230,27 @@ class PersistedCollectionRuntime< pullResponse.latestRowVersion, ) if (pullResponse.requiresFullReload || !pullResponse.deltas) { - await this.reloadActiveSubsetsUnsafe() + await this.reloadActiveSubsetsUnsafe(adapter) return } for (const delta of pullResponse.deltas) { - await this.invalidateFromCommittedTxUnsafe({ - type: `tx:committed`, - term: pullResponse.latestTerm, - seq: pullResponse.latestSeq, - txId: delta.txId, - latestRowVersion: delta.latestRowVersion, - requiresFullReload: false, - changedRows: delta.changedRows, - deletedKeys: delta.deletedKeys, - rowMetadataMutations: delta.rowMetadataMutations, - collectionMetadataMutations: delta.collectionMetadataMutations, - }) + await this.invalidateFromCommittedTxUnsafe( + { + type: `tx:committed`, + term: pullResponse.latestTerm, + seq: pullResponse.latestSeq, + txId: delta.txId, + latestRowVersion: delta.latestRowVersion, + requiresFullReload: false, + changedRows: delta.changedRows, + deletedKeys: delta.deletedKeys, + rowMetadataMutations: delta.rowMetadataMutations, + collectionMetadataMutations: delta.collectionMetadataMutations, + }, + adapter, + true, + ) } return } @@ -2073,7 +2259,7 @@ class PersistedCollectionRuntime< } } - await this.truncateAndReloadUnsafe() + await this.truncateAndReloadUnsafe(adapter) if (this.mode === `sync-present`) { for (const options of this.activeSubsets.values()) { @@ -2082,7 +2268,9 @@ class PersistedCollectionRuntime< } } - private async truncateAndReloadUnsafe(): Promise { + private async truncateAndReloadUnsafe( + adapter: HydrationPersistenceAdapter, + ): Promise { if (this.syncControls.begin && this.syncControls.commit) { this.withInternalApply(() => { this.syncControls.begin?.({ immediate: true }) @@ -2091,21 +2279,30 @@ class PersistedCollectionRuntime< }) } - await this.reloadActiveSubsetsUnsafe() + await this.reloadActiveSubsetsUnsafe(adapter) } private async invalidateFromCommittedTxUnsafe( txCommitted: TxCommitted, + adapter: HydrationPersistenceAdapter, + hydrationScopeAlreadyActive = false, ): Promise { + const reloadActiveSubsets = () => + hydrationScopeAlreadyActive + ? this.reloadActiveSubsetsUnsafe(adapter) + : this.runInHydrationScope((scopedAdapter) => + this.reloadActiveSubsetsUnsafe(scopedAdapter), + ) + if (txCommitted.requiresFullReload) { - await this.reloadActiveSubsetsUnsafe() + await reloadActiveSubsets() return } const changedKeyCount = txCommitted.changedRows.length + txCommitted.deletedKeys.length if (changedKeyCount > TARGETED_INVALIDATION_KEY_LIMIT) { - await this.reloadActiveSubsetsUnsafe() + await reloadActiveSubsets() return } @@ -2120,7 +2317,7 @@ class PersistedCollectionRuntime< // Has paginated subsets — fall back to full reload. // Targeted invalidation for paginated subsets is deferred to a future iteration. - await this.reloadActiveSubsetsUnsafe() + await reloadActiveSubsets() } private async applyTargetedInvalidationUnsafe( @@ -2179,7 +2376,9 @@ class PersistedCollectionRuntime< }) } - private async reloadActiveSubsetsUnsafe(): Promise { + private async reloadActiveSubsetsUnsafe( + adapter: HydrationPersistenceAdapter, + ): Promise { const lifecycleGeneration = this.lifecycleGeneration const activeSubsetOptions = this.activeSubsets.size > 0 @@ -2189,10 +2388,11 @@ class PersistedCollectionRuntime< this.hydratingGeneration = lifecycleGeneration try { const mergedRows = new Map() - const collectionMetadata = await this.loadCollectionMetadataSnapshot() + const collectionMetadata = + await this.loadCollectionMetadataSnapshot(adapter) if (lifecycleGeneration !== this.lifecycleGeneration) return for (const options of activeSubsetOptions) { - const subsetRows = await this.loadSubsetRowsUnsafe(options) + const subsetRows = await this.loadSubsetRowsUnsafe(options, adapter) if (lifecycleGeneration !== this.lifecycleGeneration) return for (const row of subsetRows) { mergedRows.set(row.key, { @@ -2216,8 +2416,8 @@ class PersistedCollectionRuntime< } } - await this.flushQueuedHydrationTransactionsUnsafe() - await this.flushQueuedTxCommittedUnsafe() + await this.flushQueuedHydrationTransactionsUnsafe(adapter) + await this.flushQueuedTxCommittedUnsafe(adapter) } private attachIndexLifecycleListeners(): void { @@ -2242,6 +2442,7 @@ class PersistedCollectionRuntime< private async bootstrapPersistedIndexes( indexMetadataSnapshot?: Array, + adapter: HydrationPersistenceAdapter = this.persistence.adapter, ): Promise { const collection = this.collection if (!collection && !indexMetadataSnapshot) { @@ -2251,7 +2452,7 @@ class PersistedCollectionRuntime< const indexMetadata = indexMetadataSnapshot ?? collection?.getIndexMetadata() ?? [] for (const metadata of indexMetadata) { - await this.ensurePersistedIndex(metadata) + await this.ensurePersistedIndex(metadata, adapter) } } @@ -2270,11 +2471,12 @@ class PersistedCollectionRuntime< private async ensurePersistedIndex( indexMetadata: CollectionIndexMetadata, + adapter: HydrationPersistenceAdapter = this.persistence.adapter, ): Promise { const spec = this.buildPersistedIndexSpec(indexMetadata) try { - await this.persistence.adapter.ensureIndex( + await adapter.ensureIndex( this.collectionId, indexMetadata.signature, spec, @@ -2288,6 +2490,7 @@ class PersistedCollectionRuntime< this.collectionId, indexMetadata.signature, spec, + adapter, ) } catch (error) { console.warn( diff --git a/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts b/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts index 69f29fc604..f76e1f310d 100644 --- a/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts +++ b/packages/db-sqlite-persistence-core/src/sqlite-core-adapter.ts @@ -8,12 +8,14 @@ import { InvalidPersistedStorageKeyEncodingError, } from './errors' import { + SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY, createPersistedTableName, decodePersistedStorageKey, encodePersistedStorageKey, } from './persisted' import type { LoadSubsetOptions } from '@tanstack/db' import type { + HydrationPersistenceAdapter, PersistedIndexSpec, PersistedRowScanOptions, PersistedScannedRow, @@ -71,6 +73,144 @@ export type SQLitePullSinceResult = deltas: Array, TKey>> } +type ScheduledOperationKind = `regular` | `hydrate` + +type ScheduledOperation = { + kind: ScheduledOperationKind + task: () => Promise + resolve: (value: T) => void + reject: (error: unknown) => void +} + +class SharedPersistenceScheduler { + private readonly regularQueue: Array> = [] + private readonly hydrateQueue: Array> = [] + private running = false + private lastCompletedKind: ScheduledOperationKind | undefined + + runRegular(task: () => Promise): Promise { + return this.enqueue(`regular`, task) + } + + runHydrate(task: () => Promise): Promise { + return this.enqueue(`hydrate`, task) + } + + private enqueue( + kind: ScheduledOperationKind, + task: () => Promise, + ): Promise { + const result = new Promise((resolve, reject) => { + const operation: ScheduledOperation = { + kind, + task, + resolve, + reject, + } + const queue = kind === `hydrate` ? this.hydrateQueue : this.regularQueue + queue.push(operation as ScheduledOperation) + }) + this.drain() + return result + } + + private drain(): void { + if (this.running) return + + const operation = this.takeNext() + if (!operation) return + + this.running = true + void this.execute(operation) + } + + private async execute(operation: ScheduledOperation): Promise { + try { + operation.resolve(await operation.task()) + } catch (error) { + operation.reject(error) + } finally { + this.lastCompletedKind = operation.kind + this.running = false + this.drain() + } + } + + private takeNext(): ScheduledOperation | undefined { + // Hydrates get priority after the currently running non-preemptible unit. + // While both lanes remain queued, alternate one regular operation after + // each hydrate (K=1), preserving FIFO order within each lane. + if (this.hydrateQueue.length > 0) { + if ( + this.regularQueue.length > 0 && + this.lastCompletedKind === `hydrate` + ) { + return this.regularQueue.shift() + } + return this.hydrateQueue.shift() + } + return this.regularQueue.shift() + } +} + +const sharedPersistenceSchedulers = new WeakMap< + object, + SharedPersistenceScheduler +>() + +function getSharedPersistenceScheduler( + key: object, +): SharedPersistenceScheduler { + let scheduler = sharedPersistenceSchedulers.get(key) + if (!scheduler) { + scheduler = new SharedPersistenceScheduler() + sharedPersistenceSchedulers.set(key, scheduler) + } + return scheduler +} + +function getSharedLogicalSchedulingKey(value: unknown): object | undefined { + if ((typeof value !== `object` && typeof value !== `function`) || !value) { + return undefined + } + const key = ( + value as { + [SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY]?: unknown + } + )[SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY] + return key !== null && (typeof key === `object` || typeof key === `function`) + ? key + : undefined +} + +function observeSharedLogicalSchedulingSupport( + driver: SQLiteDriver, + onSupport: (key: object) => void, +): SQLiteDriver { + const observe = (promise: Promise): Promise => { + const key = getSharedLogicalSchedulingKey(promise) + if (key) onSupport(key) + return promise + } + + return { + exec: (sql) => observe(driver.exec(sql)), + query: (sql: string, params: ReadonlyArray = []) => + observe(driver.query(sql, params)), + run: (sql, params = []) => observe(driver.run(sql, params)), + transaction: (fn: (transactionDriver: SQLiteDriver) => Promise) => + observe(driver.transaction(fn)), + transactionWithDriver: ( + fn: (transactionDriver: SQLiteDriver) => Promise, + ) => + observe( + driver.transactionWithDriver + ? driver.transactionWithDriver(fn) + : driver.transaction(fn), + ), + } +} + const DEFAULT_SCHEMA_VERSION = 1 const DEFAULT_PULL_SINCE_RELOAD_THRESHOLD = 128 @@ -1023,6 +1163,8 @@ function buildIndexName(collectionId: string, signature: string): string { export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { private readonly driver: SQLiteDriver + private scheduler: SharedPersistenceScheduler | undefined + private readonly hydrationAdapter: HydrationPersistenceAdapter private readonly schemaVersion: number private readonly schemaMismatchPolicy: SQLiteCoreAdapterSchemaMismatchPolicy private readonly appliedTxPruneMaxRows: number | undefined @@ -1078,13 +1220,52 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { ) } - this.driver = options.driver + const schedulingKey = getSharedLogicalSchedulingKey(options.driver) + this.scheduler = schedulingKey + ? getSharedPersistenceScheduler(schedulingKey) + : undefined + this.driver = schedulingKey + ? options.driver + : observeSharedLogicalSchedulingSupport(options.driver, (key) => { + this.scheduler ??= getSharedPersistenceScheduler(key) + }) this.schemaVersion = schemaVersion this.schemaMismatchPolicy = options.schemaMismatchPolicy ?? `sync-present-reset` this.appliedTxPruneMaxRows = options.appliedTxPruneMaxRows this.appliedTxPruneMaxAgeSeconds = options.appliedTxPruneMaxAgeSeconds this.pullSinceReloadThreshold = pullSinceReloadThreshold + this.hydrationAdapter = { + loadSubset: (collectionId, loadOptions, context) => + this.loadSubsetUnscheduled(collectionId, loadOptions, context), + applyCommittedTx: (collectionId, tx) => + this.applyCommittedTxUnscheduled(collectionId, tx), + loadCollectionMetadata: (collectionId) => + this.loadCollectionMetadataUnscheduled(collectionId), + scanRows: (collectionId, scanOptions) => + this.scanRowsUnscheduled(collectionId, scanOptions), + ensureIndex: (collectionId, signature, spec) => + this.ensureIndexUnscheduled(collectionId, signature, spec), + markIndexRemoved: (collectionId, signature) => + this.markIndexRemovedUnscheduled(collectionId, signature), + getStreamPosition: (collectionId) => + this.getStreamPositionUnscheduled(collectionId), + pullSince: (collectionId, fromRowVersion) => + this.pullSinceUnscheduled(collectionId, fromRowVersion), + runInHydrationScope: async (task) => task(this.hydrationAdapter), + } + } + + runInHydrationScope( + task: (adapter: HydrationPersistenceAdapter) => Promise, + ): Promise { + return this.scheduler + ? this.scheduler.runHydrate(() => task(this.hydrationAdapter)) + : Promise.resolve().then(() => task(this.hydrationAdapter)) + } + + private runRegular(task: () => Promise): Promise { + return this.scheduler ? this.scheduler.runRegular(task) : task() } private runInTransaction( @@ -1097,7 +1278,23 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { return this.driver.transaction(fn) } - async loadSubset( + loadSubset( + collectionId: string, + options: LoadSubsetOptions, + ctx?: { requiredIndexSignatures?: ReadonlyArray }, + ): Promise< + Array<{ + key: string | number + value: Record + metadata?: unknown + }> + > { + return this.runRegular(() => + this.loadSubsetUnscheduled(collectionId, options, ctx), + ) + } + + private async loadSubsetUnscheduled( collectionId: string, options: LoadSubsetOptions, ctx?: { requiredIndexSignatures?: ReadonlyArray }, @@ -1159,7 +1356,16 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { })) } - async applyCommittedTx(collectionId: string, tx: PersistedTx): Promise { + applyCommittedTx(collectionId: string, tx: PersistedTx): Promise { + return this.runRegular(() => + this.applyCommittedTxUnscheduled(collectionId, tx), + ) + } + + private async applyCommittedTxUnscheduled( + collectionId: string, + tx: PersistedTx, + ): Promise { const tableMapping = await this.ensureCollectionReady(collectionId) const collectionTableSql = quoteIdentifier(tableMapping.tableName) const tombstoneTableSql = quoteIdentifier(tableMapping.tombstoneTableName) @@ -1379,7 +1585,15 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { }) } - async loadCollectionMetadata( + loadCollectionMetadata( + collectionId: string, + ): Promise> { + return this.runRegular(() => + this.loadCollectionMetadataUnscheduled(collectionId), + ) + } + + private async loadCollectionMetadataUnscheduled( collectionId: string, ): Promise> { const rows = await this.driver.query<{ key: string; value: string }>( @@ -1395,7 +1609,16 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { })) } - async scanRows( + scanRows( + collectionId: string, + options?: PersistedRowScanOptions, + ): Promise> { + return this.runRegular(() => + this.scanRowsUnscheduled(collectionId, options), + ) + } + + private async scanRowsUnscheduled( collectionId: string, options?: PersistedRowScanOptions, ): Promise> { @@ -1418,7 +1641,17 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { })) } - async ensureIndex( + ensureIndex( + collectionId: string, + signature: string, + spec: PersistedIndexSpec, + ): Promise { + return this.runRegular(() => + this.ensureIndexUnscheduled(collectionId, signature, spec), + ) + } + + private async ensureIndexUnscheduled( collectionId: string, signature: string, spec: PersistedIndexSpec, @@ -1478,7 +1711,13 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { }) } - async markIndexRemoved( + markIndexRemoved(collectionId: string, signature: string): Promise { + return this.runRegular(() => + this.markIndexRemovedUnscheduled(collectionId, signature), + ) + } + + private async markIndexRemovedUnscheduled( collectionId: string, signature: string, ): Promise { @@ -1508,7 +1747,17 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { } } - async getStreamPosition(collectionId: string): Promise<{ + getStreamPosition(collectionId: string): Promise<{ + latestTerm: number + latestSeq: number + latestRowVersion: number + }> { + return this.runRegular(() => + this.getStreamPositionUnscheduled(collectionId), + ) + } + + private async getStreamPositionUnscheduled(collectionId: string): Promise<{ latestTerm: number latestSeq: number latestRowVersion: number @@ -1547,7 +1796,16 @@ export class SQLiteCorePersistenceAdapter implements PersistenceAdapter { } } - async pullSince( + pullSince( + collectionId: string, + fromRowVersion: number, + ): Promise> { + return this.runRegular(() => + this.pullSinceUnscheduled(collectionId, fromRowVersion), + ) + } + + private async pullSinceUnscheduled( collectionId: string, fromRowVersion: number, ): Promise> { diff --git a/packages/db-sqlite-persistence-core/tests/persisted.test.ts b/packages/db-sqlite-persistence-core/tests/persisted.test.ts index 313cdad860..50ddd16833 100644 --- a/packages/db-sqlite-persistence-core/tests/persisted.test.ts +++ b/packages/db-sqlite-persistence-core/tests/persisted.test.ts @@ -19,6 +19,7 @@ import { persistedCollectionOptions, } from '../src' import type { + CollectionReset, PersistedCollectionCoordinator, PersistedCollectionPersistence, PersistedSyncWrappedOptions, @@ -45,9 +46,9 @@ import type { LoadSubsetOptions, SyncConfig } from '@tanstack/db' * cleanup, and restart. They compare durable state, public rows, metadata, * request options, sequence evidence, errors, and late-work fencing. * - * Driver SQL behavior, browser page ownership, native runtimes, and the shared - * conformance portfolio have separate owners. This file models persistence - * protocol state, not a particular SQLite engine. + * Known omissions: driver SQL behavior, browser page ownership, native + * runtimes, and the shared conformance portfolio have separate owners. This + * file models persistence protocol state, not a particular SQLite engine. */ type Todo = { @@ -205,7 +206,7 @@ function createNoopAdapter(): PersistenceAdapter { } type CoordinatorHarness = PersistedCollectionCoordinator & { - emit: (payload: TxCommitted, senderId?: string) => void + emit: (payload: TxCommitted | CollectionReset, senderId?: string) => void pullSinceCalls: number setPullSinceResponse: (response: PullSinceResponse) => void } @@ -279,6 +280,17 @@ async function flushAsyncWork(delayMs: number = 0): Promise { await new Promise((resolve) => setTimeout(resolve, delayMs)) } +function createDeferred(): { + promise: Promise + resolve: (value: T) => void +} { + let resolve!: (value: T) => void + const promise = new Promise((settle) => { + resolve = settle + }) + return { promise, resolve } +} + describe(`persistedCollectionOptions`, () => { it(`provides a sync-absent loopback configuration with persisted utils`, async () => { const adapter = createRecordingAdapter() @@ -468,6 +480,73 @@ describe(`persistedCollectionOptions`, () => { }) }) + // Focused collection-reset refinement: metadata and rows must come from one + // hydration scope. The adapter makes an interleaved v2 write possible only + // outside that scope, so the public v1 metadata and row are the independent + // coherence checkpoint. This fixed history does not model arbitrary resets. + it(`keeps a collection-reset reload inside one hydration scope`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `Initial row` }]) + adapter.collectionMetadata.set(`snapshot`, `initial`) + const coordinator = createCoordinatorHarness() + const loadSubset = adapter.loadSubset.bind(adapter) + let inHydrationScope = false + let interleaveArmed = false + let interleaveRan = false + + const runInterleavedWrite = () => { + interleaveRan = true + adapter.collectionMetadata.set(`snapshot`, `v2`) + adapter.rows.set(`1`, { id: `1`, title: `v2 row` }) + } + + adapter.loadSubset = async (...args) => { + if (interleaveArmed && !inHydrationScope && !interleaveRan) { + runInterleavedWrite() + } + return loadSubset(...args) + } + adapter.runInHydrationScope = async (task) => { + inHydrationScope = true + try { + return await task(adapter) + } finally { + inHydrationScope = false + if (interleaveArmed && !interleaveRan) runInterleavedWrite() + } + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => { + markReady() + }, + }, + persistence: { adapter, coordinator }, + }), + ) + + await collection.preload() + adapter.collectionMetadata.set(`snapshot`, `v1`) + adapter.rows.set(`1`, { id: `1`, title: `v1 row` }) + interleaveArmed = true + + coordinator.emit({ + type: `collection:reset`, + schemaVersion: 1, + resetEpoch: 1, + }) + + await vi.waitFor(() => expect(interleaveRan).toBe(true)) + expect(collection._state.syncedCollectionMetadata.get(`snapshot`)).toBe( + `v1`, + ) + expect(collection.get(`1`)?.title).toBe(`v1 row`) + await collection.cleanup() + }) + it(`persists metadata-only wrapped sync transactions`, async () => { const adapter = createRecordingAdapter() @@ -1008,6 +1087,204 @@ describe(`persistedCollectionOptions`, () => { } }) + // Focused receipt-ownership refinements. A source receipt created by the + // hydration operation belongs to its waiter even if it rejects before the + // waiter snapshots; a receipt created after hydration work returns does not. + // The public load result and exact rejection identity distinguish those two + // boundaries without treating every pending source receipt as related. + it(`propagates an operation-owned receipt rejection that settles before the hydration waiter snapshots`, async () => { + const adapter = createRecordingAdapter() + const hydrateLoadEntered = createDeferred() + const allowHydrateLoad = createDeferred() + let gateHydrationLoad = true + adapter.loadSubset = async () => { + if (gateHydrationLoad) { + gateHydrationLoad = false + hydrateLoadEntered.resolve() + await allowHydrateLoad.promise + } + return [] + } + let remoteBegin: (() => void) | undefined + let remoteWrite: + | ((message: { type: `insert`; value: Todo }) => void) + | undefined + let remoteCommit: + | ((signal?: AbortSignal) => true | Promise) + | undefined + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present-settled-receipt-boundary`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + remoteBegin = begin + remoteWrite = write as (message: { + type: `insert` + value: Todo + }) => void + remoteCommit = commit + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + }), + ) + let load: Promise | undefined + let receipt: Promise | undefined + const abortController = new AbortController() + + try { + collection.startSyncImmediate() + await collection.stateWhenReady() + load = Promise.resolve(collection._sync.loadSubset({ limit: 1 })) + await hydrateLoadEntered.promise + + let keyReads = 0 + const establishingRow = { + get id() { + keyReads++ + if (keyReads === 2) abortController.abort() + return `establishing` + }, + title: `Abort during buffered replay`, + } + remoteBegin?.() + remoteWrite?.({ type: `insert`, value: establishingRow }) + const applied = remoteCommit?.(abortController.signal) + if (!(applied instanceof Promise)) { + throw new Error(`expected a buffered establishing receipt`) + } + receipt = applied + void receipt.catch(() => undefined) + + allowHydrateLoad.resolve() + const [loadResult, receiptResult] = await Promise.allSettled([ + load, + receipt, + ]) + + expect(keyReads).toBeGreaterThanOrEqual(2) + expect(abortController.signal.aborted).toBe(true) + expect(receiptResult.status).toBe(`rejected`) + expect(loadResult.status).toBe(`rejected`) + if ( + loadResult.status === `rejected` && + receiptResult.status === `rejected` + ) { + expect(loadResult.reason).toBe(receiptResult.reason) + } + } finally { + abortController.abort() + allowHydrateLoad.resolve() + await receipt?.catch(() => undefined) + await load?.catch(() => undefined) + await collection.cleanup() + } + }) + + it(`does not adopt an unrelated source receipt created after hydration work returns`, async () => { + const adapter = createRecordingAdapter() + const mutationEntered = createDeferred() + const releaseMutation = createDeferred() + const unrelatedStarted = createDeferred<{ + abortController: AbortController + receipt: Promise + }>() + const trace: Array = [] + let probeActive = false + let remoteBegin: (() => void) | undefined + let remoteWrite: + | ((message: { type: `insert`; value: Todo }) => void) + | undefined + let remoteCommit: + | ((signal?: AbortSignal) => true | Promise) + | undefined + + adapter.runInHydrationScope = async (task) => { + if (!probeActive) return task(adapter) + probeActive = false + + const result = await task(adapter) + trace.push(`hydrate-task-returned`) + + const abortController = new AbortController() + remoteBegin?.() + remoteWrite?.({ + type: `insert`, + value: { id: `unrelated`, title: `Outside hydrate boundary` }, + }) + const receipt = remoteCommit?.(abortController.signal) + if (!(receipt instanceof Promise)) { + throw new Error(`expected a pending unrelated receipt`) + } + trace.push(`unrelated-receipt-created`) + unrelatedStarted.resolve({ abortController, receipt }) + return result + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present-unrelated-receipt-boundary`, + getKey: (item) => item.id, + syncMode: `on-demand`, + sync: { + sync: ({ begin, write, commit, markReady }) => { + remoteBegin = begin + remoteWrite = write as (message: { + type: `insert` + value: Todo + }) => void + remoteCommit = commit + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + onInsert: async () => { + mutationEntered.resolve() + await releaseMutation.promise + }, + }), + ) + let mutation: ReturnType | undefined + let load: Promise | undefined + let unrelated: + | { abortController: AbortController; receipt: Promise } + | undefined + + try { + collection.startSyncImmediate() + await collection.stateWhenReady() + mutation = collection.insert({ id: `local`, title: `Persisting gate` }) + await mutationEntered.promise + + probeActive = true + load = Promise.resolve(collection._sync.loadSubset({ limit: 1 })) + unrelated = await unrelatedStarted.promise + expect(trace).toEqual([ + `hydrate-task-returned`, + `unrelated-receipt-created`, + ]) + + // Give the public load continuation the opportunity to snapshot receipts. + await flushAsyncWork() + unrelated.abortController.abort() + await unrelated.receipt.catch(() => undefined) + + await expect(load).resolves.toBeUndefined() + } finally { + unrelated?.abortController.abort() + releaseMutation.resolve() + await mutation?.isPersisted.promise.catch(() => undefined) + await load?.catch(() => undefined) + await collection.cleanup() + } + }) + it(`preserves row metadata set before a metadata-less insert in the same sync transaction`, async () => { const adapter = createRecordingAdapter() const ownership = { queryCollection: { owners: [`gc:q1`] } } @@ -1289,6 +1566,126 @@ describe(`persistedCollectionOptions`, () => { }) }) + // Focused R7 causal-replay witness: after hydration releases its buffer, the + // source receipt must replay without awaiting the persisting operation whose + // callback is itself awaiting that receipt. Persistence reach, both public + // settlements, and the final source row expose the otherwise hidden cycle. + it(`replays a buffered source receipt without blocking its persisting predecessor`, async () => { + const adapter = createRecordingAdapter() + const hydrateLoadEntered = createDeferred() + const allowHydrateLoad = createDeferred() + let gateHydrationLoad = true + const replayState = { persisted: false } + + adapter.loadSubset = async () => { + if (gateHydrationLoad) { + gateHydrationLoad = false + hydrateLoadEntered.resolve() + await allowHydrateLoad.promise + } + return [] + } + const applyCommittedTx = adapter.applyCommittedTx + adapter.applyCommittedTx = async (...args) => { + replayState.persisted = true + await applyCommittedTx(...args) + } + adapter.runInHydrationScope = (task) => task(adapter) + + let remoteBegin: (() => void) | undefined + let remoteWrite: + | ((message: { type: `update`; value: Todo }) => void) + | undefined + let remoteCommit: (() => true | Promise) | undefined + const sourceReady = createDeferred() + const bufferedCommitReturned = createDeferred<{ + receipt: Promise + }>() + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present-buffered-causal-replay`, + getKey: (item) => item.id, + sync: { + sync: ({ begin, write, commit, markReady }) => { + remoteBegin = begin + remoteWrite = write as (message: { + type: `update` + value: Todo + }) => void + remoteCommit = commit + sourceReady.resolve() + markReady() + return { loadSubset: () => true } + }, + }, + persistence: { adapter }, + onInsert: async () => { + if (!remoteBegin || !remoteWrite || !remoteCommit) { + throw new Error(`source sync is not ready`) + } + remoteBegin() + remoteWrite({ + type: `update`, + value: { id: `source-row`, title: `Buffered during hydrate` }, + }) + const applied = remoteCommit() + if (applied === true) { + throw new Error(`source commit was not buffered during hydration`) + } + bufferedCommitReturned.resolve({ receipt: applied }) + await applied + }, + }), + ) + + const preload = Promise.resolve(collection.preload()) + void preload.catch(() => undefined) + let mutationPersisted: Promise | undefined + let bufferedReceipt: Promise | undefined + + try { + await hydrateLoadEntered.promise + await sourceReady.promise + + const mutation = collection.insert({ id: `local`, title: `Pending` }) + mutationPersisted = mutation.isPersisted.promise + void mutationPersisted.catch(() => undefined) + const bufferedCommit = await bufferedCommitReturned.promise + bufferedReceipt = bufferedCommit.receipt + void bufferedReceipt.catch(() => undefined) + + allowHydrateLoad.resolve() + + let causalCycleObserved = false + for ( + let attempt = 0; + attempt < 100 && !replayState.persisted; + attempt++ + ) { + causalCycleObserved = collection._state.pendingSyncedTransactions.some( + (transaction) => + transaction.committed && transaction.applied.isPending(), + ) + if (causalCycleObserved) break + await Promise.resolve() + } + + expect(causalCycleObserved).toBe(false) + expect(replayState.persisted).toBe(true) + await expect(bufferedReceipt).resolves.toBeUndefined() + await expect(mutationPersisted).resolves.toBeDefined() + await expect(preload).resolves.toBeUndefined() + expect(stripVirtualProps(collection.get(`source-row`))).toEqual({ + id: `source-row`, + title: `Buffered during hydrate`, + }) + } finally { + allowHydrateLoad.resolve() + await collection.cleanup() + } + }) + it(`discards a hydration-buffered transaction aborted before replay`, async () => { const adapter = createRecordingAdapter() let resolveLoadSubset: (() => void) | undefined @@ -1720,6 +2117,153 @@ describe(`persistedCollectionOptions`, () => { }) }) + // Focused invalidation-reload refinements. Whether recovery follows a + // sequence gap or a contiguous committed notification, metadata and rows + // must be read inside one hydration scope. The adapter schedules v2 only + // outside the scope; coherent public v1 state is the checkpoint. + it(`keeps sequence-gap recovery inside one hydration scope`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `Initial row` }]) + adapter.collectionMetadata.set(`snapshot`, `initial`) + const coordinator = createCoordinatorHarness() + coordinator.setPullSinceResponse({ + type: `rpc:pullSince:res`, + rpcId: `pull-gap-scope`, + ok: true, + latestTerm: 1, + latestSeq: 1, + latestRowVersion: 1, + requiresFullReload: true, + }) + const loadSubset = adapter.loadSubset.bind(adapter) + let inHydrationScope = false + let interleaveArmed = false + let interleaveRan = false + + const runInterleavedWrite = () => { + interleaveRan = true + adapter.collectionMetadata.set(`snapshot`, `v2`) + adapter.rows.set(`1`, { id: `1`, title: `v2 row` }) + } + + adapter.loadSubset = async (...args) => { + if (interleaveArmed && !inHydrationScope && !interleaveRan) { + runInterleavedWrite() + } + return loadSubset(...args) + } + adapter.runInHydrationScope = async (task) => { + inHydrationScope = true + try { + return await task(adapter) + } finally { + inHydrationScope = false + if (interleaveArmed && !interleaveRan) runInterleavedWrite() + } + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => { + markReady() + }, + }, + persistence: { adapter, coordinator }, + }), + ) + + await collection.preload() + adapter.collectionMetadata.set(`snapshot`, `v1`) + adapter.rows.set(`1`, { id: `1`, title: `v1 row` }) + interleaveArmed = true + + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 2, + txId: `tx-gap-scope`, + latestRowVersion: 2, + requiresFullReload: false, + changedRows: [], + deletedKeys: [], + }) + + await vi.waitFor(() => expect(interleaveRan).toBe(true)) + expect(collection._state.syncedCollectionMetadata.get(`snapshot`)).toBe( + `v1`, + ) + expect(collection.get(`1`)?.title).toBe(`v1 row`) + await collection.cleanup() + }) + + it(`keeps contiguous committed reload inside one hydration scope`, async () => { + const adapter = createRecordingAdapter([{ id: `1`, title: `Initial row` }]) + adapter.collectionMetadata.set(`snapshot`, `initial`) + const coordinator = createCoordinatorHarness() + const loadSubset = adapter.loadSubset.bind(adapter) + let inHydrationScope = false + let interleaveArmed = false + let interleaveRan = false + + const runInterleavedWrite = () => { + interleaveRan = true + adapter.collectionMetadata.set(`snapshot`, `v2`) + adapter.rows.set(`1`, { id: `1`, title: `v2 row` }) + } + + adapter.loadSubset = async (...args) => { + if (interleaveArmed && !inHydrationScope && !interleaveRan) { + runInterleavedWrite() + } + return loadSubset(...args) + } + adapter.runInHydrationScope = async (task) => { + inHydrationScope = true + try { + return await task(adapter) + } finally { + inHydrationScope = false + if (interleaveArmed && !interleaveRan) runInterleavedWrite() + } + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present`, + getKey: (item) => item.id, + sync: { + sync: ({ markReady }) => { + markReady() + }, + }, + persistence: { adapter, coordinator }, + }), + ) + + await collection.preload() + adapter.collectionMetadata.set(`snapshot`, `v1`) + adapter.rows.set(`1`, { id: `1`, title: `v1 row` }) + interleaveArmed = true + + coordinator.emit({ + type: `tx:committed`, + term: 1, + seq: 1, + txId: `tx-contiguous-reload-scope`, + latestRowVersion: 1, + requiresFullReload: true, + }) + + await vi.waitFor(() => expect(interleaveRan).toBe(true)) + expect(collection._state.syncedCollectionMetadata.get(`snapshot`)).toBe( + `v1`, + ) + expect(collection.get(`1`)?.title).toBe(`v1 row`) + await collection.cleanup() + }) + it(`removes deleted rows after tx:committed invalidation reload`, async () => { const adapter = createRecordingAdapter([ { id: `1`, title: `Keep` }, @@ -1769,6 +2313,95 @@ describe(`persistedCollectionOptions`, () => { expect(collection.get(`2`)).toBeUndefined() }) + // Focused lifecycle-fencing witness: generation-zero startup is held across + // cleanup and rebound, then released while generation one is still loading. + // Zero ensure-index calls for the rebound signatures prove stale bootstrap + // and listener work did not cross the public lifecycle boundary. + it(`does not let stale startup install index work on a rebound lifecycle`, async () => { + const adapter = createRecordingAdapter() + const g0MetadataEntered = createDeferred() + const allowG0Metadata = createDeferred() + const g1MetadataEntered = createDeferred() + const allowG1Metadata = createDeferred() + let metadataCalls = 0 + adapter.loadCollectionMetadata = async () => { + metadataCalls++ + if (metadataCalls === 1) { + g0MetadataEntered.resolve() + await allowG0Metadata.promise + } else if (metadataCalls === 2) { + g1MetadataEntered.resolve() + await allowG1Metadata.promise + } + return [] + } + + const collection = createCollection( + persistedCollectionOptions({ + id: `sync-present-startup-index-generation`, + getKey: (item) => item.id, + defaultIndexType: BasicIndex, + sync: { + sync: ({ markReady }) => { + markReady() + }, + }, + persistence: { adapter }, + }), + ) + const stalePreload = Promise.resolve(collection.preload()) + void stalePreload.catch(() => undefined) + let freshReady: Promise | undefined + + try { + await g0MetadataEntered.promise + await collection.cleanup() + + const reboundIndex = collection.createIndex((row) => row.title, { + name: `rebound-bootstrap`, + }) + const reboundSignature = collection + .getIndexMetadata() + .find((metadata) => metadata.indexId === reboundIndex.id)?.signature + expect(reboundSignature).toBeDefined() + freshReady = collection.stateWhenReady() + + allowG0Metadata.resolve() + await g1MetadataEntered.promise + + const staleBootstrapCalls = adapter.ensureIndexCalls.filter( + (call) => call.signature === reboundSignature, + ) + const listenerIndex = collection.createIndex((row) => row.id, { + name: `rebound-listener`, + }) + const listenerSignature = collection + .getIndexMetadata() + .find((metadata) => metadata.indexId === listenerIndex.id)?.signature + expect(listenerSignature).toBeDefined() + const staleListenerCalls = adapter.ensureIndexCalls.filter( + (call) => call.signature === listenerSignature, + ) + + expect({ + staleBootstrapCalls: staleBootstrapCalls.length, + staleListenerCalls: staleListenerCalls.length, + }).toEqual({ + staleBootstrapCalls: 0, + staleListenerCalls: 0, + }) + + allowG1Metadata.resolve() + await freshReady + } finally { + allowG0Metadata.resolve() + allowG1Metadata.resolve() + await stalePreload.catch(() => undefined) + await freshReady?.catch(() => undefined) + await collection.cleanup() + } + }) + it(`does not let a stale invalidation reload overwrite a restarted lifecycle`, async () => { const adapter = createRecordingAdapter([{ id: `1`, title: `Initial` }]) const coordinator = createCoordinatorHarness() diff --git a/packages/db-sqlite-persistence-core/tests/shared-logical-scheduling.test.ts b/packages/db-sqlite-persistence-core/tests/shared-logical-scheduling.test.ts new file mode 100644 index 0000000000..65ef56c241 --- /dev/null +++ b/packages/db-sqlite-persistence-core/tests/shared-logical-scheduling.test.ts @@ -0,0 +1,151 @@ +/** + * # Which adapters share one logical scheduling boundary? + * + * Contract and source: the RFC #1659 driver protocol keys scheduling by the + * exact shared driver identity. Two fresh core adapters over that identity + * must not interleave a complete hydration scope with regular adapter work. + * Transparent wrappers must forward the identity unchanged, including when + * the identity itself is a function object. + * + * Independent relation and legal domain: a two-adapter history starts one + * hydration metadata query, holds it at the driver boundary, then requests one + * regular metadata query. Before release, exactly the first query may be + * admitted; after release, both operations must finish. The three legal driver + * forms are direct, transparently wrapped, and direct with a function-valued + * key. This relation records admissions without copying the production + * scheduler or invoking the opaque key. + * + * Production boundary and checkpoint: both operations use + * `createSQLiteCorePersistenceAdapter`; the hydration operation enters through + * `runInHydrationScope`. The first checkpoint is immediately after the regular + * request enters the adapter, while the first query remains held. Losing + * wrapper identity or treating a function key as a getter synchronously admits + * the second query and fails the exact admission assertion. + * + * Known omissions: this focused contract test does not establish K=1 lane + * fairness, SQL result correctness, eventual progress under arbitrary I/O, or + * cross-process coordination. Those belong to the shared-driver oracle and + * provider refinements. + */ +import { describe, expect, it } from 'vitest' +import { + SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY, + createSQLiteCorePersistenceAdapter, + forwardSQLiteDriverSharedLogicalScheduling, +} from '../src' +import type { SQLiteDriver } from '../src' + +type Deferred = { + promise: Promise + resolve: () => void +} + +function createDeferred(): Deferred { + let resolve!: () => void + const promise = new Promise((settle) => { + resolve = settle + }) + return { promise, resolve } +} + +class FirstQueryGatedDriver implements SQLiteDriver { + readonly [SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY]: object + readonly admissions: Array = [] + readonly firstQueryEntered = createDeferred() + private readonly firstQueryGate = createDeferred() + private holdFirstQuery = true + + constructor(sharedLogicalSchedulingKey: object = {}) { + this[SQLITE_DRIVER_SHARED_LOGICAL_SCHEDULING_KEY] = + sharedLogicalSchedulingKey + } + + exec(): Promise { + this.admissions.push(`exec`) + return Promise.resolve() + } + + async query(): Promise> { + this.admissions.push(`query`) + if (this.holdFirstQuery) { + this.holdFirstQuery = false + this.firstQueryEntered.resolve() + await this.firstQueryGate.promise + } + return [] + } + + run(): Promise { + this.admissions.push(`run`) + return Promise.resolve() + } + + transaction( + fn: (transactionDriver: SQLiteDriver) => Promise, + ): Promise { + this.admissions.push(`transaction`) + return fn(this) + } + + releaseFirstQuery(): void { + this.firstQueryGate.resolve() + } +} + +function createTransparentWrapper(driver: SQLiteDriver): SQLiteDriver { + return forwardSQLiteDriverSharedLogicalScheduling(driver, { + exec: (sql) => driver.exec(sql), + query: (sql: string, params: ReadonlyArray = []) => + driver.query(sql, params), + run: (sql, params = []) => driver.run(sql, params), + transaction: (fn: (transactionDriver: SQLiteDriver) => Promise) => + driver.transaction(fn), + transactionWithDriver: ( + fn: (transactionDriver: SQLiteDriver) => Promise, + ) => + driver.transactionWithDriver + ? driver.transactionWithDriver(fn) + : driver.transaction(fn), + }) +} + +describe(`shared logical scheduling`, () => { + it.each([ + { + name: `direct driver`, + key: {}, + wrap: (driver: SQLiteDriver) => driver, + }, + { + name: `transparent delegating driver`, + key: {}, + wrap: createTransparentWrapper, + }, + { + name: `function-valued key`, + key: () => undefined, + wrap: (driver: SQLiteDriver) => driver, + }, + ])( + `shares a scheduler across two fresh adapters through a $name`, + async ({ key, wrap }) => { + const underlying = new FirstQueryGatedDriver(key) + const driver = wrap(underlying) + const hydrateAdapter = createSQLiteCorePersistenceAdapter({ driver }) + const regularAdapter = createSQLiteCorePersistenceAdapter({ driver }) + + const hydrate = hydrateAdapter.runInHydrationScope!(async (scoped) => { + await scoped.loadCollectionMetadata!(`hydrate`) + }) + await underlying.firstQueryEntered.promise + + const regular = regularAdapter.loadCollectionMetadata!(`regular`) + + expect(underlying.admissions).toEqual([`query`]) + + underlying.releaseFirstQuery() + await Promise.all([hydrate, regular]) + expect(underlying.admissions.length).toBeGreaterThan(1) + }, + ) +}) diff --git a/packages/electron-db-sqlite-persistence/src/electron-coordinator.ts b/packages/electron-db-sqlite-persistence/src/electron-coordinator.ts index a4c6bb7fe8..efd242dfdc 100644 --- a/packages/electron-db-sqlite-persistence/src/electron-coordinator.ts +++ b/packages/electron-db-sqlite-persistence/src/electron-coordinator.ts @@ -1,6 +1,7 @@ import { safeRandomUUID } from '@tanstack/db-sqlite-persistence-core' import type { ApplyLocalMutationsResponse, + HydrationPersistenceAdapter, PersistedCollectionCoordinator, PersistedIndexSpec, PersistedMutationEnvelope, @@ -221,9 +222,15 @@ export class ElectronCollectionCoordinator implements PersistedCollectionCoordin collectionId: string, signature: string, spec: PersistedIndexSpec, + scopedAdapter?: HydrationPersistenceAdapter, ): Promise { if (this.isLeader(collectionId)) { - await this.requireAdapter().ensureIndex(collectionId, signature, spec) + // A scoped adapter is a leader-local capability and never crosses RPC. + await (scopedAdapter ?? this.requireAdapter()).ensureIndex( + collectionId, + signature, + spec, + ) return } @@ -270,13 +277,19 @@ export class ElectronCollectionCoordinator implements PersistedCollectionCoordin async pullSince( collectionId: string, fromRowVersion: number, + scopedAdapter?: HydrationPersistenceAdapter, ): Promise { if (this.isLeader(collectionId)) { - return this.handlePullSince(collectionId, { - type: `rpc:pullSince:req`, - rpcId: safeRandomUUID(), - fromRowVersion, - }) + // A scoped adapter is a leader-local capability and never crosses RPC. + return this.handlePullSince( + collectionId, + { + type: `rpc:pullSince:req`, + rpcId: safeRandomUUID(), + fromRowVersion, + }, + scopedAdapter, + ) } return this.sendRPC(collectionId, { @@ -733,10 +746,11 @@ export class ElectronCollectionCoordinator implements PersistedCollectionCoordin rpcId: string fromRowVersion: number }, + scopedAdapter?: HydrationPersistenceAdapter, ): Promise { const state = this.collections.get(collectionId) - const adapter = this.requireAdapter() + const adapter = scopedAdapter ?? this.requireAdapter() if (!adapter.pullSince) { return { type: `rpc:pullSince:res`, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 328d4fe5ad..450ddd4bfe 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1072,6 +1072,9 @@ importers: '@journeyapps/wa-sqlite': specifier: ^1.4.1 version: 1.5.0 + '@playwright/test': + specifier: ^1.60.0 + version: 1.60.0 '@types/better-sqlite3': specifier: ^7.6.13 version: 7.6.13