diff --git a/README.md b/README.md
index 0e0d34b9..99dfe67f 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,7 @@
- 
+ 
@@ -471,7 +471,7 @@ The Settings → AI panel shows a live GPU status badge with adapter details and
| **Document Export** | docx + jszip | Word-compatible `.docx` generation (lazy-loaded) |
| **PWA** | Service Worker + Web App Manifest v3 | Offline support, installability, Workbox chunking |
| **i18n** | Custom React Context (`I18nContext.tsx`) | 2849 keys × 19 locales (de/en/es/fr/it + ar/he/fa RTL Beta + ja/zh/pt/el/fi/sv/hu/is/eu Beta); EN fallback; `localStorage` persistence |
-| **Testing** | Vitest 4.x (5807+ tests / 529 files) + Playwright E2E | Unit/integration + cross-browser E2E; Stryker mutation (manual workflow) |
+| **Testing** | Vitest 4.x (5807+ tests / 531 files) + Playwright E2E | Unit/integration + cross-browser E2E; Stryker mutation (manual workflow) |
| **Code Quality** | Biome (lint + format) + TypeScript 7 (tsgo) strict | `--error-on-warnings` in CI; zero `any` policy |
| **Visualization** | Force-directed graph | Interactive character relationship network |
| **Desktop** | Tauri v2 | Cross-platform installer; auto-updater via `latest.json` |
@@ -508,7 +508,7 @@ WorldScript-Studio/
│ ├── sw.js # PWA Service Worker
│ └── manifest.json # PWA Web App Manifest v3
├── tests/
-│ ├── unit/ # Vitest unit tests (5807+ tests, 529 files) — count spans tests/, components/, packages/*/tests/, not just this folder
+│ ├── unit/ # Vitest unit tests (5807+ tests, 531 files) — count spans tests/, components/, packages/*/tests/, not just this folder
│ │ ├── ai/ # aiSmallModules, aiCoreFallbackPaths
│ │ └── settings/ # WebLlmPanel, AiSections
│ └── e2e/ # Playwright specs + helpers.ts
@@ -667,7 +667,7 @@ The main pipeline is [`.github/workflows/ci.yml`](.github/workflows/ci.yml). Opt
| `scorecard` | weekly + `main` push | OpenSSF Scorecard — SARIF uploaded to GitHub Code Scanning |
**Current test metrics (2026-07-28):**
-- **5807+ unit tests** across **529 test files** — all passing
+- **5807+ unit tests** across **531 test files** — all passing
- Coverage thresholds: lines ≥ 74 · branches ≥ 60 · functions ≥ 67 · statements ≥ 72 — enforced in CI (see Codecov badge for live metrics)
- i18n: **2849 keys × 19 locales** (en/de/fr/es/it + ar/he/fa RTL Beta + ja/zh/pt/el/fi/sv/hu/is/eu Beta)
diff --git a/features/featureCatalog.ts b/features/featureCatalog.ts
index 52933b29..d78b9485 100644
--- a/features/featureCatalog.ts
+++ b/features/featureCatalog.ts
@@ -430,7 +430,7 @@ const RAW_FEATURE_CATALOG: CatalogEntryInput[] = [
description: 'Conditionally shows DuckDB query tab',
},
],
- implementedIn: ['services/duckdb/', 'workers/duckdbWorker.ts', 'hooks/useDuckDb.ts'],
+ implementedIn: ['services/duckdb/', 'workers/v2/duckdb.worker.ts', 'hooks/useDuckDb.ts'],
drifts: [],
},
diff --git a/packages/worker-bus/src/workerBus.ts b/packages/worker-bus/src/workerBus.ts
index b99bb73e..f7e4b964 100644
--- a/packages/worker-bus/src/workerBus.ts
+++ b/packages/worker-bus/src/workerBus.ts
@@ -32,6 +32,8 @@ export class WorkerBus {
private readonly dlq: DeadLetterQueue;
private readonly progress = new ProgressEmitter();
private readonly tokens = new Map();
+ // QNBS-v3: [Tracks which pool each in-flight task is running on, so terminatePool() can cancel exactly the affected tasks instead of leaving their result promises hanging forever.]
+ private readonly taskPools = new Map();
private readonly listeners = new Set();
private running = false;
private pumpScheduled = false;
@@ -49,6 +51,11 @@ export class WorkerBus {
this.schedulePump();
}
+ // QNBS-v3: [Lets callers (e.g. ensureDuckDbPool()) detect a pool removed via terminatePool() and re-register it, instead of assuming a non-null bus always has every pool.]
+ hasPool(poolId: string): boolean {
+ return this.pools.has(poolId);
+ }
+
registerPool(
poolId: string,
capabilities: readonly WorkerCapability[],
@@ -195,6 +202,25 @@ export class WorkerBus {
this.progress.clear();
}
+ /**
+ * Terminate all workers in a single named pool without shutting down the rest of the bus.
+ * QNBS-v3: unlike shutdown() (tears down every pool), this only affects `poolId` — added so
+ * duckdbClient's terminate()/shutdown() can stop the DuckDB worker without also killing
+ * in-flight inference/webllm/plugin work on the shared bus (docs/adr/0014-worker-generation-
+ * duplication.md's migration follow-up). The pool is removed from the registry; a later task
+ * routed to its capabilities fails with NO_POOL until the bus re-registers it.
+ * QNBS-v3: [Cancels every in-flight task routed to this pool before removing it — otherwise their result promises would await a RESULT message from a worker that's already been terminated and never arrives.]
+ */
+ async terminatePool(poolId: string): Promise {
+ const pool = this.pools.get(poolId);
+ if (!pool) return;
+ for (const [taskId, taskPoolId] of this.taskPools) {
+ if (taskPoolId === poolId) this.cancel(taskId, 'Pool terminated');
+ }
+ await pool.terminateAll();
+ this.pools.delete(poolId);
+ }
+
// ---------------------------------------------------------------------------
// Internal
// ---------------------------------------------------------------------------
@@ -220,6 +246,19 @@ export class WorkerBus {
};
}
+ this.taskPools.set(task.taskId, pool.poolId);
+ try {
+ return await this.runOnPool(task, token, pool);
+ } finally {
+ this.taskPools.delete(task.taskId);
+ }
+ }
+
+ private async runOnPool(
+ task: WorkerTask,
+ token: CancellationToken,
+ pool: WorkerPool,
+ ): Promise> {
const worker = await pool.acquire(token.signal);
const startedAt = performance.now();
const queueTimeMs = Math.round(startedAt - task.createdAt);
diff --git a/packages/worker-bus/tests/workerBus.test.ts b/packages/worker-bus/tests/workerBus.test.ts
index 45f494ba..c0c83332 100644
--- a/packages/worker-bus/tests/workerBus.test.ts
+++ b/packages/worker-bus/tests/workerBus.test.ts
@@ -209,6 +209,103 @@ describe('WorkerBus', () => {
expect(pool).toBeUndefined();
});
+ it('terminatePool terminates and removes only the named pool, leaving others intact', async () => {
+ bus.registerPool('other', ['db.duckdb'], {
+ maxWorkers: 1,
+ minWorkers: 1,
+ idleTimeoutMs: 120_000,
+ workerScript: '/other.worker.js',
+ capabilities: ['db.duckdb'],
+ labels: {},
+ });
+ const pools = (bus as unknown as { pools: Map }).pools;
+ const terminateAllSpy = vi.spyOn(pools.get('fake')!, 'terminateAll');
+ vi.spyOn(pools.get('other')!, 'acquire').mockReturnValue(new Promise(() => {}));
+
+ await bus.terminatePool('fake');
+
+ expect(terminateAllSpy).toHaveBeenCalled();
+ expect(pools.get('fake')).toBeUndefined();
+ expect(pools.get('other')).toBeDefined();
+ });
+
+ it('terminatePool is a no-op for an unknown pool id', async () => {
+ await expect(bus.terminatePool('does-not-exist')).resolves.toBeUndefined();
+ });
+
+ it('terminatePool cancels only the in-flight tasks routed to that pool', async () => {
+ // QNBS-v3: [beforeEach's mocked acquire() ignores the abort signal entirely (returns a
+ // promise that never settles); override it here with a signal-aware implementation
+ // mirroring WorkerPool's real waitForIdle() so cancellation can actually be observed
+ // unsticking the caller, not just verified as "cancel() was called".]
+ const fakePool = (bus as unknown as { pools: Map }).pools.get('fake')!;
+ vi.spyOn(fakePool, 'acquire').mockImplementation(
+ (signal) =>
+ new Promise((_resolve, reject) => {
+ signal?.addEventListener('abort', () => reject(new Error('Aborted')), { once: true });
+ }),
+ );
+ bus.registerPool('other', ['db.duckdb'], {
+ maxWorkers: 1,
+ minWorkers: 1,
+ idleTimeoutMs: 120_000,
+ workerScript: '/other.worker.js',
+ capabilities: ['db.duckdb'],
+ labels: {},
+ });
+ const otherPool = (bus as unknown as { pools: Map }).pools.get('other')!;
+ vi.spyOn(otherPool, 'acquire').mockReturnValue(new Promise(() => {}));
+
+ const fakeHandle = bus.enqueue('test.task', {}, { capabilities: ['inference.text'] });
+ const otherHandle = bus.enqueue('other.task', {}, { capabilities: ['db.duckdb'] });
+ otherHandle.result.catch(() => {});
+
+ const cancelSpy = vi.spyOn(bus, 'cancel');
+ await bus.terminatePool('fake');
+
+ expect(cancelSpy).toHaveBeenCalledTimes(1);
+ expect(cancelSpy).toHaveBeenCalledWith(fakeHandle.taskId, 'Pool terminated');
+ // Without the fix, this task would hang forever at pool.acquire() — terminatePool()
+ // cancelling it lets the result promise settle instead.
+ await expect(fakeHandle.result).rejects.toThrow('Aborted');
+ });
+
+ it('hasPool reports true for a registered pool and false for an unknown one', () => {
+ expect(bus.hasPool('fake')).toBe(true);
+ expect(bus.hasPool('does-not-exist')).toBe(false);
+ });
+
+ it('hasPool returns false after terminatePool removes the pool', async () => {
+ await bus.terminatePool('fake');
+ expect(bus.hasPool('fake')).toBe(false);
+ });
+
+ it('isolates a throwing subscriber so a later subscriber still receives the same event', async () => {
+ // QNBS-v3: [emit() is only reached by 'circuit-breaker-open'/'backpressure-rejected' events
+ // (grep-verified — regular enqueue/completion never calls this.emit()); reuse the
+ // same circuit-breaker trigger as 'rejects when circuit breaker open' above, adding a
+ // throwing listener ahead of a recording one to prove the try/catch isolates it.]
+ const events: Array<{ kind: string }> = [];
+ bus.subscribe(() => {
+ throw new Error('listener boom');
+ });
+ bus.subscribe((ev: unknown) => events.push(ev as { kind: string }));
+
+ const cb = (
+ bus as unknown as { getCircuitBreaker: (t: string) => { recordFailure: () => void } }
+ ).getCircuitBreaker('fragile.task');
+ cb.recordFailure();
+ cb.recordFailure();
+ cb.recordFailure();
+
+ const handle = bus.enqueue('fragile.task', {});
+ await expect(handle.result).rejects.toThrow('Circuit breaker is open');
+
+ expect(events).toContainEqual(
+ expect.objectContaining({ kind: 'circuit-breaker-open', taskType: 'fragile.task' }),
+ );
+ });
+
it('progress callback receives emitted progress', async () => {
const progressUpdates: Array<{ stage: string; progress: number }> = [];
let capturedTaskId = '';
diff --git a/services/duckdb/duckdbClient.ts b/services/duckdb/duckdbClient.ts
index 5751f0d4..62757b8c 100644
--- a/services/duckdb/duckdbClient.ts
+++ b/services/duckdb/duckdbClient.ts
@@ -1,80 +1,124 @@
-// QNBS-v3: Singleton proxy for the DuckDB-WASM worker.
-// Exposes typed query/exec/init/shutdown helpers with AbortSignal cancellation.
-// Worker is instantiated lazily on first call to init().
-
-import type {
- DuckDbRequest,
- DuckDbRequestType,
- DuckDbResponse,
- DuckDbWorkerEvent,
-} from '../../workers/duckdbWorker';
-
-let worker: Worker | null = null;
-const pendingResolvers = new Map void>();
-let messageIdCounter = 0;
-// QNBS-v3: Settable by useDuckDb to surface OPFS fallback state in Redux.
-let opfsFallbackCb: ((reason: string) => void) | null = null;
+// QNBS-v3: Singleton proxy for DuckDB analytics — routes through the shared WorkerBus v2 'duckdb'
+// pool (docs/adr/0014-worker-generation-duplication.md migration, superseded by
+// ADR-0015). Public API is unchanged from the v1-backed client (init/query/exec/
+// shutdown/terminate/setOpfsFallbackHandler) so all 5 consumers (useDuckDb.ts,
+// duckdbAnalytics.ts, duckdbMigration.ts, ragVectorMigration.ts, telemetryService.ts)
+// need no changes — only the transport underneath swapped.
+
+import { ensureDuckDbPool, getWorkerBus } from '../workerBusManager';
+
+export type DuckDbRequestType = 'INIT' | 'QUERY' | 'EXEC' | 'SHUTDOWN';
+
+export interface DuckDbResponse {
+ messageId: string;
+ ok: boolean;
+ rows?: Record[];
+ error?: string;
+ latencyMs?: number;
+}
+
+const TASK_TYPE: Record = {
+ INIT: 'db.duckdb.init',
+ QUERY: 'db.duckdb.query',
+ EXEC: 'db.duckdb.exec',
+ SHUTDOWN: 'db.duckdb.shutdown',
+};
+let messageIdCounter = 0;
function generateMessageId(): string {
return `duckdb-${Date.now()}-${++messageIdCounter}`;
}
-function getWorker(): Worker {
- if (!worker) {
- // QNBS-v3: Vite import.meta.url pattern — same as inference.worker.ts.
- worker = new Worker(new URL('../../workers/duckdbWorker.ts', import.meta.url), {
- type: 'module',
- });
- worker.addEventListener('message', (event: MessageEvent) => {
- const data = event.data as DuckDbResponse | DuckDbWorkerEvent;
- // QNBS-v3: OPFS_FALLBACK is out-of-band — no messageId resolver to call.
- if ((data as DuckDbWorkerEvent).type === 'OPFS_FALLBACK') {
- opfsFallbackCb?.((data as DuckDbWorkerEvent).reason);
- return;
- }
- const { messageId } = data as DuckDbResponse;
- const resolve = pendingResolvers.get(messageId);
- if (resolve) {
- pendingResolvers.delete(messageId);
- resolve(data as DuckDbResponse);
- }
- });
- worker.addEventListener('error', (event) => {
- // Reject all pending promises on fatal worker error
- const error = event.message ?? 'DuckDB worker crashed';
- for (const [id, resolve] of pendingResolvers) {
- resolve({ messageId: id, ok: false, error });
- }
- pendingResolvers.clear();
- worker = null;
- });
- }
- return worker;
+// QNBS-v3: Settable by useDuckDb to surface OPFS fallback state in Redux.
+let opfsFallbackCb: ((reason: string) => void) | null = null;
+
+function errorMessage(err: unknown): string {
+ const code = (err as { code?: string } | undefined)?.code;
+ if (code === 'CIRCUIT_OPEN') return 'DuckDB temporarily unavailable (circuit open)';
+ return err instanceof Error ? err.message : String(err);
}
-function send(
+async function send(
type: DuckDbRequestType,
sql?: string,
params?: readonly unknown[],
signal?: AbortSignal,
+ isReinitRetry = false,
): Promise {
const messageId = generateMessageId();
- const w = getWorker();
+ const start = Date.now();
- return new Promise((resolve) => {
- pendingResolvers.set(messageId, resolve);
+ // QNBS-v3: attach synchronously, before the `await` below — v1 set up its abort listener
+ // inside a synchronous Promise executor, so a caller aborting in the same tick as the
+ // call was always caught. This function is async and awaits ensureDuckDbPool() before
+ // a handle exists, which opens a gap; a same-tick abort must not be lost inside it.
+ let abortedEarly = signal?.aborted ?? false;
+ let handleRef: { cancel: (reason?: string) => void } | undefined;
+ const onAbort = (): void => {
+ abortedEarly = true;
+ handleRef?.cancel('Aborted');
+ };
+ signal?.addEventListener('abort', onAbort, { once: true });
- if (signal) {
- signal.addEventListener('abort', () => {
- w.postMessage({ type: 'WORKER_CANCEL', messageId });
- pendingResolvers.delete(messageId);
- resolve({ messageId, ok: false, error: 'Aborted' });
- });
- }
+ try {
+ const bus = await ensureDuckDbPool();
+ if (!bus) return { messageId, ok: false, error: 'WorkerBus v2 unavailable' };
- const req: DuckDbRequest = { messageId, type, sql, params };
- w.postMessage(req);
- });
+ // QNBS-v3: v1 never retried a failed QUERY/EXEC (caller decided) — disabling the bus's
+ // default 2-retry here keeps that behavior instead of silently stacking under
+ // duckdbAnalytics.ts's own withDuckDbRetry() app-level retry
+ // (app/listenerMiddleware.ts).
+ const handle = bus.enqueue<
+ { sql?: string | undefined; params?: readonly unknown[] | undefined },
+ unknown
+ >(
+ TASK_TYPE[type],
+ { sql, params },
+ {
+ capabilities: ['db.duckdb'],
+ retryPolicy: { maxRetries: 0 },
+ onProgress: (p) => {
+ if (p.stage === 'opfs-fallback') opfsFallbackCb?.(p.message ?? 'OPFS unavailable');
+ },
+ },
+ );
+ handleRef = handle;
+ if (abortedEarly) handle.cancel('Aborted');
+
+ try {
+ const raw = await handle.result;
+ const latencyMs = Date.now() - start;
+ if (type === 'QUERY') {
+ return { messageId, ok: true, rows: raw as Record[], latencyMs };
+ }
+ return { messageId, ok: true, latencyMs };
+ } catch (err) {
+ const latencyMs = Date.now() - start;
+ if (signal?.aborted) return { messageId, ok: false, error: 'Aborted', latencyMs };
+ const message = errorMessage(err);
+ // QNBS-v3: [The duckdb pool's worker can idle-terminate (minWorkers:0) or crash-restart; a
+ // freshly spawned worker's module state has no `connection` until INIT runs again.
+ // Transparently re-init once and retry, instead of surfacing a confusing "not
+ // initialized" error for what looks like a routine idle-timeout respawn.]
+ if (
+ !isReinitRetry &&
+ type !== 'INIT' &&
+ type !== 'SHUTDOWN' &&
+ message === 'DuckDB not initialized'
+ ) {
+ const reinit = await send('INIT', undefined, undefined, signal, true);
+ if (reinit.ok) {
+ return send(type, sql, params, signal, true);
+ }
+ }
+ return { messageId, ok: false, error: message, latencyMs };
+ }
+ } finally {
+ // QNBS-v3: [{once:true} only self-removes once the listener actually fires — a caller that
+ // reuses one AbortSignal across many queries (a common hook/thunk pattern) would
+ // otherwise accumulate one listener per call for the signal's whole lifetime.]
+ signal?.removeEventListener('abort', onAbort);
+ }
}
export const duckdbClient = {
@@ -98,14 +142,17 @@ export const duckdbClient = {
return send('SHUTDOWN', undefined, undefined, signal);
},
- /** Terminate the worker immediately (no flush). */
+ /**
+ * Terminate the DuckDB pool immediately (no flush).
+ * QNBS-v3: scoped via WorkerBus.terminatePool('duckdb') so this doesn't tear down the shared
+ * bus's other pools (inference/webllm/plugin) — bus.shutdown() would kill all of them.
+ */
terminate(): void {
- worker?.terminate();
- worker = null;
- pendingResolvers.clear();
+ const bus = getWorkerBus();
+ void bus?.terminatePool('duckdb');
},
- /** Register a callback invoked when the worker falls back to in-memory (OPFS unavailable). */
+ /** Register a callback invoked when DuckDB falls back to in-memory (OPFS unavailable). */
setOpfsFallbackHandler(cb: ((reason: string) => void) | null): void {
opfsFallbackCb = cb;
},
diff --git a/services/workerBusManager.ts b/services/workerBusManager.ts
index eccf3f18..a8004663 100644
--- a/services/workerBusManager.ts
+++ b/services/workerBusManager.ts
@@ -46,6 +46,27 @@ export async function initWorkerBus(): Promise {
}
}
+// QNBS-v3: [Shared with reRegisterDuckDbPool() below so a pool removed via terminatePool() gets re-registered with identical options, not a drifted copy.]
+async function duckDbPoolOptions() {
+ const { WORKER_IDLE_TIMEOUT_MS } = await import('@domain/worker-bus');
+ return {
+ // QNBS-v3: DuckDB is single-writer — minWorkers 0 keeps the thread alive only while in use.
+ maxWorkers: 1,
+ minWorkers: 0,
+ idleTimeoutMs: WORKER_IDLE_TIMEOUT_MS,
+ workerScript: new URL('../workers/v2/duckdb.worker.ts', import.meta.url).href,
+ capabilities: ['db.duckdb'] as const,
+ labels: { pool: 'duckdb', version: 'v2' },
+ };
+}
+
+/** Re-register the 'duckdb' pool if it was removed via terminatePool() — a no-op if it's already present. */
+async function reRegisterDuckDbPool(bus: WorkerBus): Promise {
+ if (bus.hasPool('duckdb')) return;
+ const options = await duckDbPoolOptions();
+ bus.registerPool('duckdb', options.capabilities, options);
+}
+
async function doInitWorkerBus(): Promise {
try {
const {
@@ -78,7 +99,6 @@ async function doInitWorkerBus(): Promise {
// QNBS-v3: new URL(path, import.meta.url) lets Vite emit the worker script as a proper
// asset URL. The .ts extension is allowed — Vite transforms it during build.
const inferenceUrl = new URL('../workers/v2/inference.worker.ts', import.meta.url).href;
- const duckdbUrl = new URL('../workers/v2/duckdb.worker.ts', import.meta.url).href;
// QNBS-v3: P1-1 — dedicated WebLLM (WebGPU) worker. Separate pool keeps @mlc-ai/web-llm out
// of the transformers.js worker bundle and isolates the GPU lifecycle.
const webllmUrl = new URL('../workers/v2/webllm.worker.ts', import.meta.url).href;
@@ -96,18 +116,11 @@ async function doInitWorkerBus(): Promise {
},
});
+ const duckdbOptions = await duckDbPoolOptions();
registry.register({
poolId: 'duckdb',
- // QNBS-v3: DuckDB is single-writer — minWorkers 0 keeps the thread alive only while in use.
- capabilities: ['db.duckdb'],
- options: {
- maxWorkers: 1,
- minWorkers: 0,
- idleTimeoutMs: WORKER_IDLE_TIMEOUT_MS,
- workerScript: duckdbUrl,
- capabilities: ['db.duckdb'],
- labels: { pool: 'duckdb', version: 'v2' },
- },
+ capabilities: duckdbOptions.capabilities,
+ options: duckdbOptions,
});
// QNBS-v3: P1-1 — WebLLM pool. maxWorkers:1 (single heavy GPU consumer; tab-leader election
@@ -164,6 +177,23 @@ export async function ensureWebLlmPool(): Promise {
return _bus;
}
+/**
+ * Ensure the DuckDB worker pool is available, initializing the WorkerBus on demand.
+ * QNBS-v3: mirrors ensureWebLlmPool()'s decoupling from `enableWorkerBusV2` — DuckDB analytics
+ * defaults on and was never gated by that flag in the v1 worker it replaces
+ * (docs/adr/0014-worker-generation-duplication.md), so toggling an experimental infra flag off
+ * must not silently break analytics. Returns null only if init failed.
+ */
+export async function ensureDuckDbPool(): Promise {
+ if (_bus === null) {
+ await initWorkerBus();
+ return _bus;
+ }
+ // QNBS-v3: [terminatePool('duckdb') can remove the pool while the bus itself stays alive — re-register it here instead of assuming a non-null bus always has every pool.]
+ await reRegisterDuckDbPool(_bus);
+ return _bus;
+}
+
/**
* Shut down the WorkerBus v2 and terminate all worker threads.
* Called when the feature flag is disabled or the app unmounts.
diff --git a/tests/unit/duckdbClient.test.ts b/tests/unit/duckdbClient.test.ts
index ee1726b4..34392674 100644
--- a/tests/unit/duckdbClient.test.ts
+++ b/tests/unit/duckdbClient.test.ts
@@ -1,130 +1,370 @@
-import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
-// QNBS-v3: duckdbClient instantiates a Worker via new URL(…, import.meta.url).
-// We mock the Worker constructor so no real worker is spun up in jsdom.
+// QNBS-v3: duckdbClient now routes through WorkerBus v2 (workers/v2/duckdb.worker.ts) instead of
+// spawning a raw Worker directly — mock services/workerBusManager's ensureDuckDbPool()/
+// getWorkerBus() instead of the global Worker constructor.
-const mockPostMessage = vi.fn();
-const mockTerminate = vi.fn();
-const mockAddEventListener = vi.fn();
+const { mockEnqueue, mockTerminatePool, mockEnsureDuckDbPool, mockGetWorkerBus } = vi.hoisted(
+ () => ({
+ mockEnqueue: vi.fn(),
+ mockTerminatePool: vi.fn(),
+ mockEnsureDuckDbPool: vi.fn(),
+ mockGetWorkerBus: vi.fn(),
+ }),
+);
-class MockWorker {
- postMessage = mockPostMessage;
- terminate = mockTerminate;
- addEventListener = mockAddEventListener;
+vi.mock('../../services/workerBusManager', () => ({
+ ensureDuckDbPool: mockEnsureDuckDbPool,
+ getWorkerBus: mockGetWorkerBus,
+}));
+
+const { duckdbClient } = await import('../../services/duckdb/duckdbClient');
+
+function deferred() {
+ let resolve!: (v: T) => void;
+ let reject!: (e: unknown) => void;
+ const promise = new Promise((res, rej) => {
+ resolve = res;
+ reject = rej;
+ });
+ return { promise, resolve, reject };
}
-vi.stubGlobal('Worker', MockWorker);
+function makeHandle(result: Promise, cancel: ReturnType = vi.fn()) {
+ return { taskId: 't1', result, progress: (async function* () {})(), cancel };
+}
-// Re-import after stubbing so the module captures MockWorker
-const { duckdbClient } = await import('../../services/duckdb/duckdbClient');
+function makeBus() {
+ return { enqueue: mockEnqueue, terminatePool: mockTerminatePool };
+}
beforeEach(() => {
vi.clearAllMocks();
- // Reset internal worker state between tests
- duckdbClient.terminate();
+ mockEnsureDuckDbPool.mockResolvedValue(makeBus());
+ mockGetWorkerBus.mockReturnValue(makeBus());
});
describe('duckdbClient.init', () => {
- it('posts an INIT message with a messageId', async () => {
- const promise = duckdbClient.init();
+ it('enqueues db.duckdb.init with the db.duckdb capability and returns {ok:true}', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.resolve({ ok: true })));
+
+ const res = await duckdbClient.init();
- // Simulate worker responding synchronously via the message event listener
- const listenerCall = mockAddEventListener.mock.calls.find(([ev]) => ev === 'message');
- expect(listenerCall).toBeDefined();
- const onMessage = listenerCall![1] as (e: MessageEvent) => void;
+ expect(mockEnqueue).toHaveBeenCalledWith(
+ 'db.duckdb.init',
+ expect.anything(),
+ expect.objectContaining({ capabilities: ['db.duckdb'] }),
+ );
+ expect(res.ok).toBe(true);
+ });
- // Retrieve the messageId from the postMessage call
- const req = mockPostMessage.mock.calls[0]?.[0] as { messageId: string; type: string };
- expect(req.type).toBe('INIT');
+ it('returns {ok:false} without enqueuing when the pool is unavailable', async () => {
+ mockEnsureDuckDbPool.mockResolvedValue(null);
- onMessage({ data: { messageId: req.messageId, ok: true } } as MessageEvent);
- const res = await promise;
+ const res = await duckdbClient.init();
- expect(res.ok).toBe(true);
- expect(res.messageId).toBe(req.messageId);
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'WorkerBus v2 unavailable' }));
+ expect(mockEnqueue).not.toHaveBeenCalled();
});
});
describe('duckdbClient.query', () => {
- it('posts a QUERY message and returns rows', async () => {
- const promise = duckdbClient.query('SELECT 1 AS n');
+ it('enqueues db.duckdb.query and wraps the raw row array as {ok:true, rows}', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.resolve([{ n: 1 }])));
- const listenerCall = mockAddEventListener.mock.calls.find(([ev]) => ev === 'message');
- const onMessage = listenerCall![1] as (e: MessageEvent) => void;
- const req = mockPostMessage.mock.calls[0]?.[0] as { messageId: string; type: string };
+ const res = await duckdbClient.query('SELECT 1 AS n');
- expect(req.type).toBe('QUERY');
+ const [taskType, payload] = mockEnqueue.mock.calls[0] as [string, unknown];
+ expect(taskType).toBe('db.duckdb.query');
+ expect(payload).toEqual({ sql: 'SELECT 1 AS n', params: undefined });
+ expect(res).toEqual(expect.objectContaining({ ok: true, rows: [{ n: 1 }] }));
+ });
- onMessage({
- data: { messageId: req.messageId, ok: true, rows: [{ n: 1 }] },
- } as MessageEvent);
+ it('returns {ok:false, error} (not a thrown exception) when the task rejects', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.reject(new Error('bad sql'))));
- const res = await promise;
- expect(res.ok).toBe(true);
- expect(res.rows).toEqual([{ n: 1 }]);
+ const res = await duckdbClient.query('garbage');
+
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'bad sql' }));
+ });
+
+ it('maps a CIRCUIT_OPEN error code to a friendly message', async () => {
+ const err = Object.assign(new Error('circuit open'), { code: 'CIRCUIT_OPEN' });
+ mockEnqueue.mockReturnValue(makeHandle(Promise.reject(err)));
+
+ const res = await duckdbClient.query('SELECT 1');
+
+ expect(res).toEqual(
+ expect.objectContaining({
+ ok: false,
+ error: 'DuckDB temporarily unavailable (circuit open)',
+ }),
+ );
+ });
+
+ it('stringifies a rejection that is not an Error instance', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.reject('plain string rejection')));
+
+ const res = await duckdbClient.query('SELECT 1');
+
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'plain string rejection' }));
+ });
+
+ it('disables WorkerBus-level retries (retryPolicy maxRetries:0) to avoid stacking under withDuckDbRetry', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.resolve([])));
+
+ await duckdbClient.query('SELECT 1');
+
+ expect(mockEnqueue).toHaveBeenCalledWith(
+ 'db.duckdb.query',
+ expect.anything(),
+ expect.objectContaining({ retryPolicy: { maxRetries: 0 } }),
+ );
+ });
+
+ describe('auto-reinit on "DuckDB not initialized"', () => {
+ // QNBS-v3: [Regression coverage for the idle-respawn/crash-restart gap — a fresh duckdb pool
+ // worker has no `connection` until INIT runs again on it.]
+ it('re-inits once and retries the original query, then succeeds', async () => {
+ let queryCallCount = 0;
+ mockEnqueue.mockImplementation((taskType: string) => {
+ if (taskType === 'db.duckdb.query') {
+ queryCallCount++;
+ if (queryCallCount === 1) {
+ return makeHandle(Promise.reject(new Error('DuckDB not initialized')));
+ }
+ return makeHandle(Promise.resolve([{ n: 1 }]));
+ }
+ if (taskType === 'db.duckdb.init') {
+ return makeHandle(Promise.resolve({ ok: true }));
+ }
+ throw new Error(`unexpected taskType ${taskType}`);
+ });
+
+ const res = await duckdbClient.query('SELECT 1 AS n');
+
+ expect(res).toEqual(expect.objectContaining({ ok: true, rows: [{ n: 1 }] }));
+ expect(queryCallCount).toBe(2);
+ const initCalls = mockEnqueue.mock.calls.filter(([t]) => t === 'db.duckdb.init');
+ expect(initCalls).toHaveLength(1);
+ });
+
+ it('attempts exactly one reinit — does not loop when the retry also fails', async () => {
+ mockEnqueue.mockImplementation((taskType: string) => {
+ if (taskType === 'db.duckdb.query') {
+ return makeHandle(Promise.reject(new Error('DuckDB not initialized')));
+ }
+ if (taskType === 'db.duckdb.init') {
+ return makeHandle(Promise.resolve({ ok: true }));
+ }
+ throw new Error(`unexpected taskType ${taskType}`);
+ });
+
+ const res = await duckdbClient.query('SELECT 1');
+
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'DuckDB not initialized' }));
+ const initCalls = mockEnqueue.mock.calls.filter(([t]) => t === 'db.duckdb.init');
+ expect(initCalls).toHaveLength(1);
+ });
+
+ it('does not retry the original query when the reinit INIT task itself rejects', async () => {
+ // QNBS-v3: [reinit.ok is only false when send('INIT', ...) itself hits the catch branch —
+ // i.e. the INIT task's handle.result *rejects*. Distinct from the 'INIT enqueue
+ // resolves but with an app-level failure shape' case, which send() doesn't model:
+ // any resolved handle.result is always treated as {ok:true}.]
+ mockEnqueue.mockImplementation((taskType: string) => {
+ if (taskType === 'db.duckdb.query') {
+ return makeHandle(Promise.reject(new Error('DuckDB not initialized')));
+ }
+ if (taskType === 'db.duckdb.init') {
+ return makeHandle(Promise.reject(new Error('worker spawn failed')));
+ }
+ throw new Error(`unexpected taskType ${taskType}`);
+ });
+
+ const res = await duckdbClient.query('SELECT 1');
+
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'DuckDB not initialized' }));
+ const queryCalls = mockEnqueue.mock.calls.filter(([t]) => t === 'db.duckdb.query');
+ expect(queryCalls).toHaveLength(1);
+ });
+
+ it('does not reinit-retry unrelated error messages', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.reject(new Error('syntax error'))));
+
+ const res = await duckdbClient.query('garbage');
+
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'syntax error' }));
+ const initCalls = mockEnqueue.mock.calls.filter(([t]) => t === 'db.duckdb.init');
+ expect(initCalls).toHaveLength(0);
+ });
+
+ it('does not recursively reinit when INIT itself fails with "DuckDB not initialized"', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.reject(new Error('DuckDB not initialized'))));
+
+ await duckdbClient.init();
+
+ const initCalls = mockEnqueue.mock.calls.filter(([t]) => t === 'db.duckdb.init');
+ expect(initCalls).toHaveLength(1);
+ });
});
});
describe('duckdbClient.exec', () => {
- it('posts an EXEC message', async () => {
- const promise = duckdbClient.exec('CREATE TABLE t (id INTEGER)');
+ it('enqueues db.duckdb.exec and returns {ok:true}', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.resolve({ ok: true })));
- const listenerCall = mockAddEventListener.mock.calls.find(([ev]) => ev === 'message');
- const onMessage = listenerCall![1] as (e: MessageEvent) => void;
- const req = mockPostMessage.mock.calls[0]?.[0] as { messageId: string; type: string };
+ const res = await duckdbClient.exec('CREATE TABLE t (id INTEGER)');
- expect(req.type).toBe('EXEC');
-
- onMessage({ data: { messageId: req.messageId, ok: true } } as MessageEvent);
- const res = await promise;
+ expect((mockEnqueue.mock.calls[0] as [string])[0]).toBe('db.duckdb.exec');
expect(res.ok).toBe(true);
});
});
describe('duckdbClient.shutdown', () => {
- it('posts a SHUTDOWN message', async () => {
- const promise = duckdbClient.shutdown();
-
- const listenerCall = mockAddEventListener.mock.calls.find(([ev]) => ev === 'message');
- const onMessage = listenerCall![1] as (e: MessageEvent) => void;
- const req = mockPostMessage.mock.calls[0]?.[0] as { messageId: string; type: string };
+ it('enqueues db.duckdb.shutdown', async () => {
+ mockEnqueue.mockReturnValue(makeHandle(Promise.resolve({ ok: true })));
- expect(req.type).toBe('SHUTDOWN');
+ const res = await duckdbClient.shutdown();
- onMessage({ data: { messageId: req.messageId, ok: true } } as MessageEvent);
- const res = await promise;
+ expect((mockEnqueue.mock.calls[0] as [string])[0]).toBe('db.duckdb.shutdown');
expect(res.ok).toBe(true);
});
});
describe('duckdbClient.terminate', () => {
- it('calls worker.terminate()', async () => {
- // Ensure the worker is created by making a call first
- const promise = duckdbClient.query('SELECT 1');
- const listenerCall = mockAddEventListener.mock.calls.find(([ev]) => ev === 'message');
- const onMessage = listenerCall![1] as (e: MessageEvent) => void;
- const req = mockPostMessage.mock.calls[0]?.[0] as { messageId: string };
- onMessage({ data: { messageId: req.messageId, ok: true, rows: [] } } as MessageEvent);
- await promise;
-
+ it('terminates only the "duckdb" pool, not the whole shared bus', () => {
duckdbClient.terminate();
- expect(mockTerminate).toHaveBeenCalled();
+
+ expect(mockTerminatePool).toHaveBeenCalledWith('duckdb');
+ });
+
+ it('is a no-op when the bus was never initialized', () => {
+ mockGetWorkerBus.mockReturnValue(null);
+
+ expect(() => duckdbClient.terminate()).not.toThrow();
+ expect(mockTerminatePool).not.toHaveBeenCalled();
});
});
describe('duckdbClient abort', () => {
- it('posts WORKER_CANCEL when AbortSignal fires', async () => {
+ it('cancels the task handle when the AbortSignal fires mid-flight', async () => {
+ const cancel = vi.fn();
+ const { promise: resultPromise, reject } = deferred();
+ mockEnqueue.mockReturnValue(makeHandle(resultPromise, cancel));
+
const controller = new AbortController();
- void duckdbClient.query('SELECT 1', undefined, controller.signal);
+ const resPromise = duckdbClient.query('SELECT 1', undefined, controller.signal);
+ controller.abort();
+
+ // QNBS-v3: cancel() fires after the async ensureDuckDbPool()/enqueue() chain settles, not
+ // synchronously — unlike v1's Promise-executor version. waitFor tolerates that gap
+ // without hardcoding a specific number of microtask ticks.
+ await vi.waitFor(() => expect(cancel).toHaveBeenCalledWith('Aborted'));
- const req = mockPostMessage.mock.calls[0]?.[0] as { messageId: string };
+ reject(Object.assign(new Error('Task was cancelled'), { code: 'CANCELLED' }));
+ const res = await resPromise;
+
+ expect(res).toEqual(expect.objectContaining({ ok: false, error: 'Aborted' }));
+ });
+
+ it('registers the abort listener synchronously, so an already-fired-before-enqueue abort is not lost', async () => {
+ const cancel = vi.fn();
+ const { promise: resultPromise } = deferred();
+ mockEnqueue.mockReturnValue(makeHandle(resultPromise, cancel));
+
+ const controller = new AbortController();
controller.abort();
+ void duckdbClient.query('SELECT 1', undefined, controller.signal);
- // The cancel message should be posted after abort
- const cancelMsg = mockPostMessage.mock.calls.find(
- ([m]) => (m as { type: string }).type === 'WORKER_CANCEL',
+ await vi.waitFor(() => expect(cancel).toHaveBeenCalledWith('Aborted'));
+ });
+
+ it('does not miss an abort fired in the same tick as the call, before ensureDuckDbPool resolves', async () => {
+ const cancel = vi.fn();
+ const { promise: resultPromise } = deferred();
+ mockEnqueue.mockReturnValue(makeHandle(resultPromise, cancel));
+ // QNBS-v3: regression guard for the real timing bug this PR fixed — the abort listener must
+ // be attached before the `await ensureDuckDbPool()` inside send() yields, so an
+ // abort fired synchronously right after the call (same tick, no microtask flush
+ // in between) still gets caught once the handle exists.
+ let resolveEnsurePool!: (bus: ReturnType) => void;
+ mockEnsureDuckDbPool.mockReturnValue(
+ new Promise((res) => {
+ resolveEnsurePool = res;
+ }),
);
- expect(cancelMsg).toBeDefined();
- expect((cancelMsg![0] as { messageId: string }).messageId).toBe(req.messageId);
+
+ const controller = new AbortController();
+ void duckdbClient.query('SELECT 1', undefined, controller.signal);
+ controller.abort();
+ resolveEnsurePool(makeBus());
+
+ await vi.waitFor(() => expect(cancel).toHaveBeenCalledWith('Aborted'));
+ });
+
+ it('detaches its abort listener once the call settles, so a reused signal does not accumulate listeners', async () => {
+ // QNBS-v3: [regression guard — {once:true} only self-removes when the signal actually
+ // fires; a caller reusing one AbortSignal across many non-aborted queries must not
+ // leak one listener per call for the signal's whole lifetime.]
+ mockEnqueue.mockReturnValue(makeHandle(Promise.resolve([{ n: 1 }])));
+ const controller = new AbortController();
+ const removeSpy = vi.spyOn(controller.signal, 'removeEventListener');
+
+ await duckdbClient.query('SELECT 1', undefined, controller.signal);
+
+ expect(removeSpy).toHaveBeenCalledWith('abort', expect.any(Function));
+ });
+});
+
+describe('duckdbClient OPFS fallback', () => {
+ // QNBS-v3: afterEach (not end-of-test) so the handler resets even if an assertion throws first
+ afterEach(() => {
+ duckdbClient.setOpfsFallbackHandler(null);
+ });
+
+ it('forwards the opfs-fallback progress stage to setOpfsFallbackHandler', async () => {
+ let capturedOnProgress: ((p: { stage: string; message?: string }) => void) | undefined;
+ mockEnqueue.mockImplementation((_type, _payload, opts) => {
+ capturedOnProgress = opts.onProgress;
+ return makeHandle(Promise.resolve({ ok: true }));
+ });
+ const onFallback = vi.fn();
+ duckdbClient.setOpfsFallbackHandler(onFallback);
+
+ await duckdbClient.init();
+ capturedOnProgress?.({ stage: 'opfs-fallback', message: 'OPFS unavailable in private mode' });
+
+ expect(onFallback).toHaveBeenCalledWith('OPFS unavailable in private mode');
+ });
+
+ it('defaults to a generic message when the opfs-fallback progress event omits one', async () => {
+ let capturedOnProgress: ((p: { stage: string; message?: string }) => void) | undefined;
+ mockEnqueue.mockImplementation((_type, _payload, opts) => {
+ capturedOnProgress = opts.onProgress;
+ return makeHandle(Promise.resolve({ ok: true }));
+ });
+ const onFallback = vi.fn();
+ duckdbClient.setOpfsFallbackHandler(onFallback);
+
+ await duckdbClient.init();
+ capturedOnProgress?.({ stage: 'opfs-fallback' });
+
+ expect(onFallback).toHaveBeenCalledWith('OPFS unavailable');
+ });
+
+ it('ignores non-opfs-fallback progress stages', async () => {
+ let capturedOnProgress: ((p: { stage: string; message?: string }) => void) | undefined;
+ mockEnqueue.mockImplementation((_type, _payload, opts) => {
+ capturedOnProgress = opts.onProgress;
+ return makeHandle(Promise.resolve({ ok: true }));
+ });
+ const onFallback = vi.fn();
+ duckdbClient.setOpfsFallbackHandler(onFallback);
+
+ await duckdbClient.init();
+ capturedOnProgress?.({ stage: 'loading', message: 'irrelevant' });
+
+ expect(onFallback).not.toHaveBeenCalled();
});
});
diff --git a/tests/unit/duckdbWorkerHandler.test.ts b/tests/unit/duckdbWorkerHandler.test.ts
index 5ebe5fa3..18f8db18 100644
--- a/tests/unit/duckdbWorkerHandler.test.ts
+++ b/tests/unit/duckdbWorkerHandler.test.ts
@@ -145,6 +145,24 @@ describe('duckdb.worker handlers', () => {
expect(mockConnect).toHaveBeenCalledTimes(2);
});
+ it('still falls back and emits the original OPFS error when close() itself rejects', async () => {
+ const opfsClose = vi.fn().mockRejectedValue(new Error('close also failed'));
+ const opfsQuery = vi.fn().mockRejectedValue(new Error('ATTACH failed'));
+ mockConnect
+ .mockResolvedValueOnce({ query: opfsQuery, close: opfsClose })
+ .mockResolvedValueOnce(mockConnection());
+ mockRegisterFileHandle.mockResolvedValue(undefined);
+ stubOpfsDirectory(vi.fn().mockResolvedValue({}));
+ const emitProgress = vi.fn();
+
+ await expect(initDuckDb(emitProgress)).resolves.toBeUndefined();
+
+ expect(opfsClose).toHaveBeenCalled();
+ // The original ATTACH error reaches emitProgress, not the close() failure.
+ expect(emitProgress).toHaveBeenCalledWith('opfs-fallback', 1, 'ATTACH failed');
+ expect(mockConnect).toHaveBeenCalledTimes(2);
+ });
+
it('does not emit opfs-fallback when OPFS is simply unsupported (no navigator.storage)', async () => {
mockConnect.mockResolvedValue(mockConnection());
stubNoOpfs();
@@ -198,6 +216,67 @@ describe('duckdb.worker handlers', () => {
'bad sql',
);
});
+
+ it('binds params via a prepared statement instead of dropping them', async () => {
+ const row = { toJSON: () => ({ id: 1 }) };
+ const stmtQuery = vi.fn().mockResolvedValue({ toArray: () => [row] });
+ const stmtClose = vi.fn().mockResolvedValue(undefined);
+ const prepare = vi.fn().mockResolvedValue({ query: stmtQuery, close: stmtClose });
+ const rawQuery = vi.fn();
+ mockConnect.mockResolvedValue({
+ query: rawQuery,
+ close: vi.fn().mockResolvedValue(undefined),
+ prepare,
+ });
+ stubNoOpfs();
+ await initDuckDb();
+
+ const result = await handleQuery(
+ makeCtx({ payload: { sql: 'select * from t where id = ?', params: [1] } }),
+ );
+
+ expect(prepare).toHaveBeenCalledWith('select * from t where id = ?');
+ expect(stmtQuery).toHaveBeenCalledWith(1);
+ expect(stmtClose).toHaveBeenCalled();
+ expect(rawQuery).not.toHaveBeenCalled();
+ expect(result).toEqual([{ id: 1 }]);
+ });
+
+ it('closes the prepared statement even when the bound query rejects', async () => {
+ const stmtQuery = vi.fn().mockRejectedValue(new Error('bind failed'));
+ const stmtClose = vi.fn().mockResolvedValue(undefined);
+ const prepare = vi.fn().mockResolvedValue({ query: stmtQuery, close: stmtClose });
+ mockConnect.mockResolvedValue({
+ query: vi.fn(),
+ close: vi.fn().mockResolvedValue(undefined),
+ prepare,
+ });
+ stubNoOpfs();
+ await initDuckDb();
+
+ await expect(
+ handleQuery(makeCtx({ payload: { sql: 'select ?', params: ['x'] } })),
+ ).rejects.toThrow('bind failed');
+ expect(stmtClose).toHaveBeenCalled();
+ });
+
+ it('does not use a prepared statement when params is an empty array', async () => {
+ const row = { toJSON: () => ({ id: 1 }) };
+ const rawQuery = vi.fn().mockResolvedValue({ toArray: () => [row] });
+ const prepare = vi.fn();
+ mockConnect.mockResolvedValue({
+ query: rawQuery,
+ close: vi.fn().mockResolvedValue(undefined),
+ prepare,
+ });
+ stubNoOpfs();
+ await initDuckDb();
+
+ await handleQuery(makeCtx({ payload: { sql: 'select 1', params: [] } }));
+
+ expect(prepare).not.toHaveBeenCalled();
+ expect(rawQuery).toHaveBeenCalledWith('select 1');
+ });
});
describe('handleExec', () => {
@@ -218,6 +297,35 @@ describe('duckdb.worker handlers', () => {
expect(result).toEqual({ ok: true });
expect(connectionQuery).toHaveBeenCalledWith('create table t (id int)');
});
+
+ it('binds params via a prepared statement instead of dropping them', async () => {
+ const stmtQuery = vi.fn().mockResolvedValue(undefined);
+ const stmtClose = vi.fn().mockResolvedValue(undefined);
+ const prepare = vi.fn().mockResolvedValue({ query: stmtQuery, close: stmtClose });
+ const rawQuery = vi.fn();
+ mockConnect.mockResolvedValue({
+ query: rawQuery,
+ close: vi.fn().mockResolvedValue(undefined),
+ prepare,
+ });
+ stubNoOpfs();
+ await initDuckDb();
+
+ const result = await handleExec(
+ makeCtx({
+ payload: {
+ sql: 'insert into ai_telemetry values (?, ?, ?, ?, ?)',
+ params: ['task', 'backend', 'model', 12, true],
+ },
+ }),
+ );
+
+ expect(prepare).toHaveBeenCalledWith('insert into ai_telemetry values (?, ?, ?, ?, ?)');
+ expect(stmtQuery).toHaveBeenCalledWith('task', 'backend', 'model', 12, true);
+ expect(stmtClose).toHaveBeenCalled();
+ expect(rawQuery).not.toHaveBeenCalled();
+ expect(result).toEqual({ ok: true });
+ });
});
describe('handleShutdown', () => {
diff --git a/tests/unit/inferenceWorkerHandlerV2.test.ts b/tests/unit/inferenceWorkerHandlerV2.test.ts
index 07f16091..575c34ac 100644
--- a/tests/unit/inferenceWorkerHandlerV2.test.ts
+++ b/tests/unit/inferenceWorkerHandlerV2.test.ts
@@ -26,7 +26,8 @@ function makeCtx(overrides: Partial = {}): WorkerHandlerCo
describe('inference.worker (v2) handleInference', () => {
beforeEach(() => {
- vi.clearAllMocks();
+ // QNBS-v3: resetAllMocks (not clearAllMocks) clears mock implementations too, preventing state leakage between tests
+ vi.resetAllMocks();
});
afterEach(() => {
diff --git a/tests/unit/workerBusManager.test.ts b/tests/unit/workerBusManager.test.ts
index 20cf41a7..151f3717 100644
--- a/tests/unit/workerBusManager.test.ts
+++ b/tests/unit/workerBusManager.test.ts
@@ -4,10 +4,14 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
// QNBS-v3: vi.hoisted ensures these references are available inside the vi.mock factory
// which is hoisted before all imports.
-const { mockShutdown, mockRegisterPool, mockInstall, MockWorkBus, MockRegistry } = vi.hoisted(
- () => {
+const { mockShutdown, mockRegisterPool, mockHasPool, mockInstall, MockWorkBus, MockRegistry } =
+ vi.hoisted(() => {
const mockShutdown = vi.fn().mockResolvedValue(undefined);
const mockRegisterPool = vi.fn();
+ // QNBS-v3: [Defaults to true (pool present) so existing tests that don't care about
+ // re-registration behavior are unaffected; the dedicated ensureDuckDbPool tests
+ // override this per-case.]
+ const mockHasPool = vi.fn(() => true);
const mockInstall = vi.fn();
// QNBS-v3: Regular function (not arrow) so new MockWorkBus() works as a constructor.
@@ -15,6 +19,7 @@ const { mockShutdown, mockRegisterPool, mockInstall, MockWorkBus, MockRegistry }
const MockWorkBus = vi.fn(function (this: Record) {
this['shutdown'] = mockShutdown;
this['registerPool'] = mockRegisterPool;
+ this['hasPool'] = mockHasPool;
this['enqueue'] = vi.fn();
this['cancel'] = vi.fn(() => true);
this['getTelemetry'] = vi.fn(() => ({
@@ -39,9 +44,8 @@ const { mockShutdown, mockRegisterPool, mockInstall, MockWorkBus, MockRegistry }
this['install'] = mockInstall;
});
- return { mockShutdown, mockRegisterPool, mockInstall, MockWorkBus, MockRegistry };
- },
-);
+ return { mockShutdown, mockRegisterPool, mockHasPool, mockInstall, MockWorkBus, MockRegistry };
+ });
vi.mock('@domain/worker-bus', () => ({
WorkerBus: MockWorkBus,
@@ -69,6 +73,8 @@ describe('workerBusManager', () => {
MockRegistry.mockClear();
mockShutdown.mockClear();
mockRegisterPool.mockClear();
+ mockHasPool.mockClear();
+ mockHasPool.mockReturnValue(true);
mockInstall.mockClear();
});
@@ -143,4 +149,45 @@ describe('workerBusManager', () => {
await initWorkerBus();
expect(getLegacyAdapter()).not.toBeNull();
});
+
+ describe('ensureDuckDbPool', () => {
+ it('initializes the bus when not yet running', async () => {
+ const { ensureDuckDbPool, isWorkerBusReady } = await import(
+ '../../services/workerBusManager'
+ );
+ const bus = await ensureDuckDbPool();
+ expect(bus).not.toBeNull();
+ expect(isWorkerBusReady()).toBe(true);
+ });
+
+ it('does not re-register the pool when it is already present', async () => {
+ const { initWorkerBus, ensureDuckDbPool } = await import('../../services/workerBusManager');
+ await initWorkerBus();
+ mockRegisterPool.mockClear();
+ mockHasPool.mockReturnValue(true);
+
+ await ensureDuckDbPool();
+
+ expect(mockHasPool).toHaveBeenCalledWith('duckdb');
+ expect(mockRegisterPool).not.toHaveBeenCalled();
+ });
+
+ it('re-registers the duckdb pool when the bus is alive but the pool was removed', async () => {
+ // QNBS-v3: [Simulates the gap terminatePool('duckdb') can leave — the bus stays non-null
+ // but the pool is gone; ensureDuckDbPool must not just trust `_bus !== null`.]
+ const { initWorkerBus, ensureDuckDbPool } = await import('../../services/workerBusManager');
+ await initWorkerBus();
+ mockRegisterPool.mockClear();
+ mockHasPool.mockReturnValue(false);
+
+ const bus = await ensureDuckDbPool();
+
+ expect(bus).not.toBeNull();
+ expect(mockRegisterPool).toHaveBeenCalledWith(
+ 'duckdb',
+ expect.arrayContaining(['db.duckdb']),
+ expect.objectContaining({ workerScript: expect.stringContaining('duckdb.worker') }),
+ );
+ });
+ });
});
diff --git a/workers/duckdbWorker.ts b/workers/duckdbWorker.ts
deleted file mode 100644
index 30899b36..00000000
--- a/workers/duckdbWorker.ts
+++ /dev/null
@@ -1,219 +0,0 @@
-///
-// QNBS-v3: Off-main-thread DuckDB-WASM worker. OPFS persistence with in-memory fallback.
-// Uses duckdb-eh bundle (no SharedArrayBuffer required — no COOP/COEP needed).
-// Message protocol mirrors inference.worker.ts for consistency.
-
-export type DuckDbRequestType = 'INIT' | 'QUERY' | 'EXEC' | 'MIGRATE' | 'SHUTDOWN';
-
-// QNBS-v3: Out-of-band event emitted when OPFS attach fails — no messageId, fires independently.
-export interface DuckDbWorkerEvent {
- type: 'OPFS_FALLBACK';
- reason: string;
-}
-
-export interface DuckDbRequest {
- messageId: string;
- type: DuckDbRequestType;
- sql?: string | undefined;
- params?: readonly unknown[] | undefined;
-}
-
-export interface DuckDbResponse {
- messageId: string;
- ok: boolean;
- rows?: Record[];
- error?: string;
- latencyMs?: number;
-}
-
-// QNBS-v3: Lazy module reference — DuckDB bundle is ~2 MB; only load when INIT is received.
-let duckdbModule: typeof import('@duckdb/duckdb-wasm') | null = null;
-let db: import('@duckdb/duckdb-wasm').AsyncDuckDB | null = null;
-let connection: import('@duckdb/duckdb-wasm').AsyncDuckDBConnection | null = null;
-
-async function getDuckDb() {
- if (!duckdbModule) {
- duckdbModule = await import('@duckdb/duckdb-wasm');
- }
- return duckdbModule;
-}
-
-// QNBS-v3: Security guard — only process messages from the same origin (mirrors inference.worker.ts).
-function isTrustedWorkerMessage(event: MessageEvent): boolean {
- return event.origin === '' || event.origin === globalThis.location?.origin;
-}
-
-async function isOPFSSupported(): Promise {
- try {
- const root = await navigator.storage?.getDirectory?.();
- if (!root) return false;
- await root.getFileHandle('__duckdb_opfs_test__', { create: true });
- return true;
- } catch {
- return false;
- }
-}
-
-// QNBS-v3 (F-09): self-hosted, same-origin DuckDB-WASM assets (scripts/copy-duckdb-assets.mjs) replacing an unversioned CDN that was already dead under worker-src 'self' blob:' CSP; absolute path so resolution is unambiguous from either the main thread or this nested worker context.
-const DUCKDB_ASSET_BASE = `${import.meta.env.BASE_URL}duckdb/`;
-const SELF_HOSTED_BUNDLES = {
- mvp: {
- mainModule: `${DUCKDB_ASSET_BASE}duckdb-mvp.wasm`,
- mainWorker: `${DUCKDB_ASSET_BASE}duckdb-browser-mvp.worker.js`,
- },
- eh: {
- mainModule: `${DUCKDB_ASSET_BASE}duckdb-eh.wasm`,
- mainWorker: `${DUCKDB_ASSET_BASE}duckdb-browser-eh.worker.js`,
- },
-};
-
-async function initDuckDb(): Promise {
- const { AsyncDuckDB, selectBundle, ConsoleLogger } = await getDuckDb();
-
- const bundle = await selectBundle(SELF_HOSTED_BUNDLES);
- const logger = new ConsoleLogger();
- // QNBS-v3: bundle.mainWorker can be null for non-browser bundles; guard before constructing Worker.
- if (!bundle.mainWorker) throw new Error('DuckDB bundle has no worker URL');
- const worker = new Worker(bundle.mainWorker);
- const newDb = new AsyncDuckDB(logger, worker);
- await newDb.instantiate(bundle.mainModule);
-
- // QNBS-v3: OPFS for persistence when available; fall back to in-memory (private browsing, old Safari).
- const useOpfs = await isOPFSSupported();
- if (useOpfs) {
- try {
- const { DuckDBDataProtocol } = await getDuckDb();
- const opfsRoot = await navigator.storage.getDirectory();
- const fileHandle = await opfsRoot.getFileHandle('worldscript_analytics.duckdb', {
- create: true,
- });
- await newDb.registerFileHandle(
- 'worldscript_analytics.duckdb',
- fileHandle,
- DuckDBDataProtocol.BROWSER_FSACCESS,
- true,
- );
- connection = await newDb.connect();
- await connection.query("ATTACH 'worldscript_analytics.duckdb' AS analytics (TYPE duckdb)");
- } catch (opfsErr) {
- // QNBS-v3: Notify main thread so the hook can set persistence-mode badge in Redux.
- self.postMessage({
- type: 'OPFS_FALLBACK',
- reason: opfsErr instanceof Error ? opfsErr.message : String(opfsErr),
- } satisfies DuckDbWorkerEvent);
- connection = await newDb.connect();
- }
- } else {
- connection = await newDb.connect();
- }
-
- db = newDb;
-}
-
-async function handleQuery(req: DuckDbRequest): Promise {
- const start = Date.now();
- if (!connection) {
- return { messageId: req.messageId, ok: false, error: 'DuckDB not initialized', latencyMs: 0 };
- }
- try {
- const result = await connection.query(req.sql ?? '');
- const rows = result.toArray().map((row: { toJSON(): Record }) => row.toJSON());
- return { messageId: req.messageId, ok: true, rows, latencyMs: Date.now() - start };
- } catch (err) {
- return {
- messageId: req.messageId,
- ok: false,
- error: err instanceof Error ? err.message : String(err),
- latencyMs: Date.now() - start,
- };
- }
-}
-
-async function handleExec(req: DuckDbRequest): Promise {
- const start = Date.now();
- if (!connection) {
- return { messageId: req.messageId, ok: false, error: 'DuckDB not initialized', latencyMs: 0 };
- }
- try {
- await connection.query(req.sql ?? '');
- return { messageId: req.messageId, ok: true, latencyMs: Date.now() - start };
- } catch (err) {
- return {
- messageId: req.messageId,
- ok: false,
- error: err instanceof Error ? err.message : String(err),
- latencyMs: Date.now() - start,
- };
- }
-}
-
-async function handleShutdown(req: DuckDbRequest): Promise {
- try {
- await connection?.close();
- await db?.terminate();
- connection = null;
- db = null;
- return { messageId: req.messageId, ok: true };
- } catch (err) {
- return {
- messageId: req.messageId,
- ok: false,
- error: err instanceof Error ? err.message : String(err),
- };
- }
-}
-
-// QNBS-v3: In-flight AbortController map (mirrors inference.worker.ts; DuckDB cancellation via close/reopen).
-const abortMap = new Map();
-
-self.addEventListener('message', (event: MessageEvent) => {
- if (!isTrustedWorkerMessage(event)) return;
-
- // QNBS-v3: Check raw type as string before narrowing to DuckDbRequest so WORKER_CANCEL is reachable.
- const raw = event.data as { type?: string; messageId?: string };
-
- if (raw.type === 'WORKER_CANCEL' && raw.messageId) {
- abortMap.get(raw.messageId)?.abort();
- abortMap.delete(raw.messageId);
- return;
- }
-
- const req = event.data as DuckDbRequest;
-
- const controller = new AbortController();
- abortMap.set(req.messageId, controller);
-
- let task: Promise;
-
- switch (req.type) {
- case 'INIT':
- task = initDuckDb()
- .then(() => ({ messageId: req.messageId, ok: true }))
- .catch((err: unknown) => ({
- messageId: req.messageId,
- ok: false,
- error: err instanceof Error ? err.message : String(err),
- }));
- break;
- case 'QUERY':
- task = handleQuery(req);
- break;
- case 'EXEC':
- task = handleExec(req);
- break;
- case 'SHUTDOWN':
- task = handleShutdown(req);
- break;
- default:
- task = Promise.resolve({
- messageId: req.messageId,
- ok: false,
- error: `Unknown request type: ${String(req.type)}`,
- });
- }
-
- void task.then((response) => {
- abortMap.delete(req.messageId);
- self.postMessage(response);
- });
-});
diff --git a/workers/v2/duckdb.worker.ts b/workers/v2/duckdb.worker.ts
index 891886ea..28789984 100644
--- a/workers/v2/duckdb.worker.ts
+++ b/workers/v2/duckdb.worker.ts
@@ -1,15 +1,21 @@
///
-// QNBS-v3: WorkerBus v2 DuckDB worker. Wraps legacy duckdbWorker.ts logic
-// in the typed bootstrap protocol.
+// QNBS-v3: [Sole DuckDB worker generation post-consolidation (docs/adr/0014); formerly wrapped the now-deleted legacy workers/duckdbWorker.ts.]
+import type { AsyncDuckDB, AsyncDuckDBConnection } from '@duckdb/duckdb-wasm';
import {
registerTaskHandler,
type WorkerHandlerContext,
} from '../../packages/worker-bus/src/workerBootstrap';
+import { createLogger } from '../../services/logger';
+
+// QNBS-v3: [services/logger.ts is worker-safe — its window.localStorage/Tauri touches are all
+// guarded (typeof window !== 'undefined' / dynamic import) — so this matches
+// workerBusManager.ts's log.warn/log.error convention instead of raw console.warn.]
+const log = createLogger('duckdb.worker');
let duckdbModule: typeof import('@duckdb/duckdb-wasm') | null = null;
-let db: import('@duckdb/duckdb-wasm').AsyncDuckDB | null = null;
-let connection: import('@duckdb/duckdb-wasm').AsyncDuckDBConnection | null = null;
+let db: AsyncDuckDB | null = null;
+let connection: AsyncDuckDBConnection | null = null;
async function getDuckDb() {
if (!duckdbModule) {
@@ -29,7 +35,7 @@ async function isOPFSSupported(): Promise {
}
}
-// QNBS-v3 (F-09): self-hosted, same-origin DuckDB-WASM assets — see workers/duckdbWorker.ts for the full rationale (unpinned CDN URL, supply-chain trust, already CSP-dead code).
+// QNBS-v3 (F-09): [Self-hosted, same-origin DuckDB-WASM assets — replaces an unpinned CDN URL that was already CSP-dead (supply-chain trust); see docs/adr/0013-csp-wasm-and-blob-frames.md.]
const DUCKDB_ASSET_BASE = `${import.meta.env.BASE_URL}duckdb/`;
const SELF_HOSTED_BUNDLES = {
mvp: {
@@ -58,7 +64,7 @@ export async function initDuckDb(
const useOpfs = await isOPFSSupported();
if (useOpfs) {
// QNBS-v3: [Tracked separately from `connection` so a failed ATTACH can close this partial connection instead of leaking it.]
- let opfsConnection: import('@duckdb/duckdb-wasm').AsyncDuckDBConnection | null = null;
+ let opfsConnection: AsyncDuckDBConnection | null = null;
try {
const { DuckDBDataProtocol } = await getDuckDb();
const opfsRoot = await navigator.storage.getDirectory();
@@ -77,8 +83,13 @@ export async function initDuckDb(
);
connection = opfsConnection;
} catch (opfsErr) {
+ // QNBS-v3: [Cleanup is best-effort — a rejecting close() must not block the fallback below or replace the original OPFS error.]
+ try {
+ await opfsConnection?.close();
+ } catch (closeErr) {
+ log.warn('Failed to close partial OPFS connection', closeErr);
+ }
// QNBS-v3: [No bare postMessage from inside a task handler — OPFS-unavailable is surfaced via the progress channel duckdbClient's INIT adapter listens on instead.]
- await opfsConnection?.close();
emitProgress?.(
'opfs-fallback',
1,
@@ -93,19 +104,50 @@ export async function initDuckDb(
db = newDb;
}
+interface DuckDbQueryResult {
+ toArray(): Array<{ toJSON(): Record }>;
+}
+
+interface PreparableConnection {
+ prepare(sql: string): Promise<{
+ query(...params: unknown[]): Promise;
+ close(): Promise;
+ }>;
+}
+
+// QNBS-v3: [tsgo doesn't resolve AsyncDuckDBConnection.prepare() through this package's nested
+// `export *` chain (present and correct in @duckdb/duckdb-wasm's own .d.ts, verified
+// directly) — a narrow structural type sidesteps the gap, the same class of tsgo/external-
+// package resolution issue CLAUDE.md documents for the transformers.js path alias.]
+function asPreparable(conn: AsyncDuckDBConnection): PreparableConnection {
+ return conn as unknown as PreparableConnection;
+}
+
+// QNBS-v3: [Bind via a prepared statement when params are present — a raw connection.query(sql) silently drops them, leaving an unbound query and a SQL-injection surface.]
+async function runSql(sql: string, params?: readonly unknown[]): Promise {
+ if (!connection) throw new Error('DuckDB not initialized');
+ if (params && params.length > 0) {
+ const stmt = await asPreparable(connection).prepare(sql);
+ try {
+ return await stmt.query(...params);
+ } finally {
+ await stmt.close();
+ }
+ }
+ return connection.query(sql);
+}
+
export async function handleQuery(ctx: WorkerHandlerContext): Promise {
const { payload } = ctx;
- const req = payload as { sql?: string };
- if (!connection) throw new Error('DuckDB not initialized');
- const result = await connection.query(req.sql ?? '');
+ const req = payload as { sql?: string; params?: readonly unknown[] };
+ const result = await runSql(req.sql ?? '', req.params);
return result.toArray().map((row: { toJSON(): Record }) => row.toJSON());
}
export async function handleExec(ctx: WorkerHandlerContext): Promise {
const { payload } = ctx;
- const req = payload as { sql?: string };
- if (!connection) throw new Error('DuckDB not initialized');
- await connection.query(req.sql ?? '');
+ const req = payload as { sql?: string; params?: readonly unknown[] };
+ await runSql(req.sql ?? '', req.params);
return { ok: true };
}