diff --git a/templates/full/index.ts b/templates/full/index.ts index e97b609..36db44a 100644 --- a/templates/full/index.ts +++ b/templates/full/index.ts @@ -136,78 +136,132 @@ interface WasmModule { let wasmModule: WasmModule; -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: WasmModule) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: WasmModule = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule(): Promise { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit Promise>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } @@ -381,9 +435,7 @@ export const normalize = awaitInit(async (query: string): Promise => { // Sync versions export function parseSync(query: string): ParseResult { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); // Input validation if (query === null || query === undefined) { @@ -447,9 +499,7 @@ export function parseSync(query: string): ParseResult { } export function parsePlPgSQLSync(query: string): ParseResult { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; @@ -475,9 +525,7 @@ export function parsePlPgSQLSync(query: string): ParseResult { } export function fingerprintSync(query: string): string { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; @@ -502,9 +550,7 @@ export function fingerprintSync(query: string): string { } export function normalizeSync(query: string): string { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; @@ -553,9 +599,7 @@ export const scan = awaitInit(async (query: string): Promise => { }); export function scanSync(query: string): ScanResult { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; diff --git a/templates/index.ts b/templates/index.ts index 4eb9784..dbcffed 100644 --- a/templates/index.ts +++ b/templates/index.ts @@ -124,78 +124,132 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: any) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: any = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } diff --git a/versions/13/src/index.ts b/versions/13/src/index.ts index f39ac71..8a82ff3 100644 --- a/versions/13/src/index.ts +++ b/versions/13/src/index.ts @@ -131,78 +131,132 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: any) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: any = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } diff --git a/versions/13/test/wasm-recovery.test.js b/versions/13/test/wasm-recovery.test.js index 6d16d37..8894778 100644 --- a/versions/13/test/wasm-recovery.test.js +++ b/versions/13/test/wasm-recovery.test.js @@ -22,21 +22,55 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(parseSync('SELECT 1').stmts.length, 1); }); - // Without the 10-failure cap, about 45 overflows on one instance made the next one spin forever. - it('stops sync calls on the stale instance after 10 real overflows in one synchronous loop', async () => { + it('keeps sync calls working through 60 overflows in one synchronous loop', async () => { await loadModule(); - const errors = []; + const results = []; for (let i = 0; i < 60; i++) { - try { - parseSync(DEEP); - } catch (error) { - errors.push(error instanceof RangeError ? 'overflow' : error.message); + for (const query of [DEEP, 'SELECT 1']) { + try { + results.push(parseSync(query).stmts.length); + } catch (error) { + results.push(error instanceof RangeError ? 'overflow' : error.message); + } } } - assert.deepEqual(errors.slice(0, 10), Array(10).fill('overflow')); - for (const message of errors.slice(10)) assert.match(message, /Call `?loadModule\(\)`? first/); - await loadModule(); - assert.equal(parseSync('SELECT 1').stmts.length, 1); + assert.deepEqual(results, Array(60).fill(['overflow', 1]).flat()); + }); + + it('gives the next sync call a fresh instance after the memory limit is hit', () => { + assert.throws(() => parseSync('SELECT 1;\n'.repeat(5000000)), /out of WASM memory/); + assert.equal(parseSync('SELECT 1;\n'.repeat(1000000)).stmts.length, 1000000); + }); + + // Each replaced instance stays allocated until the synchronous run yields. + it('stops replacing after repeated out-of-memory failures in one synchronous run, and recovers after an await', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + const huge = 'SELECT 1;\\n'.repeat(5000000); + const messages = []; + let peak = 0; + for (let i = 0; i < 10; i++) { + try { lib.parseSync(huge); } catch (error) { messages.push(error.message + ' / ' + error.cause?.message); } + peak = Math.max(peak, process.memoryUsage().rss); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + const after = lib.parseSync('SELECT 1').stmts.length; + process.stdout.write('\\nresult=' + JSON.stringify({ messages, peak, after })); + })();`; + const output = execFileSync(process.execPath, ['-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const { messages, peak, after } = JSON.parse(output.match(/^result=(.*)$/m)[1]); + assert.equal(messages.length, 10); + assert.match(messages[0], /out of WASM memory/); + assert.match(messages[1], /out of WASM memory/); + for (const message of messages.slice(2)) { + assert.match(message, /^WASM memory limit hit repeatedly in one synchronous run; yield \(await\) to free memory \/ .*out of WASM memory/); + } + assert.ok(peak < 2.5 * 1024 ** 3, `peak RSS ${Math.round(peak / 1024 ** 2)} MiB`); + assert.equal(after, 1); }); it('frees each replaced instance, so WASM memory stays at one instance', () => { @@ -118,16 +152,22 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(loads, 3); }); - it('lets loadModule() reject on a failed reload and retry on the next call', async () => { + it('names the cause when a fresh instance cannot start, and loadModule() retries', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + const isolated = loadIsolated((real, options) => { + if (++loads === 2 || loads === 3) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await assert.rejects(isolated.loadModule(), /simulated load failure/); + assert.throws(() => isolated.parseSync(DEEP), RangeError); + assert.throws(() => isolated.parseSync('SELECT 1'), (error) => { + assert.match(error.message, /Could not start a fresh WASM instance/); + assert.match(error.cause.message, /simulated instantiate failure/); + return true; + }); await isolated.loadModule(); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); - assert.equal(loads, 3); + assert.equal(loads, 4); }); it('keeps sync calls in the same synchronous loop working after a failure', () => { @@ -145,18 +185,16 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.deepEqual(results, expected); }); - it('retries a failed fresh load on the next failure, for callers that never await', async () => { + it('retries a fresh instance that failed to start on the next sync call', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); - await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(loads, 2); - assert.throws(() => isolated.parseSync(DEEP)); - assert.equal(loads, 3); + const isolated = loadIsolated((real, options) => { + if (++loads === 2) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); + assert.throws(() => isolated.parseSync(DEEP), RangeError); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); }); it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { @@ -201,9 +239,15 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => let used; let frees = 0; let throwNext; - const isolated = loadIsolated(async (real, ...args) => { + const isolated = loadIsolated((real, options) => { const id = ++loads; - const module = await real(...args); + const ready = real(options); + // A replacement instance is built synchronously; the first load is async. + if (typeof options._malloc === 'function') return instrument(options, id) && ready; + return ready.then((module) => instrument(module, id)); + }); + + function instrument(module, id) { const free = module._free; module._free = (ptr) => { frees++; @@ -221,7 +265,7 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => }; } return module; - }); + } async function failOnce(name, query, thrown) { await isolated.loadModule(); @@ -242,73 +286,38 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => assert.equal(exitCodeAfter, 0, `${name} changed process.exitCode`); assert.equal(frees, freesBefore, `${name} freed memory in the instance that threw`); assert.equal(loads, loadsBefore + 1, `${name} did not start a fresh instance`); - const stale = used; const syncName = name.endsWith('Sync') ? name : 'parseSync'; - assert.ok(isolated[syncName](query), `${syncName} failed while the fresh instance loads`); - assert.equal(used, stale, 'a sync call did not run on the stale instance'); + assert.ok(isolated[syncName](query), `${syncName} failed right after the failure`); + assert.equal(used, loads, `${syncName} ran on the instance that threw`); assert.ok(frees > freesBefore, 'a successful call did not free its memory'); - // A sync caller gets the fresh instance once it awaits loadModule(); async calls wait by themselves. - if (name.endsWith('Sync')) await isolated.loadModule(); assert.ok(await isolated[name](query), `${name} failed after recovery`); - assert.equal(used, loads, `${name} ran on the stale instance, not the fresh one`); + assert.equal(used, loads, `${name} ran on the instance that threw`); return caught; } - it('starts one fresh instance for many failures in one synchronous loop', async () => { + it('starts a fresh instance for every failure in one synchronous loop', async () => { await isolated.loadModule(); const loadsBefore = loads; - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 30; i++) { throwNext = new RangeError('Maximum call stack size exceeded'); assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads, 'a sync call ran on the instance that threw'); } - assert.equal(loads, loadsBefore + 1); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); + assert.equal(loads, loadsBefore + 30); }); - it('stops sync calls on a stale instance after 10 failures until the fresh one loads', async () => { + it('never runs a queued call on the instance that threw', async () => { await isolated.loadModule(); - for (let i = 0; i < 9; i++) { - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - } - assert.ok(isolated.parseSync('SELECT 1')); throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - assert.throws(() => isolated.parseSync('SELECT 1'), /Call `?loadModule\(\)`? first/); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); - }); - - it('makes loadModule() and queued calls wait again when a queued call breaks the fresh instance', async () => { - await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); const after = isolated.parse('SELECT 1'); await assert.rejects(failing, RangeError); assert.ok(await after); - assert.equal(used, loads, 'a queued call ran on the instance that just threw'); - await ready; - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); - }); - - it('makes loadModule() wait again when a call queued ahead of it breaks the fresh instance', async () => { + assert.equal(used, loads, 'a queued call ran on the instance that threw'); await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); - await assert.rejects(failing, RangeError); - await ready; assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); + assert.equal(used, loads); }); it('names a thrown object that has no message', async () => { @@ -330,11 +339,19 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => const caught = await failOnce(name, query, thrown); assert.ok(caught instanceof Error); assert.match(caught.message, /out of WASM memory/); - assert.equal(caught.cause, thrown); + assert.deepEqual(caught.cause, thrown); }); } }); +// Makes the loader's synchronous instantiation fail, as if the new instance's memory could not be allocated. +function failInstantiate(options) { + const instantiateWasm = options.instantiateWasm; + if (!instantiateWasm) return; + options.instantiateWasm = (imports) => + instantiateWasm(imports, () => { throw new Error('simulated instantiate failure'); }); +} + // A copy of the wrapper whose WASM factory is wrapped, leaving the shared copy alone. function loadIsolated(wrapFactory) { const indexPath = require.resolve('../wasm/index.cjs'); diff --git a/versions/14/src/index.ts b/versions/14/src/index.ts index f39ac71..8a82ff3 100644 --- a/versions/14/src/index.ts +++ b/versions/14/src/index.ts @@ -131,78 +131,132 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: any) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: any = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } diff --git a/versions/14/test/wasm-recovery.test.js b/versions/14/test/wasm-recovery.test.js index 6d16d37..8894778 100644 --- a/versions/14/test/wasm-recovery.test.js +++ b/versions/14/test/wasm-recovery.test.js @@ -22,21 +22,55 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(parseSync('SELECT 1').stmts.length, 1); }); - // Without the 10-failure cap, about 45 overflows on one instance made the next one spin forever. - it('stops sync calls on the stale instance after 10 real overflows in one synchronous loop', async () => { + it('keeps sync calls working through 60 overflows in one synchronous loop', async () => { await loadModule(); - const errors = []; + const results = []; for (let i = 0; i < 60; i++) { - try { - parseSync(DEEP); - } catch (error) { - errors.push(error instanceof RangeError ? 'overflow' : error.message); + for (const query of [DEEP, 'SELECT 1']) { + try { + results.push(parseSync(query).stmts.length); + } catch (error) { + results.push(error instanceof RangeError ? 'overflow' : error.message); + } } } - assert.deepEqual(errors.slice(0, 10), Array(10).fill('overflow')); - for (const message of errors.slice(10)) assert.match(message, /Call `?loadModule\(\)`? first/); - await loadModule(); - assert.equal(parseSync('SELECT 1').stmts.length, 1); + assert.deepEqual(results, Array(60).fill(['overflow', 1]).flat()); + }); + + it('gives the next sync call a fresh instance after the memory limit is hit', () => { + assert.throws(() => parseSync('SELECT 1;\n'.repeat(5000000)), /out of WASM memory/); + assert.equal(parseSync('SELECT 1;\n'.repeat(1000000)).stmts.length, 1000000); + }); + + // Each replaced instance stays allocated until the synchronous run yields. + it('stops replacing after repeated out-of-memory failures in one synchronous run, and recovers after an await', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + const huge = 'SELECT 1;\\n'.repeat(5000000); + const messages = []; + let peak = 0; + for (let i = 0; i < 10; i++) { + try { lib.parseSync(huge); } catch (error) { messages.push(error.message + ' / ' + error.cause?.message); } + peak = Math.max(peak, process.memoryUsage().rss); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + const after = lib.parseSync('SELECT 1').stmts.length; + process.stdout.write('\\nresult=' + JSON.stringify({ messages, peak, after })); + })();`; + const output = execFileSync(process.execPath, ['-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const { messages, peak, after } = JSON.parse(output.match(/^result=(.*)$/m)[1]); + assert.equal(messages.length, 10); + assert.match(messages[0], /out of WASM memory/); + assert.match(messages[1], /out of WASM memory/); + for (const message of messages.slice(2)) { + assert.match(message, /^WASM memory limit hit repeatedly in one synchronous run; yield \(await\) to free memory \/ .*out of WASM memory/); + } + assert.ok(peak < 2.5 * 1024 ** 3, `peak RSS ${Math.round(peak / 1024 ** 2)} MiB`); + assert.equal(after, 1); }); it('frees each replaced instance, so WASM memory stays at one instance', () => { @@ -118,16 +152,22 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(loads, 3); }); - it('lets loadModule() reject on a failed reload and retry on the next call', async () => { + it('names the cause when a fresh instance cannot start, and loadModule() retries', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + const isolated = loadIsolated((real, options) => { + if (++loads === 2 || loads === 3) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await assert.rejects(isolated.loadModule(), /simulated load failure/); + assert.throws(() => isolated.parseSync(DEEP), RangeError); + assert.throws(() => isolated.parseSync('SELECT 1'), (error) => { + assert.match(error.message, /Could not start a fresh WASM instance/); + assert.match(error.cause.message, /simulated instantiate failure/); + return true; + }); await isolated.loadModule(); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); - assert.equal(loads, 3); + assert.equal(loads, 4); }); it('keeps sync calls in the same synchronous loop working after a failure', () => { @@ -145,18 +185,16 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.deepEqual(results, expected); }); - it('retries a failed fresh load on the next failure, for callers that never await', async () => { + it('retries a fresh instance that failed to start on the next sync call', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); - await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(loads, 2); - assert.throws(() => isolated.parseSync(DEEP)); - assert.equal(loads, 3); + const isolated = loadIsolated((real, options) => { + if (++loads === 2) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); + assert.throws(() => isolated.parseSync(DEEP), RangeError); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); }); it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { @@ -201,9 +239,15 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => let used; let frees = 0; let throwNext; - const isolated = loadIsolated(async (real, ...args) => { + const isolated = loadIsolated((real, options) => { const id = ++loads; - const module = await real(...args); + const ready = real(options); + // A replacement instance is built synchronously; the first load is async. + if (typeof options._malloc === 'function') return instrument(options, id) && ready; + return ready.then((module) => instrument(module, id)); + }); + + function instrument(module, id) { const free = module._free; module._free = (ptr) => { frees++; @@ -221,7 +265,7 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => }; } return module; - }); + } async function failOnce(name, query, thrown) { await isolated.loadModule(); @@ -242,73 +286,38 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => assert.equal(exitCodeAfter, 0, `${name} changed process.exitCode`); assert.equal(frees, freesBefore, `${name} freed memory in the instance that threw`); assert.equal(loads, loadsBefore + 1, `${name} did not start a fresh instance`); - const stale = used; const syncName = name.endsWith('Sync') ? name : 'parseSync'; - assert.ok(isolated[syncName](query), `${syncName} failed while the fresh instance loads`); - assert.equal(used, stale, 'a sync call did not run on the stale instance'); + assert.ok(isolated[syncName](query), `${syncName} failed right after the failure`); + assert.equal(used, loads, `${syncName} ran on the instance that threw`); assert.ok(frees > freesBefore, 'a successful call did not free its memory'); - // A sync caller gets the fresh instance once it awaits loadModule(); async calls wait by themselves. - if (name.endsWith('Sync')) await isolated.loadModule(); assert.ok(await isolated[name](query), `${name} failed after recovery`); - assert.equal(used, loads, `${name} ran on the stale instance, not the fresh one`); + assert.equal(used, loads, `${name} ran on the instance that threw`); return caught; } - it('starts one fresh instance for many failures in one synchronous loop', async () => { + it('starts a fresh instance for every failure in one synchronous loop', async () => { await isolated.loadModule(); const loadsBefore = loads; - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 30; i++) { throwNext = new RangeError('Maximum call stack size exceeded'); assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads, 'a sync call ran on the instance that threw'); } - assert.equal(loads, loadsBefore + 1); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); + assert.equal(loads, loadsBefore + 30); }); - it('stops sync calls on a stale instance after 10 failures until the fresh one loads', async () => { + it('never runs a queued call on the instance that threw', async () => { await isolated.loadModule(); - for (let i = 0; i < 9; i++) { - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - } - assert.ok(isolated.parseSync('SELECT 1')); throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - assert.throws(() => isolated.parseSync('SELECT 1'), /Call `?loadModule\(\)`? first/); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); - }); - - it('makes loadModule() and queued calls wait again when a queued call breaks the fresh instance', async () => { - await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); const after = isolated.parse('SELECT 1'); await assert.rejects(failing, RangeError); assert.ok(await after); - assert.equal(used, loads, 'a queued call ran on the instance that just threw'); - await ready; - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); - }); - - it('makes loadModule() wait again when a call queued ahead of it breaks the fresh instance', async () => { + assert.equal(used, loads, 'a queued call ran on the instance that threw'); await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); - await assert.rejects(failing, RangeError); - await ready; assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); + assert.equal(used, loads); }); it('names a thrown object that has no message', async () => { @@ -330,11 +339,19 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => const caught = await failOnce(name, query, thrown); assert.ok(caught instanceof Error); assert.match(caught.message, /out of WASM memory/); - assert.equal(caught.cause, thrown); + assert.deepEqual(caught.cause, thrown); }); } }); +// Makes the loader's synchronous instantiation fail, as if the new instance's memory could not be allocated. +function failInstantiate(options) { + const instantiateWasm = options.instantiateWasm; + if (!instantiateWasm) return; + options.instantiateWasm = (imports) => + instantiateWasm(imports, () => { throw new Error('simulated instantiate failure'); }); +} + // A copy of the wrapper whose WASM factory is wrapped, leaving the shared copy alone. function loadIsolated(wrapFactory) { const indexPath = require.resolve('../wasm/index.cjs'); diff --git a/versions/15/src/index.ts b/versions/15/src/index.ts index f39ac71..8a82ff3 100644 --- a/versions/15/src/index.ts +++ b/versions/15/src/index.ts @@ -131,78 +131,132 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: any) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: any = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } diff --git a/versions/15/test/wasm-recovery.test.js b/versions/15/test/wasm-recovery.test.js index 6d16d37..8894778 100644 --- a/versions/15/test/wasm-recovery.test.js +++ b/versions/15/test/wasm-recovery.test.js @@ -22,21 +22,55 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(parseSync('SELECT 1').stmts.length, 1); }); - // Without the 10-failure cap, about 45 overflows on one instance made the next one spin forever. - it('stops sync calls on the stale instance after 10 real overflows in one synchronous loop', async () => { + it('keeps sync calls working through 60 overflows in one synchronous loop', async () => { await loadModule(); - const errors = []; + const results = []; for (let i = 0; i < 60; i++) { - try { - parseSync(DEEP); - } catch (error) { - errors.push(error instanceof RangeError ? 'overflow' : error.message); + for (const query of [DEEP, 'SELECT 1']) { + try { + results.push(parseSync(query).stmts.length); + } catch (error) { + results.push(error instanceof RangeError ? 'overflow' : error.message); + } } } - assert.deepEqual(errors.slice(0, 10), Array(10).fill('overflow')); - for (const message of errors.slice(10)) assert.match(message, /Call `?loadModule\(\)`? first/); - await loadModule(); - assert.equal(parseSync('SELECT 1').stmts.length, 1); + assert.deepEqual(results, Array(60).fill(['overflow', 1]).flat()); + }); + + it('gives the next sync call a fresh instance after the memory limit is hit', () => { + assert.throws(() => parseSync('SELECT 1;\n'.repeat(5000000)), /out of WASM memory/); + assert.equal(parseSync('SELECT 1;\n'.repeat(1000000)).stmts.length, 1000000); + }); + + // Each replaced instance stays allocated until the synchronous run yields. + it('stops replacing after repeated out-of-memory failures in one synchronous run, and recovers after an await', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + const huge = 'SELECT 1;\\n'.repeat(5000000); + const messages = []; + let peak = 0; + for (let i = 0; i < 10; i++) { + try { lib.parseSync(huge); } catch (error) { messages.push(error.message + ' / ' + error.cause?.message); } + peak = Math.max(peak, process.memoryUsage().rss); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + const after = lib.parseSync('SELECT 1').stmts.length; + process.stdout.write('\\nresult=' + JSON.stringify({ messages, peak, after })); + })();`; + const output = execFileSync(process.execPath, ['-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const { messages, peak, after } = JSON.parse(output.match(/^result=(.*)$/m)[1]); + assert.equal(messages.length, 10); + assert.match(messages[0], /out of WASM memory/); + assert.match(messages[1], /out of WASM memory/); + for (const message of messages.slice(2)) { + assert.match(message, /^WASM memory limit hit repeatedly in one synchronous run; yield \(await\) to free memory \/ .*out of WASM memory/); + } + assert.ok(peak < 2.5 * 1024 ** 3, `peak RSS ${Math.round(peak / 1024 ** 2)} MiB`); + assert.equal(after, 1); }); it('frees each replaced instance, so WASM memory stays at one instance', () => { @@ -118,16 +152,22 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(loads, 3); }); - it('lets loadModule() reject on a failed reload and retry on the next call', async () => { + it('names the cause when a fresh instance cannot start, and loadModule() retries', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + const isolated = loadIsolated((real, options) => { + if (++loads === 2 || loads === 3) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await assert.rejects(isolated.loadModule(), /simulated load failure/); + assert.throws(() => isolated.parseSync(DEEP), RangeError); + assert.throws(() => isolated.parseSync('SELECT 1'), (error) => { + assert.match(error.message, /Could not start a fresh WASM instance/); + assert.match(error.cause.message, /simulated instantiate failure/); + return true; + }); await isolated.loadModule(); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); - assert.equal(loads, 3); + assert.equal(loads, 4); }); it('keeps sync calls in the same synchronous loop working after a failure', () => { @@ -145,18 +185,16 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.deepEqual(results, expected); }); - it('retries a failed fresh load on the next failure, for callers that never await', async () => { + it('retries a fresh instance that failed to start on the next sync call', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); - await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(loads, 2); - assert.throws(() => isolated.parseSync(DEEP)); - assert.equal(loads, 3); + const isolated = loadIsolated((real, options) => { + if (++loads === 2) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); + assert.throws(() => isolated.parseSync(DEEP), RangeError); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); }); it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { @@ -201,9 +239,15 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => let used; let frees = 0; let throwNext; - const isolated = loadIsolated(async (real, ...args) => { + const isolated = loadIsolated((real, options) => { const id = ++loads; - const module = await real(...args); + const ready = real(options); + // A replacement instance is built synchronously; the first load is async. + if (typeof options._malloc === 'function') return instrument(options, id) && ready; + return ready.then((module) => instrument(module, id)); + }); + + function instrument(module, id) { const free = module._free; module._free = (ptr) => { frees++; @@ -221,7 +265,7 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => }; } return module; - }); + } async function failOnce(name, query, thrown) { await isolated.loadModule(); @@ -242,73 +286,38 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => assert.equal(exitCodeAfter, 0, `${name} changed process.exitCode`); assert.equal(frees, freesBefore, `${name} freed memory in the instance that threw`); assert.equal(loads, loadsBefore + 1, `${name} did not start a fresh instance`); - const stale = used; const syncName = name.endsWith('Sync') ? name : 'parseSync'; - assert.ok(isolated[syncName](query), `${syncName} failed while the fresh instance loads`); - assert.equal(used, stale, 'a sync call did not run on the stale instance'); + assert.ok(isolated[syncName](query), `${syncName} failed right after the failure`); + assert.equal(used, loads, `${syncName} ran on the instance that threw`); assert.ok(frees > freesBefore, 'a successful call did not free its memory'); - // A sync caller gets the fresh instance once it awaits loadModule(); async calls wait by themselves. - if (name.endsWith('Sync')) await isolated.loadModule(); assert.ok(await isolated[name](query), `${name} failed after recovery`); - assert.equal(used, loads, `${name} ran on the stale instance, not the fresh one`); + assert.equal(used, loads, `${name} ran on the instance that threw`); return caught; } - it('starts one fresh instance for many failures in one synchronous loop', async () => { + it('starts a fresh instance for every failure in one synchronous loop', async () => { await isolated.loadModule(); const loadsBefore = loads; - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 30; i++) { throwNext = new RangeError('Maximum call stack size exceeded'); assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads, 'a sync call ran on the instance that threw'); } - assert.equal(loads, loadsBefore + 1); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); + assert.equal(loads, loadsBefore + 30); }); - it('stops sync calls on a stale instance after 10 failures until the fresh one loads', async () => { + it('never runs a queued call on the instance that threw', async () => { await isolated.loadModule(); - for (let i = 0; i < 9; i++) { - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - } - assert.ok(isolated.parseSync('SELECT 1')); throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - assert.throws(() => isolated.parseSync('SELECT 1'), /Call `?loadModule\(\)`? first/); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); - }); - - it('makes loadModule() and queued calls wait again when a queued call breaks the fresh instance', async () => { - await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); const after = isolated.parse('SELECT 1'); await assert.rejects(failing, RangeError); assert.ok(await after); - assert.equal(used, loads, 'a queued call ran on the instance that just threw'); - await ready; - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); - }); - - it('makes loadModule() wait again when a call queued ahead of it breaks the fresh instance', async () => { + assert.equal(used, loads, 'a queued call ran on the instance that threw'); await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); - await assert.rejects(failing, RangeError); - await ready; assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); + assert.equal(used, loads); }); it('names a thrown object that has no message', async () => { @@ -330,11 +339,19 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => const caught = await failOnce(name, query, thrown); assert.ok(caught instanceof Error); assert.match(caught.message, /out of WASM memory/); - assert.equal(caught.cause, thrown); + assert.deepEqual(caught.cause, thrown); }); } }); +// Makes the loader's synchronous instantiation fail, as if the new instance's memory could not be allocated. +function failInstantiate(options) { + const instantiateWasm = options.instantiateWasm; + if (!instantiateWasm) return; + options.instantiateWasm = (imports) => + instantiateWasm(imports, () => { throw new Error('simulated instantiate failure'); }); +} + // A copy of the wrapper whose WASM factory is wrapped, leaving the shared copy alone. function loadIsolated(wrapFactory) { const indexPath = require.resolve('../wasm/index.cjs'); diff --git a/versions/16/src/index.ts b/versions/16/src/index.ts index f39ac71..8a82ff3 100644 --- a/versions/16/src/index.ts +++ b/versions/16/src/index.ts @@ -131,78 +131,132 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: any) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: any = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } diff --git a/versions/16/test/wasm-recovery.test.js b/versions/16/test/wasm-recovery.test.js index 6d16d37..8894778 100644 --- a/versions/16/test/wasm-recovery.test.js +++ b/versions/16/test/wasm-recovery.test.js @@ -22,21 +22,55 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(parseSync('SELECT 1').stmts.length, 1); }); - // Without the 10-failure cap, about 45 overflows on one instance made the next one spin forever. - it('stops sync calls on the stale instance after 10 real overflows in one synchronous loop', async () => { + it('keeps sync calls working through 60 overflows in one synchronous loop', async () => { await loadModule(); - const errors = []; + const results = []; for (let i = 0; i < 60; i++) { - try { - parseSync(DEEP); - } catch (error) { - errors.push(error instanceof RangeError ? 'overflow' : error.message); + for (const query of [DEEP, 'SELECT 1']) { + try { + results.push(parseSync(query).stmts.length); + } catch (error) { + results.push(error instanceof RangeError ? 'overflow' : error.message); + } } } - assert.deepEqual(errors.slice(0, 10), Array(10).fill('overflow')); - for (const message of errors.slice(10)) assert.match(message, /Call `?loadModule\(\)`? first/); - await loadModule(); - assert.equal(parseSync('SELECT 1').stmts.length, 1); + assert.deepEqual(results, Array(60).fill(['overflow', 1]).flat()); + }); + + it('gives the next sync call a fresh instance after the memory limit is hit', () => { + assert.throws(() => parseSync('SELECT 1;\n'.repeat(5000000)), /out of WASM memory/); + assert.equal(parseSync('SELECT 1;\n'.repeat(1000000)).stmts.length, 1000000); + }); + + // Each replaced instance stays allocated until the synchronous run yields. + it('stops replacing after repeated out-of-memory failures in one synchronous run, and recovers after an await', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + const huge = 'SELECT 1;\\n'.repeat(5000000); + const messages = []; + let peak = 0; + for (let i = 0; i < 10; i++) { + try { lib.parseSync(huge); } catch (error) { messages.push(error.message + ' / ' + error.cause?.message); } + peak = Math.max(peak, process.memoryUsage().rss); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + const after = lib.parseSync('SELECT 1').stmts.length; + process.stdout.write('\\nresult=' + JSON.stringify({ messages, peak, after })); + })();`; + const output = execFileSync(process.execPath, ['-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const { messages, peak, after } = JSON.parse(output.match(/^result=(.*)$/m)[1]); + assert.equal(messages.length, 10); + assert.match(messages[0], /out of WASM memory/); + assert.match(messages[1], /out of WASM memory/); + for (const message of messages.slice(2)) { + assert.match(message, /^WASM memory limit hit repeatedly in one synchronous run; yield \(await\) to free memory \/ .*out of WASM memory/); + } + assert.ok(peak < 2.5 * 1024 ** 3, `peak RSS ${Math.round(peak / 1024 ** 2)} MiB`); + assert.equal(after, 1); }); it('frees each replaced instance, so WASM memory stays at one instance', () => { @@ -118,16 +152,22 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(loads, 3); }); - it('lets loadModule() reject on a failed reload and retry on the next call', async () => { + it('names the cause when a fresh instance cannot start, and loadModule() retries', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + const isolated = loadIsolated((real, options) => { + if (++loads === 2 || loads === 3) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await assert.rejects(isolated.loadModule(), /simulated load failure/); + assert.throws(() => isolated.parseSync(DEEP), RangeError); + assert.throws(() => isolated.parseSync('SELECT 1'), (error) => { + assert.match(error.message, /Could not start a fresh WASM instance/); + assert.match(error.cause.message, /simulated instantiate failure/); + return true; + }); await isolated.loadModule(); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); - assert.equal(loads, 3); + assert.equal(loads, 4); }); it('keeps sync calls in the same synchronous loop working after a failure', () => { @@ -145,18 +185,16 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.deepEqual(results, expected); }); - it('retries a failed fresh load on the next failure, for callers that never await', async () => { + it('retries a fresh instance that failed to start on the next sync call', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); - await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(loads, 2); - assert.throws(() => isolated.parseSync(DEEP)); - assert.equal(loads, 3); + const isolated = loadIsolated((real, options) => { + if (++loads === 2) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); + assert.throws(() => isolated.parseSync(DEEP), RangeError); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); }); it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { @@ -201,9 +239,15 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => let used; let frees = 0; let throwNext; - const isolated = loadIsolated(async (real, ...args) => { + const isolated = loadIsolated((real, options) => { const id = ++loads; - const module = await real(...args); + const ready = real(options); + // A replacement instance is built synchronously; the first load is async. + if (typeof options._malloc === 'function') return instrument(options, id) && ready; + return ready.then((module) => instrument(module, id)); + }); + + function instrument(module, id) { const free = module._free; module._free = (ptr) => { frees++; @@ -221,7 +265,7 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => }; } return module; - }); + } async function failOnce(name, query, thrown) { await isolated.loadModule(); @@ -242,73 +286,38 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => assert.equal(exitCodeAfter, 0, `${name} changed process.exitCode`); assert.equal(frees, freesBefore, `${name} freed memory in the instance that threw`); assert.equal(loads, loadsBefore + 1, `${name} did not start a fresh instance`); - const stale = used; const syncName = name.endsWith('Sync') ? name : 'parseSync'; - assert.ok(isolated[syncName](query), `${syncName} failed while the fresh instance loads`); - assert.equal(used, stale, 'a sync call did not run on the stale instance'); + assert.ok(isolated[syncName](query), `${syncName} failed right after the failure`); + assert.equal(used, loads, `${syncName} ran on the instance that threw`); assert.ok(frees > freesBefore, 'a successful call did not free its memory'); - // A sync caller gets the fresh instance once it awaits loadModule(); async calls wait by themselves. - if (name.endsWith('Sync')) await isolated.loadModule(); assert.ok(await isolated[name](query), `${name} failed after recovery`); - assert.equal(used, loads, `${name} ran on the stale instance, not the fresh one`); + assert.equal(used, loads, `${name} ran on the instance that threw`); return caught; } - it('starts one fresh instance for many failures in one synchronous loop', async () => { + it('starts a fresh instance for every failure in one synchronous loop', async () => { await isolated.loadModule(); const loadsBefore = loads; - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 30; i++) { throwNext = new RangeError('Maximum call stack size exceeded'); assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads, 'a sync call ran on the instance that threw'); } - assert.equal(loads, loadsBefore + 1); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); + assert.equal(loads, loadsBefore + 30); }); - it('stops sync calls on a stale instance after 10 failures until the fresh one loads', async () => { + it('never runs a queued call on the instance that threw', async () => { await isolated.loadModule(); - for (let i = 0; i < 9; i++) { - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - } - assert.ok(isolated.parseSync('SELECT 1')); throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - assert.throws(() => isolated.parseSync('SELECT 1'), /Call `?loadModule\(\)`? first/); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); - }); - - it('makes loadModule() and queued calls wait again when a queued call breaks the fresh instance', async () => { - await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); const after = isolated.parse('SELECT 1'); await assert.rejects(failing, RangeError); assert.ok(await after); - assert.equal(used, loads, 'a queued call ran on the instance that just threw'); - await ready; - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); - }); - - it('makes loadModule() wait again when a call queued ahead of it breaks the fresh instance', async () => { + assert.equal(used, loads, 'a queued call ran on the instance that threw'); await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); - await assert.rejects(failing, RangeError); - await ready; assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); + assert.equal(used, loads); }); it('names a thrown object that has no message', async () => { @@ -330,11 +339,19 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => const caught = await failOnce(name, query, thrown); assert.ok(caught instanceof Error); assert.match(caught.message, /out of WASM memory/); - assert.equal(caught.cause, thrown); + assert.deepEqual(caught.cause, thrown); }); } }); +// Makes the loader's synchronous instantiation fail, as if the new instance's memory could not be allocated. +function failInstantiate(options) { + const instantiateWasm = options.instantiateWasm; + if (!instantiateWasm) return; + options.instantiateWasm = (imports) => + instantiateWasm(imports, () => { throw new Error('simulated instantiate failure'); }); +} + // A copy of the wrapper whose WASM factory is wrapped, leaving the shared copy alone. function loadIsolated(wrapFactory) { const indexPath = require.resolve('../wasm/index.cjs'); diff --git a/versions/17/src/index.ts b/versions/17/src/index.ts index f39ac71..8a82ff3 100644 --- a/versions/17/src/index.ts +++ b/versions/17/src/index.ts @@ -131,78 +131,132 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: any) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: any = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } diff --git a/versions/17/test/wasm-recovery.test.js b/versions/17/test/wasm-recovery.test.js index 6d16d37..8894778 100644 --- a/versions/17/test/wasm-recovery.test.js +++ b/versions/17/test/wasm-recovery.test.js @@ -22,21 +22,55 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(parseSync('SELECT 1').stmts.length, 1); }); - // Without the 10-failure cap, about 45 overflows on one instance made the next one spin forever. - it('stops sync calls on the stale instance after 10 real overflows in one synchronous loop', async () => { + it('keeps sync calls working through 60 overflows in one synchronous loop', async () => { await loadModule(); - const errors = []; + const results = []; for (let i = 0; i < 60; i++) { - try { - parseSync(DEEP); - } catch (error) { - errors.push(error instanceof RangeError ? 'overflow' : error.message); + for (const query of [DEEP, 'SELECT 1']) { + try { + results.push(parseSync(query).stmts.length); + } catch (error) { + results.push(error instanceof RangeError ? 'overflow' : error.message); + } } } - assert.deepEqual(errors.slice(0, 10), Array(10).fill('overflow')); - for (const message of errors.slice(10)) assert.match(message, /Call `?loadModule\(\)`? first/); - await loadModule(); - assert.equal(parseSync('SELECT 1').stmts.length, 1); + assert.deepEqual(results, Array(60).fill(['overflow', 1]).flat()); + }); + + it('gives the next sync call a fresh instance after the memory limit is hit', () => { + assert.throws(() => parseSync('SELECT 1;\n'.repeat(5000000)), /out of WASM memory/); + assert.equal(parseSync('SELECT 1;\n'.repeat(1000000)).stmts.length, 1000000); + }); + + // Each replaced instance stays allocated until the synchronous run yields. + it('stops replacing after repeated out-of-memory failures in one synchronous run, and recovers after an await', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + const huge = 'SELECT 1;\\n'.repeat(5000000); + const messages = []; + let peak = 0; + for (let i = 0; i < 10; i++) { + try { lib.parseSync(huge); } catch (error) { messages.push(error.message + ' / ' + error.cause?.message); } + peak = Math.max(peak, process.memoryUsage().rss); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + const after = lib.parseSync('SELECT 1').stmts.length; + process.stdout.write('\\nresult=' + JSON.stringify({ messages, peak, after })); + })();`; + const output = execFileSync(process.execPath, ['-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const { messages, peak, after } = JSON.parse(output.match(/^result=(.*)$/m)[1]); + assert.equal(messages.length, 10); + assert.match(messages[0], /out of WASM memory/); + assert.match(messages[1], /out of WASM memory/); + for (const message of messages.slice(2)) { + assert.match(message, /^WASM memory limit hit repeatedly in one synchronous run; yield \(await\) to free memory \/ .*out of WASM memory/); + } + assert.ok(peak < 2.5 * 1024 ** 3, `peak RSS ${Math.round(peak / 1024 ** 2)} MiB`); + assert.equal(after, 1); }); it('frees each replaced instance, so WASM memory stays at one instance', () => { @@ -118,16 +152,22 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(loads, 3); }); - it('lets loadModule() reject on a failed reload and retry on the next call', async () => { + it('names the cause when a fresh instance cannot start, and loadModule() retries', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + const isolated = loadIsolated((real, options) => { + if (++loads === 2 || loads === 3) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await assert.rejects(isolated.loadModule(), /simulated load failure/); + assert.throws(() => isolated.parseSync(DEEP), RangeError); + assert.throws(() => isolated.parseSync('SELECT 1'), (error) => { + assert.match(error.message, /Could not start a fresh WASM instance/); + assert.match(error.cause.message, /simulated instantiate failure/); + return true; + }); await isolated.loadModule(); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); - assert.equal(loads, 3); + assert.equal(loads, 4); }); it('keeps sync calls in the same synchronous loop working after a failure', () => { @@ -145,18 +185,16 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.deepEqual(results, expected); }); - it('retries a failed fresh load on the next failure, for callers that never await', async () => { + it('retries a fresh instance that failed to start on the next sync call', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); - await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(loads, 2); - assert.throws(() => isolated.parseSync(DEEP)); - assert.equal(loads, 3); + const isolated = loadIsolated((real, options) => { + if (++loads === 2) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); + assert.throws(() => isolated.parseSync(DEEP), RangeError); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); }); it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { @@ -201,9 +239,15 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => let used; let frees = 0; let throwNext; - const isolated = loadIsolated(async (real, ...args) => { + const isolated = loadIsolated((real, options) => { const id = ++loads; - const module = await real(...args); + const ready = real(options); + // A replacement instance is built synchronously; the first load is async. + if (typeof options._malloc === 'function') return instrument(options, id) && ready; + return ready.then((module) => instrument(module, id)); + }); + + function instrument(module, id) { const free = module._free; module._free = (ptr) => { frees++; @@ -221,7 +265,7 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => }; } return module; - }); + } async function failOnce(name, query, thrown) { await isolated.loadModule(); @@ -242,73 +286,38 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => assert.equal(exitCodeAfter, 0, `${name} changed process.exitCode`); assert.equal(frees, freesBefore, `${name} freed memory in the instance that threw`); assert.equal(loads, loadsBefore + 1, `${name} did not start a fresh instance`); - const stale = used; const syncName = name.endsWith('Sync') ? name : 'parseSync'; - assert.ok(isolated[syncName](query), `${syncName} failed while the fresh instance loads`); - assert.equal(used, stale, 'a sync call did not run on the stale instance'); + assert.ok(isolated[syncName](query), `${syncName} failed right after the failure`); + assert.equal(used, loads, `${syncName} ran on the instance that threw`); assert.ok(frees > freesBefore, 'a successful call did not free its memory'); - // A sync caller gets the fresh instance once it awaits loadModule(); async calls wait by themselves. - if (name.endsWith('Sync')) await isolated.loadModule(); assert.ok(await isolated[name](query), `${name} failed after recovery`); - assert.equal(used, loads, `${name} ran on the stale instance, not the fresh one`); + assert.equal(used, loads, `${name} ran on the instance that threw`); return caught; } - it('starts one fresh instance for many failures in one synchronous loop', async () => { + it('starts a fresh instance for every failure in one synchronous loop', async () => { await isolated.loadModule(); const loadsBefore = loads; - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 30; i++) { throwNext = new RangeError('Maximum call stack size exceeded'); assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads, 'a sync call ran on the instance that threw'); } - assert.equal(loads, loadsBefore + 1); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); + assert.equal(loads, loadsBefore + 30); }); - it('stops sync calls on a stale instance after 10 failures until the fresh one loads', async () => { + it('never runs a queued call on the instance that threw', async () => { await isolated.loadModule(); - for (let i = 0; i < 9; i++) { - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - } - assert.ok(isolated.parseSync('SELECT 1')); throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - assert.throws(() => isolated.parseSync('SELECT 1'), /Call `?loadModule\(\)`? first/); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); - }); - - it('makes loadModule() and queued calls wait again when a queued call breaks the fresh instance', async () => { - await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); const after = isolated.parse('SELECT 1'); await assert.rejects(failing, RangeError); assert.ok(await after); - assert.equal(used, loads, 'a queued call ran on the instance that just threw'); - await ready; - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); - }); - - it('makes loadModule() wait again when a call queued ahead of it breaks the fresh instance', async () => { + assert.equal(used, loads, 'a queued call ran on the instance that threw'); await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); - await assert.rejects(failing, RangeError); - await ready; assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); + assert.equal(used, loads); }); it('names a thrown object that has no message', async () => { @@ -330,11 +339,19 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => const caught = await failOnce(name, query, thrown); assert.ok(caught instanceof Error); assert.match(caught.message, /out of WASM memory/); - assert.equal(caught.cause, thrown); + assert.deepEqual(caught.cause, thrown); }); } }); +// Makes the loader's synchronous instantiation fail, as if the new instance's memory could not be allocated. +function failInstantiate(options) { + const instantiateWasm = options.instantiateWasm; + if (!instantiateWasm) return; + options.instantiateWasm = (imports) => + instantiateWasm(imports, () => { throw new Error('simulated instantiate failure'); }); +} + // A copy of the wrapper whose WASM factory is wrapped, leaving the shared copy alone. function loadIsolated(wrapFactory) { const indexPath = require.resolve('../wasm/index.cjs'); diff --git a/versions/18/src/index.ts b/versions/18/src/index.ts index 56c93d6..3a33f84 100644 --- a/versions/18/src/index.ts +++ b/versions/18/src/index.ts @@ -143,78 +143,132 @@ interface WasmModule { let wasmModule: WasmModule; -// After a WASM call throws, the instance is stale: async calls wait for its -// replacement, while sync calls, which cannot wait, keep using it until the -// replacement is ready. One replacement loads at a time; the next call retries -// a failed load. -let stale = false; +// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a +// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and +// can leave the instance corrupt. callWasm() therefore swaps in a fresh instance +// before rethrowing. It builds that instance synchronously from the compiled +// module, so the very next call, sync or async, already runs on a clean one. +let compiled: WebAssembly.Module | undefined; let loading: Promise | undefined; -// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +// Counts WASM calls that threw, so cleanup skips memory of the replaced instance. let failures = 0; -// Sync calls stop using a stale instance after this many failures on it: past -// about 45, a stack overflow can make the next one spin forever. -const MAX_STALE_FAILURES = 10; -let staleFailures = 0; +// Why the last synchronous replacement failed; the next loadModule() retries. +let replaceError: unknown; +// A replaced instance is freed only after the current synchronous run yields. +// At most one that grew past half the 1 GiB heap limit may wait for that, so +// repeated out-of-memory failures can't pile up gigabytes; small ones (stack +// overflows on deep nesting) are not limited. +const LARGE_INSTANCE_BYTES = 512 * 1024 * 1024; +let largeUnfreed = 0; + +// Reads the .wasm file the way the Emscripten loader does: from disk in Node, else with fetch. +async function compileWasm(url: string): Promise { + const proc = (globalThis as any).process; + if (proc?.versions?.node && proc.type !== 'renderer') { + const fs = await import('node:fs'); + return WebAssembly.compile(fs.readFileSync(url.startsWith('file://') ? new URL(url) : url)); + } + const response = await fetch(url, { credentials: 'same-origin' }); + if (!response.ok) throw new Error(`Failed to load ${url}: HTTP ${response.status}`); + return WebAssembly.compile(await response.arrayBuffer()); +} function instantiate(): Promise { if (loading) return loading; - const promise = PgQueryModule().then( - (module: WasmModule) => { - wasmModule = module; - stale = false; - staleFailures = 0; - loading = undefined; - }, - (error: unknown) => { - loading = undefined; - throw error; - } - ); + const promise = (async () => { + let wasmUrl: string | undefined; + const module: WasmModule = await PgQueryModule({ + locateFile: (path: string, dir: string) => (wasmUrl = dir + path), + }); + compiled ??= await compileWasm(wasmUrl); + wasmModule = module; + replaceError = undefined; + })(); loading = promise; - // Rejects only for the callers awaiting it. - promise.catch(() => {}); + promise.then( + () => { loading = undefined; }, + () => { loading = undefined; } + ); return promise; } -instantiate(); +// Rejects only for the callers awaiting it. +instantiate().catch(() => {}); + +function replaceInstance(): void { + let error: unknown = new Error('the WASM loader did not instantiate synchronously'); + const module: any = { + instantiateWasm(imports: WebAssembly.Imports, receive: (instance: WebAssembly.Instance) => void) { + try { + receive(new WebAssembly.Instance(compiled, imports)); + error = undefined; + } catch (e) { + error = e; + } + return {}; + }, + }; + // receive() sets the exports and memory views, so the instance is usable right + // away; the loader runs the rest of its start-up (stdio streams) on a later microtask. + PgQueryModule(module).catch(() => {}); + wasmModule = error === undefined ? module : undefined; + replaceError = error; +} -// An exception out of a WASM call (V8 stack overflow on deeply nested SQL, a -// trap, or exit() once the heap limit is hit) skips libpg_query's cleanup and -// leaves the instance corrupt, so replace it with a fresh one. function callWasm(fn: () => number): number { const proc = (globalThis as any).process; const exitCode = proc?.exitCode; try { return fn(); - } catch (error) { + } catch (thrown) { failures++; - stale = true; - if (++staleFailures >= MAX_STALE_FAILURES) wasmModule = undefined; - instantiate(); - if (proc) proc.exitCode = exitCode; - if (error instanceof Error) throw error; - const reason = typeof error === 'object' && error !== null - ? ((error as any).message ?? (error as any).constructor?.name) - : error; - throw new Error( - `libpg_query terminated (${reason}), most likely out of WASM memory; parse large inputs in smaller pieces`, - { cause: error } + const error = thrown instanceof Error ? thrown : new Error( + `libpg_query terminated (${describeThrown(thrown)}), most likely out of WASM memory; parse large inputs in smaller pieces`, + // A copy: Emscripten's own exit object keeps the whole instance alive. + { cause: typeof thrown === 'object' && thrown !== null ? { ...thrown } : thrown } ); + if (wasmModule.HEAPU8.length > LARGE_INSTANCE_BYTES && largeUnfreed++ === 0) { + queueMicrotask(() => { largeUnfreed = 0; }); + } + if (largeUnfreed <= 1) { + replaceInstance(); + } else { + wasmModule = undefined; + replaceError = new Error( + 'WASM memory limit hit repeatedly in one synchronous run; yield (await) to free memory', + { cause: error } + ); + } + if (proc) proc.exitCode = exitCode; + throw error; } } +function describeThrown(thrown: unknown) { + return typeof thrown === 'object' && thrown !== null + ? ((thrown as any).message ?? (thrown as any).constructor?.name) + : thrown; +} + function ensureLoaded() { - if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); + if (wasmModule) return; + // A replacement that was skipped or failed to start is retried here, for callers that never await. + if (compiled && largeUnfreed <= 1) { + replaceInstance(); + if (wasmModule) return; + throw new Error('Could not start a fresh WASM instance after a failed call; await `loadModule()` to retry', { cause: replaceError }); + } + if (replaceError !== undefined) throw replaceError; + throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule(): Promise { - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); } function awaitInit Promise>(fn: T): T { return (async (...args: Parameters) => { - // Re-checked after every await: another call may have failed in between. - while (!wasmModule || stale) await instantiate(); + while (!wasmModule) await instantiate(); return fn(...args); }) as T; } @@ -388,9 +442,7 @@ export const normalize = awaitInit(async (query: string): Promise => { // Sync versions export function parseSync(query: string): ParseResult { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); // Input validation if (query === null || query === undefined) { @@ -454,9 +506,7 @@ export function parseSync(query: string): ParseResult { } export function parsePlPgSQLSync(query: string): ParseResult { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; @@ -482,9 +532,7 @@ export function parsePlPgSQLSync(query: string): ParseResult { } export function fingerprintSync(query: string): string { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; @@ -509,9 +557,7 @@ export function fingerprintSync(query: string): string { } export function normalizeSync(query: string): string { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; @@ -560,9 +606,7 @@ export const scan = awaitInit(async (query: string): Promise => { }); export function scanSync(query: string): ScanResult { - if (!wasmModule) { - throw new Error('WASM module not initialized. Call loadModule() first.'); - } + ensureLoaded(); const queryPtr = stringToPtr(query); const failuresBefore = failures; let resultPtr = 0; diff --git a/versions/18/test/wasm-recovery.test.js b/versions/18/test/wasm-recovery.test.js index 6d16d37..8894778 100644 --- a/versions/18/test/wasm-recovery.test.js +++ b/versions/18/test/wasm-recovery.test.js @@ -22,21 +22,55 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(parseSync('SELECT 1').stmts.length, 1); }); - // Without the 10-failure cap, about 45 overflows on one instance made the next one spin forever. - it('stops sync calls on the stale instance after 10 real overflows in one synchronous loop', async () => { + it('keeps sync calls working through 60 overflows in one synchronous loop', async () => { await loadModule(); - const errors = []; + const results = []; for (let i = 0; i < 60; i++) { - try { - parseSync(DEEP); - } catch (error) { - errors.push(error instanceof RangeError ? 'overflow' : error.message); + for (const query of [DEEP, 'SELECT 1']) { + try { + results.push(parseSync(query).stmts.length); + } catch (error) { + results.push(error instanceof RangeError ? 'overflow' : error.message); + } } } - assert.deepEqual(errors.slice(0, 10), Array(10).fill('overflow')); - for (const message of errors.slice(10)) assert.match(message, /Call `?loadModule\(\)`? first/); - await loadModule(); - assert.equal(parseSync('SELECT 1').stmts.length, 1); + assert.deepEqual(results, Array(60).fill(['overflow', 1]).flat()); + }); + + it('gives the next sync call a fresh instance after the memory limit is hit', () => { + assert.throws(() => parseSync('SELECT 1;\n'.repeat(5000000)), /out of WASM memory/); + assert.equal(parseSync('SELECT 1;\n'.repeat(1000000)).stmts.length, 1000000); + }); + + // Each replaced instance stays allocated until the synchronous run yields. + it('stops replacing after repeated out-of-memory failures in one synchronous run, and recovers after an await', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + const huge = 'SELECT 1;\\n'.repeat(5000000); + const messages = []; + let peak = 0; + for (let i = 0; i < 10; i++) { + try { lib.parseSync(huge); } catch (error) { messages.push(error.message + ' / ' + error.cause?.message); } + peak = Math.max(peak, process.memoryUsage().rss); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + const after = lib.parseSync('SELECT 1').stmts.length; + process.stdout.write('\\nresult=' + JSON.stringify({ messages, peak, after })); + })();`; + const output = execFileSync(process.execPath, ['-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const { messages, peak, after } = JSON.parse(output.match(/^result=(.*)$/m)[1]); + assert.equal(messages.length, 10); + assert.match(messages[0], /out of WASM memory/); + assert.match(messages[1], /out of WASM memory/); + for (const message of messages.slice(2)) { + assert.match(message, /^WASM memory limit hit repeatedly in one synchronous run; yield \(await\) to free memory \/ .*out of WASM memory/); + } + assert.ok(peak < 2.5 * 1024 ** 3, `peak RSS ${Math.round(peak / 1024 ** 2)} MiB`); + assert.equal(after, 1); }); it('frees each replaced instance, so WASM memory stays at one instance', () => { @@ -118,16 +152,22 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.equal(loads, 3); }); - it('lets loadModule() reject on a failed reload and retry on the next call', async () => { + it('names the cause when a fresh instance cannot start, and loadModule() retries', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + const isolated = loadIsolated((real, options) => { + if (++loads === 2 || loads === 3) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await assert.rejects(isolated.loadModule(), /simulated load failure/); + assert.throws(() => isolated.parseSync(DEEP), RangeError); + assert.throws(() => isolated.parseSync('SELECT 1'), (error) => { + assert.match(error.message, /Could not start a fresh WASM instance/); + assert.match(error.cause.message, /simulated instantiate failure/); + return true; + }); await isolated.loadModule(); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); - assert.equal(loads, 3); + assert.equal(loads, 4); }); it('keeps sync calls in the same synchronous loop working after a failure', () => { @@ -145,18 +185,16 @@ describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { assert.deepEqual(results, expected); }); - it('retries a failed fresh load on the next failure, for callers that never await', async () => { + it('retries a fresh instance that failed to start on the next sync call', async () => { let loads = 0; - const isolated = loadIsolated((real, ...args) => - ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); - await isolated.loadModule(); - assert.throws(() => isolated.parseSync(DEEP)); - await new Promise((resolve) => setImmediate(resolve)); - assert.equal(loads, 2); - assert.throws(() => isolated.parseSync(DEEP)); - assert.equal(loads, 3); + const isolated = loadIsolated((real, options) => { + if (++loads === 2) failInstantiate(options); + return real(options); + }); await isolated.loadModule(); + assert.throws(() => isolated.parseSync(DEEP), RangeError); assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); }); it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { @@ -201,9 +239,15 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => let used; let frees = 0; let throwNext; - const isolated = loadIsolated(async (real, ...args) => { + const isolated = loadIsolated((real, options) => { const id = ++loads; - const module = await real(...args); + const ready = real(options); + // A replacement instance is built synchronously; the first load is async. + if (typeof options._malloc === 'function') return instrument(options, id) && ready; + return ready.then((module) => instrument(module, id)); + }); + + function instrument(module, id) { const free = module._free; module._free = (ptr) => { frees++; @@ -221,7 +265,7 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => }; } return module; - }); + } async function failOnce(name, query, thrown) { await isolated.loadModule(); @@ -242,73 +286,38 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => assert.equal(exitCodeAfter, 0, `${name} changed process.exitCode`); assert.equal(frees, freesBefore, `${name} freed memory in the instance that threw`); assert.equal(loads, loadsBefore + 1, `${name} did not start a fresh instance`); - const stale = used; const syncName = name.endsWith('Sync') ? name : 'parseSync'; - assert.ok(isolated[syncName](query), `${syncName} failed while the fresh instance loads`); - assert.equal(used, stale, 'a sync call did not run on the stale instance'); + assert.ok(isolated[syncName](query), `${syncName} failed right after the failure`); + assert.equal(used, loads, `${syncName} ran on the instance that threw`); assert.ok(frees > freesBefore, 'a successful call did not free its memory'); - // A sync caller gets the fresh instance once it awaits loadModule(); async calls wait by themselves. - if (name.endsWith('Sync')) await isolated.loadModule(); assert.ok(await isolated[name](query), `${name} failed after recovery`); - assert.equal(used, loads, `${name} ran on the stale instance, not the fresh one`); + assert.equal(used, loads, `${name} ran on the instance that threw`); return caught; } - it('starts one fresh instance for many failures in one synchronous loop', async () => { + it('starts a fresh instance for every failure in one synchronous loop', async () => { await isolated.loadModule(); const loadsBefore = loads; - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 30; i++) { throwNext = new RangeError('Maximum call stack size exceeded'); assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads, 'a sync call ran on the instance that threw'); } - assert.equal(loads, loadsBefore + 1); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); + assert.equal(loads, loadsBefore + 30); }); - it('stops sync calls on a stale instance after 10 failures until the fresh one loads', async () => { + it('never runs a queued call on the instance that threw', async () => { await isolated.loadModule(); - for (let i = 0; i < 9; i++) { - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - } - assert.ok(isolated.parseSync('SELECT 1')); throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - assert.throws(() => isolated.parseSync('SELECT 1'), /Call `?loadModule\(\)`? first/); - await isolated.loadModule(); - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads); - }); - - it('makes loadModule() and queued calls wait again when a queued call breaks the fresh instance', async () => { - await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); const after = isolated.parse('SELECT 1'); await assert.rejects(failing, RangeError); assert.ok(await after); - assert.equal(used, loads, 'a queued call ran on the instance that just threw'); - await ready; - assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); - }); - - it('makes loadModule() wait again when a call queued ahead of it breaks the fresh instance', async () => { + assert.equal(used, loads, 'a queued call ran on the instance that threw'); await isolated.loadModule(); - throwNext = new RangeError('Maximum call stack size exceeded'); - assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); - const failing = isolated.parse('SELECT 1'); - throwNext = new RangeError('Maximum call stack size exceeded'); - const ready = isolated.loadModule(); - await assert.rejects(failing, RangeError); - await ready; assert.ok(isolated.parseSync('SELECT 1')); - assert.equal(used, loads, 'loadModule() resolved before the fresh instance loaded'); + assert.equal(used, loads); }); it('names a thrown object that has no message', async () => { @@ -330,11 +339,19 @@ describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => const caught = await failOnce(name, query, thrown); assert.ok(caught instanceof Error); assert.match(caught.message, /out of WASM memory/); - assert.equal(caught.cause, thrown); + assert.deepEqual(caught.cause, thrown); }); } }); +// Makes the loader's synchronous instantiation fail, as if the new instance's memory could not be allocated. +function failInstantiate(options) { + const instantiateWasm = options.instantiateWasm; + if (!instantiateWasm) return; + options.instantiateWasm = (imports) => + instantiateWasm(imports, () => { throw new Error('simulated instantiate failure'); }); +} + // A copy of the wrapper whose WASM factory is wrapped, leaving the shared copy alone. function loadIsolated(wrapFactory) { const indexPath = require.resolve('../wasm/index.cjs');