From 0b8e7186d47d23d7f7b5cfda994cce687b171e9c Mon Sep 17 00:00:00 2001 From: Istvan Matejcsok <119620946+matejcsok-ee@users.noreply.github.com> Date: Wed, 19 Aug 2026 15:48:41 +0200 Subject: [PATCH] findings(E-1,E-2,E-3,E-6): ngspice service generations + credit-bounded event transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adapted from codex/asyncify-execution-owner-core 3753320. Service side mirrors the occ-service shape (E-1 watchdogs, E-2 fail-all + boot-death fix — onerror now rejects the in-flight boot waiter instead of stranding it, E-3 onmessageerror terminal, Blob URL revoked, per-generation evtQueue cleared on retirement). E-6 transport bounds (worker hunks re-applied inside the emscripten-6 em-pthread else-branch — the codex file predates that split, so this is a re-application, not a cherry-pick): - batch cut at 512 lines / 1 MiB exact JSON-UTF-8 bytes, measured before a line is retained; a single line > 1 MiB flushes the accepted prefix then stops the event stream terminally (never retained); - posting gated by a 64-frame / 8 MiB unacked credit window; each frame carries { eventSequence, eventBytes } and is released only by an exact { sequence, bytes } ack; any mismatched ack is terminal; - the service mirrors the same 64-frame / 8 MiB bound on its pre-handler queue, acks after handing a frame to __ngspiceOnEvent, and retires the generation on invalid credit; { fatal } frames retire the worker. Tests: ngspice-service.test.ts (11, ported) — watchdogs, crash/bootError/ decode-fault settlement + recovery, out-of-order ids, sync postMessage throw, stale-generation event drops, fatal-frame retirement. tests/tools/ ngspice-worker-batch-unit.ts (node:vm over the production worker source; `npm run ngspice:worker-batch`) — bounded ordered chunks, byte-pressure flush, 100k-chunk credit storm, over-limit line, exact ack lease. e2e harness twin updated to speak the ack protocol (adds __ngspiceServiceTestHooks). Co-Authored-By: Claude Fable 5 --- tests/kicad/utils/ngspice-service.ts | 613 ++++++++++++++++-- tests/package.json | 3 +- tests/tools/ngspice-worker-batch-unit.ts | 196 ++++++ .../src/wasm/ngspice-service.test.ts | 434 +++++++++++++ web/standalone/src/wasm/ngspice-service.ts | 287 ++++++-- web/standalone/src/wasm/ngspice-worker.js | 174 ++++- 6 files changed, 1592 insertions(+), 115 deletions(-) create mode 100644 tests/tools/ngspice-worker-batch-unit.ts create mode 100644 web/standalone/src/wasm/ngspice-service.test.ts diff --git a/tests/kicad/utils/ngspice-service.ts b/tests/kicad/utils/ngspice-service.ts index 9b646d4..7b63cbd 100644 --- a/tests/kicad/utils/ngspice-service.ts +++ b/tests/kicad/utils/ngspice-service.ts @@ -21,7 +21,16 @@ import type { Page } from '@playwright/test'; * forwarded to globalThis.__ngspiceOnEvent (the editor client stub's * dispatcher, when integrated) — specs assert live streaming by comparing * event timestamps against run boundaries; - * - request/response summaries are appended to window.__ngspiceLog. + * - request/response summaries are appended to window.__ngspiceLog. Each + * carries the sequence assigned when the request was issued. The test hook + * offers a scan-then-subscribe receipt so a response cannot land in the + * gap between an array scan and listener installation; + * - Worker generations own their Blob URL, boot deadline, response deadlines, + * pending calls, and queued events. Retirement settles and cleans only that + * exact generation; + * - the native simulator publishes its run generation only after the final + * plot, operating-point, and canvas refresh calls. The harness stores an + * atomic scan-then-subscribe receipt for that applied generation. * * The worker fetches ngspice_service.js lazily on the FIRST request — specs * assert the lazy-load boundary by watching network requests. @@ -32,103 +41,534 @@ const NGSPICE_WORKER_SRC = fs.readFileSync( 'web', 'standalone', 'src', 'wasm', 'ngspice-worker.js'), 'utf8'); -export async function installNgspiceServiceStub(page: Page): Promise { - await page.addInitScript((workerSrc: string) => { +export interface NgspiceHarnessWatchdogs { + bootTimeoutMs?: number; + responseTimeoutMs?: number; +} + +export async function installNgspiceServiceStub( + page: Page, + watchdogs: NgspiceHarnessWatchdogs = {}, +): Promise { + const validDeadline = (value: number | undefined, fallback: number, name: string): number => { + const deadline = value ?? fallback; + if (!Number.isSafeInteger(deadline) || deadline < 1) + throw new Error(`${name} must be a positive safe integer`); + return deadline; + }; + const bootTimeoutMs = validDeadline(watchdogs.bootTimeoutMs, 2 * 60_000, 'bootTimeoutMs'); + const responseTimeoutMs = validDeadline( + watchdogs.responseTimeoutMs, + 30 * 60_000, + 'responseTimeoutMs', + ); + + await page.addInitScript((options: { + workerSrc: string; + bootTimeoutMs: number; + responseTimeoutMs: number; + }) => { if ((globalThis as any).ngspiceService) return; + const { workerSrc, bootTimeoutMs, responseTimeoutMs } = options; + const t0 = Date.now(); (window as any).__ngspiceEvents = []; (window as any).__ngspiceLog = []; - let workerP: Promise | null = null; - const pending = new Map void>(); - let nextId = 1; + interface WorkerSlot { + generation: number; + worker?: Worker; + workerUrl?: string; + failed: boolean; + ready: Promise; + bootTimer?: ReturnType; + rejectBoot?: (reason?: unknown) => void; + removeBootListener?: () => void; + /** The exact lifecycle transition used by Worker.onmessageerror. */ + failDecode: () => void; + } + interface PendingRequest { + generation: number; + resolve: (res: any) => void; + timer: ReturnType; + } + interface RequestSummary { + sequence: number; + kind: string; + cmd?: string; + name?: string; + ret?: number; + error?: string; + length?: number; + t: number; + } + interface RequestCriteria { + kind?: string; + name?: string; + minimumLength?: number; + } + interface RequestWaiter { + after: number; + criteria: RequestCriteria; + resolve: (summary: RequestSummary) => void; + reject: (reason?: unknown) => void; + timer: ReturnType; + } + interface AppliedGenerationReceipt { + generation: number; + t: number; + } + interface AppliedGenerationWaiter { + after: number; + resolve: (receipt: AppliedGenerationReceipt) => void; + reject: (reason?: unknown) => void; + timer: ReturnType; + } + interface CancellableReceipt extends Promise { + cancel: (reason?: string) => void; + } - const evtQueue: any[] = []; - const dispatchEvt = (evt: any) => { + const RECEIPT_TIMEOUT_MS = 2 * 60_000; + const MAX_RECEIPT_TIMEOUT_MS = 5 * 60_000; + const MAX_RECEIPT_WAITERS = 128; + + let workerSlot: WorkerSlot | null = null; + let nextGeneration = 1; + const pending = new Map(); + let nextId = 1; + let maxPending = 0; + let bootMessageErrorArmed = false; + let runtimeMessageErrorThreshold: number | null = null; + const retiredGenerations: number[] = []; + let nextRequestSequence = 1; + const requestWaiters = new Set(); + const appliedGenerations: AppliedGenerationReceipt[] = []; + const appliedGenerationWaiters = new Set(); + let disposed = false; + + const validateReceiptTimeout = (timeoutMs: number): string | undefined => { + if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1 + || timeoutMs > MAX_RECEIPT_TIMEOUT_MS) { + return `receipt timeout must be an integer from 1 to ${MAX_RECEIPT_TIMEOUT_MS} ms`; + } + return undefined; + }; + + const cancellable = ( + promise: Promise, + cancel: (reason?: string) => void, + ): CancellableReceipt => { + const receipt = promise as CancellableReceipt; + Object.defineProperty(receipt, 'cancel', { value: cancel }); + return receipt; + }; + + const requestMatches = ( + summary: RequestSummary, + after: number, + criteria: RequestCriteria, + ): boolean => summary.sequence > after + && summary.error === undefined + && (criteria.kind === undefined || summary.kind === criteria.kind) + && (criteria.name === undefined || summary.name === criteria.name) + && (criteria.minimumLength === undefined + || (summary.length ?? -1) >= criteria.minimumLength); + + const rejectRequestWaiter = (waiter: RequestWaiter, reason: Error): void => { + if (!requestWaiters.delete(waiter)) return; + clearTimeout(waiter.timer); + waiter.reject(reason); + }; + + const publishRequestReceipt = (summary: RequestSummary) => { + (window as any).__ngspiceLog.push(summary); + for (const waiter of [...requestWaiters]) { + if (!requestMatches(summary, waiter.after, waiter.criteria)) continue; + requestWaiters.delete(waiter); + clearTimeout(waiter.timer); + waiter.resolve(summary); + } + }; + + const waitForRequestAfter = ( + after: number, + criteria: RequestCriteria, + timeoutMs = RECEIPT_TIMEOUT_MS, + ): CancellableReceipt => { + const rejected = (message: string) => cancellable( + Promise.reject(new Error(message)), + () => undefined, + ); + if (disposed) return rejected('ngspice receipt service was disposed'); + if (!Number.isSafeInteger(after) || after < 0) + return rejected('request checkpoint must be a non-negative integer'); + if (!criteria || typeof criteria !== 'object') + return rejected('request receipt criteria must be an object'); + if (criteria.minimumLength !== undefined + && (!Number.isSafeInteger(criteria.minimumLength) + || criteria.minimumLength < 0)) { + return rejected('minimumLength must be a non-negative integer'); + } + const timeoutError = validateReceiptTimeout(timeoutMs); + if (timeoutError) return rejected(timeoutError); + const log = (window as any).__ngspiceLog as RequestSummary[]; + const existing = log.find((entry) => + requestMatches(entry, after, criteria)); + if (existing) return cancellable(Promise.resolve(existing), () => undefined); + if (requestWaiters.size >= MAX_RECEIPT_WAITERS) + return rejected('ngspice request receipt waiter capacity exceeded'); + + let waiter!: RequestWaiter; + const promise = new Promise((resolve, reject) => { + waiter = { + after, + criteria: { ...criteria }, + resolve, + reject, + timer: setTimeout(() => rejectRequestWaiter( + waiter, + new Error(`ngspice request receipt timed out after ${timeoutMs} ms`), + ), timeoutMs), + }; + requestWaiters.add(waiter); + }); + return cancellable(promise, (reason = 'canceled') => rejectRequestWaiter( + waiter, + new Error(`ngspice request receipt ${reason}`), + )); + }; + + const rejectAppliedGenerationWaiter = ( + waiter: AppliedGenerationWaiter, + reason: Error, + ): void => { + if (!appliedGenerationWaiters.delete(waiter)) return; + clearTimeout(waiter.timer); + waiter.reject(reason); + }; + + const waitForAppliedGenerationAfter = ( + after: number, + timeoutMs = RECEIPT_TIMEOUT_MS, + ): CancellableReceipt => { + const rejected = (message: string) => cancellable( + Promise.reject(new Error(message)), + () => undefined, + ); + if (disposed) return rejected('ngspice receipt service was disposed'); + if (!Number.isSafeInteger(after) || after < 0) + return rejected('applied generation checkpoint must be a non-negative integer'); + const timeoutError = validateReceiptTimeout(timeoutMs); + if (timeoutError) return rejected(timeoutError); + const existing = appliedGenerations.find((entry) => entry.generation > after); + if (existing) return cancellable(Promise.resolve(existing), () => undefined); + if (appliedGenerationWaiters.size >= MAX_RECEIPT_WAITERS) + return rejected('ngspice applied-generation waiter capacity exceeded'); + + let waiter!: AppliedGenerationWaiter; + const promise = new Promise((resolve, reject) => { + waiter = { + after, + resolve, + reject, + timer: setTimeout(() => rejectAppliedGenerationWaiter( + waiter, + new Error(`ngspice applied generation timed out after ${timeoutMs} ms`), + ), timeoutMs), + }; + appliedGenerationWaiters.add(waiter); + }); + return cancellable(promise, (reason = 'canceled') => rejectAppliedGenerationWaiter( + waiter, + new Error(`ngspice applied-generation receipt ${reason}`), + )); + }; + + const previousAppliedHook = (globalThis as any).__pcbjamNgspiceFinalRefreshApplied; + const publishAppliedGeneration = (generation: number): void => { + if (disposed) return; + if (!Number.isSafeInteger(generation) || generation < 1) { + console.error(`[TEST-NGSPICE] ignored invalid applied generation ${generation}`); + return; + } + const previousGeneration = appliedGenerations.length + ? appliedGenerations[appliedGenerations.length - 1]!.generation + : 0; + if (generation <= previousGeneration) { + console.error(`[TEST-NGSPICE] ignored stale applied generation ${generation}`); + return; + } + const receipt = { generation, t: Date.now() - t0 }; + appliedGenerations.push(receipt); + for (const waiter of [...appliedGenerationWaiters]) { + if (generation <= waiter.after) continue; + appliedGenerationWaiters.delete(waiter); + clearTimeout(waiter.timer); + waiter.resolve(receipt); + } + if (typeof previousAppliedHook === 'function') previousAppliedHook(generation); + }; + (globalThis as any).__pcbjamNgspiceFinalRefreshApplied = publishAppliedGeneration; + + interface QueuedEventFrame { + generation: number; + evt: any; + sequence: number; + bytes: number; + } + const MAX_QUEUED_EVENT_FRAMES = 64; + const MAX_QUEUED_EVENT_BYTES = 8 * 1024 * 1024; + const evtQueue: QueuedEventFrame[] = []; + let evtQueueBytes = 0; + const ackEvent = (slot: WorkerSlot, frame: QueuedEventFrame): boolean => { + if (slot.failed || workerSlot !== slot || !slot.worker) return false; + try { + slot.worker.postMessage({ + eventAck: { sequence: frame.sequence, bytes: frame.bytes }, + }); + return true; + } catch (error) { + retireWorker( + slot, + `ngspice_service event acknowledgment failed: ${String(error)}`, + ); + return false; + } + }; + const dispatchEvt = ( + slot: WorkerSlot, + evt: any, + sequence: number, + bytes: number, + ) => { + if (slot.failed || workerSlot !== slot) return; + if (!Number.isSafeInteger(sequence) || sequence < 1 + || !Number.isSafeInteger(bytes) || bytes < 1 + || bytes > MAX_QUEUED_EVENT_BYTES) { + retireWorker(slot, 'ngspice_service sent invalid event-frame credit'); + return; + } + const frame = { generation: slot.generation, evt, sequence, bytes }; (window as any).__ngspiceEvents.push({ ...evt, t: Date.now() - t0 }); const handler = (globalThis as any).__ngspiceOnEvent; if (handler) { - while (evtQueue.length) handler(evtQueue.shift()); + while (evtQueue.length) { + const queued = evtQueue.shift()!; + evtQueueBytes -= queued.bytes; + if (queued.generation !== slot.generation) continue; + handler(queued.evt); + if (!ackEvent(slot, queued)) return; + } handler(evt); + ackEvent(slot, frame); } else { - evtQueue.push(evt); + if (evtQueue.length >= MAX_QUEUED_EVENT_FRAMES + || evtQueueBytes > MAX_QUEUED_EVENT_BYTES - bytes) { + retireWorker(slot, 'ngspice_service event-frame queue exceeded credit'); + return; + } + evtQueue.push(frame); + evtQueueBytes += bytes; } }; - const failAllPending = (why: string) => { - for (const [, resolve] of pending) resolve({ error: why }); - pending.clear(); + const failPending = (generation: number, why: string) => { + for (const [id, request] of pending) { + if (request.generation !== generation) continue; + pending.delete(id); + clearTimeout(request.timer); + request.resolve({ error: why }); + } }; - const ensureWorker = (): Promise => { - if (!workerP) { - workerP = (async () => { + const retireWorker = (slot: WorkerSlot, why: string) => { + if (slot.failed) return; + slot.failed = true; + retiredGenerations.push(slot.generation); + if (slot.bootTimer !== undefined) { + clearTimeout(slot.bootTimer); + slot.bootTimer = undefined; + } + slot.removeBootListener?.(); + slot.removeBootListener = undefined; + failPending(slot.generation, why); + for (let i = evtQueue.length - 1; i >= 0; --i) { + if (evtQueue[i]!.generation === slot.generation) { + evtQueueBytes -= evtQueue[i]!.bytes; + evtQueue.splice(i, 1); + } + } + if (workerSlot === slot) workerSlot = null; + try { slot.worker?.terminate(); } catch { /* already gone */ } + if (slot.workerUrl) { + try { URL.revokeObjectURL(slot.workerUrl); } catch { /* cleanup only */ } + slot.workerUrl = undefined; + } + const reject = slot.rejectBoot; + slot.rejectBoot = undefined; + reject?.(new Error(why)); + }; + + const ensureWorker = (): Promise => { + if (disposed) return Promise.reject(new Error('ngspice service was disposed')); + if (!workerSlot) { + const slot = { + generation: nextGeneration++, + failed: false, + } as WorkerSlot; + workerSlot = slot; + + const bootDeadline = new Promise((_resolve, reject) => { + slot.rejectBoot = reject; + slot.bootTimer = setTimeout(() => { + if (slot.failed || workerSlot !== slot) return; + const why = `ngspice_service boot timed out after ${bootTimeoutMs} ms`; + console.log(`[TEST-NGSPICE] ${why} — resetting service`); + retireWorker(slot, why); + }, bootTimeoutMs); + }); + + const boot = (async () => { const glue = new URL('ngspice_service.js', window.location.href).href; console.log(`[TEST-NGSPICE] booting ngspice_service from ${glue}`); - const worker = new Worker(URL.createObjectURL(new Blob( + slot.workerUrl = URL.createObjectURL(new Blob( [`self.NGSPICE_GLUE_URL = ${JSON.stringify(glue)};\n`, workerSrc], - { type: 'text/javascript' }))); + { type: 'text/javascript' })); + const worker = new Worker(slot.workerUrl); + slot.worker = worker; + worker.onmessage = (e) => { + if (slot.failed || workerSlot !== slot) return; const data = e.data ?? {}; - if (data.evt) { dispatchEvt(data.evt); return; } + if (data.fatal) { + failWorker(`event stream failure: ${String(data.fatal)}`); + return; + } + if (data.evt) { + dispatchEvt( + slot, + data.evt, + data.eventSequence, + data.eventBytes, + ); + return; + } if (typeof data.id !== 'number') return; - const resolve = pending.get(data.id); - if (resolve) { pending.delete(data.id); resolve(data.res); } + const request = pending.get(data.id); + if (request?.generation === slot.generation) { + pending.delete(data.id); + clearTimeout(request.timer); + request.resolve(data.res); + } }; - worker.onerror = (e) => { - console.log(`[TEST-NGSPICE] worker error: ${e.message} — resetting service`); - failAllPending(`ngspice_service crashed: ${e.message}`); - workerP = null; - try { worker.terminate(); } catch { /* already gone */ } + const failWorker = (detail: string) => { + const why = `ngspice_service crashed: ${detail}`; + console.log(`[TEST-NGSPICE] worker error: ${detail} — resetting service`); + retireWorker(slot, why); }; - // Legible boot: bound the handshake and surface worker - // death — the bare version hung to the spec timeout with - // zero evidence (occ-service.ts has the same guard). - await new Promise((resolve, reject) => { - const fail = (msg: string) => { - clearTimeout(timer); - reject(new Error(msg)); - }; - const timer = setTimeout( - () => fail('[TEST-NGSPICE] ngspice_service boot timed out after ' - + '60s (no ready/bootError from the worker)'), 60000); + worker.onerror = (e) => failWorker(e.message || 'worker error'); + slot.failDecode = () => failWorker('message decode failed'); + worker.onmessageerror = slot.failDecode; + if (bootMessageErrorArmed) { + bootMessageErrorArmed = false; + // Synthetic dispatch on Worker is engine-dependent. + // Call the same transition as the real event handler + // after rejectBoot is installed below. + queueMicrotask(() => slot.failDecode()); + } + await new Promise((resolve) => { const onFirst = (e: MessageEvent) => { if (e.data?.ready) { - worker.removeEventListener('message', onFirst); - clearTimeout(timer); + if (slot.bootTimer !== undefined) { + clearTimeout(slot.bootTimer); + slot.bootTimer = undefined; + } + slot.removeBootListener?.(); + slot.removeBootListener = undefined; + slot.rejectBoot = undefined; resolve(); } else if (e.data?.bootError) { - fail(`[TEST-NGSPICE] ngspice_service bootError: ${e.data.bootError}`); + const why = `ngspice_service boot failed: ${String(e.data.bootError)}`; + retireWorker(slot, why); } }; worker.addEventListener('message', onFirst); - worker.addEventListener('error', (e: any) => fail( - `[TEST-NGSPICE] ngspice_service worker error: ${e?.message ?? e} ` - + `(${e?.filename ?? '?'}:${e?.lineno ?? '?'})`)); - worker.addEventListener('messageerror', () => fail( - '[TEST-NGSPICE] ngspice_service worker messageerror (structured clone failed)')); + slot.removeBootListener = () => worker.removeEventListener('message', onFirst); }); + + if (slot.failed || workerSlot !== slot) { + throw new Error('ngspice_service worker retired during boot'); + } console.log('[TEST-NGSPICE] ngspice_service ready'); - return worker; - })().catch((e) => { workerP = null; throw e; }); + return slot; + })(); + slot.ready = Promise.race([boot, bootDeadline]).catch((e) => { + retireWorker(slot, `ngspice_service unavailable: ${String(e)}`); + throw e; + }); } - return workerP; + return workerSlot.ready; + }; + + const post = (slot: WorkerSlot, req: any): Promise => { + const worker = slot.worker; + if (!worker || slot.failed || workerSlot !== slot) { + return Promise.resolve({ error: 'ngspice_service worker is unavailable' }); + } + const id = nextId++; + return new Promise((resolve) => { + const timer = setTimeout(() => { + if (pending.get(id)?.generation !== slot.generation) return; + const why = `ngspice_service response timed out after ${responseTimeoutMs} ms`; + console.log(`[TEST-NGSPICE] ${why} — resetting service`); + retireWorker(slot, why); + }, responseTimeoutMs); + pending.set(id, { generation: slot.generation, resolve, timer }); + let generationPending = 0; + for (const request of pending.values()) { + if (request.generation === slot.generation) generationPending++; + } + maxPending = Math.max(maxPending, generationPending); + try { + worker.postMessage({ id, req }); + if (runtimeMessageErrorThreshold !== null + && generationPending >= runtimeMessageErrorThreshold) { + runtimeMessageErrorThreshold = null; + slot.failDecode(); + } + } catch (error) { + pending.delete(id); + clearTimeout(timer); + resolve({ error: `ngspice_service request failed: ${String(error)}` }); + } + }); }; const request = async (req: any) => { - let worker: Worker; + // Assign at issue time, before worker boot or posting. A late + // response from a request issued before a new simulation's + // checkpoint can therefore never satisfy that simulation. + if (disposed) return { error: 'ngspice service was disposed' }; + const requestSequence = nextRequestSequence++; + let slot: WorkerSlot; try { - worker = await ensureWorker(); + slot = await ensureWorker(); } catch (e) { - return { error: `ngspice_service unavailable: ${e}` }; + const res = { error: `ngspice_service unavailable: ${e}` }; + publishRequestReceipt({ + sequence: requestSequence, + kind: String(req.kind), + cmd: req.cmd, + name: req.name, + error: res.error, + t: Date.now() - t0, + }); + return res; } - const id = nextId++; - const res: any = await new Promise((resolve) => { - pending.set(id, resolve); - worker.postMessage({ id, req }); - }); - (window as any).__ngspiceLog.push({ + const res: any = await post(slot, req); + publishRequestReceipt({ + sequence: requestSequence, kind: req.kind, cmd: req.cmd, name: req.name, @@ -140,6 +580,63 @@ export async function installNgspiceServiceStub(page: Page): Promise { return res; }; + const dispose = (): void => { + if (disposed) return; + disposed = true; + if (workerSlot) retireWorker(workerSlot, 'ngspice service was disposed'); + for (const waiter of [...requestWaiters]) { + rejectRequestWaiter(waiter, new Error('ngspice request receipt canceled by teardown')); + } + for (const waiter of [...appliedGenerationWaiters]) { + rejectAppliedGenerationWaiter( + waiter, + new Error('ngspice applied-generation receipt canceled by teardown'), + ); + } + if ((globalThis as any).__pcbjamNgspiceFinalRefreshApplied + === publishAppliedGeneration) { + (globalThis as any).__pcbjamNgspiceFinalRefreshApplied = previousAppliedHook; + } + }; + + (globalThis as any).__ngspiceServiceTestHooks = { + requestCheckpoint() { + return nextRequestSequence - 1; + }, + waitForRequestAfter, + appliedGenerationCheckpoint() { + return appliedGenerations.length + ? appliedGenerations[appliedGenerations.length - 1]!.generation + : 0; + }, + waitForAppliedGenerationAfter, + dispose, + messageErrorDuringNextBoot() { + bootMessageErrorArmed = true; + }, + messageErrorWhenPendingAtLeast(count: number) { + if (!Number.isSafeInteger(count) || count < 1) + throw new Error('pending threshold must be a positive safe integer'); + runtimeMessageErrorThreshold = count; + }, + snapshot() { + return { + activeGeneration: workerSlot?.generation ?? null, + pending: pending.size, + maxPending, + retiredGenerations: [...retiredGenerations], + bootFaultArmed: bootMessageErrorArmed, + runtimeFaultArmed: runtimeMessageErrorThreshold !== null, + lastRequestSequence: nextRequestSequence - 1, + requestReceiptWaiters: requestWaiters.size, + appliedGenerations: appliedGenerations.map((entry) => entry.generation), + appliedGenerationWaiters: appliedGenerationWaiters.size, + disposed, + }; + }, + }; + (globalThis as any).ngspiceService = { request }; - }, NGSPICE_WORKER_SRC); + window.addEventListener('pagehide', dispose, { once: true }); + }, { workerSrc: NGSPICE_WORKER_SRC, bootTimeoutMs, responseTimeoutMs }); } diff --git a/tests/package.json b/tests/package.json index 6d0423c..020030e 100644 --- a/tests/package.json +++ b/tests/package.json @@ -29,7 +29,8 @@ "3d:check:parity": "tsx tools/screenshots/compare-dirs.ts --old 3d-regression/baseline --new 3d-regression/output/webgl --out 3d-regression/output/diff/parity --floors 3d-regression/floors.json --level webgl-vs-native --label 3d-parity", "3d:review": "tsx tools/screenshots/compare-dirs.ts --old 3d-regression/baseline --new 3d-regression/output/webgl --out 3d-regression/output/diff/parity-review --floors 3d-regression/floors.json --level webgl-vs-native --label 3d-parity --artifacts always", "3d:test:webgl": "playwright test --project=wx-chromium e2e/3d-webgl.spec.ts", - "tools:contract": "tsx tools/cli-contract.ts" + "tools:contract": "tsx tools/cli-contract.ts", + "ngspice:worker-batch": "tsx tools/ngspice-worker-batch-unit.ts" }, "devDependencies": { "@playwright/test": "^1.62.1", diff --git a/tests/tools/ngspice-worker-batch-unit.ts b/tests/tools/ngspice-worker-batch-unit.ts new file mode 100644 index 0000000..0d563e1 --- /dev/null +++ b/tests/tools/ngspice-worker-batch-unit.ts @@ -0,0 +1,196 @@ +/** + * Behavioral reducer for ngspice-worker.js output batching and transport + * credits. It executes the production worker source in a VM with a synchronous + * fake native module, where queued microtasks cannot hide retained bursts. + */ +import { strict as assert } from "node:assert"; +import { readFileSync } from "node:fs"; +import * as path from "node:path"; +import { fileURLToPath } from "node:url"; +import vm from "node:vm"; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const repo = path.resolve(here, "../.."); +const workerSource = readFileSync( + path.join(repo, "web/standalone/src/wasm/ngspice-worker.js"), + "utf8", +); + +type Frame = { + id?: number; + evt?: { kind: string; lines?: string[] }; + fatal?: string; + res?: { error?: string }; + eventSequence?: number; + eventBytes?: number; +}; + +type WorkerHarness = { + frames: Frame[]; + emit(kind: number, text: string, a: number, b: number): void; + message(data: unknown): Promise; +}; + +async function createWorkerHarness(): Promise { + const frames: Frame[] = []; + let messageHandler!: (event: { data: unknown }) => Promise; + const module: Record = { + init: () => 0, + circ: () => 0, + command: () => 0, + getVecInfo: () => ({ found: false }), + curPlot: () => "", + allPlots: () => [], + allVecs: () => [], + running: () => false, + cmInputPath: () => undefined, + }; + const workerGlobal: Record = { + NGSPICE_GLUE_URL: "https://pcbjam.test/ngspice_service.js", + addEventListener: () => undefined, + }; + Object.defineProperty(workerGlobal, "onmessage", { + set(value) { messageHandler = value as typeof messageHandler; }, + }); + const context = vm.createContext({ + self: workerGlobal, + console, + Blob, + URL, + RangeError, + String, + Number, + Map, + JSON, + Promise, + queueMicrotask, + importScripts: () => undefined, + NgspiceService: () => Promise.resolve(module), + postMessage: (frame: Frame) => frames.push(structuredClone(frame)), + structuredClone, + }); + + vm.runInContext(workerSource, context, { filename: "ngspice-worker.js" }); + await Promise.resolve(); + assert.equal(typeof module.ngspiceEmit, "function"); + assert.equal(typeof messageHandler, "function"); + frames.length = 0; // discard ready + return { + frames, + emit: module.ngspiceEmit, + message: (data) => messageHandler({ data }), + }; +} + +async function acknowledgeAll(harness: WorkerHarness): Promise { + for (const frame of harness.frames) { + if (!frame.eventSequence) continue; + await harness.message({ + eventAck: { + sequence: frame.eventSequence, + bytes: frame.eventBytes, + }, + }); + } +} + +async function main(): Promise { +const bounded = await createWorkerHarness(); +// 1,025 synchronous lines cannot wait for a microtask: 512, 512 flush on the +// line bound, and the final line flushes at the queued microtask. +for (let i = 0; i < 1_025; ++i) bounded.emit(0, `line-${i}`, 0, 0); +assert.deepEqual( + bounded.frames.map((frame) => frame.evt?.lines?.length), + [512, 512], +); +await Promise.resolve(); +assert.deepEqual( + bounded.frames.map((frame) => frame.evt?.lines?.length), + [512, 512, 1], +); +assert.deepEqual( + bounded.frames.flatMap((frame) => frame.evt?.lines ?? []), + Array.from({ length: 1_025 }, (_, i) => `line-${i}`), +); +await acknowledgeAll(bounded); +console.log("ok synchronous output flushes in bounded ordered line chunks"); + +bounded.frames.length = 0; +const wide = "x".repeat(400_000); +bounded.emit(0, `${wide}-0`, 0, 0); +bounded.emit(0, `${wide}-1`, 0, 0); +bounded.emit(0, `${wide}-2`, 0, 0); // crossing line flushes first two +assert.equal(bounded.frames.length, 1); +assert.deepEqual( + bounded.frames[0]!.evt!.lines!.map((line) => line.at(-1)), + ["0", "1"], +); +assert.ok(bounded.frames[0]!.eventBytes! <= 1024 * 1024); +await Promise.resolve(); +assert.equal(bounded.frames[1]!.evt!.lines!.length, 1); +assert.ok(bounded.frames[1]!.eventBytes! <= 1024 * 1024); +await acknowledgeAll(bounded); +console.log("ok UTF-8 byte pressure flushes before retaining the crossing line"); + +const storm = await createWorkerHarness(); +const chunk = "z".repeat(900_000); +for (let i = 0; i < 100_000; ++i) { + try { + storm.emit(0, `${chunk}-${i}`, 0, 0); + } catch { + // Continue attempts deliberately: terminal state must remain inert and + // must never post another retained frame. + } +} +await Promise.resolve(); +const stormEvents = storm.frames.filter((frame) => frame.evt); +assert.ok(stormEvents.length <= 64); +assert.ok( + stormEvents.reduce((sum, frame) => sum + frame.eventBytes!, 0) + <= 8 * 1024 * 1024, +); +assert.equal(storm.frames.filter((frame) => frame.fatal).length, 1); +assert.match(storm.frames.find((frame) => frame.fatal)!.fatal!, /unacknowledged/); +console.log("ok 100,000 synchronous chunk attempts cannot exceed transport credit"); + +const oversize = await createWorkerHarness(); +assert.throws( + () => oversize.emit(0, "y".repeat(1024 * 1024), 0, 0), + /ngspice event line exceeds 1048576 UTF-8 bytes/, +); +assert.deepEqual(oversize.frames, [{ + fatal: "ngspice event line exceeds 1048576 UTF-8 bytes", +}]); +assert.throws( + () => oversize.emit(0, "late", 0, 0), + /ngspice event line exceeds 1048576 UTF-8 bytes/, +); +await oversize.message({ id: 91, req: { kind: "running" } }); +assert.deepEqual(oversize.frames.at(-1), { + id: 91, + res: { error: "ngspice event line exceeds 1048576 UTF-8 bytes" }, +}); +console.log("ok a single over-limit line is never retained and terminalizes requests"); + +const ackMismatch = await createWorkerHarness(); +ackMismatch.emit(2, "", 0, 0); +const credited = ackMismatch.frames[0]!; +await assert.rejects( + ackMismatch.message({ + eventAck: { + sequence: credited.eventSequence, + bytes: credited.eventBytes! + 1, + }, + }), + /acknowledgment did not match an exact frame/, +); +assert.equal(ackMismatch.frames.filter((frame) => frame.fatal).length, 1); +console.log("ok acknowledgment must match the exact sequence and byte lease"); + +console.log("ngspice-worker-batch-unit: all green"); +} + +main().catch((error) => { + console.error(error); + process.exitCode = 1; +}); diff --git a/web/standalone/src/wasm/ngspice-service.test.ts b/web/standalone/src/wasm/ngspice-service.test.ts new file mode 100644 index 0000000..f81a1fb --- /dev/null +++ b/web/standalone/src/wasm/ngspice-service.test.ts @@ -0,0 +1,434 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("./ngspice-worker.js?raw", () => ({ default: "// fake ngspice worker" })); +vi.mock("./wasm-assets", () => ({ + resolveWasmBase: vi.fn(async () => "/wasm"), +})); + +import { + installNgspiceService, + type NgspiceRequest, + type NgspiceResponse, +} from "./ngspice-service"; +import { resolveWasmBase } from "./wasm-assets"; + +const mockedResolveWasmBase = vi.mocked(resolveWasmBase); + +const TEST_BOOT_TIMEOUT_MS = 1_000; +const TEST_RESPONSE_TIMEOUT_MS = 5_000; + +type MessageListener = (event: MessageEvent) => void; + +class FakeWorker { + static instances: FakeWorker[] = []; + + onmessage: MessageListener | null = null; + onerror: ((event: ErrorEvent) => void) | null = null; + onmessageerror: ((event: MessageEvent) => void) | null = null; + readonly postMessage = vi.fn(); + readonly terminate = vi.fn(); + private readonly messageListeners = new Set(); + + constructor() { + FakeWorker.instances.push(this); + } + + addEventListener(type: string, listener: EventListenerOrEventListenerObject): void { + if (type === "message") this.messageListeners.add(listener as MessageListener); + } + + removeEventListener(type: string, listener: EventListenerOrEventListenerObject): void { + if (type === "message") this.messageListeners.delete(listener as MessageListener); + } + + emitMessage(data: unknown): void { + const event = { data } as MessageEvent; + this.onmessage?.(event); + for (const listener of [...this.messageListeners]) listener(event); + } + + emitError(message: string): void { + this.onerror?.({ message } as ErrorEvent); + } + + emitMessageError(): void { + this.onmessageerror?.({} as MessageEvent); + } +} + +const commandRequest = (cmd = "run"): NgspiceRequest => ({ kind: "command", cmd }); + +const service = () => { + const installed = globalThis.ngspiceService; + if (!installed) throw new Error("ngspice service was not installed"); + return installed; +}; + +async function waitForWorker(index: number): Promise { + await vi.waitFor(() => expect(FakeWorker.instances.length).toBeGreaterThan(index)); + return FakeWorker.instances[index]!; +} + +async function readyRequest( + workerIndex: number, + requestBody: NgspiceRequest = commandRequest(), +): Promise<{ worker: FakeWorker; request: Promise; id: number }> { + const request = service().request(requestBody); + const worker = await waitForWorker(workerIndex); + worker.emitMessage({ ready: true }); + await vi.waitFor(() => expect(worker.postMessage).toHaveBeenCalledTimes(1)); + const [{ id }] = worker.postMessage.mock.calls[0] as [{ id: number }]; + return { worker, request, id }; +} + +describe("ngspice service worker lifetime", () => { + beforeEach(() => { + FakeWorker.instances = []; + mockedResolveWasmBase.mockReset(); + mockedResolveWasmBase.mockResolvedValue("/wasm"); + vi.stubGlobal("window", { location: { href: "https://pcbjam.test/editor" } }); + vi.stubGlobal("Worker", FakeWorker); + let nextBlob = 1; + vi.spyOn(URL, "createObjectURL").mockImplementation( + () => `blob:ngspice-test-${nextBlob++}`, + ); + vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => undefined); + delete globalThis.ngspiceService; + delete globalThis.__ngspiceOnEvent; + installNgspiceService(vi.fn(), { + bootTimeoutMs: TEST_BOOT_TIMEOUT_MS, + responseTimeoutMs: TEST_RESPONSE_TIMEOUT_MS, + }); + }); + + afterEach(() => { + delete globalThis.ngspiceService; + delete globalThis.__ngspiceOnEvent; + vi.useRealTimers(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it("bounds a never-ready generation, retires it, and boots a fresh worker", async () => { + vi.useFakeTimers(); + + const timedOut = service().request(commandRequest("never ready")); + await vi.advanceTimersByTimeAsync(0); + const deadWorker = FakeWorker.instances[0]!; + expect(deadWorker).toBeDefined(); + + await vi.advanceTimersByTimeAsync(TEST_BOOT_TIMEOUT_MS); + await expect(timedOut).resolves.toEqual({ + error: expect.stringContaining( + `ngspice_service boot timed out after ${TEST_BOOT_TIMEOUT_MS} ms`, + ), + }); + expect(deadWorker.terminate).toHaveBeenCalledTimes(1); + expect(URL.revokeObjectURL).toHaveBeenCalledWith("blob:ngspice-test-1"); + expect(vi.getTimerCount()).toBe(0); + + // The old listener was removed and the slot is retired. A late ready frame + // cannot resurrect this generation. + deadWorker.emitMessage({ ready: true }); + + const recovered = service().request(commandRequest("fresh")); + await vi.advanceTimersByTimeAsync(0); + const freshWorker = FakeWorker.instances[1]!; + expect(freshWorker).toBeDefined(); + freshWorker.emitMessage({ ready: true }); + await vi.advanceTimersByTimeAsync(0); + const [{ id }] = freshWorker.postMessage.mock.calls[0] as [{ id: number }]; + freshWorker.emitMessage({ id, res: { ret: 0 } }); + + await expect(recovered).resolves.toEqual({ ret: 0 }); + expect(freshWorker.terminate).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("bounds delivery resolution and never creates a late retired worker", async () => { + vi.useFakeTimers(); + let releaseBase!: (base: string) => void; + mockedResolveWasmBase.mockImplementationOnce( + () => new Promise((resolve) => { + releaseBase = resolve; + }), + ); + + const timedOut = service().request(commandRequest("resolve forever")); + await vi.advanceTimersByTimeAsync(0); + expect(FakeWorker.instances).toHaveLength(0); + + await vi.advanceTimersByTimeAsync(TEST_BOOT_TIMEOUT_MS); + await expect(timedOut).resolves.toEqual({ + error: expect.stringContaining( + `ngspice_service boot timed out after ${TEST_BOOT_TIMEOUT_MS} ms`, + ), + }); + expect(FakeWorker.instances).toHaveLength(0); + expect(vi.getTimerCount()).toBe(0); + + // Delivery may still finish because the underlying fetch is not abortable + // here. Its retired generation must not create a Worker or replace the + // fresh slot which the next exact request owns. + releaseBase("/stale-wasm"); + await vi.advanceTimersByTimeAsync(0); + expect(FakeWorker.instances).toHaveLength(0); + + const recovered = service().request(commandRequest("fresh")); + await vi.advanceTimersByTimeAsync(0); + const freshWorker = FakeWorker.instances[0]!; + expect(freshWorker).toBeDefined(); + freshWorker.emitMessage({ ready: true }); + await vi.advanceTimersByTimeAsync(0); + const [{ id }] = freshWorker.postMessage.mock.calls[0] as [{ id: number }]; + freshWorker.emitMessage({ id, res: { ret: 0 } }); + + await expect(recovered).resolves.toEqual({ ret: 0 }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("retires a ready-but-silent generation and settles all concurrent ids", async () => { + vi.useFakeTimers(); + + const first = service().request(commandRequest("silent one")); + await vi.advanceTimersByTimeAsync(0); + const deadWorker = FakeWorker.instances[0]!; + deadWorker.emitMessage({ ready: true }); + await vi.advanceTimersByTimeAsync(0); + + const second = service().request(commandRequest("silent two")); + await vi.advanceTimersByTimeAsync(0); + expect(deadWorker.postMessage).toHaveBeenCalledTimes(2); + const [{ id: firstId }] = deadWorker.postMessage.mock.calls[0] as [ + { id: number }, + ]; + const [{ id: secondId }] = deadWorker.postMessage.mock.calls[1] as [ + { id: number }, + ]; + expect(firstId).not.toBe(secondId); + + await vi.advanceTimersByTimeAsync(TEST_RESPONSE_TIMEOUT_MS); + const timeout = + `ngspice_service response timed out after ${TEST_RESPONSE_TIMEOUT_MS} ms`; + await expect(first).resolves.toEqual({ error: timeout }); + await expect(second).resolves.toEqual({ error: timeout }); + expect(deadWorker.terminate).toHaveBeenCalledTimes(1); + expect(URL.revokeObjectURL).toHaveBeenCalledWith("blob:ngspice-test-1"); + expect(vi.getTimerCount()).toBe(0); + + const recovered = service().request(commandRequest("recovered")); + await vi.advanceTimersByTimeAsync(0); + const freshWorker = FakeWorker.instances[1]!; + freshWorker.emitMessage({ ready: true }); + await vi.advanceTimersByTimeAsync(0); + const [{ id: freshId }] = freshWorker.postMessage.mock.calls[0] as [ + { id: number }, + ]; + let recoveredSettled = false; + const observed = recovered.then((response) => { + recoveredSettled = true; + return response; + }); + + // Even a late old-generation callback carrying the current numeric id is + // ignored. The replacement remains live until its own Worker answers. + deadWorker.emitMessage({ id: freshId, res: { ret: 99 } }); + await Promise.resolve(); + expect(recoveredSettled).toBe(false); + expect(freshWorker.terminate).not.toHaveBeenCalled(); + + freshWorker.emitMessage({ id: freshId, res: { ret: 0 } }); + await expect(observed).resolves.toEqual({ ret: 0 }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("settles a first request when the worker crashes before ready and retries", async () => { + const firstRequest = service().request(commandRequest("first")); + const firstWorker = await waitForWorker(0); + + firstWorker.emitError("boot trap"); + + await expect(firstRequest).resolves.toEqual({ + error: expect.stringContaining("ngspice_service crashed: boot trap"), + }); + expect(firstWorker.terminate).toHaveBeenCalledTimes(1); + + const retry = await readyRequest(1, commandRequest("retry")); + retry.worker.emitMessage({ id: retry.id, res: { ret: 0 } }); + await expect(retry.request).resolves.toEqual({ ret: 0 }); + }); + + it("fails every exact pending request on a runtime crash and ignores stale callbacks", async () => { + const first = await readyRequest(0, commandRequest("one")); + const alsoPending = service().request(commandRequest("two")); + await vi.waitFor(() => expect(first.worker.postMessage).toHaveBeenCalledTimes(2)); + + first.worker.emitError("wasm trap"); + + await expect(first.request).resolves.toEqual({ + error: "ngspice_service crashed: wasm trap", + }); + await expect(alsoPending).resolves.toEqual({ + error: "ngspice_service crashed: wasm trap", + }); + expect(first.worker.terminate).toHaveBeenCalledTimes(1); + + const second = await readyRequest(1, commandRequest("recovered")); + let secondSettled = false; + const observedSecond = second.request.then((response) => { + secondSettled = true; + return response; + }); + + first.worker.emitMessage({ id: second.id, res: { ret: 99 } }); + first.worker.emitError("late old error"); + await Promise.resolve(); + expect(secondSettled).toBe(false); + expect(second.worker.terminate).not.toHaveBeenCalled(); + + second.worker.emitMessage({ id: second.id, res: { ret: 0 } }); + await expect(observedSecond).resolves.toEqual({ ret: 0 }); + }); + + it("keeps requests concurrent and correlates out-of-order responses by id", async () => { + const first = await readyRequest(0, commandRequest("slow")); + const secondRequest = service().request(commandRequest("fast")); + await vi.waitFor(() => expect(first.worker.postMessage).toHaveBeenCalledTimes(2)); + const [{ id: secondId }] = first.worker.postMessage.mock.calls[1] as [ + { id: number }, + ]; + + first.worker.emitMessage({ id: secondId, res: { ret: 2 } }); + await expect(secondRequest).resolves.toEqual({ ret: 2 }); + + let firstSettled = false; + const observedFirst = first.request.then((response) => { + firstSettled = true; + return response; + }); + await Promise.resolve(); + expect(firstSettled).toBe(false); + + first.worker.emitMessage({ id: first.id, res: { ret: 1 } }); + await expect(observedFirst).resolves.toEqual({ ret: 1 }); + }); + + it("turns bootError into a response and keeps the next generation retryable", async () => { + const failedRequest = service().request(commandRequest()); + const failedWorker = await waitForWorker(0); + + failedWorker.emitMessage({ bootError: "initialization failed" }); + + await expect(failedRequest).resolves.toEqual({ + error: expect.stringContaining( + "ngspice_service boot failed: initialization failed", + ), + }); + expect(failedWorker.terminate).toHaveBeenCalledTimes(1); + + const retry = await readyRequest(1); + retry.worker.emitMessage({ id: retry.id, res: { ret: 0 } }); + await expect(retry.request).resolves.toEqual({ ret: 0 }); + }); + + it("fails all boot and runtime waiters on decode errors, then recovers", async () => { + const bootRequest = service().request(commandRequest("boot-one")); + const alsoBooting = service().request(commandRequest("boot-two")); + const bootWorker = await waitForWorker(0); + bootWorker.emitMessageError(); + await expect(bootRequest).resolves.toEqual({ + error: expect.stringContaining("ngspice_service crashed: message decode failed"), + }); + await expect(alsoBooting).resolves.toEqual({ + error: expect.stringContaining("ngspice_service crashed: message decode failed"), + }); + + const runtime = await readyRequest(1, commandRequest("runtime")); + const alsoPending = service().request(commandRequest("also-runtime")); + await vi.waitFor(() => expect(runtime.worker.postMessage).toHaveBeenCalledTimes(2)); + runtime.worker.emitMessageError(); + await expect(runtime.request).resolves.toEqual({ + error: "ngspice_service crashed: message decode failed", + }); + await expect(alsoPending).resolves.toEqual({ + error: "ngspice_service crashed: message decode failed", + }); + + const recovered = await readyRequest(2, commandRequest("recovered")); + recovered.worker.emitMessage({ id: recovered.id, res: { ret: 0 } }); + await expect(recovered.request).resolves.toEqual({ ret: 0 }); + }); + + it("settles a synchronous postMessage failure without leaking its pending id", async () => { + const failedRequest = service().request(commandRequest("bad post")); + const worker = await waitForWorker(0); + worker.postMessage.mockImplementationOnce(() => { + throw new DOMException("cannot clone", "DataCloneError"); + }); + worker.emitMessage({ ready: true }); + + await expect(failedRequest).resolves.toEqual({ + error: expect.stringContaining("ngspice_service request failed: DataCloneError"), + }); + + const retry = service().request(commandRequest("good post")); + await vi.waitFor(() => expect(worker.postMessage).toHaveBeenCalledTimes(2)); + const [{ id }] = worker.postMessage.mock.calls[1] as [{ id: number }]; + worker.emitMessage({ id, res: { ret: 0 } }); + await expect(retry).resolves.toEqual({ ret: 0 }); + + // A late reply for the failed request cannot consume the recovered request. + const [{ id: failedId }] = worker.postMessage.mock.calls[0] as [{ id: number }]; + expect(id).not.toBe(failedId); + }); + + it("drops queued events and late event callbacks from a retired generation", async () => { + const first = await readyRequest(0); + first.worker.emitMessage({ + evt: { kind: "char", lines: ["old"] }, + eventSequence: 1, + eventBytes: 31, + }); + first.worker.emitError("worker died"); + await expect(first.request).resolves.toEqual({ + error: "ngspice_service crashed: worker died", + }); + + const events: string[] = []; + globalThis.__ngspiceOnEvent = (event) => events.push(event.lines?.[0] ?? event.kind); + + const second = await readyRequest(1); + first.worker.emitMessage({ + evt: { kind: "char", lines: ["late old"] }, + eventSequence: 2, + eventBytes: 36, + }); + second.worker.emitMessage({ + evt: { kind: "char", lines: ["new"] }, + eventSequence: 1, + eventBytes: 31, + }); + second.worker.emitMessage({ id: second.id, res: { ret: 0 } }); + + await expect(second.request).resolves.toEqual({ ret: 0 }); + expect(events).toEqual(["new"]); + }); + + it("retires a worker whose bounded event stream reports a fatal line", async () => { + const first = await readyRequest(0, commandRequest("oversize output")); + first.worker.emitMessage({ + fatal: "ngspice event line exceeds 1048576 UTF-8 bytes", + }); + + await expect(first.request).resolves.toEqual({ + error: "ngspice_service crashed: event stream failure: " + + "ngspice event line exceeds 1048576 UTF-8 bytes", + }); + expect(first.worker.terminate).toHaveBeenCalledTimes(1); + expect(URL.revokeObjectURL).toHaveBeenCalledWith("blob:ngspice-test-1"); + + const recovered = await readyRequest(1, commandRequest("fresh generation")); + recovered.worker.emitMessage({ id: recovered.id, res: { ret: 0 } }); + await expect(recovered.request).resolves.toEqual({ ret: 0 }); + }); +}); diff --git a/web/standalone/src/wasm/ngspice-service.ts b/web/standalone/src/wasm/ngspice-service.ts index 2c8b2b1..357a3e2 100644 --- a/web/standalone/src/wasm/ngspice-service.ts +++ b/web/standalone/src/wasm/ngspice-service.ts @@ -83,107 +83,304 @@ export function ngspiceWorkerBlobParts(glueHref: string): string[] { ]; } -export function installNgspiceService(log: (msg: string) => void): void { +export interface NgspiceServiceWatchdogs { + /** Maximum time from the first request until a new generation announces `ready`. */ + bootTimeoutMs?: number; + /** Maximum time for any one request in a ready generation to answer. */ + responseTimeoutMs?: number; +} + +// These are last-resort failure bounds, not normal scheduling deadlines. +// SPICE startup and foreground simulations can be expensive on slow devices, +// so production defaults deliberately leave a large margin. +export const NGSPICE_BOOT_TIMEOUT_MS = 2 * 60_000; +export const NGSPICE_RESPONSE_TIMEOUT_MS = 30 * 60_000; + +export function installNgspiceService( + log: (msg: string) => void, + watchdogs: NgspiceServiceWatchdogs = {}, +): void { if (globalThis.ngspiceService) return; + const bootTimeoutMs = + watchdogs.bootTimeoutMs ?? NGSPICE_BOOT_TIMEOUT_MS; + const responseTimeoutMs = + watchdogs.responseTimeoutMs ?? NGSPICE_RESPONSE_TIMEOUT_MS; + + interface WorkerSlot { + generation: number; + worker?: Worker; + workerUrl?: string; + failed: boolean; + ready: Promise; + bootTimer?: ReturnType; + rejectBoot?: (reason?: unknown) => void; + removeBootListener?: () => void; + } + + interface PendingRequest { + generation: number; + resolve: (res: NgspiceResponse) => void; + timer: ReturnType; + } + let nextId = 1; - const pending = new Map void>(); - let workerP: Promise | null = null; + let nextGeneration = 1; + const pending = new Map(); + let workerSlot: WorkerSlot | null = null; // Events can arrive before the client stub installs __ngspiceOnEvent // (the handler comes with the first editor-side ngSpice_Init). - const evtQueue: NgspiceEvent[] = []; - const dispatchEvt = (evt: NgspiceEvent) => { + interface QueuedEventFrame { + generation: number; + evt: NgspiceEvent; + sequence: number; + bytes: number; + } + const MAX_QUEUED_EVENT_FRAMES = 64; + const MAX_QUEUED_EVENT_BYTES = 8 * 1024 * 1024; + const evtQueue: QueuedEventFrame[] = []; + let evtQueueBytes = 0; + const ackEvent = (slot: WorkerSlot, frame: QueuedEventFrame): boolean => { + if (slot.failed || workerSlot !== slot || !slot.worker) return false; + try { + slot.worker.postMessage({ + eventAck: { sequence: frame.sequence, bytes: frame.bytes }, + }); + return true; + } catch (error) { + retireWorker(slot, `ngspice_service event acknowledgment failed: ${String(error)}`); + return false; + } + }; + const dispatchEvt = ( + slot: WorkerSlot, + evt: NgspiceEvent, + sequence: number, + bytes: number, + ) => { + if (slot.failed || workerSlot !== slot) return; + if (!Number.isSafeInteger(sequence) || sequence < 1 + || !Number.isSafeInteger(bytes) || bytes < 1 + || bytes > MAX_QUEUED_EVENT_BYTES) { + retireWorker(slot, "ngspice_service sent invalid event-frame credit"); + return; + } + const frame = { generation: slot.generation, evt, sequence, bytes }; const handler = globalThis.__ngspiceOnEvent; if (handler) { - while (evtQueue.length) handler(evtQueue.shift()!); + while (evtQueue.length) { + const queued = evtQueue.shift()!; + evtQueueBytes -= queued.bytes; + if (queued.generation !== slot.generation) continue; + handler(queued.evt); + if (!ackEvent(slot, queued)) return; + } handler(evt); + ackEvent(slot, frame); } else { - evtQueue.push(evt); + if (evtQueue.length >= MAX_QUEUED_EVENT_FRAMES + || evtQueueBytes > MAX_QUEUED_EVENT_BYTES - bytes) { + retireWorker(slot, "ngspice_service event-frame queue exceeded credit"); + return; + } + evtQueue.push(frame); + evtQueueBytes += bytes; } }; - const failAllPending = (why: string) => { - for (const [, resolve] of pending) resolve({ error: why }); - pending.clear(); + const failPending = (generation: number, why: string): void => { + for (const [id, request] of pending) { + if (request.generation !== generation) continue; + pending.delete(id); + clearTimeout(request.timer); + request.resolve({ error: why }); + } }; - const ensureWorker = (): Promise => { - if (!workerP) { - workerP = (async () => { + const retireWorker = (slot: WorkerSlot, why: string): void => { + if (slot.failed) return; + slot.failed = true; + if (slot.bootTimer !== undefined) { + clearTimeout(slot.bootTimer); + slot.bootTimer = undefined; + } + slot.removeBootListener?.(); + slot.removeBootListener = undefined; + failPending(slot.generation, why); + for (let i = evtQueue.length - 1; i >= 0; --i) { + if (evtQueue[i]!.generation === slot.generation) { + evtQueueBytes -= evtQueue[i]!.bytes; + evtQueue.splice(i, 1); + } + } + if (workerSlot === slot) workerSlot = null; + try { + slot.worker?.terminate(); + } catch { + /* already gone */ + } + if (slot.workerUrl) { + try { + URL.revokeObjectURL(slot.workerUrl); + } catch { + /* URL cleanup must not prevent exact wait settlement */ + } + slot.workerUrl = undefined; + } + const reject = slot.rejectBoot; + slot.rejectBoot = undefined; + reject?.(new Error(why)); + }; + + const ensureWorker = (): Promise => { + if (!workerSlot) { + const slot = { + generation: nextGeneration++, + failed: false, + } as WorkerSlot; + // Publish the generation before its async boot reaches the first await. + // This also lets every continuation test exact slot ownership directly. + workerSlot = slot; + + // The editor is parked for this entire operation, including delivery + // discovery. Start the generation deadline before resolveWasmBase(): a + // hung manifest/CDN lookup must settle the exact wait just like a Worker + // which never announces ready. + const bootDeadline = new Promise((_resolve, reject) => { + slot.rejectBoot = reject; + slot.bootTimer = setTimeout(() => { + if (slot.failed || workerSlot !== slot) return; + const why = + `ngspice_service boot timed out after ${bootTimeoutMs} ms`; + log(`[ngspice] ${why} — resetting service`); + retireWorker(slot, why); + }, bootTimeoutMs); + }); + + const boot = (async () => { const base = await resolveWasmBase("ngspice_service"); + if (slot.failed || workerSlot !== slot) { + throw new Error( + "ngspice_service worker retired during delivery resolution", + ); + } const glue = new URL(`${base}/ngspice_service.js`, window.location.href).href; log(`[ngspice] booting ngspice_service from ${base}`); - const worker = new Worker( - URL.createObjectURL( - new Blob(ngspiceWorkerBlobParts(glue), { type: "text/javascript" }), - ), + slot.workerUrl = URL.createObjectURL( + new Blob(ngspiceWorkerBlobParts(glue), { type: "text/javascript" }), ); + const worker = new Worker(slot.workerUrl); + slot.worker = worker; worker.onmessage = (e) => { + if (slot.failed || workerSlot !== slot) return; const data = e.data ?? {}; + if (data.fatal) { + failWorker(`event stream failure: ${String(data.fatal)}`); + return; + } if (data.evt) { - dispatchEvt(data.evt as NgspiceEvent); + dispatchEvt( + slot, + data.evt as NgspiceEvent, + data.eventSequence, + data.eventBytes, + ); return; } if (typeof data.id !== "number") return; - const resolve = pending.get(data.id); - if (resolve) { + const request = pending.get(data.id); + if (request?.generation === slot.generation) { pending.delete(data.id); - resolve(data.res as NgspiceResponse); + clearTimeout(request.timer); + request.resolve(data.res as NgspiceResponse); } }; // A dead worker (hard ngspice fault) must not strand the editor // suspended in an EM_ASYNC_JS bridge: fail everything in flight and // make the next request boot a fresh worker. - worker.onerror = (e) => { - log(`[ngspice] worker error: ${e.message} — resetting service`); - failAllPending(`ngspice_service crashed: ${e.message}`); - workerP = null; - try { - worker.terminate(); - } catch { - /* already gone */ - } + const failWorker = (detail: string) => { + const why = `ngspice_service crashed: ${detail}`; + log(`[ngspice] worker error: ${detail} — resetting service`); + retireWorker(slot, why); }; + worker.onerror = (e) => failWorker(e.message || "worker error"); + worker.onmessageerror = () => failWorker("message decode failed"); - await new Promise((resolve, reject) => { + await new Promise((resolve) => { const onFirst = (e: MessageEvent) => { if (e.data?.ready) { - worker.removeEventListener("message", onFirst); + if (slot.bootTimer !== undefined) { + clearTimeout(slot.bootTimer); + slot.bootTimer = undefined; + } + slot.removeBootListener?.(); + slot.removeBootListener = undefined; + slot.rejectBoot = undefined; resolve(); } else if (e.data?.bootError) { - reject(new Error(e.data.bootError)); + const why = `ngspice_service boot failed: ${String(e.data.bootError)}`; + retireWorker(slot, why); } }; worker.addEventListener("message", onFirst); + slot.removeBootListener = () => + worker.removeEventListener("message", onFirst); }); + if (slot.failed || workerSlot !== slot) { + throw new Error("ngspice_service worker retired during boot"); + } log("[ngspice] ngspice_service ready"); - return worker; - })().catch((e) => { - workerP = null; // a failed boot must stay retryable + return slot; + })(); + + slot.ready = Promise.race([boot, bootDeadline]).catch((e) => { + // A late failure from a retired generation must not clear a replacement + // which a re-entrant caller has already started. + retireWorker(slot, `ngspice_service unavailable: ${String(e)}`); throw e; }); } - return workerP; + return workerSlot.ready; + }; + + const post = (slot: WorkerSlot, req: NgspiceRequest): Promise => { + const worker = slot.worker; + if (!worker || slot.failed || workerSlot !== slot) { + return Promise.resolve({ error: "ngspice_service worker is unavailable" }); + } + const id = nextId++; + return new Promise((resolve) => { + const timer = setTimeout(() => { + if (pending.get(id)?.generation !== slot.generation) return; + const why = + `ngspice_service response timed out after ${responseTimeoutMs} ms`; + log(`[ngspice] ${why} — resetting service`); + retireWorker(slot, why); + }, responseTimeoutMs); + pending.set(id, { generation: slot.generation, resolve, timer }); + try { + worker.postMessage({ id, req }); + } catch (error) { + pending.delete(id); + clearTimeout(timer); + resolve({ error: `ngspice_service request failed: ${String(error)}` }); + } + }); }; const request = async (req: NgspiceRequest): Promise => { - let worker: Worker; + let slot: WorkerSlot; try { - worker = await ensureWorker(); + slot = await ensureWorker(); } catch (e) { return { error: `ngspice_service unavailable: ${e}` }; } - - const id = nextId++; - return new Promise((resolve) => { - pending.set(id, resolve); - worker.postMessage({ id, req }); - }); + return post(slot, req); }; globalThis.ngspiceService = { request }; diff --git a/web/standalone/src/wasm/ngspice-worker.js b/web/standalone/src/wasm/ngspice-worker.js index 2ef1470..cc16899 100644 --- a/web/standalone/src/wasm/ngspice-worker.js +++ b/web/standalone/src/wasm/ngspice-worker.js @@ -11,11 +11,13 @@ * host -> worker { id, req } req.kind: init | circ | command | * get_vec_info | cur_plot | all_plots | * all_vecs | running | cm_input_path + * host -> worker { eventAck: { sequence, bytes } } releases event credit * worker -> host { id, res } - * worker -> host { evt } unsolicited event stream: + * worker -> host { evt, eventSequence, eventBytes } unsolicited events: * { evt: { kind: "char"|"stat", lines: [...] } } batched console/status * { evt: { kind: "bg", finished: bool } } BGThreadRunning * { evt: { kind: "exit", status, immediate, quit } } ControlledExit + * worker -> host { fatal } terminal event-transport failure * boot: one-shot { ready: true } | { bootError }. */ const GLUE = self.NGSPICE_GLUE_URL; @@ -50,27 +52,169 @@ const modP = NgspiceService({ // --- event stream ----------------------------------------------------------- // char/stat lines are batched per microtask: a chatty simulation can emit // thousands of SendChar lines per second, and one postMessage per line would -// swamp the editor's main thread. bg/exit events flush the pending batch first -// so relative order is preserved. +// swamp the editor's main thread. A native call can emit synchronously for the +// whole request, so a microtask is not itself a memory bound. Measure the exact +// UTF-8 size of JSON.stringify(lines) before retaining each line and flush at +// either limit. bg/exit events flush first so relative order is preserved. +// E-6: batches are cut at MAX_EVENT_BATCH_* and posting is gated by the +// MAX_EVENT_UNACKED_* credit window — the host acks each frame with its exact +// { sequence, bytes } after taking ownership. const EVT_CHAR = 0, EVT_STAT = 1, EVT_BG = 2, EVT_EXIT = 3; -let pendingLines = null; // { kind, lines } of the open batch +const MAX_EVENT_BATCH_LINES = 512; +const MAX_EVENT_BATCH_UTF8_BYTES = 1024 * 1024; +const MAX_EVENT_UNACKED_FRAMES = 64; +const MAX_EVENT_UNACKED_UTF8_BYTES = 8 * 1024 * 1024; +let pendingLines = null; // { kind, lines, utf8Bytes } of the open batch let flushQueued = false; +let eventStreamFailure = null; +let nextEventSequence = 1; +let unackedEventBytes = 0; +const unackedEvents = new Map(); + +function utf8Bytes(text) { + let bytes = 0; + for (let i = 0; i < text.length; ++i) { + const unit = text.charCodeAt(i); + if (unit <= 0x7f) { + bytes += 1; + } else if (unit <= 0x7ff) { + bytes += 2; + } else if (unit >= 0xd800 && unit <= 0xdbff) { + const next = i + 1 < text.length ? text.charCodeAt(i + 1) : 0; + if (next >= 0xdc00 && next <= 0xdfff) { + bytes += 4; + ++i; + } else { + bytes += 3; + } + } else { + bytes += 3; + } + } + return bytes; +} + +// Exact byte count for one JSON string without allocating its serialized form. +// Modern JSON.stringify escapes lone surrogates as \udxxx; paired surrogates +// become one four-byte UTF-8 scalar. Control characters use either a two-byte +// short escape or a six-byte \u00xx escape. +function jsonStringUtf8Bytes(text) { + let bytes = 2; // opening and closing quotes + for (let i = 0; i < text.length; ++i) { + const unit = text.charCodeAt(i); + if (unit === 0x22 || unit === 0x5c) { + bytes += 2; + } else if (unit <= 0x1f) { + bytes += unit === 0x08 || unit === 0x09 || unit === 0x0a + || unit === 0x0c || unit === 0x0d ? 2 : 6; + } else if (unit <= 0x7f) { + bytes += 1; + } else if (unit <= 0x7ff) { + bytes += 2; + } else if (unit >= 0xd800 && unit <= 0xdbff) { + const next = i + 1 < text.length ? text.charCodeAt(i + 1) : 0; + if (next >= 0xdc00 && next <= 0xdfff) { + bytes += 4; + ++i; + } else { + bytes += 6; + } + } else if (unit >= 0xdc00 && unit <= 0xdfff) { + bytes += 6; + } else { + bytes += 3; + } + } + return bytes; +} function flushLines() { flushQueued = false; if (pendingLines) { const batch = pendingLines; pendingLines = null; - postMessage({ evt: { kind: batch.kind, lines: batch.lines } }); + postEvent({ kind: batch.kind, lines: batch.lines }); } } +function stopEventStream(reason) { + if (!eventStreamFailure) { + // Detach the not-yet-transferred batch before reporting terminal state. + // Frames already posted remain bounded by the unacknowledged credit set. + pendingLines = null; + flushQueued = false; + eventStreamFailure = reason; + postMessage({ fatal: reason }); + } + throw new RangeError(eventStreamFailure); +} + +function postEvent(evt) { + const eventBytes = utf8Bytes(JSON.stringify(evt)); + if (unackedEvents.size >= MAX_EVENT_UNACKED_FRAMES + || eventBytes > MAX_EVENT_UNACKED_UTF8_BYTES + || unackedEventBytes > MAX_EVENT_UNACKED_UTF8_BYTES - eventBytes) { + stopEventStream( + `ngspice event transport exceeded ${MAX_EVENT_UNACKED_FRAMES} frames or ` + + `${MAX_EVENT_UNACKED_UTF8_BYTES} unacknowledged UTF-8 bytes`); + } + if (nextEventSequence > Number.MAX_SAFE_INTEGER) { + stopEventStream("ngspice event sequence space exhausted"); + } + + const eventSequence = nextEventSequence++; + unackedEvents.set(eventSequence, eventBytes); + unackedEventBytes += eventBytes; + try { + postMessage({ evt, eventSequence, eventBytes }); + } catch (error) { + unackedEvents.delete(eventSequence); + unackedEventBytes -= eventBytes; + stopEventStream(`ngspice event postMessage failed: ${String(error)}`); + } +} + +function acknowledgeEvent(ack) { + const sequence = ack && ack.sequence; + const bytes = ack && ack.bytes; + const retained = unackedEvents.get(sequence); + if (!Number.isSafeInteger(sequence) || !Number.isSafeInteger(bytes) + || retained === undefined || retained !== bytes + || bytes < 0 || bytes > unackedEventBytes) { + stopEventStream("ngspice event acknowledgment did not match an exact frame"); + } + unackedEvents.delete(sequence); + unackedEventBytes -= bytes; +} + function onEmit(kind, text, a, b) { + if (eventStreamFailure) throw new RangeError(eventStreamFailure); if (kind === EVT_CHAR || kind === EVT_STAT) { const k = kind === EVT_CHAR ? "char" : "stat"; + const line = String(text); + const lineBytes = jsonStringUtf8Bytes(line); + if (lineBytes + 2 > MAX_EVENT_BATCH_UTF8_BYTES) { + // Every earlier line was accepted while capacity existed. Transfer it + // before refusing this line, which is measured but never retained. + flushLines(); + stopEventStream( + `ngspice event line exceeds ${MAX_EVENT_BATCH_UTF8_BYTES} UTF-8 bytes`); + } if (pendingLines && pendingLines.kind !== k) flushLines(); - if (!pendingLines) pendingLines = { kind: k, lines: [] }; - pendingLines.lines.push(text); + if (pendingLines) { + const nextBytes = pendingLines.utf8Bytes + 1 + lineBytes; + if (pendingLines.lines.length >= MAX_EVENT_BATCH_LINES + || nextBytes > MAX_EVENT_BATCH_UTF8_BYTES) { + flushLines(); + } + } + if (!pendingLines) pendingLines = { kind: k, lines: [], utf8Bytes: 2 }; + pendingLines.utf8Bytes += (pendingLines.lines.length ? 1 : 0) + lineBytes; + pendingLines.lines.push(line); + if (pendingLines.lines.length >= MAX_EVENT_BATCH_LINES + || pendingLines.utf8Bytes >= MAX_EVENT_BATCH_UTF8_BYTES) { + flushLines(); + } if (!flushQueued) { flushQueued = true; queueMicrotask(flushLines); @@ -79,9 +223,9 @@ function onEmit(kind, text, a, b) { } flushLines(); if (kind === EVT_BG) { - postMessage({ evt: { kind: "bg", finished: !!a } }); + postEvent({ kind: "bg", finished: !!a }); } else if (kind === EVT_EXIT) { - postMessage({ evt: { kind: "exit", status: a, immediate: !!(b & 1), quit: !!(b & 2) } }); + postEvent({ kind: "exit", status: a, immediate: !!(b & 1), quit: !!(b & 2) }); } } @@ -91,9 +235,17 @@ modP.then((mod) => { }, (e) => postMessage({ bootError: String(e) })); // --- request dispatch ------------------------------------------------------- -onmessage = async (e) => { - const { id, req } = e.data; +self.onmessage = async (e) => { + const { id, req, eventAck } = e.data; + if (eventAck) { + acknowledgeEvent(eventAck); + return; + } if (typeof id !== "number") return; + if (eventStreamFailure) { + postMessage({ id, res: { error: eventStreamFailure } }); + return; + } let res; const transfer = []; try {