From b4b1bbfc7b389fb3f1821e86719aa2a3290de1e0 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:21:24 +0200 Subject: [PATCH 01/10] test(worker-bus): export v2 duckdb/inference handlers + parity gate tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Neither workers/v2/duckdb.worker.ts's handlers nor workers/v2/inference.worker.ts's handleInference were exported or unit-tested against real logic (only webllm.worker.ts had that, tests/unit/webllmWorkerHandler.test.ts) despite both already being registered as live WorkerBus v2 pools with zero production callers — docs/adr/0014-worker-generation-duplication.md's deferred migration needs these to actually match v1 before any consumer cuts over. Testing surfaced a real gap: initDuckDb()'s OPFS-unavailable catch block silently swallowed the failure, unlike v1's out-of-band OPFS_FALLBACK message that lets the UI warn users their analytics won't persist. Fixed by threading ctx.emitProgress into initDuckDb() and emitting an 'opfs-fallback' progress stage from the catch block — reuses the existing progress channel WebLLM already uses for download progress, no new message-protocol/schema change needed. First PR of the ADR-0014 consolidation sprint (5 PRs total, see .claude/plans/worker-generation-v1-vs-enumerated-fox.md). Co-Authored-By: Claude Sonnet 5 --- tests/unit/duckdbWorkerHandler.test.ts | 233 ++++++++++++++++++++ tests/unit/inferenceWorkerHandlerV2.test.ts | 174 +++++++++++++++ workers/v2/duckdb.worker.ts | 35 ++- workers/v2/inference.worker.ts | 4 +- 4 files changed, 436 insertions(+), 10 deletions(-) create mode 100644 tests/unit/duckdbWorkerHandler.test.ts create mode 100644 tests/unit/inferenceWorkerHandlerV2.test.ts diff --git a/tests/unit/duckdbWorkerHandler.test.ts b/tests/unit/duckdbWorkerHandler.test.ts new file mode 100644 index 00000000..c1e0f10e --- /dev/null +++ b/tests/unit/duckdbWorkerHandler.test.ts @@ -0,0 +1,233 @@ +// @vitest-environment jsdom +// QNBS-v3: parity gate (PR 0, worker-generation consolidation, docs/adr/0014-worker-generation- +// duplication.md) — unit tests for the v2 DuckDB worker's handler functions, exercised +// directly (not via a real worker/WorkerBus) against a mocked @duckdb/duckdb-wasm, to +// catch response-shape / OPFS-fallback-signal drift from v1 before any consumer cuts over. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { WorkerHandlerContext } from '../../packages/worker-bus/src/workerBootstrap'; + +class MockWorker { + postMessage = vi.fn(); + terminate = vi.fn(); + addEventListener = vi.fn(); +} +vi.stubGlobal('Worker', MockWorker); + +// QNBS-v3: vi.hoisted — the worker module is imported statically below, so the mock factory runs +// during the hoisted import phase; plain consts would hit a TDZ ReferenceError. +const { + mockSelectBundle, + mockInstantiate, + mockRegisterFileHandle, + mockConnect, + mockDbTerminate, + MockAsyncDuckDB, +} = vi.hoisted(() => { + const mockInstantiate = vi.fn(); + const mockRegisterFileHandle = vi.fn(); + const mockConnect = vi.fn(); + const mockDbTerminate = vi.fn(); + // QNBS-v3: must be a `function`, not an arrow — arrow functions aren't constructable, and + // `initDuckDb()` calls `new AsyncDuckDB(...)`. + const MockAsyncDuckDB = vi.fn().mockImplementation(function MockAsyncDuckDBCtor() { + return { + instantiate: mockInstantiate, + registerFileHandle: mockRegisterFileHandle, + connect: mockConnect, + terminate: mockDbTerminate, + }; + }); + return { + mockSelectBundle: vi.fn(), + mockInstantiate, + mockRegisterFileHandle, + mockConnect, + mockDbTerminate, + MockAsyncDuckDB, + }; +}); + +vi.mock('@duckdb/duckdb-wasm', () => ({ + AsyncDuckDB: MockAsyncDuckDB, + selectBundle: mockSelectBundle, + ConsoleLogger: vi.fn(), + DuckDBDataProtocol: { BROWSER_FSACCESS: 'browser-fsaccess' }, +})); + +// QNBS-v3: imported AFTER the mock is declared so the worker picks up the mocked module. +import { + handleExec, + handleQuery, + handleShutdown, + initDuckDb, +} from '../../workers/v2/duckdb.worker'; + +function makeCtx(overrides: Partial = {}): WorkerHandlerContext { + return { + taskId: 't1', + taskType: 'db.duckdb.query', + payload: {}, + signal: new AbortController().signal, + emitProgress: vi.fn(), + ...overrides, + }; +} + +function stubOpfsDirectory(getFileHandle: ReturnType): void { + Object.defineProperty(globalThis.navigator, 'storage', { + value: { getDirectory: vi.fn().mockResolvedValue({ getFileHandle }) }, + configurable: true, + }); +} + +function stubNoOpfs(): void { + Object.defineProperty(globalThis.navigator, 'storage', { + value: undefined, + configurable: true, + }); +} + +// QNBS-v3: always include `close` — the shared beforeEach calls the real handleShutdown() to +// reset module state, which unconditionally calls `connection?.close()`. +function mockConnection(overrides: { query?: ReturnType } = {}) { + return { query: overrides.query ?? vi.fn(), close: vi.fn().mockResolvedValue(undefined) }; +} + +describe('duckdb.worker handlers', () => { + beforeEach(async () => { + // QNBS-v3: connection/db are module-scoped in duckdb.worker.ts and persist across tests in + // this file — reset via handleShutdown() before each test so "not initialized" + // assertions aren't polluted by a previous test's initDuckDb() call. + await handleShutdown(); + vi.clearAllMocks(); + mockSelectBundle.mockResolvedValue({ mainModule: 'mvp.wasm', mainWorker: 'mvp.worker.js' }); + mockInstantiate.mockResolvedValue(undefined); + }); + + afterEach(() => { + stubNoOpfs(); + }); + + describe('initDuckDb', () => { + it('attaches OPFS persistence and does not emit opfs-fallback when it succeeds', async () => { + const connectionQuery = vi.fn().mockResolvedValue(undefined); + mockConnect.mockResolvedValue(mockConnection({ query: connectionQuery })); + mockRegisterFileHandle.mockResolvedValue(undefined); + stubOpfsDirectory(vi.fn().mockResolvedValue({})); + const emitProgress = vi.fn(); + + await initDuckDb(emitProgress); + + expect(connectionQuery).toHaveBeenCalledWith(expect.stringContaining('ATTACH')); + expect(emitProgress).not.toHaveBeenCalled(); + }); + + it('emits opfs-fallback and still connects when registerFileHandle throws', async () => { + mockConnect.mockResolvedValue(mockConnection()); + mockRegisterFileHandle.mockRejectedValue(new Error('registerFileHandle denied')); + stubOpfsDirectory(vi.fn().mockResolvedValue({})); + const emitProgress = vi.fn(); + + await initDuckDb(emitProgress); + + expect(emitProgress).toHaveBeenCalledWith('opfs-fallback', 1, 'registerFileHandle denied'); + // Falls back to a plain (in-memory) connect() despite the OPFS failure. + expect(mockConnect).toHaveBeenCalledTimes(1); + }); + + it('does not emit opfs-fallback when OPFS is simply unsupported (no navigator.storage)', async () => { + mockConnect.mockResolvedValue(mockConnection()); + stubNoOpfs(); + const emitProgress = vi.fn(); + + await initDuckDb(emitProgress); + + expect(emitProgress).not.toHaveBeenCalled(); + expect(mockConnect).toHaveBeenCalledTimes(1); + }); + + it('works when emitProgress is omitted (registerTaskHandler always passes ctx.emitProgress, but keep the param optional/safe)', async () => { + mockConnect.mockResolvedValue( + mockConnection({ query: vi.fn().mockRejectedValue(new Error('boom')) }), + ); + mockRegisterFileHandle.mockResolvedValue(undefined); + stubOpfsDirectory(vi.fn().mockResolvedValue({})); + + await expect(initDuckDb()).resolves.toBeUndefined(); + }); + }); + + describe('handleQuery', () => { + it('throws when DuckDB is not initialized', async () => { + await expect(handleQuery(makeCtx({ payload: { sql: 'select 1' } }))).rejects.toThrow( + 'DuckDB not initialized', + ); + }); + + it('returns a raw array of plain-object rows (not a {ok,rows} wrapper)', async () => { + const row = { toJSON: () => ({ id: 1 }) }; + mockConnect.mockResolvedValue( + mockConnection({ query: vi.fn().mockResolvedValue({ toArray: () => [row] }) }), + ); + stubNoOpfs(); + await initDuckDb(); + + const result = await handleQuery(makeCtx({ payload: { sql: 'select 1' } })); + + expect(result).toEqual([{ id: 1 }]); + }); + + it('throws (does not resolve {ok:false}) when the query rejects', async () => { + mockConnect.mockResolvedValue( + mockConnection({ query: vi.fn().mockRejectedValue(new Error('bad sql')) }), + ); + stubNoOpfs(); + await initDuckDb(); + + await expect(handleQuery(makeCtx({ payload: { sql: 'garbage' } }))).rejects.toThrow( + 'bad sql', + ); + }); + }); + + describe('handleExec', () => { + it('throws when DuckDB is not initialized', async () => { + await expect( + handleExec(makeCtx({ payload: { sql: 'insert into t values (1)' } })), + ).rejects.toThrow('DuckDB not initialized'); + }); + + it('returns {ok:true} on success', async () => { + const connectionQuery = vi.fn().mockResolvedValue(undefined); + mockConnect.mockResolvedValue(mockConnection({ query: connectionQuery })); + stubNoOpfs(); + await initDuckDb(); + + const result = await handleExec(makeCtx({ payload: { sql: 'create table t (id int)' } })); + + expect(result).toEqual({ ok: true }); + expect(connectionQuery).toHaveBeenCalledWith('create table t (id int)'); + }); + }); + + describe('handleShutdown', () => { + it('closes the connection and terminates the db, returning {ok:true}', async () => { + const close = vi.fn().mockResolvedValue(undefined); + mockConnect.mockResolvedValue({ query: vi.fn(), close }); + stubNoOpfs(); + await initDuckDb(); + + const result = await handleShutdown(); + + expect(result).toEqual({ ok: true }); + expect(close).toHaveBeenCalled(); + expect(mockDbTerminate).toHaveBeenCalled(); + + // A query after shutdown must throw 'DuckDB not initialized' again. + await expect(handleQuery(makeCtx({ payload: { sql: 'select 1' } }))).rejects.toThrow( + 'DuckDB not initialized', + ); + }); + }); +}); diff --git a/tests/unit/inferenceWorkerHandlerV2.test.ts b/tests/unit/inferenceWorkerHandlerV2.test.ts new file mode 100644 index 00000000..aadb8001 --- /dev/null +++ b/tests/unit/inferenceWorkerHandlerV2.test.ts @@ -0,0 +1,174 @@ +// @vitest-environment jsdom +// QNBS-v3: parity gate (PR 0, worker-generation consolidation, docs/adr/0014-worker-generation- +// duplication.md) — unit tests for the v2 inference worker's handler, exercised directly +// (not via a real worker/WorkerBus) against a mocked @huggingface/transformers, to catch +// drift from v1 (workers/inference.worker.ts) before any consumer cuts over. File name is +// distinct from the existing v1 tests/unit/inferenceWorker.test.ts until v1 is deleted. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { WorkerHandlerContext } from '../../packages/worker-bus/src/workerBootstrap'; + +const { mockPipelineFactory } = vi.hoisted(() => ({ mockPipelineFactory: vi.fn() })); + +vi.mock('@huggingface/transformers', () => ({ + pipeline: mockPipelineFactory, +})); + +// QNBS-v3: imported AFTER the mock is declared so the worker picks up the mocked factory. +import { handleInference } from '../../workers/v2/inference.worker'; + +function makeCtx(overrides: Partial = {}): WorkerHandlerContext { + return { + taskId: 't1', + taskType: 'inference.embed', + payload: { task: 'feature-extraction', modelId: 'm', input: 'hello world' }, + signal: new AbortController().signal, + emitProgress: vi.fn(), + ...overrides, + }; +} + +describe('inference.worker (v2) handleInference', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('throws Aborted when the signal is already aborted', async () => { + const controller = new AbortController(); + controller.abort(); + await expect(handleInference(makeCtx({ signal: controller.signal }))).rejects.toThrow( + 'Aborted', + ); + expect(mockPipelineFactory).not.toHaveBeenCalled(); + }); + + it('throws Aborted when the signal aborts between model load and inference', async () => { + const controller = new AbortController(); + const pipe = vi.fn(); + mockPipelineFactory.mockImplementation(async () => { + controller.abort(); + return pipe; + }); + + await expect( + handleInference( + makeCtx({ + signal: controller.signal, + payload: { task: 'feature-extraction', modelId: 'abort-model', input: 'x' }, + }), + ), + ).rejects.toThrow('Aborted'); + expect(pipe).not.toHaveBeenCalled(); + }); + + it('feature-extraction: normalizes a Float32Array result to a flat number[]', async () => { + const pipe = vi.fn().mockResolvedValue(new Float32Array([0.1, 0.2, 0.3])); + mockPipelineFactory.mockResolvedValue(pipe); + const emitProgress = vi.fn(); + + const result = await handleInference( + makeCtx({ + payload: { task: 'feature-extraction', modelId: 'embed-1', input: 'x' }, + emitProgress, + }), + ); + + expect(result).toEqual([Math.fround(0.1), Math.fround(0.2), Math.fround(0.3)]); + expect(pipe).toHaveBeenCalledWith( + 'x', + expect.objectContaining({ pooling: 'mean', normalize: true }), + ); + expect(emitProgress).toHaveBeenNthCalledWith(1, 'loading', 0.2, 'Loading model'); + expect(emitProgress).toHaveBeenNthCalledWith(2, 'inference', 0.6, 'Running inference'); + expect(emitProgress).toHaveBeenNthCalledWith(3, 'done', 1.0, 'Complete'); + }); + + it('feature-extraction: normalizes a {data: Float32Array} result to a flat number[]', async () => { + const pipe = vi.fn().mockResolvedValue({ data: new Float32Array([1, 2]) }); + mockPipelineFactory.mockResolvedValue(pipe); + + const result = await handleInference( + makeCtx({ payload: { task: 'feature-extraction', modelId: 'embed-2', input: 'x' } }), + ); + + expect(result).toEqual([1, 2]); + }); + + it('feature-extraction: normalizes a plain iterable result to a flat number[]', async () => { + const pipe = vi.fn().mockResolvedValue([4, 5, 6]); + mockPipelineFactory.mockResolvedValue(pipe); + + const result = await handleInference( + makeCtx({ payload: { task: 'feature-extraction', modelId: 'embed-3', input: 'x' } }), + ); + + expect(result).toEqual([4, 5, 6]); + }); + + it('sentiment-analysis: returns "LABEL:score" formatted to 4 decimals', async () => { + const pipe = vi.fn().mockResolvedValue([{ label: 'POSITIVE', score: 0.5 }]); + mockPipelineFactory.mockResolvedValue(pipe); + + const result = await handleInference( + makeCtx({ payload: { task: 'sentiment-analysis', modelId: 'sent-1', input: 'great!' } }), + ); + + expect(result).toBe('POSITIVE:0.5000'); + expect(pipe).toHaveBeenCalledWith('great!', {}); + }); + + it('sentiment-analysis: defaults to NEUTRAL:0.0000 when the pipeline result is empty', async () => { + const pipe = vi.fn().mockResolvedValue([]); + mockPipelineFactory.mockResolvedValue(pipe); + + const result = await handleInference( + makeCtx({ payload: { task: 'sentiment-analysis', modelId: 'sent-2', input: 'x' } }), + ); + + expect(result).toBe('NEUTRAL:0.0000'); + }); + + it('summarization: returns summary_text', async () => { + const pipe = vi.fn().mockResolvedValue([{ summary_text: 'a short summary' }]); + mockPipelineFactory.mockResolvedValue(pipe); + + const result = await handleInference( + makeCtx({ payload: { task: 'summarization', modelId: 'sum-1', input: 'long text...' } }), + ); + + expect(result).toBe('a short summary'); + }); + + it('text-generation: returns generated_text', async () => { + const pipe = vi.fn().mockResolvedValue([{ generated_text: 'once upon a time' }]); + mockPipelineFactory.mockResolvedValue(pipe); + + const result = await handleInference( + makeCtx({ payload: { task: 'text-generation', modelId: 'gen-1', input: 'prompt' } }), + ); + + expect(result).toBe('once upon a time'); + }); + + it('passes inferenceOptions through to the pipeline call for non-embedding tasks', async () => { + const pipe = vi.fn().mockResolvedValue([{ generated_text: 'x' }]); + mockPipelineFactory.mockResolvedValue(pipe); + + await handleInference( + makeCtx({ + payload: { + task: 'text-generation', + modelId: 'gen-2', + input: 'prompt', + inferenceOptions: { max_new_tokens: 32 }, + }, + }), + ); + + expect(pipe).toHaveBeenCalledWith('prompt', { max_new_tokens: 32 }); + }); +}); diff --git a/workers/v2/duckdb.worker.ts b/workers/v2/duckdb.worker.ts index 228e384e..2743ce7c 100644 --- a/workers/v2/duckdb.worker.ts +++ b/workers/v2/duckdb.worker.ts @@ -42,7 +42,11 @@ const SELF_HOSTED_BUNDLES = { }, }; -async function initDuckDb(): Promise { +// QNBS-v3: exported for tests/unit/duckdbWorkerHandler.test.ts (parity gate, PR 0 of the +// worker-generation consolidation) — verifies real handler logic without a live worker. +export async function initDuckDb( + emitProgress?: (stage: string, progress: number, message?: string) => void, +): Promise { const { AsyncDuckDB, selectBundle, ConsoleLogger } = await getDuckDb(); const bundle = await selectBundle(SELF_HOSTED_BUNDLES); @@ -68,7 +72,16 @@ async function initDuckDb(): Promise { ); connection = await newDb.connect(); await connection.query("ATTACH 'worldscript_analytics.duckdb' AS analytics (TYPE duckdb)"); - } catch { + } catch (opfsErr) { + // QNBS-v3: mirrors legacy workers/duckdbWorker.ts's out-of-band OPFS_FALLBACK message — + // v2 has no bare postMessage escape hatch from inside a task handler, so this + // reuses the existing progress channel instead (duckdbClient's INIT adapter + // listens for the 'opfs-fallback' stage and forwards it to setOpfsFallbackHandler). + emitProgress?.( + 'opfs-fallback', + 1, + opfsErr instanceof Error ? opfsErr.message : String(opfsErr), + ); connection = await newDb.connect(); } } else { @@ -78,7 +91,7 @@ async function initDuckDb(): Promise { db = newDb; } -async function handleQuery(ctx: WorkerHandlerContext): Promise { +export async function handleQuery(ctx: WorkerHandlerContext): Promise { const { payload } = ctx; const req = payload as { sql?: string }; if (!connection) throw new Error('DuckDB not initialized'); @@ -86,7 +99,7 @@ async function handleQuery(ctx: WorkerHandlerContext): Promise { return result.toArray().map((row: { toJSON(): Record }) => row.toJSON()); } -async function handleExec(ctx: WorkerHandlerContext): Promise { +export async function handleExec(ctx: WorkerHandlerContext): Promise { const { payload } = ctx; const req = payload as { sql?: string }; if (!connection) throw new Error('DuckDB not initialized'); @@ -94,7 +107,7 @@ async function handleExec(ctx: WorkerHandlerContext): Promise { return { ok: true }; } -async function handleShutdown(): Promise { +export async function handleShutdown(): Promise { await connection?.close(); await db?.terminate(); connection = null; @@ -102,10 +115,14 @@ async function handleShutdown(): Promise { return { ok: true }; } -registerTaskHandler('db.duckdb.init', async () => { - await initDuckDb(); - return { ok: true }; -}, ['db.duckdb']); +registerTaskHandler( + 'db.duckdb.init', + async (ctx) => { + await initDuckDb(ctx.emitProgress); + return { ok: true }; + }, + ['db.duckdb'], +); registerTaskHandler('db.duckdb.query', handleQuery, ['db.duckdb']); registerTaskHandler('db.duckdb.exec', handleExec, ['db.duckdb']); diff --git a/workers/v2/inference.worker.ts b/workers/v2/inference.worker.ts index 261eb717..a0eb7e3a 100644 --- a/workers/v2/inference.worker.ts +++ b/workers/v2/inference.worker.ts @@ -42,7 +42,9 @@ async function loadPipeline(task: string, modelId: string, quantized = true) { }); } -async function handleInference(ctx: WorkerHandlerContext): Promise { +// QNBS-v3: exported for tests/unit/inferenceWorkerHandlerV2.test.ts (parity gate, PR 0 of the +// worker-generation consolidation) — verifies real handler logic without a live worker. +export async function handleInference(ctx: WorkerHandlerContext): Promise { const { payload, signal, emitProgress } = ctx; const req = payload as { task: string; From bcc41c2757ba6fd9ba6e1d10d3a7fb3dddf7a47f Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:44:35 +0200 Subject: [PATCH 02/10] fix(worker-bus): close leaked OPFS connection on ATTACH failure + QNBS-v3 format CodeRabbit review on #286 found a real bug: initDuckDb()'s OPFS catch block overwrote `connection` with the fallback connection without closing the first (partially-attached) one, leaking a DuckDB connection on every OPFS-attach failure. Fixed by tracking the OPFS attempt in its own local variable and closing it in the catch path before falling back. New regression test asserts close() is called. Also collapsed several multi-line QNBS-v3 comments in workers/v2/duckdb.worker.ts, workers/v2/inference.worker.ts, and both new test files to the required single bracketed-line format. Co-Authored-By: Claude Sonnet 5 --- tests/unit/duckdbWorkerHandler.test.ts | 37 +++++++++++++-------- tests/unit/inferenceWorkerHandlerV2.test.ts | 8 ++--- workers/v2/duckdb.worker.ts | 18 +++++----- workers/v2/inference.worker.ts | 3 +- 4 files changed, 36 insertions(+), 30 deletions(-) diff --git a/tests/unit/duckdbWorkerHandler.test.ts b/tests/unit/duckdbWorkerHandler.test.ts index c1e0f10e..5ebe5fa3 100644 --- a/tests/unit/duckdbWorkerHandler.test.ts +++ b/tests/unit/duckdbWorkerHandler.test.ts @@ -1,8 +1,5 @@ // @vitest-environment jsdom -// QNBS-v3: parity gate (PR 0, worker-generation consolidation, docs/adr/0014-worker-generation- -// duplication.md) — unit tests for the v2 DuckDB worker's handler functions, exercised -// directly (not via a real worker/WorkerBus) against a mocked @duckdb/duckdb-wasm, to -// catch response-shape / OPFS-fallback-signal drift from v1 before any consumer cuts over. +// QNBS-v3: [Parity gate for the v2 DuckDB worker handlers (docs/adr/0014) — catches response-shape/OPFS-fallback drift from v1 before any consumer cuts over.] import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { WorkerHandlerContext } from '../../packages/worker-bus/src/workerBootstrap'; @@ -14,8 +11,7 @@ class MockWorker { } vi.stubGlobal('Worker', MockWorker); -// QNBS-v3: vi.hoisted — the worker module is imported statically below, so the mock factory runs -// during the hoisted import phase; plain consts would hit a TDZ ReferenceError. +// QNBS-v3: [vi.hoisted — the mock factory runs during the hoisted import phase, so plain consts would hit a TDZ ReferenceError.] const { mockSelectBundle, mockInstantiate, @@ -28,8 +24,7 @@ const { const mockRegisterFileHandle = vi.fn(); const mockConnect = vi.fn(); const mockDbTerminate = vi.fn(); - // QNBS-v3: must be a `function`, not an arrow — arrow functions aren't constructable, and - // `initDuckDb()` calls `new AsyncDuckDB(...)`. + // QNBS-v3: [Must be a `function`, not an arrow — arrow functions aren't constructable, and initDuckDb() calls `new AsyncDuckDB(...)`.] const MockAsyncDuckDB = vi.fn().mockImplementation(function MockAsyncDuckDBCtor() { return { instantiate: mockInstantiate, @@ -55,7 +50,7 @@ vi.mock('@duckdb/duckdb-wasm', () => ({ DuckDBDataProtocol: { BROWSER_FSACCESS: 'browser-fsaccess' }, })); -// QNBS-v3: imported AFTER the mock is declared so the worker picks up the mocked module. +// QNBS-v3: [Imported after the mock is declared so the worker picks up the mocked module.] import { handleExec, handleQuery, @@ -88,17 +83,14 @@ function stubNoOpfs(): void { }); } -// QNBS-v3: always include `close` — the shared beforeEach calls the real handleShutdown() to -// reset module state, which unconditionally calls `connection?.close()`. +// QNBS-v3: [Always include `close` — the shared beforeEach calls handleShutdown() to reset state, which unconditionally calls `connection?.close()`.] function mockConnection(overrides: { query?: ReturnType } = {}) { return { query: overrides.query ?? vi.fn(), close: vi.fn().mockResolvedValue(undefined) }; } describe('duckdb.worker handlers', () => { beforeEach(async () => { - // QNBS-v3: connection/db are module-scoped in duckdb.worker.ts and persist across tests in - // this file — reset via handleShutdown() before each test so "not initialized" - // assertions aren't polluted by a previous test's initDuckDb() call. + // QNBS-v3: [connection/db are module-scoped and persist across tests — reset first so "not initialized" assertions aren't polluted by a prior test's initDuckDb().] await handleShutdown(); vi.clearAllMocks(); mockSelectBundle.mockResolvedValue({ mainModule: 'mvp.wasm', mainWorker: 'mvp.worker.js' }); @@ -136,6 +128,23 @@ describe('duckdb.worker handlers', () => { expect(mockConnect).toHaveBeenCalledTimes(1); }); + it('closes the partially-initialized OPFS connection before falling back when ATTACH fails', async () => { + const opfsClose = vi.fn().mockResolvedValue(undefined); + 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 initDuckDb(emitProgress); + + expect(opfsClose).toHaveBeenCalled(); + 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(); diff --git a/tests/unit/inferenceWorkerHandlerV2.test.ts b/tests/unit/inferenceWorkerHandlerV2.test.ts index aadb8001..07f16091 100644 --- a/tests/unit/inferenceWorkerHandlerV2.test.ts +++ b/tests/unit/inferenceWorkerHandlerV2.test.ts @@ -1,9 +1,5 @@ // @vitest-environment jsdom -// QNBS-v3: parity gate (PR 0, worker-generation consolidation, docs/adr/0014-worker-generation- -// duplication.md) — unit tests for the v2 inference worker's handler, exercised directly -// (not via a real worker/WorkerBus) against a mocked @huggingface/transformers, to catch -// drift from v1 (workers/inference.worker.ts) before any consumer cuts over. File name is -// distinct from the existing v1 tests/unit/inferenceWorker.test.ts until v1 is deleted. +// QNBS-v3: [Parity gate for the v2 inference worker handler (docs/adr/0014) — catches drift from v1 before any consumer cuts over; named distinctly from tests/unit/inferenceWorker.test.ts until v1 is deleted.] import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { WorkerHandlerContext } from '../../packages/worker-bus/src/workerBootstrap'; @@ -14,7 +10,7 @@ vi.mock('@huggingface/transformers', () => ({ pipeline: mockPipelineFactory, })); -// QNBS-v3: imported AFTER the mock is declared so the worker picks up the mocked factory. +// QNBS-v3: [Imported after the mock is declared so the worker picks up the mocked factory.] import { handleInference } from '../../workers/v2/inference.worker'; function makeCtx(overrides: Partial = {}): WorkerHandlerContext { diff --git a/workers/v2/duckdb.worker.ts b/workers/v2/duckdb.worker.ts index 2743ce7c..891886ea 100644 --- a/workers/v2/duckdb.worker.ts +++ b/workers/v2/duckdb.worker.ts @@ -42,8 +42,7 @@ const SELF_HOSTED_BUNDLES = { }, }; -// QNBS-v3: exported for tests/unit/duckdbWorkerHandler.test.ts (parity gate, PR 0 of the -// worker-generation consolidation) — verifies real handler logic without a live worker. +// QNBS-v3: [Exported for tests/unit/duckdbWorkerHandler.test.ts to verify real handler logic without a live worker.] export async function initDuckDb( emitProgress?: (stage: string, progress: number, message?: string) => void, ): Promise { @@ -58,6 +57,8 @@ 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; try { const { DuckDBDataProtocol } = await getDuckDb(); const opfsRoot = await navigator.storage.getDirectory(); @@ -70,13 +71,14 @@ export async function initDuckDb( DuckDBDataProtocol.BROWSER_FSACCESS, true, ); - connection = await newDb.connect(); - await connection.query("ATTACH 'worldscript_analytics.duckdb' AS analytics (TYPE duckdb)"); + opfsConnection = await newDb.connect(); + await opfsConnection.query( + "ATTACH 'worldscript_analytics.duckdb' AS analytics (TYPE duckdb)", + ); + connection = opfsConnection; } catch (opfsErr) { - // QNBS-v3: mirrors legacy workers/duckdbWorker.ts's out-of-band OPFS_FALLBACK message — - // v2 has no bare postMessage escape hatch from inside a task handler, so this - // reuses the existing progress channel instead (duckdbClient's INIT adapter - // listens for the 'opfs-fallback' stage and forwards it to setOpfsFallbackHandler). + // 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, diff --git a/workers/v2/inference.worker.ts b/workers/v2/inference.worker.ts index a0eb7e3a..88633733 100644 --- a/workers/v2/inference.worker.ts +++ b/workers/v2/inference.worker.ts @@ -42,8 +42,7 @@ async function loadPipeline(task: string, modelId: string, quantized = true) { }); } -// QNBS-v3: exported for tests/unit/inferenceWorkerHandlerV2.test.ts (parity gate, PR 0 of the -// worker-generation consolidation) — verifies real handler logic without a live worker. +// QNBS-v3: [Exported for tests/unit/inferenceWorkerHandlerV2.test.ts to verify real handler logic without a live worker.] export async function handleInference(ctx: WorkerHandlerContext): Promise { const { payload, signal, emitProgress } = ctx; const req = payload as { From bc371b36e9890d880d2b77ab42237776cc32bd24 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:59:57 +0200 Subject: [PATCH 03/10] feat(duckdb): migrate duckdbClient onto WorkerBus v2, delete v1 worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second of 5 PRs consolidating the v1/WorkerBus-v2 worker duplication (ADR-0014). services/duckdb/duckdbClient.ts's public API (init/query/ exec/shutdown/terminate/setOpfsFallbackHandler) is unchanged — its 5 real consumers (useDuckDb.ts, duckdbAnalytics.ts, duckdbMigration.ts, ragVectorMigration.ts, telemetryService.ts) need no changes. Only the internals swapped from a raw `new Worker(duckdbWorker.ts)` to routing through the shared WorkerBus v2 'duckdb' pool via a new ensureDuckDbPool() (mirrors ensureWebLlmPool()'s decoupling from the enableWorkerBusV2 flag — analytics is core, not experimental infra). The v2 worker returns raw values and throws on failure, unlike v1's {ok,rows,error} wrapper — the adapter in duckdbClient.ts's send() translates at that one boundary instead of touching ~20 call sites across 5 files. WorkerBus's default 2-retry is disabled per-call (retryPolicy: {maxRetries:0}) to avoid stacking under duckdbAnalytics.ts's existing withDuckDbRetry() app-level retry. Two things surfaced while wiring this up, both fixed here: - WorkerBus.shutdown() tears down all 4 pools with no way to scope to just 'duckdb' — added WorkerBus.terminatePool(poolId) so duckdbClient.terminate() doesn't kill in-flight inference/webllm/ plugin work on the shared bus. - send() is async and awaits ensureDuckDbPool() before a task handle exists; naively attaching the AbortSignal listener after that await (like a first draft did) can lose a same-tick abort, unlike v1's synchronous Promise-executor version. Fixed by attaching the listener synchronously up front and applying it once the handle is ready — regression-tested. workers/duckdbWorker.ts deleted; features/featureCatalog.ts's enableDuckDbAnalytics entry retargeted to workers/v2/duckdb.worker.ts. Co-Authored-By: Claude Sonnet 5 --- features/featureCatalog.ts | 2 +- packages/worker-bus/src/workerBus.ts | 15 ++ packages/worker-bus/tests/workerBus.test.ts | 24 ++ services/duckdb/duckdbClient.ts | 157 +++++++----- services/workerBusManager.ts | 12 + tests/unit/duckdbClient.test.ts | 264 ++++++++++++++------ workers/duckdbWorker.ts | 219 ---------------- 7 files changed, 332 insertions(+), 361 deletions(-) delete mode 100644 workers/duckdbWorker.ts 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..84b45627 100644 --- a/packages/worker-bus/src/workerBus.ts +++ b/packages/worker-bus/src/workerBus.ts @@ -195,6 +195,21 @@ 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. + */ + async terminatePool(poolId: string): Promise { + const pool = this.pools.get(poolId); + if (!pool) return; + await pool.terminateAll(); + this.pools.delete(poolId); + } + // --------------------------------------------------------------------------- // Internal // --------------------------------------------------------------------------- diff --git a/packages/worker-bus/tests/workerBus.test.ts b/packages/worker-bus/tests/workerBus.test.ts index 45f494ba..74ae557c 100644 --- a/packages/worker-bus/tests/workerBus.test.ts +++ b/packages/worker-bus/tests/workerBus.test.ts @@ -209,6 +209,30 @@ 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('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..1315002f 100644 --- a/services/duckdb/duckdbClient.ts +++ b/services/duckdb/duckdbClient.ts @@ -1,80 +1,102 @@ -// 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, ): 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; + signal?.addEventListener( + 'abort', + () => { + abortedEarly = true; + handleRef?.cancel('Aborted'); + }, + { once: true }, + ); - if (signal) { - signal.addEventListener('abort', () => { - w.postMessage({ type: 'WORKER_CANCEL', messageId }); - pendingResolvers.delete(messageId); - resolve({ messageId, ok: false, error: 'Aborted' }); - }); - } + const bus = await ensureDuckDbPool(); + if (!bus) return { messageId, ok: false, error: 'WorkerBus v2 unavailable' }; + + // 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'); - const req: DuckDbRequest = { messageId, type, sql, params }; - w.postMessage(req); - }); + 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 }; + return { messageId, ok: false, error: errorMessage(err), latencyMs }; + } } export const duckdbClient = { @@ -98,14 +120,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..2e973c24 100644 --- a/services/workerBusManager.ts +++ b/services/workerBusManager.ts @@ -164,6 +164,18 @@ 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; +} + /** * 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..c592b26c 100644 --- a/tests/unit/duckdbClient.test.ts +++ b/tests/unit/duckdbClient.test.ts @@ -1,130 +1,244 @@ import { 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 { mockEnqueue, mockTerminatePool, mockEnsureDuckDbPool, mockGetWorkerBus } = vi.hoisted( + () => ({ + mockEnqueue: vi.fn(), + mockTerminatePool: vi.fn(), + mockEnsureDuckDbPool: vi.fn(), + mockGetWorkerBus: vi.fn(), + }), +); + +vi.mock('../../services/workerBusManager', () => ({ + ensureDuckDbPool: mockEnsureDuckDbPool, + getWorkerBus: mockGetWorkerBus, +})); -const mockPostMessage = vi.fn(); -const mockTerminate = vi.fn(); -const mockAddEventListener = vi.fn(); +const { duckdbClient } = await import('../../services/duckdb/duckdbClient'); -class MockWorker { - postMessage = mockPostMessage; - terminate = mockTerminate; - addEventListener = mockAddEventListener; +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('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('duckdbClient.exec', () => { - it('posts an EXEC message', async () => { - const promise = duckdbClient.exec('CREATE TABLE t (id INTEGER)'); - - 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.exec and returns {ok:true}', async () => { + mockEnqueue.mockReturnValue(makeHandle(Promise.resolve({ ok: true }))); - expect(req.type).toBe('EXEC'); + const res = await duckdbClient.exec('CREATE TABLE t (id INTEGER)'); - 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(); + it('enqueues db.duckdb.shutdown', 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.shutdown(); - expect(req.type).toBe('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')); + }); +}); + +describe('duckdbClient OPFS fallback', () => { + 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'); + duckdbClient.setOpfsFallbackHandler(null); + }); + + 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(); + duckdbClient.setOpfsFallbackHandler(null); }); }); 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); - }); -}); From dbebf5cba7ee064986f17b59eec07d9d51bba819 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 20:15:39 +0200 Subject: [PATCH 04/10] fix(worker-bus): make OPFS-connection cleanup best-effort on ATTACH failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit's second pass on #286 flagged this as an outside-diff-range comment (not a resolvable inline thread) on bcc41c27's own fix: the new `await opfsConnection?.close()` call could itself reject, which would skip emitProgress('opfs-fallback', ...) and the fallback newDb.connect(), and replace the original ATTACH error with the close failure — turning a graceful degrade into a hard init failure. Wrapped the cleanup in its own try/catch (best-effort, logged via console.warn, never blocks the fallback path) so a failing close() can't prevent the fallback connection or swallow the real error. New regression test asserts both still happen when close() rejects. Co-Authored-By: Claude Sonnet 5 --- tests/unit/duckdbWorkerHandler.test.ts | 18 ++++++++++++++++++ workers/v2/duckdb.worker.ts | 7 ++++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/tests/unit/duckdbWorkerHandler.test.ts b/tests/unit/duckdbWorkerHandler.test.ts index 5ebe5fa3..b942e2ec 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(); diff --git a/workers/v2/duckdb.worker.ts b/workers/v2/duckdb.worker.ts index 891886ea..d7a5880a 100644 --- a/workers/v2/duckdb.worker.ts +++ b/workers/v2/duckdb.worker.ts @@ -77,8 +77,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) { + console.warn('[duckdb.worker] 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, From 871cfbaa3c0afdeb7b2a15614a017ab361f14534 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 20:27:08 +0200 Subject: [PATCH 05/10] feat(ai): migrate localEmbeddingService onto WorkerBus v2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third of 5 PRs consolidating the v1/WorkerBus-v2 worker duplication (ADR-0014). services/ai/localEmbeddingService.ts's public API (embedText/embedBatch/cosineSimilarity/clearEmbeddingCache) is unchanged — its consumers (localRagService.ts, crossProjectIndexService.ts, ragPromptAssembly.ts, ragVectorMigration.ts, proForgeMemoryBank.ts, loraEvaluationService.ts, loraDatasetBuilder.ts, inferenceGateway.ts) need no changes. Internals swapped from a dedicated `new Worker(inference.worker.ts)` instance to the shared WorkerBus v2 'inference' pool via ensureInferencePool() (same decoupling-from- enableWorkerBusV2 pattern as ensureWebLlmPool/ensureDuckDbPool). v2's handler already throws on failure (matching embedText's existing throw-on-failure contract), so no response-shape translation is needed here — just swap the transport. Deleted ~70 lines of hand-rolled health-check/restart/backoff logic (30s ping, exponential backoff up to 5 attempts), superseded by WorkerPool's generic health check. Capped the shared 'inference' pool's maxWorkers to 2 (was MAX_WORKERS_INFERENCE=4) in workerBusManager.ts: each pool replica independently loads its own transformers.js pipeline with no cross-replica cache sharing, so 4 concurrent workers under a burst (e.g. embedBatch's micro-batching) could mean 4x the model memory footprint. workers/inference.worker.ts (v1) is NOT deleted yet — localNlpService.ts still uses it directly; removal is PR 4 of this sequence, after that migration lands too. Co-Authored-By: Claude Sonnet 5 --- services/ai/localEmbeddingService.ts | 145 ++--------------------- services/workerBusManager.ts | 3 +- tests/unit/localEmbeddingService.test.ts | 127 ++++++++++---------- 3 files changed, 77 insertions(+), 198 deletions(-) diff --git a/services/ai/localEmbeddingService.ts b/services/ai/localEmbeddingService.ts index b3bb9782..990ae993 100644 --- a/services/ai/localEmbeddingService.ts +++ b/services/ai/localEmbeddingService.ts @@ -1,9 +1,6 @@ -// QNBS-v3: Semantic embedding service — routes to inference.worker.ts via WorkerBus channels. -// Uses Xenova/all-MiniLM-L6-v2 (384-dim, L2-normalized) for semantic RAG and cross-project search. -// Adapted from CannaGuide-2025 embeddingService.ts patterns. +// QNBS-v3: [Semantic embedding service — routes to workers/v2/inference.worker.ts via WorkerBus v2 (docs/adr/0014 migration). Xenova/all-MiniLM-L6-v2, 384-dim, L2-normalized.] -// QNBS-v3: logger import was missing — restartWorker() references logger.error on restart-limit. -import { logger } from '../logger'; +import { ensureInferencePool } from '../workerBusManager'; const EMBEDDING_MODEL = 'Xenova/all-MiniLM-L6-v2'; const MAX_INPUT_CHARS = 512; @@ -22,95 +19,6 @@ function makeCacheKey(text: string): string { export type EmbeddingVector = Float32Array; -// QNBS-v3: The inference worker is loaded lazily to avoid a 2 MB bundle on app start. -let workerInstance: Worker | null = null; -// QNBS-v3: Health check — 30s ping interval; worker restarts if no pong within PONG_TIMEOUT_MS. -const PING_INTERVAL_MS = 30_000; -const PONG_TIMEOUT_MS = 5_000; -let pingTimer: ReturnType | null = null; -let pongTimeoutTimer: ReturnType | null = null; - -// QNBS-v3: Exponential backoff for worker restart to prevent infinite spin on permanent failure. -let restartAttemptCount = 0; -const MAX_RESTART_ATTEMPTS = 5; -const MAX_RESTART_BACKOFF_MS = 60_000; - -function getRestartBackoffMs(): number { - return Math.min(2 ** restartAttemptCount * 1000, MAX_RESTART_BACKOFF_MS); -} - -function clearWorkerHealthTimers(): void { - if (pingTimer) { - clearInterval(pingTimer); - pingTimer = null; - } - if (pongTimeoutTimer) { - clearTimeout(pongTimeoutTimer); - pongTimeoutTimer = null; - } -} - -function restartWorker(): void { - clearWorkerHealthTimers(); - if (workerInstance) { - workerInstance.terminate(); - workerInstance = null; - } - if (import.meta.env?.DEV) { - console.warn('[localEmbeddingService] Inference worker restarted (missed health check pong)'); - } - restartAttemptCount++; - if (restartAttemptCount > MAX_RESTART_ATTEMPTS) { - logger.error( - `[localEmbeddingService] Worker restart limit (${MAX_RESTART_ATTEMPTS}) reached. ` + - 'Embedding service is offline. Reload the app to retry.', - ); - return; - } - const backoff = getRestartBackoffMs(); - if (backoff > 0) { - setTimeout(() => startWorkerHealthCheck(), backoff); - } else { - startWorkerHealthCheck(); - } -} - -function startWorkerHealthCheck(): void { - if (typeof Worker === 'undefined') return; - pingTimer = setInterval(() => { - const w = workerInstance; - if (!w) return; - w.postMessage({ type: 'WORKER_PING' }); - pongTimeoutTimer = setTimeout(() => { - // QNBS-v3: No pong received — worker is dead or hung; restart with backoff. - restartWorker(); - }, PONG_TIMEOUT_MS); - - const pongHandler = (ev: MessageEvent<{ type?: string }>) => { - if (ev.data?.type === 'WORKER_PONG') { - // QNBS-v3: Successful pong resets the restart counter. - restartAttemptCount = 0; - if (pongTimeoutTimer) { - clearTimeout(pongTimeoutTimer); - pongTimeoutTimer = null; - } - w.removeEventListener('message', pongHandler); - } - }; - w.addEventListener('message', pongHandler); - }, PING_INTERVAL_MS); -} - -function getWorker(): Worker { - if (!workerInstance) { - workerInstance = new Worker(new URL('../../workers/inference.worker.ts', import.meta.url), { - type: 'module', - }); - startWorkerHealthCheck(); - } - return workerInstance; -} - function truncate(text: string): string { if (text.length <= MAX_INPUT_CHARS) return text; // QNBS-v3: Silent truncation — warn in dev builds only. @@ -129,30 +37,15 @@ function l2Normalize(vec: number[]): EmbeddingVector { return new Float32Array(vec.map((v) => v / magnitude)); } -function postToWorker( - task: string, - modelId: string, - input: string, -): Promise<{ ok: boolean; result?: number[]; error?: string }> { - return new Promise((resolve) => { - const messageId = `emb-${Date.now()}-${Math.random().toString(36).slice(2, 7)}`; - const worker = getWorker(); - - const handler = (event: MessageEvent) => { - const data = event.data as { - messageId: string; - ok: boolean; - result?: number[]; - error?: string; - }; - if (data.messageId !== messageId) return; - worker.removeEventListener('message', handler); - resolve(data); - }; - - worker.addEventListener('message', handler); - worker.postMessage({ messageId, task, modelId, input }); - }); +async function requestEmbedding(task: string, modelId: string, input: string): Promise { + const bus = await ensureInferencePool(); + if (!bus) throw new Error('WorkerBus v2 unavailable'); + const handle = bus.enqueue<{ task: string; modelId: string; input: string }, number[]>( + 'inference.embed', + { task, modelId, input }, + { capabilities: ['inference.embed'] }, + ); + return handle.result; } export async function embedText(text: string): Promise { @@ -167,11 +60,8 @@ export async function embedText(text: string): Promise { return cached; } - const response = await postToWorker('feature-extraction', EMBEDDING_MODEL, truncated); - if (!response.ok || !response.result) { - throw new Error(response.error ?? 'embedding failed'); - } - const vector = l2Normalize(response.result); + const raw = await requestEmbedding('feature-extraction', EMBEDDING_MODEL, truncated); + const vector = l2Normalize(raw); // QNBS-v3: Evict the oldest (first) entry when at capacity before inserting. if (embeddingCache.size >= EMBEDDING_CACHE_MAX) { @@ -206,12 +96,3 @@ export function cosineSimilarity(a: EmbeddingVector, b: EmbeddingVector): number // Both vectors are already L2-normalised, so cosine = dot product return Math.max(-1, Math.min(1, dot)); } - -// QNBS-v3: Used in testing to reset the worker instance without affecting production code. -export function _resetWorkerForTest(): void { - clearWorkerHealthTimers(); - if (workerInstance) { - workerInstance.terminate(); - workerInstance = null; - } -} diff --git a/services/workerBusManager.ts b/services/workerBusManager.ts index 2e973c24..ede6b7d6 100644 --- a/services/workerBusManager.ts +++ b/services/workerBusManager.ts @@ -87,7 +87,8 @@ async function doInitWorkerBus(): Promise { poolId: 'inference', capabilities: ['inference.text', 'inference.embed'], options: { - maxWorkers: MAX_WORKERS_INFERENCE, + // QNBS-v3: [Capped below MAX_WORKERS_INFERENCE — each replica loads its own transformers.js pipeline (no cross-replica cache sharing), so 4 concurrent workers could mean 4x the model memory footprint under a burst.] + maxWorkers: Math.min(2, MAX_WORKERS_INFERENCE), minWorkers: MIN_WORKERS, idleTimeoutMs: WORKER_IDLE_TIMEOUT_MS, workerScript: inferenceUrl, diff --git a/tests/unit/localEmbeddingService.test.ts b/tests/unit/localEmbeddingService.test.ts index 314b1f06..45310d42 100644 --- a/tests/unit/localEmbeddingService.test.ts +++ b/tests/unit/localEmbeddingService.test.ts @@ -1,99 +1,96 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; - -// QNBS-v3: Self-resolving MockWorker — when postMessage is called it immediately fires -// all registered 'message' handlers with a configurable response. This avoids -// async timing issues with concurrent Promise.all batches in embedBatch. -let mockResponse: { ok: boolean; result?: number[]; error?: string } = { - ok: true, - result: [0.5, 0.5], -}; - -let postMessageCalls: Array<{ messageId: string; input: string; task: string }> = []; - -class MockWorker { - private handlers: Array<(e: MessageEvent) => void> = []; - - addEventListener(_type: string, handler: (e: MessageEvent) => void) { - this.handlers.push(handler); - } - - removeEventListener(_type: string, handler: (e: MessageEvent) => void) { - const idx = this.handlers.indexOf(handler); - if (idx >= 0) this.handlers.splice(idx, 1); - } - - postMessage(msg: { messageId: string; input: string; task: string }) { - postMessageCalls.push(msg); - // QNBS-v3: Respond synchronously so Promise.all batches resolve without microtask tricks. - const response = { ...mockResponse, messageId: msg.messageId }; - const event = { data: response } as MessageEvent; - for (const handler of [...this.handlers]) { - handler(event); - } - } +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +// QNBS-v3: [localEmbeddingService now routes through WorkerBus v2 — mock ensureInferencePool() instead of the global Worker constructor.] + +const { mockEnqueue, mockEnsureInferencePool } = vi.hoisted(() => ({ + mockEnqueue: vi.fn(), + mockEnsureInferencePool: vi.fn(), +})); - terminate() {} +vi.mock('../../services/workerBusManager', () => ({ + ensureInferencePool: mockEnsureInferencePool, +})); + +const { clearEmbeddingCache, cosineSimilarity, embedBatch, embedText } = await import( + '../../services/ai/localEmbeddingService' +); + +function makeHandle(result: Promise) { + return { taskId: 't1', result, progress: (async function* () {})(), cancel: vi.fn() }; } -vi.stubGlobal('Worker', MockWorker); +function makeBus() { + return { enqueue: mockEnqueue }; +} -import { - _resetWorkerForTest, - cosineSimilarity, - embedBatch, - embedText, -} from '../../services/ai/localEmbeddingService'; +let requestCalls: Array<{ task: string; modelId: string; input: string }> = []; +let nextResult: number[] = [0.5, 0.5]; beforeEach(() => { - _resetWorkerForTest(); - mockResponse = { ok: true, result: [0.5, 0.5] }; - postMessageCalls = []; -}); - -afterEach(() => { - _resetWorkerForTest(); + vi.clearAllMocks(); + clearEmbeddingCache(); + requestCalls = []; + nextResult = [0.5, 0.5]; + mockEnsureInferencePool.mockResolvedValue(makeBus()); + mockEnqueue.mockImplementation((_taskType: string, payload: unknown) => { + requestCalls.push(payload as { task: string; modelId: string; input: string }); + return makeHandle(Promise.resolve(nextResult)); + }); }); // ─── embedText ────────────────────────────────────────────────────────────── describe('embedText', () => { - it('returns a Float32Array for a successful worker response', async () => { - mockResponse = { ok: true, result: [0.5, 0.5] }; + it('returns a Float32Array for a successful task result', async () => { + nextResult = [0.5, 0.5]; const vec = await embedText('Hello world'); expect(vec).toBeInstanceOf(Float32Array); expect(vec.length).toBe(2); }); it('L2-normalises the returned vector (magnitude ≈ 1)', async () => { - mockResponse = { ok: true, result: [3, 4] }; // magnitude = 5 → normalised to [0.6, 0.8] + nextResult = [3, 4]; // magnitude = 5 → normalised to [0.6, 0.8] const vec = await embedText('Normalise me'); const magnitude = Math.sqrt(vec[0]! ** 2 + vec[1]! ** 2); expect(magnitude).toBeCloseTo(1, 5); }); - it('throws when worker returns ok:false with error message', async () => { - mockResponse = { ok: false, error: 'OOM' }; + it('propagates the error message when the task rejects', async () => { + mockEnqueue.mockReturnValue(makeHandle(Promise.reject(new Error('OOM')))); await expect(embedText('fail case')).rejects.toThrow('OOM'); }); - it('throws with generic message when error field is absent', async () => { - mockResponse = { ok: false }; - await expect(embedText('fail case 2')).rejects.toThrow('embedding failed'); + it('throws WorkerBus v2 unavailable without enqueuing when the pool is unavailable', async () => { + mockEnsureInferencePool.mockResolvedValue(null); + await expect(embedText('fail case 2')).rejects.toThrow('WorkerBus v2 unavailable'); + expect(mockEnqueue).not.toHaveBeenCalled(); }); it('truncates input silently when text exceeds 512 chars', async () => { await embedText('a'.repeat(600)); - expect(postMessageCalls[0]?.input.length).toBe(512); + expect(requestCalls[0]?.input.length).toBe(512); }); it('does not truncate input at exactly 512 chars', async () => { await embedText('b'.repeat(512)); - expect(postMessageCalls[0]?.input.length).toBe(512); + expect(requestCalls[0]?.input.length).toBe(512); }); - it('sends feature-extraction task to worker', async () => { + it('enqueues inference.embed with the feature-extraction task and inference.embed capability', async () => { await embedText('test'); - expect(postMessageCalls[0]?.task).toBe('feature-extraction'); + expect(requestCalls[0]?.task).toBe('feature-extraction'); + expect(mockEnqueue).toHaveBeenCalledWith( + 'inference.embed', + expect.anything(), + expect.objectContaining({ capabilities: ['inference.embed'] }), + ); + }); + + it('returns the cached vector on a second call for the same text without re-enqueuing', async () => { + const first = await embedText('cache me'); + const second = await embedText('cache me'); + expect(second).toBe(first); + expect(mockEnqueue).toHaveBeenCalledTimes(1); }); }); @@ -101,7 +98,7 @@ describe('embedText', () => { describe('embedBatch', () => { it('returns an array of Float32Arrays, one per input', async () => { - mockResponse = { ok: true, result: [0.5, 0.5] }; + nextResult = [0.5, 0.5]; const results = await embedBatch(['a', 'b', 'c']); expect(results).toHaveLength(3); for (const vec of results) { @@ -110,16 +107,16 @@ describe('embedBatch', () => { }); it('returns correct count for 9 texts (spans two micro-batches of 8+1)', async () => { - mockResponse = { ok: true, result: [0.1, 0.2] }; + nextResult = [0.1, 0.2]; const texts = Array.from({ length: 9 }, (_, i) => `text-${i}`); const results = await embedBatch(texts); expect(results).toHaveLength(9); }); - it('sends exactly N worker messages for N input texts', async () => { - mockResponse = { ok: true, result: [0.1, 0.2] }; + it('sends exactly N enqueue calls for N input texts', async () => { + nextResult = [0.1, 0.2]; await embedBatch(['x', 'y', 'z']); - expect(postMessageCalls).toHaveLength(3); + expect(requestCalls).toHaveLength(3); }); it('returns empty array for empty input', async () => { From 36c5f7cb553fc1751c4fc65551d683a111ec5acf Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 21:23:14 +0200 Subject: [PATCH 06/10] fix(worker-bus): pool-termination lifecycle, idle-respawn recovery, SQL param binding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three real findings from CodeRabbit's review of #287 (duckdbClient migration onto WorkerBus v2), all confirmed against current code before fixing: 1. CRITICAL — params silently dropped (workers/v2/duckdb.worker.ts). duckdbClient.send() enqueues {sql, params}, but handleQuery/ handleExec only read req.sql and called connection.query(sql) directly, ignoring params entirely and leaving an unbound-query/ SQL-injection surface flagged by static analysis. This was already live: services/ai/telemetryService.ts's writeToDuckDb() passes real bound params ([taskType, backend, modelId, latencyMs, success]) that were being silently discarded. Fixed via DuckDB-WASM prepared statements (connection.prepare(sql) + stmt.query(...params) + stmt.close()) when params is non-empty; unchanged direct connection.query(sql) otherwise. Needed a narrow structural type (PreparableConnection) since tsgo doesn't resolve AsyncDuckDBConnection.prepare() through this package's nested `export *` chain, despite it being present in the package's own .d.ts — same class of tsgo/external-package gap CLAUDE.md already documents for the transformers.js path alias. 2. MAJOR — DuckDB pool worker losing its connection on respawn (services/duckdb/duckdbClient.ts). The duckdb pool's minWorkers:0 lets its worker idle-terminate; a freshly spawned replacement worker's module state has no `connection` until INIT runs again on it, so the next QUERY/EXEC after a respawn failed with "DuckDB not initialized". Fixed by transparently re-sending INIT once and retrying the original call when send() sees that specific error message (bounded to one retry, INIT/SHUTDOWN excluded from the check to avoid any recursive loop). 3. MAJOR — terminatePool() had no lifecycle contract (packages/worker-bus/src/workerBus.ts + services/workerBusManager.ts). Removing a pool didn't settle tasks already routed to it — their result promises would await a RESULT message from a worker that no longer exists. Fixed by tracking task->pool assignment (new taskPools map, set/cleared around runTask's pool-acquire phase) and cancelling matching tasks before deletion. Separately, ensureDuckDbPool() assumed a non-null bus always has every pool, so any call after duckdbClient.terminate() would fail with NO_POOL forever; added WorkerBus.hasPool() and a re-registration path (duckDbPoolOptions()/reRegisterDuckDbPool(), shared with doInitWorkerBus() to avoid option drift between the two registration sites). Co-Authored-By: Claude Sonnet 5 --- README.md | 8 +- packages/worker-bus/src/workerBus.ts | 24 ++++++ packages/worker-bus/tests/workerBus.test.ts | 37 +++++++++ services/duckdb/duckdbClient.ts | 19 ++++- services/workerBusManager.ts | 42 +++++++--- tests/unit/duckdbClient.test.ts | 65 +++++++++++++++ tests/unit/duckdbWorkerHandler.test.ts | 90 +++++++++++++++++++++ tests/unit/workerBusManager.test.ts | 57 +++++++++++-- workers/v2/duckdb.worker.ts | 55 ++++++++++--- 9 files changed, 363 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 0e0d34b9..99dfe67f 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ IndexedDB v8 PWA v3.0 i18n 19 locales — 2849 keys - 5807+ tests / 529 files + 5807+ tests / 531 files Codecov Coverage License MIT CI Status @@ -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/packages/worker-bus/src/workerBus.ts b/packages/worker-bus/src/workerBus.ts index 84b45627..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[], @@ -202,10 +209,14 @@ export class WorkerBus { * 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); } @@ -235,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 74ae557c..f1f9c580 100644 --- a/packages/worker-bus/tests/workerBus.test.ts +++ b/packages/worker-bus/tests/workerBus.test.ts @@ -233,6 +233,43 @@ describe('WorkerBus', () => { 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('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 1315002f..b561fe72 100644 --- a/services/duckdb/duckdbClient.ts +++ b/services/duckdb/duckdbClient.ts @@ -43,6 +43,7 @@ async function send( sql?: string, params?: readonly unknown[], signal?: AbortSignal, + isReinitRetry = false, ): Promise { const messageId = generateMessageId(); const start = Date.now(); @@ -95,7 +96,23 @@ async function send( } catch (err) { const latencyMs = Date.now() - start; if (signal?.aborted) return { messageId, ok: false, error: 'Aborted', latencyMs }; - return { messageId, ok: false, error: errorMessage(err), 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 }; } } diff --git a/services/workerBusManager.ts b/services/workerBusManager.ts index 2e973c24..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 @@ -172,7 +185,12 @@ export async function ensureWebLlmPool(): Promise { * must not silently break analytics. Returns null only if init failed. */ export async function ensureDuckDbPool(): Promise { - if (_bus === null) await initWorkerBus(); + 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; } diff --git a/tests/unit/duckdbClient.test.ts b/tests/unit/duckdbClient.test.ts index c592b26c..c5790a92 100644 --- a/tests/unit/duckdbClient.test.ts +++ b/tests/unit/duckdbClient.test.ts @@ -113,6 +113,71 @@ describe('duckdbClient.query', () => { 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 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', () => { diff --git a/tests/unit/duckdbWorkerHandler.test.ts b/tests/unit/duckdbWorkerHandler.test.ts index b942e2ec..18f8db18 100644 --- a/tests/unit/duckdbWorkerHandler.test.ts +++ b/tests/unit/duckdbWorkerHandler.test.ts @@ -216,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', () => { @@ -236,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/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/v2/duckdb.worker.ts b/workers/v2/duckdb.worker.ts index d7a5880a..56c89b2c 100644 --- a/workers/v2/duckdb.worker.ts +++ b/workers/v2/duckdb.worker.ts @@ -1,15 +1,15 @@ /// -// 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'; 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 +29,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 +58,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(); @@ -98,19 +98,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 }; } From 776ff4e6a6a9a3221424e50255ee734bb3a74474 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Wed, 29 Jul 2026 22:31:06 +0200 Subject: [PATCH 07/10] fix(ai): add the missing ensureInferencePool export that broke prod builds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of both Vercel deployment failures on this branch/PR chain: services/ai/localEmbeddingService.ts imports and calls ensureInferencePool() from workerBusManager.ts, but that function was never actually added there — only ensureDuckDbPool() was. Vitest never caught this because localEmbeddingService.test.ts mocks the entire workerBusManager module (vi.mock(...) replaces it wholesale), so the real file's missing export was invisible to that suite. tsgo's typecheck also passed clean locally for reasons still unclear (worth a follow-up look), but rolldown's stricter static bundling in the actual `pnpm run build:edge` (Vercel's exact build command) correctly failed hard with [MISSING_EXPORT] — reproduced locally with `NODE_OPTIONS=--max-old-space-size=3072 pnpm run build:edge`, which is what actually surfaced this. Added ensureInferencePool(), mirroring ensureDuckDbPool()'s shape (decoupled from enableWorkerBusV2, re-registers the pool if it was removed via terminatePool() while the bus stays alive). Refactored doInitWorkerBus()'s inline 'inference' pool registration to share the same inferencePoolOptions() the new function uses, avoiding the same kind of drift the duckdb pool options already guard against. New tests in workerBusManager.test.ts import the *real* module (only @domain/worker-bus is mocked) — this exact suite would have caught the missing export immediately, unlike the fully-mocked consumer-side test. Verified fixed via a full local `pnpm run build:edge` reproduction of Vercel's exact build command (previously failed with [MISSING_EXPORT], now builds clean). Co-Authored-By: Claude Sonnet 5 --- services/workerBusManager.ts | 56 +++++++++++++++++++++-------- tests/unit/workerBusManager.test.ts | 47 ++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 14 deletions(-) diff --git a/services/workerBusManager.ts b/services/workerBusManager.ts index 7fe74030..b2223c21 100644 --- a/services/workerBusManager.ts +++ b/services/workerBusManager.ts @@ -67,13 +67,35 @@ async function reRegisterDuckDbPool(bus: WorkerBus): Promise { bus.registerPool('duckdb', options.capabilities, options); } +// QNBS-v3: [Shared with reRegisterInferencePool() below, same rationale as duckDbPoolOptions().] +async function inferencePoolOptions() { + const { MAX_WORKERS_INFERENCE, MIN_WORKERS, WORKER_IDLE_TIMEOUT_MS } = await import( + '@domain/worker-bus' + ); + return { + // QNBS-v3: [Capped below MAX_WORKERS_INFERENCE — each replica loads its own transformers.js pipeline (no cross-replica cache sharing), so 4 concurrent workers could mean 4x the model memory footprint under a burst.] + maxWorkers: Math.min(2, MAX_WORKERS_INFERENCE), + minWorkers: MIN_WORKERS, + idleTimeoutMs: WORKER_IDLE_TIMEOUT_MS, + workerScript: new URL('../workers/v2/inference.worker.ts', import.meta.url).href, + capabilities: ['inference.text', 'inference.embed'] as const, + labels: { pool: 'inference', version: 'v2' }, + }; +} + +/** Re-register the 'inference' pool if it was removed via terminatePool() — a no-op if already present. */ +async function reRegisterInferencePool(bus: WorkerBus): Promise { + if (bus.hasPool('inference')) return; + const options = await inferencePoolOptions(); + bus.registerPool('inference', options.capabilities, options); +} + async function doInitWorkerBus(): Promise { try { const { WorkerBus, WorkerRegistry, MAX_WORKERS_INFERENCE, - MIN_WORKERS, WORKER_IDLE_TIMEOUT_MS, CIRCUIT_BREAKER_THRESHOLD, CIRCUIT_BREAKER_RECOVERY_MS, @@ -96,25 +118,15 @@ async function doInitWorkerBus(): Promise { const registry = new WorkerRegistry(); - // 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; // 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; + const inferenceOptions = await inferencePoolOptions(); registry.register({ poolId: 'inference', - capabilities: ['inference.text', 'inference.embed'], - options: { - // QNBS-v3: [Capped below MAX_WORKERS_INFERENCE — each replica loads its own transformers.js pipeline (no cross-replica cache sharing), so 4 concurrent workers could mean 4x the model memory footprint under a burst.] - maxWorkers: Math.min(2, MAX_WORKERS_INFERENCE), - minWorkers: MIN_WORKERS, - idleTimeoutMs: WORKER_IDLE_TIMEOUT_MS, - workerScript: inferenceUrl, - capabilities: ['inference.text', 'inference.embed'], - labels: { pool: 'inference', version: 'v2' }, - }, + capabilities: inferenceOptions.capabilities, + options: inferenceOptions, }); const duckdbOptions = await duckDbPoolOptions(); @@ -195,6 +207,22 @@ export async function ensureDuckDbPool(): Promise { return _bus; } +/** + * Ensure the shared local-inference worker pool is available, initializing the WorkerBus on + * demand. QNBS-v3: mirrors ensureDuckDbPool()/ensureWebLlmPool()'s decoupling from + * `enableWorkerBusV2` — embeddings/NLP were never gated by that flag in the v1 worker they + * replace, so toggling an experimental infra flag off must not silently break RAG/cross-project + * search. Returns null only if init failed. + */ +export async function ensureInferencePool(): Promise { + if (_bus === null) { + await initWorkerBus(); + return _bus; + } + await reRegisterInferencePool(_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/workerBusManager.test.ts b/tests/unit/workerBusManager.test.ts index 151f3717..7781e221 100644 --- a/tests/unit/workerBusManager.test.ts +++ b/tests/unit/workerBusManager.test.ts @@ -190,4 +190,51 @@ describe('workerBusManager', () => { ); }); }); + + describe('ensureInferencePool', () => { + // QNBS-v3: [Imports the real module (only @domain/worker-bus is mocked) — unlike + // localEmbeddingService.test.ts's fully-mocked workerBusManager, this actually + // exercises the export surface. This exact suite would have caught PR #288's + // missing-export bug (ensureInferencePool was imported but never defined).] + it('initializes the bus when not yet running', async () => { + const { ensureInferencePool, isWorkerBusReady } = await import( + '../../services/workerBusManager' + ); + const bus = await ensureInferencePool(); + expect(bus).not.toBeNull(); + expect(isWorkerBusReady()).toBe(true); + }); + + it('does not re-register the pool when it is already present', async () => { + const { initWorkerBus, ensureInferencePool } = await import( + '../../services/workerBusManager' + ); + await initWorkerBus(); + mockRegisterPool.mockClear(); + mockHasPool.mockReturnValue(true); + + await ensureInferencePool(); + + expect(mockHasPool).toHaveBeenCalledWith('inference'); + expect(mockRegisterPool).not.toHaveBeenCalled(); + }); + + it('re-registers the inference pool when the bus is alive but the pool was removed', async () => { + const { initWorkerBus, ensureInferencePool } = await import( + '../../services/workerBusManager' + ); + await initWorkerBus(); + mockRegisterPool.mockClear(); + mockHasPool.mockReturnValue(false); + + const bus = await ensureInferencePool(); + + expect(bus).not.toBeNull(); + expect(mockRegisterPool).toHaveBeenCalledWith( + 'inference', + expect.arrayContaining(['inference.text', 'inference.embed']), + expect.objectContaining({ workerScript: expect.stringContaining('inference.worker') }), + ); + }); + }); }); From 8a63b7c102797a18b0ea04ed54b74ab73b4674e5 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Thu, 30 Jul 2026 03:09:04 +0200 Subject: [PATCH 08/10] fix(worker-bus): guard pool re-registration failures in workerBusManager ensureDuckDbPool() and ensureInferencePool() could reject after _bus was already live because their re-registration helpers awaited a lazy import and called registerPool() without a catch, breaking the documented "null only if init failed" contract. Wrap both paths so a failed re-add logs instead of propagating, and add a maxWorkers:2 assertion to the inference re-registration test so the pool cap regression would be caught. Addresses CodeRabbit review comments on PR #288. Co-Authored-By: Claude Sonnet 5 --- services/workerBusManager.ts | 21 ++++++++++++++++++--- tests/unit/workerBusManager.test.ts | 5 ++++- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/services/workerBusManager.ts b/services/workerBusManager.ts index b2223c21..caf9a97a 100644 --- a/services/workerBusManager.ts +++ b/services/workerBusManager.ts @@ -202,8 +202,15 @@ export async function ensureDuckDbPool(): Promise { 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); + // 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. Catch so a re-registration failure logs instead of breaking the documented "null only if init failed" contract.] + try { + await reRegisterDuckDbPool(_bus); + } catch (err) { + log.error( + 'Failed to re-register duckdb pool', + err instanceof Error ? err : new Error(String(err)), + ); + } return _bus; } @@ -219,7 +226,15 @@ export async function ensureInferencePool(): Promise { await initWorkerBus(); return _bus; } - await reRegisterInferencePool(_bus); + // QNBS-v3: [catch so a re-registration failure logs instead of breaking the documented "null only if init failed" contract, mirroring ensureDuckDbPool().] + try { + await reRegisterInferencePool(_bus); + } catch (err) { + log.error( + 'Failed to re-register inference pool', + err instanceof Error ? err : new Error(String(err)), + ); + } return _bus; } diff --git a/tests/unit/workerBusManager.test.ts b/tests/unit/workerBusManager.test.ts index 7781e221..6055afe7 100644 --- a/tests/unit/workerBusManager.test.ts +++ b/tests/unit/workerBusManager.test.ts @@ -233,7 +233,10 @@ describe('workerBusManager', () => { expect(mockRegisterPool).toHaveBeenCalledWith( 'inference', expect.arrayContaining(['inference.text', 'inference.embed']), - expect.objectContaining({ workerScript: expect.stringContaining('inference.worker') }), + expect.objectContaining({ + maxWorkers: 2, + workerScript: expect.stringContaining('inference.worker'), + }), ); }); }); From a8686c4d84a9b53e2a3e5fd31205eadbae1ad7e7 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Thu, 30 Jul 2026 03:44:53 +0200 Subject: [PATCH 09/10] test(worker-bus): close Codecov patch-coverage gap on PR #288 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior CodeRabbit fix (8a63b7c1) added try/catch guards in ensureDuckDbPool()/ensureInferencePool() so a pool re-registration failure logs instead of propagating, but no test exercised the catch path — Codecov flagged 4 missing lines (83.33% patch coverage). Adds one test per function that forces registerPool() to throw and asserts the live bus is still returned and the failure is logged via services/logger.ts. Co-Authored-By: Claude Sonnet 5 --- tests/unit/workerBusManager.test.ts | 53 +++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/tests/unit/workerBusManager.test.ts b/tests/unit/workerBusManager.test.ts index 6055afe7..5f311d08 100644 --- a/tests/unit/workerBusManager.test.ts +++ b/tests/unit/workerBusManager.test.ts @@ -66,6 +66,18 @@ vi.mock('../../services/legacyWorkerBusAdapter', () => ({ }), })); +const { mockLogError } = vi.hoisted(() => ({ mockLogError: vi.fn() })); + +vi.mock('../../services/logger', () => ({ + createLogger: () => ({ + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: mockLogError, + withContext: vi.fn(), + }), +})); + describe('workerBusManager', () => { beforeEach(() => { vi.resetModules(); @@ -76,6 +88,7 @@ describe('workerBusManager', () => { mockHasPool.mockClear(); mockHasPool.mockReturnValue(true); mockInstall.mockClear(); + mockLogError.mockClear(); }); afterEach(() => { @@ -189,6 +202,25 @@ describe('workerBusManager', () => { expect.objectContaining({ workerScript: expect.stringContaining('duckdb.worker') }), ); }); + + it('logs and returns the live bus instead of rejecting when re-registration fails', async () => { + // QNBS-v3: [Regression guard for the documented "null only if init failed" contract — a + // re-registration failure must not propagate out of ensureDuckDbPool.] + const { initWorkerBus, ensureDuckDbPool } = await import('../../services/workerBusManager'); + await initWorkerBus(); + mockHasPool.mockReturnValue(false); + mockRegisterPool.mockImplementationOnce(() => { + throw new Error('registerPool boom'); + }); + + const bus = await ensureDuckDbPool(); + + expect(bus).not.toBeNull(); + expect(mockLogError).toHaveBeenCalledWith( + 'Failed to re-register duckdb pool', + expect.any(Error), + ); + }); }); describe('ensureInferencePool', () => { @@ -239,5 +271,26 @@ describe('workerBusManager', () => { }), ); }); + + it('logs and returns the live bus instead of rejecting when re-registration fails', async () => { + // QNBS-v3: [Regression guard for the documented "null only if init failed" contract — a + // re-registration failure must not propagate out of ensureInferencePool.] + const { initWorkerBus, ensureInferencePool } = await import( + '../../services/workerBusManager' + ); + await initWorkerBus(); + mockHasPool.mockReturnValue(false); + mockRegisterPool.mockImplementationOnce(() => { + throw new Error('registerPool boom'); + }); + + const bus = await ensureInferencePool(); + + expect(bus).not.toBeNull(); + expect(mockLogError).toHaveBeenCalledWith( + 'Failed to re-register inference pool', + expect.any(Error), + ); + }); }); }); From b3e5d1af2570ed3e96d7b42bd97b3392b742ec21 Mon Sep 17 00:00:00 2001 From: qnbs <155236708+qnbs@users.noreply.github.com> Date: Thu, 30 Jul 2026 03:59:34 +0200 Subject: [PATCH 10/10] docs(worker-bus): add QNBS-v3 rationale comments per CodeRabbit outside-diff-range review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses CodeRabbit's 2 still-valid findings on PR #288: the logger mock (why it's isolated from the real StructuredLogger) and the maxWorkers:2 regression assertion (why 2, not just what). The other 3 flagged spots (reRegisterInferencePool, the inference registry.register block, and the mockLogError.mockClear() line) mirror already-uncommented sibling code in the same files (reRegisterDuckDbPool, the duckdb registry.register block, 6 other .mockClear() calls) — commenting only the new instance would be less consistent, not more, so those are left as-is. Co-Authored-By: Claude Sonnet 5 --- tests/unit/workerBusManager.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/unit/workerBusManager.test.ts b/tests/unit/workerBusManager.test.ts index 5f311d08..2a17c768 100644 --- a/tests/unit/workerBusManager.test.ts +++ b/tests/unit/workerBusManager.test.ts @@ -68,6 +68,7 @@ vi.mock('../../services/legacyWorkerBusAdapter', () => ({ const { mockLogError } = vi.hoisted(() => ({ mockLogError: vi.fn() })); +// QNBS-v3: mocked separately so re-registration-failure tests can assert on log.error without asserting on the real StructuredLogger's console/IDB side effects vi.mock('../../services/logger', () => ({ createLogger: () => ({ debug: vi.fn(), @@ -262,6 +263,7 @@ describe('workerBusManager', () => { const bus = await ensureInferencePool(); expect(bus).not.toBeNull(); + // QNBS-v3: asserts the memory-safety cap (below MAX_WORKERS_INFERENCE) survives re-registration, not just initial registration expect(mockRegisterPool).toHaveBeenCalledWith( 'inference', expect.arrayContaining(['inference.text', 'inference.embed']),