findings(E-1,E-2,E-3,E-6): ngspice service generations + credit-bounded event transport
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 <noreply@anthropic.com>
This commit is contained in:
parent
3de00bb15b
commit
0b8e7186d4
6 changed files with 1591 additions and 114 deletions
|
|
@ -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<void> {
|
||||
await page.addInitScript((workerSrc: string) => {
|
||||
export interface NgspiceHarnessWatchdogs {
|
||||
bootTimeoutMs?: number;
|
||||
responseTimeoutMs?: number;
|
||||
}
|
||||
|
||||
export async function installNgspiceServiceStub(
|
||||
page: Page,
|
||||
watchdogs: NgspiceHarnessWatchdogs = {},
|
||||
): Promise<void> {
|
||||
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<Worker> | null = null;
|
||||
const pending = new Map<number, (res: any) => void>();
|
||||
let nextId = 1;
|
||||
interface WorkerSlot {
|
||||
generation: number;
|
||||
worker?: Worker;
|
||||
workerUrl?: string;
|
||||
failed: boolean;
|
||||
ready: Promise<WorkerSlot>;
|
||||
bootTimer?: ReturnType<typeof setTimeout>;
|
||||
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<typeof setTimeout>;
|
||||
}
|
||||
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<typeof setTimeout>;
|
||||
}
|
||||
interface AppliedGenerationReceipt {
|
||||
generation: number;
|
||||
t: number;
|
||||
}
|
||||
interface AppliedGenerationWaiter {
|
||||
after: number;
|
||||
resolve: (receipt: AppliedGenerationReceipt) => void;
|
||||
reject: (reason?: unknown) => void;
|
||||
timer: ReturnType<typeof setTimeout>;
|
||||
}
|
||||
interface CancellableReceipt<T> extends Promise<T> {
|
||||
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<number, PendingRequest>();
|
||||
let nextId = 1;
|
||||
let maxPending = 0;
|
||||
let bootMessageErrorArmed = false;
|
||||
let runtimeMessageErrorThreshold: number | null = null;
|
||||
const retiredGenerations: number[] = [];
|
||||
let nextRequestSequence = 1;
|
||||
const requestWaiters = new Set<RequestWaiter>();
|
||||
const appliedGenerations: AppliedGenerationReceipt[] = [];
|
||||
const appliedGenerationWaiters = new Set<AppliedGenerationWaiter>();
|
||||
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 = <T>(
|
||||
promise: Promise<T>,
|
||||
cancel: (reason?: string) => void,
|
||||
): CancellableReceipt<T> => {
|
||||
const receipt = promise as CancellableReceipt<T>;
|
||||
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<RequestSummary> => {
|
||||
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<RequestSummary>((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<AppliedGenerationReceipt> => {
|
||||
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<AppliedGenerationReceipt>((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<Worker> => {
|
||||
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<WorkerSlot> => {
|
||||
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<never>((_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<void>((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<void>((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<any> => {
|
||||
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<void> {
|
|||
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 });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
196
tests/tools/ngspice-worker-batch-unit.ts
Normal file
196
tests/tools/ngspice-worker-batch-unit.ts
Normal file
|
|
@ -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<void>;
|
||||
};
|
||||
|
||||
async function createWorkerHarness(): Promise<WorkerHarness> {
|
||||
const frames: Frame[] = [];
|
||||
let messageHandler!: (event: { data: unknown }) => Promise<void>;
|
||||
const module: Record<string, any> = {
|
||||
init: () => 0,
|
||||
circ: () => 0,
|
||||
command: () => 0,
|
||||
getVecInfo: () => ({ found: false }),
|
||||
curPlot: () => "",
|
||||
allPlots: () => [],
|
||||
allVecs: () => [],
|
||||
running: () => false,
|
||||
cmInputPath: () => undefined,
|
||||
};
|
||||
const workerGlobal: Record<string, unknown> = {
|
||||
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<void> {
|
||||
for (const frame of harness.frames) {
|
||||
if (!frame.eventSequence) continue;
|
||||
await harness.message({
|
||||
eventAck: {
|
||||
sequence: frame.eventSequence,
|
||||
bytes: frame.eventBytes,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
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;
|
||||
});
|
||||
Loading…
Reference in a new issue