perf(wa-sqlite): bypass queued materialization broadcasts

.changeset/shared-worker-materialized-fastpath.md

@@ -0,0 +1,5 @@
+---
+'@treecrdt/wa-sqlite': patch
+---
+
+Avoid queuing shared-worker materialization broadcasts as RPC calls.

.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.

.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.

packages/treecrdt-wa-sqlite-vendor/treecrdt-ext.c

@@ -4,15 +4,12 @@
 // static library.
 
 #include <sqlite3.h>
+#include <emscripten/emscripten.h>
 
 // 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);
 }

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/`.

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<Treecr
   return createTreecrdtClient({
     docId: opts.docId,
     storage: { type: 'opfs', filename: opts.filename, fallback: 'throw' },
-    runtime: { type: opts.runtime },
+    runtime:
+      opts.runtime === 'shared-worker' && opts.sharedWorkerName
+        ? { type: 'shared-worker', name: opts.sharedWorkerName }
+        : { type: opts.runtime },
   });
 }
 

packages/treecrdt-wa-sqlite/e2e/tests/cross-tab.spec.ts

@@ -12,6 +12,48 @@ async function waitForCrossTabHarness(page: Page) {
 
 type RuntimeChoice = 'auto' | 'dedicated-worker' | 'shared-worker';
 
+type SharedWorkerBroadcastCounts = {
+  push: number;
+  queuedRpc: number;
+};
+
+async function recordSharedWorkerBroadcasts(page: Page) {
+  await page.addInitScript(() => {
+    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<SharedWorkerBroadcastCounts> {
+  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()]);

packages/treecrdt-wa-sqlite/e2e/tests/lifecycle.spec.ts

@@ -2,6 +2,12 @@ import { test, expect, type Page } from '@playwright/test';
 
 type LifecycleHarness = NonNullable<Window['__treecrdtLifecycle']>;
 type LifecycleRuntime = 'direct' | 'dedicated-worker' | 'shared-worker';
+type LifecycleOptions = {
+  docId: string;
+  filename: string;
+  runtime: LifecycleRuntime;
+  sharedWorkerName?: string;
+};
 
 const scenarios: Array<{
   runtime: LifecycleRuntime;
@@ -39,37 +45,23 @@ async function support(page: Page): Promise<ReturnType<LifecycleHarness['support
   });
 }
 
-async function drop(
-  page: Page,
-  opts: { docId: string; filename: string; runtime: LifecycleRuntime },
-) {
+async function drop(page: Page, opts: LifecycleOptions) {
   await page.evaluate(async (dropOpts) => {
     const harness = window.__treecrdtLifecycle;
     if (!harness) throw new Error('__treecrdtLifecycle not available');
     await harness.drop(dropOpts);
   }, 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');
     return await harness.write(writeOpts);
   }, 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(() => {});
+    }
+  });
 });

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);