diff --git a/.changeset/shared-worker-materialized-fastpath.md b/.changeset/shared-worker-materialized-fastpath.md new file mode 100644 index 00000000..24061982 --- /dev/null +++ b/.changeset/shared-worker-materialized-fastpath.md @@ -0,0 +1,5 @@ +--- +'@treecrdt/wa-sqlite': patch +--- + +Avoid queuing shared-worker materialization broadcasts as RPC calls. diff --git a/.changeset/wa-sqlite-explicit-extension-init.md b/.changeset/wa-sqlite-explicit-extension-init.md new file mode 100644 index 00000000..76380d4f --- /dev/null +++ b/.changeset/wa-sqlite-explicit-extension-init.md @@ -0,0 +1,5 @@ +--- +'@treecrdt/wa-sqlite': patch +--- + +Initialize the statically linked TreeCRDT extension explicitly after opening SQLite. diff --git a/.changeset/wa-sqlite-opfs-init-fallback.md b/.changeset/wa-sqlite-opfs-init-fallback.md new file mode 100644 index 00000000..3c775518 --- /dev/null +++ b/.changeset/wa-sqlite-opfs-init-fallback.md @@ -0,0 +1,5 @@ +--- +'@treecrdt/wa-sqlite': patch +--- + +Close failed SQLite and OPFS resources, honor memory fallback after initialization failures, and clean up failed worker lifecycles. diff --git a/packages/treecrdt-wa-sqlite-vendor/treecrdt-ext.c b/packages/treecrdt-wa-sqlite-vendor/treecrdt-ext.c index 2aa232ef..69a43b25 100644 --- a/packages/treecrdt-wa-sqlite-vendor/treecrdt-ext.c +++ b/packages/treecrdt-wa-sqlite-vendor/treecrdt-ext.c @@ -4,15 +4,12 @@ // static library. #include +#include // The Rust extension entrypoint (static-link build ignores the sqlite3_api_routines pointer). extern int sqlite3_treecrdt_init(sqlite3 *db, char **pzErrMsg, const void *pApi); -__attribute__((used, constructor)) static void treecrdt_register_auto(void) { - // wa-sqlite builds SQLite with SQLITE_OMIT_AUTOINIT, so ensure initialization. - sqlite3_initialize(); - - // SQLite calls the registered function with (db, err, api); cast to silence - // the prototype mismatch on platforms that declare xEntryPoint as void(*)(void). - sqlite3_auto_extension((void (*)(void))sqlite3_treecrdt_init); +EMSCRIPTEN_KEEPALIVE +int treecrdt_sqlite_init(sqlite3 *db) { + return sqlite3_treecrdt_init(db, 0, 0); } diff --git a/packages/treecrdt-wa-sqlite-vendor/wa-sqlite b/packages/treecrdt-wa-sqlite-vendor/wa-sqlite index 3b3fe0ca..d7a1e3c4 160000 --- a/packages/treecrdt-wa-sqlite-vendor/wa-sqlite +++ b/packages/treecrdt-wa-sqlite-vendor/wa-sqlite @@ -1 +1 @@ -Subproject commit 3b3fe0cabf7e8edb60adfb6eac04a855561516e2 +Subproject commit d7a1e3c46d0fb910e30a487d09e03377cb50360c diff --git a/packages/treecrdt-wa-sqlite/README.md b/packages/treecrdt-wa-sqlite/README.md index 51872667..2f5b253a 100644 --- a/packages/treecrdt-wa-sqlite/README.md +++ b/packages/treecrdt-wa-sqlite/README.md @@ -13,6 +13,10 @@ pnpm --filter @treecrdt/wa-sqlite build The build copies wa-sqlite WASM/JS assets into `dist/wa-sqlite/` for Node and packages them for browser apps via the Vite plugin. +Low-level callers that open a wa-sqlite handle themselves must call +`initializeTreecrdtExtension(module, handle)` before constructing an adapter with +`createWaSqliteApi`. `createTreecrdtClient()` does this automatically. + ## Browser usage Use `createTreecrdtClient()` with OPFS or in-memory storage. Browser apps should use `@treecrdt/wa-sqlite/vite-plugin` to copy assets into `public/wa-sqlite/`. diff --git a/packages/treecrdt-wa-sqlite/e2e/src/lifecycle.ts b/packages/treecrdt-wa-sqlite/e2e/src/lifecycle.ts index 6a33fd09..02554c64 100644 --- a/packages/treecrdt-wa-sqlite/e2e/src/lifecycle.ts +++ b/packages/treecrdt-wa-sqlite/e2e/src/lifecycle.ts @@ -17,6 +17,7 @@ type LifecycleOptions = { docId: string; filename: string; runtime: LifecycleRuntime; + sharedWorkerName?: string; }; export type LifecycleState = { @@ -39,7 +40,10 @@ async function createOpfsLifecycleClient(opts: LifecycleOptions): Promise { + const NativeSharedWorker = window.SharedWorker; + const counts = { push: 0, queuedRpc: 0 }; + (window as any).__treecrdtSharedWorkerBroadcasts = counts; + + function WrappedSharedWorker(url: string | URL, options?: string | WorkerOptions) { + const worker = new NativeSharedWorker(url, options as any); + const originalPostMessage = worker.port.postMessage.bind(worker.port); + worker.port.postMessage = ((message: unknown, transferOrOptions?: unknown) => { + if (message && typeof message === 'object') { + const record = message as { method?: unknown; type?: unknown; id?: unknown }; + if (record.method === 'broadcastMaterialized') counts.queuedRpc += 1; + if (record.type === 'materialized' && typeof record.id !== 'number') { + counts.push += 1; + } + } + return originalPostMessage(message, transferOrOptions as any); + }) as typeof worker.port.postMessage; + return worker; + } + + WrappedSharedWorker.prototype = NativeSharedWorker.prototype; + Object.defineProperty(window, 'SharedWorker', { + configurable: true, + value: WrappedSharedWorker, + writable: true, + }); + }); +} + +async function sharedWorkerBroadcastCounts(page: Page): Promise { + return page.evaluate( + () => (window as any).__treecrdtSharedWorkerBroadcasts ?? { push: 0, queuedRpc: 0 }, + ); +} + async function openClient(page: Page, docId: string, filename: string, runtime: RuntimeChoice) { return page.evaluate( async ({ docId, filename, runtime }) => { @@ -92,6 +134,7 @@ for (const scenario of scenarios) { pageB.on('console', (msg) => console.log(`[pageB][${msg.type()}] ${msg.text()}`)); try { + if (scenario.runtime === 'shared-worker') await recordSharedWorkerBroadcasts(pageA); await Promise.all([waitForCrossTabHarness(pageA), waitForCrossTabHarness(pageB)]); const [summaryA, summaryB] = await Promise.all([ @@ -181,6 +224,12 @@ for (const scenario of scenarios) { expect(parentDeletedOnA.existsByNode[child.node]).toBe(true); expect(parentDeletedOnA.childrenByParent[root]).not.toContain(parent.node); expect(parentDeletedOnA.childrenByParent[root]).toContain(child.node); + + if (scenario.runtime === 'shared-worker') { + const broadcasts = await sharedWorkerBroadcastCounts(pageA); + expect(broadcasts.queuedRpc).toBe(0); + expect(broadcasts.push).toBeGreaterThanOrEqual(1); + } } finally { await Promise.allSettled([closeClient(pageA), closeClient(pageB)]); await Promise.allSettled([pageA.close(), pageB.close()]); diff --git a/packages/treecrdt-wa-sqlite/e2e/tests/lifecycle.spec.ts b/packages/treecrdt-wa-sqlite/e2e/tests/lifecycle.spec.ts index ea79f1f3..e023885e 100644 --- a/packages/treecrdt-wa-sqlite/e2e/tests/lifecycle.spec.ts +++ b/packages/treecrdt-wa-sqlite/e2e/tests/lifecycle.spec.ts @@ -2,6 +2,12 @@ import { test, expect, type Page } from '@playwright/test'; type LifecycleHarness = NonNullable; type LifecycleRuntime = 'direct' | 'dedicated-worker' | 'shared-worker'; +type LifecycleOptions = { + docId: string; + filename: string; + runtime: LifecycleRuntime; + sharedWorkerName?: string; +}; const scenarios: Array<{ runtime: LifecycleRuntime; @@ -39,10 +45,7 @@ async function support(page: Page): Promise { const harness = window.__treecrdtLifecycle; if (!harness) throw new Error('__treecrdtLifecycle not available'); @@ -50,15 +53,7 @@ async function drop( }, opts); } -async function write( - page: Page, - opts: { - docId: string; - filename: string; - runtime: LifecycleRuntime; - closeBeforeReload?: boolean; - }, -) { +async function write(page: Page, opts: LifecycleOptions & { closeBeforeReload?: boolean }) { return page.evaluate(async (writeOpts) => { const harness = window.__treecrdtLifecycle; if (!harness) throw new Error('__treecrdtLifecycle not available'); @@ -66,10 +61,7 @@ async function write( }, opts); } -async function read( - page: Page, - opts: { docId: string; filename: string; runtime: LifecycleRuntime }, -) { +async function read(page: Page, opts: LifecycleOptions) { return page.evaluate(async (readOpts) => { const harness = window.__treecrdtLifecycle; if (!harness) throw new Error('__treecrdtLifecycle not available'); @@ -143,4 +135,55 @@ test.describe('browser OPFS lifecycle', () => { }); } } + + test('releases a SharedWorker port after failed OPFS initialization', async ({ + page, + }, testInfo) => { + if (testInfo.project.name !== 'chromium-dev') test.skip(); + test.setTimeout(120_000); + page.on('console', (msg) => console.log(`[page][${msg.type()}] ${msg.text()}`)); + + const suffix = `${Date.now().toString(36)}-${Math.random().toString(16).slice(2, 8)}`; + const sharedWorkerName = `lifecycle-recovery-${suffix}`; + const failedOpts: LifecycleOptions = { + docId: `lifecycle-recovery-failed-${suffix}`, + filename: `/${'x'.repeat(512)}.db`, + runtime: 'shared-worker', + sharedWorkerName, + }; + const firstOpts: LifecycleOptions = { + docId: `lifecycle-recovery-first-${suffix}`, + filename: `/lifecycle-recovery-first-${suffix}.db`, + runtime: 'shared-worker', + sharedWorkerName, + }; + const secondOpts: LifecycleOptions = { + docId: `lifecycle-recovery-second-${suffix}`, + filename: `/lifecycle-recovery-second-${suffix}.db`, + runtime: 'shared-worker', + sharedWorkerName, + }; + + await waitForHarness(page); + const opfsSupport = await support(page); + if (!opfsSupport.available) test.skip(true, `OPFS unavailable: ${opfsSupport.reason}`); + expect(new TextEncoder().encode(failedOpts.filename).byteLength).toBeGreaterThan(512); + + try { + await expect(write(page, failedOpts)).rejects.toThrow(/sqlite3_open_v2|OPFS requested/); + + expectReloadedTree(await write(page, { ...firstOpts, closeBeforeReload: true }), { + mode: 'worker', + runtime: 'shared-worker', + }); + + expectReloadedTree(await write(page, { ...secondOpts, closeBeforeReload: true }), { + mode: 'worker', + runtime: 'shared-worker', + }); + } finally { + await drop(page, { ...firstOpts, runtime: 'direct' }).catch(() => {}); + await drop(page, { ...secondOpts, runtime: 'direct' }).catch(() => {}); + } + }); }); diff --git a/packages/treecrdt-wa-sqlite/scripts/bench.ts b/packages/treecrdt-wa-sqlite/scripts/bench.ts index ad935495..23976a03 100644 --- a/packages/treecrdt-wa-sqlite/scripts/bench.ts +++ b/packages/treecrdt-wa-sqlite/scripts/bench.ts @@ -1,7 +1,7 @@ import path from 'node:path'; import { buildWorkloads, runWorkloads } from '@treecrdt/benchmark'; import { parseBenchCliArgs, repoRootFromImportMeta, writeResult } from '@treecrdt/benchmark/node'; -import { createWaSqliteApi } from '../dist/index.js'; +import { createWaSqliteApi, initializeTreecrdtExtension } from '../dist/index.js'; import { makeDbAdapter } from '../dist/db.js'; import { loadWaSqliteNode } from '../dist/node/load-wa-sqlite.js'; @@ -13,12 +13,13 @@ async function main() { const workloadDefs = buildWorkloads(opts.workloads, opts.sizes); // wa-sqlite is browser-first; in Node we only exercise the in-memory runtime. - const { sqlite3 } = await loadWaSqliteNode(); + const { sqlite3, module } = await loadWaSqliteNode(); const docId = 'treecrdt-wa-sqlite-bench'; // Probe extension registration once so benchmark timing isn't dominated by setup errors. const probeHandle = await sqlite3.open_v2(':memory:'); try { + await initializeTreecrdtExtension(module, probeHandle); await sqlite3.exec(probeHandle, 'SELECT treecrdt_ops_since(0)'); } catch (err) { const msg = sqlite3.errmsg ? sqlite3.errmsg(probeHandle) : String(err); @@ -29,6 +30,7 @@ async function main() { const adapterFactory = async () => { const handle = await sqlite3.open_v2(':memory:'); + await initializeTreecrdtExtension(module, handle); const db = makeDbAdapter(sqlite3, handle); const api = createWaSqliteApi(db); await api.setDocId(docId); diff --git a/packages/treecrdt-wa-sqlite/src/client.ts b/packages/treecrdt-wa-sqlite/src/client.ts index 57ab3e45..618f1ce8 100644 --- a/packages/treecrdt-wa-sqlite/src/client.ts +++ b/packages/treecrdt-wa-sqlite/src/client.ts @@ -367,24 +367,8 @@ async function createWorkerClient(opts: { for (const { reject } of pending.values()) reject(err); pending.clear(); }; - worker.addEventListener('message', onMessage); - worker.addEventListener('error', onError); - - // init - const initResult = (await call('init', [ - opts.baseUrl ?? '/', - opts.filename, - opts.storage, - opts.docId, - ])) as { storage?: StorageMode; filename?: string; opfsError?: string } | undefined; - const effectiveStorage: StorageMode = initResult?.storage === 'opfs' ? 'opfs' : 'memory'; - const effectiveFilename = - initResult?.filename ?? - (effectiveStorage === 'opfs' ? (opts.filename ?? '/treecrdt.db') : ':memory:'); - if (effectiveStorage === 'opfs') { - materialized.enableCrossTab({ docId: opts.docId, filename: effectiveFilename }); - } const cleanup = () => { + if (closed) return; closed = true; materialized.close(); for (const { reject } of pending.values()) reject(closedError); @@ -393,9 +377,23 @@ async function createWorkerClient(opts: { worker.removeEventListener('message', onMessage); worker.terminate(); }; + worker.addEventListener('message', onMessage); + worker.addEventListener('error', onError); + // init + let initResult: RpcResult<'init'>; + try { + initResult = await call('init', [opts.baseUrl ?? '/', opts.filename, opts.storage, opts.docId]); + } catch (err) { + cleanup(); + throw err; + } + const { storage: effectiveStorage, filename: effectiveFilename, opfsError } = initResult; + if (effectiveStorage === 'opfs') { + materialized.enableCrossTab({ docId: opts.docId, filename: effectiveFilename }); + } if (opts.requireOpfs && effectiveStorage !== 'opfs') { - const reason = initResult?.opfsError ? `: ${initResult.opfsError}` : ''; + const reason = opfsError ? `: ${opfsError}` : ''; try { if (!terminalError) await call('close', [] as RpcParams<'close'>); } catch { @@ -410,9 +408,8 @@ async function createWorkerClient(opts: { if (closed) return; try { if (!terminalError) await call('close', [] as RpcParams<'close'>); - cleanup(); } finally { - // noop: cleanup already handles terminal teardown, and repeated close is idempotent + cleanup(); } }; @@ -420,9 +417,8 @@ async function createWorkerClient(opts: { if (closed) return; try { if (!terminalError) await call('drop', [] as RpcParams<'drop'>); - cleanup(); } finally { - // noop: cleanup already handles terminal teardown, and repeated drop is idempotent + cleanup(); } }; @@ -487,9 +483,12 @@ async function createSharedWorkerClient(opts: { }; const materialized = createClientMaterializationDispatcher({ broadcast: (event) => { - void call('broadcastMaterialized', [event]).catch(() => { + if (closed || terminalError) return; + try { + port.postMessage({ type: 'materialized', event } satisfies RpcPushMessage); + } catch { // Closing tabs can race a final materialization notification. - }); + } }, }); @@ -512,18 +511,8 @@ async function createSharedWorkerClient(opts: { for (const { reject } of pending.values()) reject(err); pending.clear(); }; - port.addEventListener('message', onMessage); - port.addEventListener('messageerror', onMessageError); - port.start(); - - const initResult = (await call('init', [ - opts.baseUrl ?? '/', - opts.filename, - opts.storage, - opts.docId, - ])) as { storage?: StorageMode; filename?: string; opfsError?: string } | undefined; - const effectiveStorage: StorageMode = initResult?.storage === 'opfs' ? 'opfs' : 'memory'; const cleanup = () => { + if (closed) return; closed = true; materialized.close(); for (const { reject } of pending.values()) reject(closedError); @@ -532,9 +521,27 @@ async function createSharedWorkerClient(opts: { port.removeEventListener('messageerror', onMessageError); port.close(); }; + port.addEventListener('message', onMessage); + port.addEventListener('messageerror', onMessageError); + port.start(); + + let initResult: RpcResult<'init'>; + try { + initResult = await call('init', [opts.baseUrl ?? '/', opts.filename, opts.storage, opts.docId]); + } catch (err) { + try { + if (!terminalError) await call('close', [] as RpcParams<'close'>); + } catch { + // Initialization may not have completed, but the shared worker still needs its port removed. + } finally { + cleanup(); + } + throw err; + } + const { storage: effectiveStorage, opfsError } = initResult; if (opts.requireOpfs && effectiveStorage !== 'opfs') { - const reason = initResult?.opfsError ? `: ${initResult.opfsError}` : ''; + const reason = opfsError ? `: ${opfsError}` : ''; try { if (!terminalError) await call('close', [] as RpcParams<'close'>); } catch { @@ -549,9 +556,8 @@ async function createSharedWorkerClient(opts: { if (closed) return; try { if (!terminalError) await call('close', [] as RpcParams<'close'>); - cleanup(); } finally { - // noop + cleanup(); } }; @@ -559,9 +565,8 @@ async function createSharedWorkerClient(opts: { if (closed) return; try { if (!terminalError) await call('drop', [] as RpcParams<'drop'>); - cleanup(); } finally { - // noop + cleanup(); } }; diff --git a/packages/treecrdt-wa-sqlite/src/extension.ts b/packages/treecrdt-wa-sqlite/src/extension.ts new file mode 100644 index 00000000..2967d921 --- /dev/null +++ b/packages/treecrdt-wa-sqlite/src/extension.ts @@ -0,0 +1,72 @@ +type WaSqliteModule = { + cwrap?: ( + name: string, + returnType: string, + argTypes: string[], + opts?: { async?: boolean }, + ) => (...args: unknown[]) => Promise | number; + retryOps?: Promise[]; + pendingOps?: Promise[]; +}; + +const initCache = new WeakMap Promise | number>(); +const SQLITE_OK = 0; +const SQLITE_ERROR = 1; + +function pendingErrorCode(error: unknown): number { + if (typeof error !== 'object' || error === null || !('code' in error)) return SQLITE_ERROR; + const code = (error as { code?: unknown }).code; + return typeof code === 'number' && code !== SQLITE_OK ? code : SQLITE_ERROR; +} + +async function runWithWaSqliteRetries(run: () => Promise | number, module: WaSqliteModule) { + while (true) { + if (module.retryOps?.length) { + try { + await Promise.all(module.retryOps); + } finally { + module.retryOps = []; + } + } + + const rc = await run(); + if (rc === SQLITE_OK || !module.retryOps?.length) { + if (module.pendingOps?.length) { + try { + await Promise.all(module.pendingOps); + } catch (error) { + return pendingErrorCode(error); + } finally { + module.pendingOps = []; + } + } + return rc; + } + + // Unlike wa-sqlite's generic retry cap, this idempotent schema initializer can safely + // continue on the same open handle while each failed attempt queues real VFS work. + } +} + +/** Initialize the statically linked TreeCRDT extension on an open wa-sqlite handle. */ +export async function initializeTreecrdtExtension( + module: WaSqliteModule, + handle: number, +): Promise { + if (!module || typeof module.cwrap !== 'function') { + throw new Error('wa-sqlite module does not expose cwrap'); + } + + let init = initCache.get(module as object); + if (!init) { + init = module.cwrap('treecrdt_sqlite_init', 'number', ['number'], { async: true }) as ( + handle: number, + ) => Promise | number; + initCache.set(module as object, init); + } + + const rc = await runWithWaSqliteRetries(() => init(handle), module); + if (rc !== 0) { + throw new Error(`TreeCRDT SQLite extension init failed (rc=${rc})`); + } +} diff --git a/packages/treecrdt-wa-sqlite/src/index.browser.ts b/packages/treecrdt-wa-sqlite/src/index.browser.ts index b7b7ac05..518f1d0f 100644 --- a/packages/treecrdt-wa-sqlite/src/index.browser.ts +++ b/packages/treecrdt-wa-sqlite/src/index.browser.ts @@ -22,3 +22,4 @@ export { export { CLIENT_CLOSED_ERROR, createTreecrdtClient } from './client.js'; export { createWaSqliteApi } from './adapter.js'; +export { initializeTreecrdtExtension } from './extension.js'; diff --git a/packages/treecrdt-wa-sqlite/src/index.node.ts b/packages/treecrdt-wa-sqlite/src/index.node.ts index c48c2889..917a3840 100644 --- a/packages/treecrdt-wa-sqlite/src/index.node.ts +++ b/packages/treecrdt-wa-sqlite/src/index.node.ts @@ -14,6 +14,7 @@ export { createTreecrdtClient } from './node/client.js'; export { CLIENT_CLOSED_ERROR } from './client.js'; export { createWaSqliteApi } from './adapter.js'; +export { initializeTreecrdtExtension } from './extension.js'; export { loadWaSqliteNode } from './node/load-wa-sqlite.js'; export { openTreecrdtDbNode } from './node/open.js'; diff --git a/packages/treecrdt-wa-sqlite/src/node/open.ts b/packages/treecrdt-wa-sqlite/src/node/open.ts index 8e8c2637..ea0de5f5 100644 --- a/packages/treecrdt-wa-sqlite/src/node/open.ts +++ b/packages/treecrdt-wa-sqlite/src/node/open.ts @@ -1,6 +1,8 @@ -import { createWaSqliteApi } from '../adapter.js'; -import { makeDbAdapter } from '../db.js'; -import type { OpenTreecrdtDbOptions, OpenTreecrdtDbResult } from '../open-core.js'; +import { + openTreecrdtDbFromLoaded, + type OpenTreecrdtDbOptions, + type OpenTreecrdtDbResult, +} from '../open-core.js'; import { loadWaSqliteNode } from './load-wa-sqlite.js'; /** Node entry: loads wa-sqlite WASM from the filesystem (in-memory only). */ @@ -10,10 +12,7 @@ export async function openTreecrdtDbNode( if (opts.storage === 'opfs' && opts.requireOpfs) { throw new Error('OPFS is not supported in Node'); } - const { sqlite3 } = await loadWaSqliteNode(opts.baseUrl); - const handle = await sqlite3.open_v2(':memory:'); - const db = makeDbAdapter(sqlite3, handle); - const api = createWaSqliteApi(db, { onMaterialized: opts.onMaterialized }); - await api.setDocId(opts.docId); - return { db, api, storage: 'memory', filename: ':memory:' }; + const loadFresh = () => loadWaSqliteNode(opts.baseUrl); + const loaded = await loadFresh(); + return openTreecrdtDbFromLoaded({ ...opts, storage: 'memory' }, loaded, loadFresh); } diff --git a/packages/treecrdt-wa-sqlite/src/open-core.ts b/packages/treecrdt-wa-sqlite/src/open-core.ts index c403ce26..2d8156a1 100644 --- a/packages/treecrdt-wa-sqlite/src/open-core.ts +++ b/packages/treecrdt-wa-sqlite/src/open-core.ts @@ -4,6 +4,7 @@ import type { Database } from './types.js'; import { makeDbAdapter } from './db.js'; import type { TreecrdtAdapter } from '@treecrdt/interface'; import type { MaterializationEvent } from '@treecrdt/interface/engine'; +import { initializeTreecrdtExtension } from './extension.js'; export type OpenTreecrdtDbOptions = { baseUrl?: string; @@ -23,33 +24,134 @@ export type OpenTreecrdtDbResult = { opfsError?: string; }; +export type LoadedWaSqlite = { sqlite3: any; module: any }; +export type LoadFreshWaSqlite = () => Promise; + +const OPFS_VFS_NAME = 'opfs'; + +function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +function memoryFallbackError(opfsFailure: unknown, fallbackFailure: unknown): Error { + const error = new Error( + `OPFS initialization failed: ${errorMessage(opfsFailure)}; memory fallback failed: ${errorMessage(fallbackFailure)}`, + ) as Error & { cause?: unknown; opfsCause?: unknown }; + error.cause = fallbackFailure; + error.opfsCause = opfsFailure; + return error; +} + +async function closeIgnoringErrors(close: (() => Promise | void) | undefined): Promise { + if (!close) return; + try { + await close(); + } catch { + // Preserve the error that made initialization fail. + } +} + +async function openInitializedDatabase( + sqlite3: any, + module: any, + filename: string, + opts: OpenTreecrdtDbOptions, + vfsName?: string, +): Promise<{ db: Database; api: TreecrdtAdapter }> { + let db: Database | undefined; + try { + const handle = vfsName + ? await sqlite3.open_v2(filename, undefined, vfsName) + : await sqlite3.open_v2(filename); + db = makeDbAdapter(sqlite3, handle); + await initializeTreecrdtExtension(module, handle); + const api = createWaSqliteApi(db, { onMaterialized: opts.onMaterialized }); + await api.setDocId(opts.docId); + return { db, api }; + } catch (err) { + await closeIgnoringErrors(db?.close ? () => db!.close!() : undefined); + throw err; + } +} + +function closeDatabaseWithVfs(db: Database, vfs: { close?: () => Promise | void }): Database { + if (!vfs.close) return db; + let closePromise: Promise | null = null; + return { + ...db, + close: () => { + closePromise ??= (async () => { + try { + await db.close?.(); + } finally { + await vfs.close!(); + } + })(); + return closePromise; + }, + }; +} + export async function openTreecrdtDbFromLoaded( opts: OpenTreecrdtDbOptions, - loaded: { sqlite3: any; module: any }, + loaded: LoadedWaSqlite, + loadFresh: LoadFreshWaSqlite, ): Promise { const { sqlite3, module } = loaded; - - let storage: 'memory' | 'opfs' = opts.storage === 'opfs' ? 'opfs' : 'memory'; let opfsError: string | undefined; + let opfsFailure: unknown; + const requestedFilename = opts.filename ?? '/treecrdt.db'; - if (storage === 'opfs') { + if (opts.storage === 'opfs') { + let vfs: { close?: () => Promise | void } | undefined; try { - const vfs = await createOpfsVfs(module, { name: 'opfs', kind: opts.opfsVfs }); - sqlite3.vfs_register(vfs, true); + const initializedVfs = await createOpfsVfs(module, { + name: OPFS_VFS_NAME, + kind: opts.opfsVfs, + }); + vfs = initializedVfs; + sqlite3.vfs_register(initializedVfs, false); + const opened = await openInitializedDatabase( + sqlite3, + module, + requestedFilename, + opts, + OPFS_VFS_NAME, + ); + return { + ...opened, + db: closeDatabaseWithVfs(opened.db, initializedVfs), + storage: 'opfs', + filename: requestedFilename, + }; } catch (err) { - opfsError = err instanceof Error ? err.message : String(err); + opfsFailure = err; + opfsError = errorMessage(err); + await closeIgnoringErrors(vfs?.close ? () => vfs!.close!() : undefined); if (opts.requireOpfs) { - throw new Error(`OPFS requested but could not be initialized: ${opfsError}`); + const requiredError = new Error( + `OPFS requested but could not be initialized: ${opfsError}`, + ) as Error & { cause?: unknown }; + requiredError.cause = err; + throw requiredError; } - storage = 'memory'; } } - const filename = storage === 'opfs' ? (opts.filename ?? '/treecrdt.db') : ':memory:'; - const handle = await sqlite3.open_v2(filename); - const db = makeDbAdapter(sqlite3, handle); - const api = createWaSqliteApi(db, { onMaterialized: opts.onMaterialized }); - await api.setDocId(opts.docId); - - return opfsError ? { db, api, storage, filename, opfsError } : { db, api, storage, filename }; + // A failed OPFS attempt leaves its registered VFS and callback state on the module even after + // the VFS is closed. Isolate the memory fallback in a fresh module instead of reusing that state. + try { + const memoryLoaded = opfsError !== undefined ? await loadFresh() : loaded; + const opened = await openInitializedDatabase( + memoryLoaded.sqlite3, + memoryLoaded.module, + ':memory:', + opts, + ); + const result = { ...opened, storage: 'memory' as const, filename: ':memory:' }; + return opfsError !== undefined ? { ...result, opfsError } : result; + } catch (fallbackFailure) { + if (opfsError === undefined) throw fallbackFailure; + throw memoryFallbackError(opfsFailure, fallbackFailure); + } } diff --git a/packages/treecrdt-wa-sqlite/src/open.ts b/packages/treecrdt-wa-sqlite/src/open.ts index 8b2f6883..60206a26 100644 --- a/packages/treecrdt-wa-sqlite/src/open.ts +++ b/packages/treecrdt-wa-sqlite/src/open.ts @@ -9,6 +9,7 @@ export type { OpenTreecrdtDbOptions, OpenTreecrdtDbResult }; /** Browser/worker entry: loads wa-sqlite assets from public URLs. */ export async function openTreecrdtDb(opts: OpenTreecrdtDbOptions): Promise { - const loaded = await loadWaSqliteBrowser({ assetsDir: opts.baseUrl }); - return openTreecrdtDbFromLoaded(opts, loaded); + const loadFresh = () => loadWaSqliteBrowser({ assetsDir: opts.baseUrl }); + const loaded = await loadFresh(); + return openTreecrdtDbFromLoaded(opts, loaded, loadFresh); } diff --git a/packages/treecrdt-wa-sqlite/src/opfs.ts b/packages/treecrdt-wa-sqlite/src/opfs.ts index ebe81ac6..424ccbc9 100644 --- a/packages/treecrdt-wa-sqlite/src/opfs.ts +++ b/packages/treecrdt-wa-sqlite/src/opfs.ts @@ -1,5 +1,6 @@ import type { Database } from './types.js'; import { makeDbAdapter } from './db.js'; +import { initializeTreecrdtExtension } from './extension.js'; export type OpfsSupport = { available: boolean; @@ -166,30 +167,54 @@ export async function openWithStorage( opts: OpenOptions, ): Promise<{ db: Database; close?: () => Promise }> { const { moduleFactory, filename = ':memory:', sqliteApi, storage } = opts; - let module = await moduleFactory(); + const module = await moduleFactory(); const sqlite3 = sqliteApi.Factory(module); let file = filename; - if (storage === 'opfs') { - const support = detectOpfsSupport(); - if (!support.available) { - throw new Error(`OPFS unsupported: ${support.reason ?? 'unknown reason'}`); + let vfs: { close?: () => Promise | void } | undefined; + let vfsName: string | undefined; + let handle: number | undefined; + try { + if (storage === 'opfs') { + const support = detectOpfsSupport(); + if (!support.available) { + throw new Error(`OPFS unsupported: ${support.reason ?? 'unknown reason'}`); + } + vfsName = 'opfs'; + vfs = await createOpfsVfs(module, { name: vfsName, kind: opts.opfsVfs }); + sqlite3.vfs_register(vfs, false); + file = filename === ':memory:' ? '/treecrdt.db' : filename; } - const vfs = await createOpfsVfs(module, { name: 'opfs', kind: opts.opfsVfs }); - sqlite3.vfs_register(vfs, true); - file = filename === ':memory:' ? '/treecrdt.db' : filename; - } - const handle = await sqlite3.open_v2(file); - const db = makeDbAdapter(sqlite3, handle); - return { - db, - close: async () => { - try { - await sqlite3.close(handle); - } catch { - /* ignore */ - } - }, - }; + const openedHandle = vfsName + ? await sqlite3.open_v2(file, undefined, vfsName) + : await sqlite3.open_v2(file); + handle = openedHandle; + const db = makeDbAdapter(sqlite3, openedHandle); + await initializeTreecrdtExtension(module, openedHandle); + let closePromise: Promise | undefined; + return { + db, + close: () => + (closePromise ??= (async () => { + try { + await db.close?.(); + } finally { + await vfs?.close?.(); + } + })()), + }; + } catch (err) { + try { + if (handle !== undefined) await sqlite3.close(handle); + } catch { + // Preserve the initialization error. + } + try { + await vfs?.close?.(); + } catch { + // Preserve the initialization error. + } + throw err; + } } diff --git a/packages/treecrdt-wa-sqlite/src/rpc.ts b/packages/treecrdt-wa-sqlite/src/rpc.ts index 0a3754a5..2159cc6c 100644 --- a/packages/treecrdt-wa-sqlite/src/rpc.ts +++ b/packages/treecrdt-wa-sqlite/src/rpc.ts @@ -27,7 +27,6 @@ export type RpcSchema = { params: [ops: Operation[]]; result: MaterializationOutcome; }; - broadcastMaterialized: { params: [event: MaterializationEvent]; result: void }; opsSince: { params: [lamport: number, root?: string]; result: unknown[] }; opRefsAll: { params: []; result: unknown[] }; opRefsChildren: { params: [parent: string]; result: unknown[] }; diff --git a/packages/treecrdt-wa-sqlite/src/shared-worker.ts b/packages/treecrdt-wa-sqlite/src/shared-worker.ts index 05800351..877f472d 100644 --- a/packages/treecrdt-wa-sqlite/src/shared-worker.ts +++ b/packages/treecrdt-wa-sqlite/src/shared-worker.ts @@ -5,6 +5,7 @@ import { type RpcInitResult, type RpcMethod, type RpcParams, + type RpcPushMessage, type RpcRequest, type RpcResult, } from './rpc.js'; @@ -55,12 +56,22 @@ function broadcastMaterialized(event: MaterializationEvent, exclude?: MessagePor } } +function isClientPushMessage(message: RpcRequest | RpcPushMessage): message is RpcPushMessage { + return 'type' in message && message.type === 'materialized'; +} + (self as unknown as SharedWorkerGlobal).onconnect = (ev: MessageEvent) => { const port = ev.ports[0]; if (!port) return; ports.add(port); - port.onmessage = (message: MessageEvent) => { - const request = message.data; + port.onmessage = (message: MessageEvent) => { + const data = message.data; + if (isClientPushMessage(data)) { + broadcastMaterialized(data.event, port); + return; + } + + const request = data; const respondSuccess = (result?: unknown) => { const transfer = request.method === 'treePayload' || request.method === 'treeParent' @@ -90,12 +101,6 @@ async function handleRequest( return (await init(baseUrl, filename, storage, docId)) as RpcResult; } - if (request.method === 'broadcastMaterialized') { - const [event] = request.params as RpcParams<'broadcastMaterialized'>; - broadcastMaterialized(event, sourcePort); - return undefined; - } - if (request.method === 'close') { await close(sourcePort); return undefined; diff --git a/packages/treecrdt-wa-sqlite/src/types.ts b/packages/treecrdt-wa-sqlite/src/types.ts index deffe270..56147f5d 100644 --- a/packages/treecrdt-wa-sqlite/src/types.ts +++ b/packages/treecrdt-wa-sqlite/src/types.ts @@ -1,7 +1,7 @@ import type { MaterializationEvent, TreecrdtEngine } from '@treecrdt/interface/engine'; import { createMaterializationDispatcher } from '@treecrdt/interface/engine'; import type { SqliteRunner } from '@treecrdt/interface/sqlite'; -import type { RpcMethod, RpcParams, RpcRequest, RpcResult } from './rpc.js'; +import type { RpcMethod, RpcParams, RpcPushMessage, RpcRequest, RpcResult } from './rpc.js'; // Minimal wa-sqlite surface needed by the adapter. Exported so consumers // don't need to import types from wa-sqlite directly. @@ -64,7 +64,7 @@ export type WorkerProxy = { }; export type MessagePortProxy = { - postMessage(msg: RpcRequest, transfer?: Transferable[]): void; + postMessage(msg: RpcRequest | RpcPushMessage, transfer?: Transferable[]): void; start: () => void; close: () => void; addEventListener: (type: 'message' | 'messageerror', fn: (ev: any) => void) => void; diff --git a/packages/treecrdt-wa-sqlite/tests/extension.test.ts b/packages/treecrdt-wa-sqlite/tests/extension.test.ts new file mode 100644 index 00000000..ad56c1ed --- /dev/null +++ b/packages/treecrdt-wa-sqlite/tests/extension.test.ts @@ -0,0 +1,146 @@ +import { expect, test, vi } from 'vitest'; +import { initializeTreecrdtExtension } from '../src/extension.js'; + +test('initializes each database handle and caches the module wrapper', async () => { + const init = vi.fn(async () => 0); + const module = { + cwrap: vi.fn(() => init), + retryOps: [] as Promise[], + }; + + await initializeTreecrdtExtension(module, 11); + await initializeTreecrdtExtension(module, 12); + + expect(module.cwrap).toHaveBeenCalledOnce(); + expect(module.cwrap).toHaveBeenCalledWith('treecrdt_sqlite_init', 'number', ['number'], { + async: true, + }); + expect(init).toHaveBeenNthCalledWith(1, 11); + expect(init).toHaveBeenNthCalledWith(2, 12); +}); + +test('waits and retries when an async VFS operation requests it', async () => { + let attempt = 0; + const module = { + cwrap: vi.fn(() => + vi.fn(async () => { + attempt += 1; + if (attempt === 1) { + module.retryOps.push(Promise.resolve()); + return 5; + } + return 0; + }), + ), + retryOps: [] as Promise[], + }; + + await initializeTreecrdtExtension(module, 21); + expect(attempt).toBe(2); + expect(module.retryOps).toEqual([]); +}); + +test('clears a rejected retry operation without calling the initializer', async () => { + const retryFailure = new Error('retry failed'); + const init = vi.fn(async () => 0); + const module = { + cwrap: vi.fn(() => init), + retryOps: [Promise.reject(retryFailure)], + }; + + await expect(initializeTreecrdtExtension(module, 22)).rejects.toBe(retryFailure); + expect(module.retryOps).toEqual([]); + expect(init).not.toHaveBeenCalled(); +}); + +test('continues through multiple queued retry phases before succeeding', async () => { + const phases: string[] = []; + let attempt = 0; + const init = vi.fn(async () => { + attempt += 1; + phases.push(`init-${attempt}`); + if (attempt <= 3) { + const phase = attempt; + module.retryOps.push( + Promise.resolve().then(() => { + phases.push(`retry-${phase}`); + }), + ); + return 5; + } + return 0; + }); + const module = { + cwrap: vi.fn(() => init), + retryOps: [] as Promise[], + }; + + await initializeTreecrdtExtension(module, 23); + + expect(init).toHaveBeenCalledTimes(4); + expect(phases).toEqual(['init-1', 'retry-1', 'init-2', 'retry-2', 'init-3', 'retry-3', 'init-4']); + expect(module.retryOps).toEqual([]); +}); + +test('waits for pending VFS work before reporting successful initialization', async () => { + let releasePending!: () => void; + const pending = new Promise((resolve) => { + releasePending = resolve; + }); + const module = { + cwrap: vi.fn(() => + vi.fn(async () => { + module.pendingOps.push(pending); + return 0; + }), + ), + retryOps: [] as Promise[], + pendingOps: [] as Promise[], + }; + + let initialized = false; + const initialization = initializeTreecrdtExtension(module, 24).then(() => { + initialized = true; + }); + await Promise.resolve(); + expect(initialized).toBe(false); + + releasePending(); + await initialization; + expect(module.pendingOps).toEqual([]); +}); + +test.each([ + { failure: Object.assign(new Error('checkpoint failed'), { code: 10 }), expectedCode: 10 }, + { failure: new Error('checkpoint failed'), expectedCode: 1 }, +])( + 'maps a rejected pending operation to SQLite code $expectedCode', + async ({ failure, expectedCode }) => { + const module = { + cwrap: vi.fn(() => + vi.fn(async () => { + module.pendingOps.push(Promise.reject(failure)); + return 0; + }), + ), + retryOps: [] as Promise[], + pendingOps: [] as Promise[], + }; + + await expect(initializeTreecrdtExtension(module, 25)).rejects.toThrow( + `TreeCRDT SQLite extension init failed (rc=${expectedCode})`, + ); + expect(module.pendingOps).toEqual([]); + }, +); + +test('fails clearly when initialization returns an SQLite error code', async () => { + const module = { + cwrap: vi.fn(() => vi.fn(async () => 10)), + retryOps: [] as Promise[], + }; + + await expect(initializeTreecrdtExtension(module, 31)).rejects.toThrow( + 'TreeCRDT SQLite extension init failed (rc=10)', + ); +}); diff --git a/packages/treecrdt-wa-sqlite/tests/open-extension.test.ts b/packages/treecrdt-wa-sqlite/tests/open-extension.test.ts new file mode 100644 index 00000000..f45e6e97 --- /dev/null +++ b/packages/treecrdt-wa-sqlite/tests/open-extension.test.ts @@ -0,0 +1,96 @@ +import { beforeEach, expect, test, vi } from 'vitest'; + +vi.mock('../src/opfs.js', () => ({ createOpfsVfs: vi.fn() })); + +import { createOpfsVfs } from '../src/opfs.js'; +import { openTreecrdtDbFromLoaded } from '../src/open-core.js'; + +function createFakeModule(initResult = 0) { + const init = vi.fn(async () => initResult); + return { + cwrap: vi.fn(() => init), + init, + retryOps: [] as Promise[], + pendingOps: [] as Promise[], + }; +} + +function createFakeSqlite() { + let nextStatement = 100; + return { + vfs_register: vi.fn(), + open_v2: vi.fn(async () => 1), + statements: vi.fn(() => { + const statement = nextStatement++; + return { + next: async () => ({ value: statement }), + return: async () => undefined, + }; + }), + bind: vi.fn(), + step: vi.fn(async () => 101), + column_text: vi.fn(), + finalize: vi.fn(), + exec: vi.fn(), + close: vi.fn(), + }; +} + +beforeEach(() => { + vi.mocked(createOpfsVfs).mockReset(); +}); + +test('initializes the extension after opening a memory database', async () => { + const sqlite3 = createFakeSqlite(); + const module = createFakeModule(); + + const opened = await openTreecrdtDbFromLoaded( + { storage: 'memory', docId: 'memory-explicit-init' }, + { sqlite3, module }, + ); + + expect(sqlite3.open_v2).toHaveBeenCalledWith(':memory:'); + expect(module.init).toHaveBeenCalledWith(1); + expect(module.init.mock.invocationCallOrder[0]).toBeLessThan( + sqlite3.statements.mock.invocationCallOrder[0]!, + ); + await opened.db.close?.(); + expect(sqlite3.close).toHaveBeenCalledWith(1); +}); + +test('closes the database when explicit extension initialization fails', async () => { + const sqlite3 = createFakeSqlite(); + const module = createFakeModule(10); + + await expect( + openTreecrdtDbFromLoaded( + { storage: 'memory', docId: 'memory-explicit-init-failure' }, + { sqlite3, module }, + ), + ).rejects.toThrow('TreeCRDT SQLite extension init failed (rc=10)'); + + expect(sqlite3.close).toHaveBeenCalledWith(1); +}); + +test('uses the named OPFS VFS and closes partial resources when initialization fails', async () => { + const sqlite3 = createFakeSqlite(); + const module = createFakeModule(10); + const vfs = { close: vi.fn() }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + + await expect( + openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/explicit-init-failure.db', + docId: 'opfs-explicit-init-failure', + }, + { sqlite3, module }, + ), + ).rejects.toThrow('TreeCRDT SQLite extension init failed (rc=10)'); + + expect(sqlite3.vfs_register).toHaveBeenCalledWith(vfs, false); + expect(sqlite3.open_v2).toHaveBeenCalledWith('/explicit-init-failure.db', undefined, 'opfs'); + expect(sqlite3.close).toHaveBeenCalledWith(1); + expect(vfs.close).toHaveBeenCalledOnce(); +}); diff --git a/packages/treecrdt-wa-sqlite/tests/open-fallback.test.ts b/packages/treecrdt-wa-sqlite/tests/open-fallback.test.ts new file mode 100644 index 00000000..f90b30e0 --- /dev/null +++ b/packages/treecrdt-wa-sqlite/tests/open-fallback.test.ts @@ -0,0 +1,243 @@ +import { beforeEach, expect, test, vi } from 'vitest'; + +vi.mock('../src/opfs.js', () => ({ createOpfsVfs: vi.fn() })); + +import { createOpfsVfs } from '../src/opfs.js'; +import { openTreecrdtDbFromLoaded } from '../src/open-core.js'; + +function createFakeModule(initResult = 0) { + return { + cwrap: vi.fn(() => vi.fn(async () => initResult)), + retryOps: [] as Promise[], + pendingOps: [] as Promise[], + }; +} + +function createFakeSqlite( + opts: { failOpen?: string; failOpenError?: Error; failInitializationHandle?: number } = {}, +) { + const statementHandles = new Map(); + let nextHandle = 1; + let nextStatement = 100; + + return { + vfs_register: vi.fn(), + open_v2: vi.fn(async (filename: string, _flags?: number, _vfs?: string) => { + if (filename === opts.failOpen) throw opts.failOpenError ?? new Error('OPFS open failed'); + return nextHandle++; + }), + statements: vi.fn((handle: number) => { + const statement = nextStatement++; + statementHandles.set(statement, handle); + return { + next: async () => ({ value: statement }), + return: async () => undefined, + }; + }), + bind: vi.fn(), + step: vi.fn(async (statement: number) => { + if (statementHandles.get(statement) === opts.failInitializationHandle) { + throw new Error('TreeCRDT initialization failed'); + } + return 101; + }), + column_text: vi.fn(), + finalize: vi.fn(), + exec: vi.fn(), + close: vi.fn(), + }; +} + +beforeEach(() => { + vi.mocked(createOpfsVfs).mockReset(); +}); + +test('keeps the memory path single-pass without creating an OPFS VFS', async () => { + const sqlite3 = createFakeSqlite(); + const loadFresh = vi.fn(); + + const opened = await openTreecrdtDbFromLoaded( + { storage: 'memory', docId: 'memory-fast-path' }, + { sqlite3, module: createFakeModule() }, + loadFresh, + ); + + expect(opened.storage).toBe('memory'); + expect(sqlite3.open_v2).toHaveBeenCalledOnce(); + expect(sqlite3.open_v2).toHaveBeenCalledWith(':memory:'); + expect(sqlite3.vfs_register).not.toHaveBeenCalled(); + expect(createOpfsVfs).not.toHaveBeenCalled(); + expect(loadFresh).not.toHaveBeenCalled(); +}); + +test('falls back to memory when opening the OPFS database fails', async () => { + const vfs = { close: vi.fn() }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + const sqlite3 = createFakeSqlite({ failOpen: '/fallback.db' }); + const memorySqlite3 = createFakeSqlite(); + const loadFresh = vi.fn(async () => ({ sqlite3: memorySqlite3, module: createFakeModule() })); + + const opened = await openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/fallback.db', + docId: 'fallback-open', + requireOpfs: false, + }, + { sqlite3, module: createFakeModule() }, + loadFresh, + ); + + expect(opened.storage).toBe('memory'); + expect(opened.filename).toBe(':memory:'); + expect(opened.opfsError).toBe('OPFS open failed'); + expect(sqlite3.open_v2).toHaveBeenCalledOnce(); + expect(sqlite3.open_v2).toHaveBeenCalledWith('/fallback.db', undefined, 'opfs'); + expect(memorySqlite3.open_v2).toHaveBeenCalledOnce(); + expect(memorySqlite3.open_v2).toHaveBeenCalledWith(':memory:'); + expect(sqlite3.vfs_register).toHaveBeenCalledWith(vfs, false); + expect(vfs.close).toHaveBeenCalledOnce(); + expect(loadFresh).toHaveBeenCalledOnce(); +}); + +test('closes a partially initialized OPFS database before falling back', async () => { + const vfs = { close: vi.fn() }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + const sqlite3 = createFakeSqlite({ failInitializationHandle: 1 }); + const memorySqlite3 = createFakeSqlite(); + const loadFresh = vi.fn(async () => ({ sqlite3: memorySqlite3, module: createFakeModule() })); + + const opened = await openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/fallback-init.db', + docId: 'fallback-init', + requireOpfs: false, + }, + { sqlite3, module: createFakeModule() }, + loadFresh, + ); + + expect(opened.storage).toBe('memory'); + expect(opened.opfsError).toBe('TreeCRDT initialization failed'); + expect(sqlite3.close).toHaveBeenCalledWith(1); + expect(memorySqlite3.open_v2).toHaveBeenCalledWith(':memory:'); + expect(vfs.close).toHaveBeenCalledOnce(); + expect(loadFresh).toHaveBeenCalledOnce(); +}); + +test('falls back after explicit extension initialization fails', async () => { + const vfs = { close: vi.fn() }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + const sqlite3 = createFakeSqlite(); + const memorySqlite3 = createFakeSqlite(); + const loadFresh = vi.fn(async () => ({ + sqlite3: memorySqlite3, + module: createFakeModule(), + })); + + const opened = await openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/fallback-extension.db', + docId: 'fallback-extension', + requireOpfs: false, + }, + { sqlite3, module: createFakeModule(10) }, + loadFresh, + ); + + expect(opened.storage).toBe('memory'); + expect(opened.opfsError).toBe('TreeCRDT SQLite extension init failed (rc=10)'); + expect(sqlite3.close).toHaveBeenCalledWith(1); + expect(vfs.close).toHaveBeenCalledOnce(); + expect(memorySqlite3.open_v2).toHaveBeenCalledWith(':memory:'); +}); + +test('preserves both errors when the fresh memory fallback also fails', async () => { + const opfsFailure = new Error('OPFS open failed'); + const fallbackFailure = new Error('memory module load failed'); + const sqlite3 = createFakeSqlite({ failOpen: '/both-fail.db', failOpenError: opfsFailure }); + const vfs = { close: vi.fn() }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + + const result = openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/both-fail.db', + docId: 'both-fail', + requireOpfs: false, + }, + { sqlite3, module: createFakeModule() }, + vi.fn().mockRejectedValue(fallbackFailure), + ); + + await expect(result).rejects.toMatchObject({ + message: + 'OPFS initialization failed: OPFS open failed; memory fallback failed: memory module load failed', + cause: fallbackFailure, + opfsCause: opfsFailure, + }); + expect(vfs.close).toHaveBeenCalledOnce(); +}); + +test('keeps the successful OPFS path single-pass and closes its database and VFS once', async () => { + const vfs = { close: vi.fn() }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + const sqlite3 = createFakeSqlite(); + const loadFresh = vi.fn(); + + const opened = await openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/success.db', + docId: 'success', + requireOpfs: true, + }, + { sqlite3, module: createFakeModule() }, + loadFresh, + ); + + expect(opened.storage).toBe('opfs'); + expect(sqlite3.open_v2).toHaveBeenCalledOnce(); + expect(sqlite3.open_v2).toHaveBeenCalledWith('/success.db', undefined, 'opfs'); + expect(sqlite3.vfs_register).toHaveBeenCalledWith(vfs, false); + expect(loadFresh).not.toHaveBeenCalled(); + + await opened.db.close?.(); + await opened.db.close?.(); + expect(sqlite3.close).toHaveBeenCalledOnce(); + expect(vfs.close).toHaveBeenCalledOnce(); +}); + +test('required OPFS closes the VFS and does not attempt memory fallback', async () => { + const vfs = { + close: vi.fn(() => { + throw new Error('VFS close failed'); + }), + }; + vi.mocked(createOpfsVfs).mockResolvedValue(vfs); + const opfsFailure = new Error('OPFS open failed'); + const sqlite3 = createFakeSqlite({ failOpen: '/required.db', failOpenError: opfsFailure }); + const loadFresh = vi.fn(); + + await expect( + openTreecrdtDbFromLoaded( + { + storage: 'opfs', + filename: '/required.db', + docId: 'required', + requireOpfs: true, + }, + { sqlite3, module: createFakeModule() }, + loadFresh, + ), + ).rejects.toMatchObject({ + message: 'OPFS requested but could not be initialized: OPFS open failed', + cause: opfsFailure, + }); + + expect(sqlite3.open_v2).toHaveBeenCalledOnce(); + expect(vfs.close).toHaveBeenCalledOnce(); + expect(loadFresh).not.toHaveBeenCalled(); +}); diff --git a/packages/treecrdt-wa-sqlite/tests/worker-cleanup.test.ts b/packages/treecrdt-wa-sqlite/tests/worker-cleanup.test.ts new file mode 100644 index 00000000..3d1bab34 --- /dev/null +++ b/packages/treecrdt-wa-sqlite/tests/worker-cleanup.test.ts @@ -0,0 +1,131 @@ +import { afterEach, expect, test, vi } from 'vitest'; + +import { createTreecrdtClient } from '../src/client.js'; +import type { RpcRequest } from '../src/rpc.js'; + +type Runtime = 'dedicated-worker' | 'shared-worker'; +type RpcResponse = + | { id: number; ok: true; result?: unknown } + | { id: number; ok: false; error: string }; + +class FakeEndpoint { + readonly listeners = new Map void>>(); + readonly requests: RpcRequest[] = []; + terminated = false; + portClosed = false; + + constructor(private readonly respond: (request: RpcRequest) => RpcResponse) {} + + addEventListener(type: string, listener: (event: any) => void) { + const listeners = this.listeners.get(type) ?? new Set(); + listeners.add(listener); + this.listeners.set(type, listeners); + } + + removeEventListener(type: string, listener: (event: any) => void) { + this.listeners.get(type)?.delete(listener); + } + + postMessage(request: RpcRequest) { + this.requests.push(request); + const response = this.respond(request); + queueMicrotask(() => { + for (const listener of this.listeners.get('message') ?? []) listener({ data: response }); + }); + } + + start() {} + + terminate() { + this.terminated = true; + } + + close() { + this.portClosed = true; + } +} + +function installEndpoint(runtime: Runtime, respond: (request: RpcRequest) => RpcResponse) { + const endpoint = new FakeEndpoint(respond); + if (runtime === 'dedicated-worker') { + vi.stubGlobal( + 'Worker', + class { + constructor() { + return endpoint; + } + }, + ); + } else { + vi.stubGlobal( + 'SharedWorker', + class { + port = endpoint; + }, + ); + } + return endpoint; +} + +function clientOptions(runtime: Runtime) { + return { + storage: { type: 'memory' as const }, + runtime: + runtime === 'dedicated-worker' + ? ({ type: runtime } as const) + : ({ type: runtime, name: 'cleanup-test' } as const), + docId: `cleanup-${runtime}`, + }; +} + +function expectCleaned(runtime: Runtime, endpoint: FakeEndpoint) { + expect(runtime === 'dedicated-worker' ? endpoint.terminated : endpoint.portClosed).toBe(true); + expect([...endpoint.listeners.values()].every((listeners) => listeners.size === 0)).toBe(true); +} + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +for (const runtime of ['dedicated-worker', 'shared-worker'] as const) { + test(`${runtime} cleans up after rejected initialization`, async () => { + const endpoint = installEndpoint(runtime, (request) => ({ + id: request.id, + ok: false, + error: 'init failed', + })); + + await expect(createTreecrdtClient(clientOptions(runtime))).rejects.toThrow('init failed'); + + expectCleaned(runtime, endpoint); + expect(endpoint.requests.map((request) => request.method)).toEqual( + runtime === 'shared-worker' ? ['init', 'close'] : ['init'], + ); + }); + + test(`${runtime} cleans up when close RPC fails`, async () => { + const endpoint = installEndpoint(runtime, (request) => + request.method === 'init' + ? { id: request.id, ok: true, result: { storage: 'memory', filename: ':memory:' } } + : { id: request.id, ok: false, error: 'close failed' }, + ); + const client = await createTreecrdtClient(clientOptions(runtime)); + + await client.close(); + + expectCleaned(runtime, endpoint); + }); + + test(`${runtime} cleans up when drop RPC fails`, async () => { + const endpoint = installEndpoint(runtime, (request) => + request.method === 'init' + ? { id: request.id, ok: true, result: { storage: 'memory', filename: ':memory:' } } + : { id: request.id, ok: false, error: 'drop failed' }, + ); + const client = await createTreecrdtClient(clientOptions(runtime)); + + await expect(client.drop()).rejects.toThrow('drop failed'); + + expectCleaned(runtime, endpoint); + }); +}