diff --git a/README.md b/README.md index 072f72b8..3dcd25af 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,16 @@ pnpm run test pnpm run clean && pnpm run build && pnpm run test ``` +### Slow recovery checks (local only) + +Run these after changing the WASM recovery code in `templates/`. Both use the +built `versions//wasm/` artifacts and are too slow for CI. + +```bash +pnpm test:stress [13 18 ...] # thousands of WASM crashes and >1 GiB inputs; memory must stay flat (a few minutes) +pnpm test:mutations [13 18 ...] # breaks each branch of the recovery code; the tests must catch every one (about 20 min) +``` + ## Troubleshooting ### Common Issues diff --git a/package.json b/package.json index ec171aa7..df537e2a 100644 --- a/package.json +++ b/package.json @@ -26,6 +26,8 @@ "publish:versions": "node scripts/publish-versions.js", "update:versions-types": "node scripts/update-versions-types.js", "copy:templates": "node scripts/copy-templates.js", + "test:stress": "node scripts/test-stress.js", + "test:mutations": "node scripts/test-mutations.js", "build:parser": "pnpm --filter @pgsql/parser build", "build:parser:lts": "PARSER_BUILD_TYPE=lts pnpm --filter @pgsql/parser build", "build:parser:full": "PARSER_BUILD_TYPE=full pnpm --filter @pgsql/parser build", diff --git a/scripts/test-mutations.js b/scripts/test-mutations.js new file mode 100644 index 00000000..76d0e038 --- /dev/null +++ b/scripts/test-mutations.js @@ -0,0 +1,111 @@ +#!/usr/bin/env node +// Checks that test/wasm-recovery.test.js covers every branch of the WASM +// recovery code: each mutation below breaks one branch of the built +// wasm/index.cjs, and the tests must fail (or hang) on every one of them. +// Too slow for CI (about 20 s per mutation); run it after changing the +// recovery code in templates/. +// +// usage: pnpm test:mutations [version ...] [--only ] (defaults to 13 and 18) + +const { spawnSync } = require('child_process'); +const fs = require('fs'); +const path = require('path'); + +const versionsDir = path.join(__dirname, '..', 'versions'); +const FINALLY = ' if (failures === failuresBefore) {'; + +const replace = (from, to) => (source) => source.replace(from, to); +const nth = (source, text, n, to) => { + let index = -1; + for (let i = 0; i <= n; i++) index = source.indexOf(text, index + 1); + return source.slice(0, index) + to + source.slice(index + text.length); +}; + +const MUTATIONS = [ + ['callWasm: never marks the instance stale', replace(' stale = true;\n', '')], + ['callWasm: does not count the failure', replace(' failures++;\n', '')], + ['callWasm: never starts a fresh instance', replace('wasmModule = undefined;\n instantiate();', 'wasmModule = undefined;\n')], + ['callWasm: drops the instance on the first failure', replace('if (++staleFailures >= MAX_STALE_FAILURES)', 'if (true)')], + ['callWasm: never drops a stale instance', replace('if (++staleFailures >= MAX_STALE_FAILURES)', 'if (false)')], + ['callWasm: cap off by one', replace('if (++staleFailures >= MAX_STALE_FAILURES)', 'if (++staleFailures > MAX_STALE_FAILURES)')], + ['callWasm: does not restore process.exitCode', replace(' if (proc)\n proc.exitCode = exitCode;\n', '')], + ['callWasm: does not remember process.exitCode', replace('const exitCode = proc?.exitCode;', 'const exitCode = undefined;')], + ['callWasm: wraps every error', replace('if (error instanceof Error)\n throw error;', 'if (false)\n throw error;')], + ['callWasm: never wraps non-Error throws', replace('if (error instanceof Error)\n throw error;', 'if (true)\n throw error;')], + ['callWasm: no constructor-name fallback', replace('(error.message ?? error.constructor?.name)', '(error.message)')], + ['instantiate: does not reset staleFailures', replace(' staleFailures = 0;\n', '')], + ['instantiate: does not clear stale', replace(' stale = false;\n staleFailures', ' staleFailures')], + ['instantiate: not single-flight', replace(' if (loading)\n return loading;\n', '')], + ['instantiate: success never clears loading', replace(' staleFailures = 0;\n loading = undefined;\n', ' staleFailures = 0;\n')], + ['instantiate: failure never clears loading', replace(' }, (error) => {\n loading = undefined;\n', ' }, (error) => {\n')], + ['instantiate: swallows a failed load', replace(' loading = undefined;\n throw error;\n', ' loading = undefined;\n')], + ['loadModule: ignores stale', replace(' while (!wasmModule || stale)\n await instantiate();\n}', ' while (!wasmModule)\n await instantiate();\n}')], + ['loadModule: waits only once', replace(' while (!wasmModule || stale)\n await instantiate();\n}', ' if (!wasmModule || stale)\n await instantiate();\n}')], + ['awaitInit: ignores stale', replace(' while (!wasmModule || stale)\n await instantiate();', ' while (!wasmModule)\n await instantiate();')], + ['awaitInit: waits only once', replace(' while (!wasmModule || stale)\n await instantiate();', ' if (!wasmModule || stale)\n await instantiate();')], +]; + +function functionAt(source, index) { + const pattern = /^(?:exports\.(\w+) = awaitInit\(async|function (\w+)\()/gm; + let name; + for (let match; (match = pattern.exec(source)) && match.index < index;) { + name = match[1] ? { name: match[1], async: true } : { name: match[2], async: false }; + } + return name; +} + +function mutationsFor(source) { + const list = [...MUTATIONS]; + for (let n = 0, index = source.indexOf(FINALLY); index !== -1; n++, index = source.indexOf(FINALLY, index + 1)) { + const fn = functionAt(source, index); + list.push([`${fn.name} cleanup: frees after the call threw`, (s) => nth(s, FINALLY, n, ' if (true) {')]); + // An async call only starts on a non-stale instance, so this one can't change async behavior. + if (!fn.async) { + list.push([`${fn.name} cleanup: skips frees whenever stale`, (s) => nth(s, FINALLY, n, ' if (!stale) {')]); + } + } + return list; +} + +const args = process.argv.slice(2); +const onlyAt = args.indexOf('--only'); +const only = onlyAt === -1 ? undefined : args.splice(onlyAt, 2)[1]; +const versions = args.length ? args : ['13', '18']; + +let problems = 0; +for (const version of versions) { + const dir = path.join(versionsDir, version); + const file = path.join(dir, 'wasm', 'index.cjs'); + if (!fs.existsSync(file)) { + console.error(`v${version}: ${file} not found; build it first`); + problems++; + continue; + } + const original = fs.readFileSync(file, 'utf8'); + const restore = () => fs.writeFileSync(file, original); + process.on('SIGINT', () => { restore(); process.exit(130); }); + const mutations = mutationsFor(original).filter(([name]) => !only || name.includes(only)); + let caught = 0; + try { + for (const [name, mutate] of mutations) { + const mutated = mutate(original); + if (mutated === original) { + console.log(`NOT APPLIED v${version} ${name}`); + problems++; + continue; + } + fs.writeFileSync(file, mutated); + const run = spawnSync(process.execPath, ['--test', 'test/wasm-recovery.test.js'], { + cwd: dir, encoding: 'utf8', timeout: 150000, stdio: ['ignore', 'pipe', 'ignore'], + }); + const result = run.error?.code === 'ETIMEDOUT' ? 'caught (hangs)' : run.status !== 0 ? 'caught' : 'MISSED'; + if (result === 'MISSED') problems++; + else caught++; + console.log(`${result.padEnd(14)} v${version} ${name}`); + } + } finally { + restore(); + } + console.log(`v${version}: ${caught}/${mutations.length} mutations caught\n`); +} +process.exit(problems ? 1 : 0); diff --git a/scripts/test-stress.js b/scripts/test-stress.js new file mode 100644 index 00000000..f357eebe --- /dev/null +++ b/scripts/test-stress.js @@ -0,0 +1,104 @@ +#!/usr/bin/env node +// Long-running recovery checks that are too slow for CI. Each scenario runs in +// a fresh Node process with GC exposed and fails if WASM memory (V8 `external`) +// grows, `SELECT 1` stops working, or process.exitCode changes. +// +// usage: pnpm test:stress [version ...] (defaults to every built version) + +const { execFileSync } = require('child_process'); +const fs = require('fs'); +const path = require('path'); + +const versionsDir = path.join(__dirname, '..', 'versions'); +const MAX_GROWTH = 32 * 1024 * 1024; + +const SCENARIOS = { + 'async: 1,000 stack overflows': ` + for (let i = 0; i < 1000; i++) { + await lib.parse(DEEP).catch(() => {}); + check(i); + }`, + 'sync: 1,000 stack overflows, awaiting loadModule() after each': ` + for (let i = 0; i < 1000; i++) { + try { lib.parseSync(DEEP); } catch {} + await lib.loadModule(); + check(i); + }`, + 'sync: 200 stack overflows in one synchronous loop, 20 times': ` + for (let i = 0; i < 20; i++) { + for (let j = 0; j < 200; j++) { try { lib.parseSync(DEEP); } catch {} } + await lib.loadModule(); + check(i); + }`, + 'async: 20 inputs over the 1 GiB WASM limit': ` + for (let i = 0; i < 20; i++) { + await lib.parse(HUGE).catch(() => {}); + check(i); + }`, + 'control: 1,000 normal parses': ` + for (let i = 0; i < 1000; i++) { + await lib.parse('SELECT 1' + '+1'.repeat(2000)); + check(i); + }`, +}; + +function runScenario(indexPath, body) { + const script = ` + const lib = require(${JSON.stringify(indexPath)}); + const DEEP = 'SELECT 1' + '+1'.repeat(100000); + const HUGE = 'SELECT 1;\\n'.repeat(5000000); + let broken = null; + function check(i) { + try { lib.parseSync('SELECT 1'); } catch { if (broken === null) broken = i; } + } + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + ${body} + await lib.loadModule(); + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + const grown = process.memoryUsage().external - base; + process.stdout.write('\\nresult=' + JSON.stringify({ grown, broken, exitCode: process.exitCode ?? null })); + })();`; + // libpg_query prints a memory dump to stdout on out-of-memory, so read the marked line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + maxBuffer: 64 * 1024 * 1024, + }); + return JSON.parse(output.match(/^result=(.*)$/m)[1]); +} + +const requested = process.argv.slice(2); +const versions = (requested.length ? requested : fs.readdirSync(versionsDir)) + .filter((v) => fs.existsSync(path.join(versionsDir, v, 'wasm', 'index.cjs'))); +if (!versions.length) { + console.error('No built versions found; run `pnpm build` in versions/ first.'); + process.exit(1); +} + +let failed = 0; +for (const version of versions) { + const indexPath = path.join(versionsDir, version, 'wasm', 'index.cjs'); + for (const [name, body] of Object.entries(SCENARIOS)) { + const start = Date.now(); + let problem; + let result; + try { + result = runScenario(indexPath, body); + if (result.grown >= MAX_GROWTH) problem = `WASM memory grew by ${(result.grown / 1048576).toFixed(1)} MB`; + else if (result.broken !== null) problem = `SELECT 1 failed from iteration ${result.broken}`; + else if (result.exitCode !== null) problem = `process.exitCode became ${result.exitCode}`; + } catch (error) { + problem = `process failed: ${error.message.split('\n')[0]}`; + } + const seconds = ((Date.now() - start) / 1000).toFixed(1); + const growth = result ? `${(result.grown / 1048576).toFixed(1)} MB` : '-'; + console.log(`${problem ? 'FAIL' : 'ok '} v${version} ${name} (${seconds}s, memory ${growth})${problem ? `: ${problem}` : ''}`); + if (problem) failed++; + } +} +process.exit(failed ? 1 : 0); diff --git a/templates/full/index.ts b/templates/full/index.ts index f41240c3..682f05b5 100644 --- a/templates/full/index.ts +++ b/templates/full/index.ts @@ -136,23 +136,78 @@ interface WasmModule { let wasmModule: WasmModule; -const initPromise = PgQueryModule().then((module: WasmModule) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule(): Promise { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit Promise>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -200,10 +255,11 @@ export const parse = awaitInit(async (query: string): Promise => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to parse query: memory allocation failed'); } @@ -241,19 +297,22 @@ export const parse = awaitInit(async (query: string): Promise => { const parseTreeStr = wasmModule.UTF8ToString(parseTreePtr); return JSON.parse(parseTreeStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); export const parsePlPgSQL = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_plpgsql(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_plpgsql(queryPtr)); const resultStr = ptrToString(resultPtr); // Success is always a JSON object; anything else is an error message @@ -263,19 +322,22 @@ export const parsePlPgSQL = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_fingerprint(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_fingerprint(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -284,19 +346,22 @@ export const fingerprint = awaitInit(async (query: string): Promise => { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } }); export const normalize = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_normalize_query(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_normalize_query(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -305,9 +370,11 @@ export const normalize = awaitInit(async (query: string): Promise => { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } }); @@ -328,10 +395,11 @@ export function parseSync(query: string): ParseResult { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to parse query: memory allocation failed'); } @@ -369,9 +437,11 @@ export function parseSync(query: string): ParseResult { const parseTreeStr = wasmModule.UTF8ToString(parseTreePtr); return JSON.parse(parseTreeStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } @@ -381,10 +451,11 @@ export function parsePlPgSQLSync(query: string): ParseResult { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_plpgsql(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_plpgsql(queryPtr)); const resultStr = ptrToString(resultPtr); // Success is always a JSON object; anything else is an error message @@ -394,9 +465,11 @@ export function parsePlPgSQLSync(query: string): ParseResult { return JSON.parse(resultStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } @@ -406,10 +479,11 @@ export function fingerprintSync(query: string): string { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_fingerprint(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_fingerprint(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -418,9 +492,11 @@ export function fingerprintSync(query: string): string { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } @@ -430,10 +506,11 @@ export function normalizeSync(query: string): string { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_normalize_query(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_normalize_query(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -442,19 +519,22 @@ export function normalizeSync(query: string): string { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } export const scan = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_scan(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_scan(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -463,9 +543,11 @@ export const scan = awaitInit(async (query: string): Promise => { return JSON.parse(resultStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } }); @@ -475,10 +557,11 @@ export function scanSync(query: string): ScanResult { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_scan(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_scan(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -487,9 +570,11 @@ export function scanSync(query: string): ScanResult { return JSON.parse(resultStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } \ No newline at end of file diff --git a/templates/index.ts b/templates/index.ts index 99241255..4a9a8929 100644 --- a/templates/index.ts +++ b/templates/index.ts @@ -124,23 +124,78 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -const initPromise = PgQueryModule().then((module: any) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -190,11 +245,12 @@ export const parse = awaitInit(async (query: string) => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -239,9 +295,11 @@ export const parse = awaitInit(async (query: string) => { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); @@ -259,11 +317,12 @@ export function parseSync(query: string) { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -308,9 +367,11 @@ export function parseSync(query: string) { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/13/package.json b/versions/13/package.json index 083666bf..1b39e210 100644 --- a/versions/13/package.json +++ b/versions/13/package.json @@ -28,7 +28,7 @@ "wasm:rebuild": "pnpm wasm:make rebuild", "wasm:clean": "pnpm wasm:make clean", "wasm:clean-cache": "pnpm wasm:make clean-cache", - "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js" + "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js test/wasm-recovery.test.js" }, "author": "Constructive ", "license": "MIT", diff --git a/versions/13/src/index.ts b/versions/13/src/index.ts index 51e15b6b..ea70bb2a 100644 --- a/versions/13/src/index.ts +++ b/versions/13/src/index.ts @@ -131,23 +131,78 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -const initPromise = PgQueryModule().then((module: any) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -197,11 +252,12 @@ export const parse = awaitInit(async (query: string) => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -246,9 +302,11 @@ export const parse = awaitInit(async (query: string) => { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); @@ -266,11 +324,12 @@ export function parseSync(query: string) { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -315,9 +374,11 @@ export function parseSync(query: string) { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/13/test/wasm-recovery.test.js b/versions/13/test/wasm-recovery.test.js new file mode 100644 index 00000000..6d16d379 --- /dev/null +++ b/versions/13/test/wasm-recovery.test.js @@ -0,0 +1,352 @@ +const { describe, it, before } = require('node:test'); +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const lib = require('../wasm/index.cjs'); +const { parse, parseSync, loadModule } = lib; + +const DEEP = 'SELECT 1' + '+1'.repeat(100000); + +// An exception out of the WASM parser skips libpg_query's cleanup. Before the +// wrapper replaced the instance, enough of them made even `SELECT 1` trap. +describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { + before(async () => { + await loadModule(); + }); + + it('stays usable after repeated stack overflows on deeply nested SQL', async () => { + for (let i = 0; i < 60; i++) { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + } + await loadModule(); + 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 () => { + await loadModule(); + const errors = []; + for (let i = 0; i < 60; i++) { + try { + parseSync(DEEP); + } catch (error) { + errors.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); + }); + + it('frees each replaced instance, so WASM memory stays at one instance', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + const deep = 'SELECT 1' + '+1'.repeat(100000); + for (let i = 0; i < 30; i++) { + try { lib.parseSync(deep); } catch {} + await lib.loadModule(); + } + const huge = 'SELECT 1;\\n'.repeat(5000000); + for (let i = 0; i < 2; i++) { + try { await lib.parse(huge); } catch {} + } + await lib.loadModule(); + lib.parseSync('SELECT 1'); + // V8 frees a dropped instance's memory on a later task, not in the same microtask run. + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + process.stdout.write('\\ngrown=' + (process.memoryUsage().external - base)); + })();`; + // A non-zero exit code (libpg_query's exit() leaking out) makes execFileSync throw. + // libpg_query prints a memory dump to stdout on out-of-memory, so read the last line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const grown = Number(output.match(/^grown=(-?\d+)$/m)?.[1]); + assert.ok(grown < 32 * 1024 * 1024, `WASM memory grew by ${grown} bytes after 32 replacements`); + }); + + it('lets a call queued behind a failing one wait for the fresh instance', async () => { + await loadModule(); + const [failed, queued] = await Promise.allSettled([ + parse(DEEP), + parse('SELECT 1'), + ]); + assert.equal(failed.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); + + it('throws an Error and keeps process.exitCode when the memory limit is hit', async () => { + const exitCode = process.exitCode; + await assert.rejects(parse('SELECT 1;\n'.repeat(5000000)), (error) => { + assert.ok(error instanceof Error); + assert.match(error.message, /out of WASM memory/); + return true; + }); + assert.equal(process.exitCode, exitCode); + assert.equal((await parse('SELECT 1')).stmts.length, 1); + }); + + for (const name of ['parsePlPgSQL', 'fingerprint', 'normalize', 'scan']) { + const skip = typeof lib[name] !== 'function' && 'not in this build'; + it(`${name}: rejects a 50 MB input, keeps process.exitCode and recovers`, { skip }, async () => { + const exitCode = process.exitCode; + await assert.rejects(lib[name]('SELECT 1;\n'.repeat(5000000))); + assert.equal(process.exitCode, exitCode); + assert.ok(await lib[name](name === 'parsePlPgSQL' + ? 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql' + : 'SELECT 1')); + assert.ok(lib[`${name}Sync`]('SELECT 1')); + }); + } + + it('retries the load when loading a fresh instance fails', async () => { + let loads = 0; + const isolated = loadIsolated((real, ...args) => + ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + await isolated.loadModule(); + await assert.rejects(isolated.parse(DEEP)); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal((await isolated.parse('SELECT 1')).stmts.length, 1); + assert.equal(loads, 3); + }); + + it('lets loadModule() reject on a failed reload and retry on the next 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 assert.rejects(isolated.loadModule(), /simulated load failure/); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); + }); + + it('keeps sync calls in the same synchronous loop working after a failure', () => { + const queries = Array.from({ length: 200 }, (_, i) => `SELECT ${i}, 'x${i}' FROM t${i} WHERE a = ${i}`); + const expected = queries.map((query) => JSON.stringify(parseSync(query))); + let failed = 0; + const results = []; + for (const [i, query] of queries.entries()) { + if (i === 10) { + try { parseSync(DEEP); } catch { failed++; } + } + results.push(JSON.stringify(parseSync(query))); + } + assert.equal(failed, 1); + assert.deepEqual(results, expected); + }); + + it('retries a failed fresh load on the next failure, for callers that never await', 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); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + }); + + it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + const failing = parse(DEEP); + const ready = loadModule(); + await assert.rejects(failing); + await ready; + assert.equal(parseSync('SELECT 1').stmts.length, 1); + }); + + it('keeps every call queued behind two failures waiting for a working instance', async () => { + await loadModule(); + const [first, second, queued] = await Promise.allSettled([parse(DEEP), parse(DEEP), parse('SELECT 1')]); + assert.equal(first.status, 'rejected'); + assert.equal(second.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); +}); + +// Every entry point, with the next WASM call made to throw: a RangeError +// stands for a stack overflow, a non-Error object for Emscripten's ExitStatus. +describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => { + const PLPGSQL = 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql'; + const entryPoints = [ + ['parse', 'SELECT 1'], + ['parseSync', 'SELECT 1'], + ['parsePlPgSQL', PLPGSQL], + ['parsePlPgSQLSync', PLPGSQL], + ['fingerprint', 'SELECT 1'], + ['fingerprintSync', 'SELECT 1'], + ['normalize', 'SELECT 1'], + ['normalizeSync', 'SELECT 1'], + ['scan', 'SELECT 1'], + ['scanSync', 'SELECT 1'], + ]; + + // Instances are numbered by load; `used` is the one the last WASM call ran on. + let loads = 0; + let used; + let frees = 0; + let throwNext; + const isolated = loadIsolated(async (real, ...args) => { + const id = ++loads; + const module = await real(...args); + const free = module._free; + module._free = (ptr) => { + frees++; + return free(ptr); + }; + for (const name of Object.keys(module).filter((key) => key.startsWith('_wasm_'))) { + const call = module[name]; + module[name] = (...callArgs) => { + used = id; + if (throwNext === undefined) return call(...callArgs); + const thrown = throwNext; + throwNext = undefined; + process.exitCode = 7; + throw thrown; + }; + } + return module; + }); + + async function failOnce(name, query, thrown) { + await isolated.loadModule(); + const loadsBefore = loads; + const exitCode = process.exitCode; + process.exitCode = 0; + const freesBefore = frees; + throwNext = thrown; + let caught; + try { + await isolated[name](query); + } catch (error) { + caught = error; + } finally { + var exitCodeAfter = process.exitCode; + process.exitCode = exitCode; + } + 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(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`); + return caught; + } + + it('starts one fresh instance for many failures in one synchronous loop', async () => { + await isolated.loadModule(); + const loadsBefore = loads; + for (let i = 0; i < 5; i++) { + throwNext = new RangeError('Maximum call stack size exceeded'); + assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + } + assert.equal(loads, loadsBefore + 1); + await isolated.loadModule(); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads); + }); + + it('stops sync calls on a stale instance after 10 failures until the fresh one loads', 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 () => { + 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'); + }); + + it('names a thrown object that has no message', async () => { + class EmscriptenSjLj {} + const caught = await failOnce('parseSync', 'SELECT 1', new EmscriptenSjLj()); + assert.match(caught.message, /libpg_query terminated \(EmscriptenSjLj\)/); + }); + + for (const [name, query] of entryPoints) { + const skip = typeof isolated[name] !== 'function' && 'not in this build'; + + it(`${name}: rethrows the original Error and recovers`, { skip }, async () => { + const thrown = new RangeError('Maximum call stack size exceeded'); + assert.equal(await failOnce(name, query, thrown), thrown); + }); + + it(`${name}: wraps a non-Error throw in an Error and recovers`, { skip }, async () => { + const thrown = { name: 'ExitStatus', message: 'Program terminated with exit(1)', status: 1 }; + 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); + }); + } +}); + +// 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'); + const factory = require.cache[require.resolve('../wasm/libpg-query.js')]; + const shared = require.cache[indexPath]; + const realFactory = factory.exports; + factory.exports = (...args) => wrapFactory(realFactory, ...args); + delete require.cache[indexPath]; + try { + return require(indexPath); + } finally { + factory.exports = realFactory; + require.cache[indexPath] = shared; + } +} diff --git a/versions/14/package.json b/versions/14/package.json index 35d8a921..0149a88a 100644 --- a/versions/14/package.json +++ b/versions/14/package.json @@ -28,7 +28,7 @@ "wasm:rebuild": "pnpm wasm:make rebuild", "wasm:clean": "pnpm wasm:make clean", "wasm:clean-cache": "pnpm wasm:make clean-cache", - "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js" + "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js test/wasm-recovery.test.js" }, "author": "Constructive ", "license": "MIT", diff --git a/versions/14/src/index.ts b/versions/14/src/index.ts index 51e15b6b..ea70bb2a 100644 --- a/versions/14/src/index.ts +++ b/versions/14/src/index.ts @@ -131,23 +131,78 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -const initPromise = PgQueryModule().then((module: any) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -197,11 +252,12 @@ export const parse = awaitInit(async (query: string) => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -246,9 +302,11 @@ export const parse = awaitInit(async (query: string) => { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); @@ -266,11 +324,12 @@ export function parseSync(query: string) { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -315,9 +374,11 @@ export function parseSync(query: string) { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/14/test/wasm-recovery.test.js b/versions/14/test/wasm-recovery.test.js new file mode 100644 index 00000000..6d16d379 --- /dev/null +++ b/versions/14/test/wasm-recovery.test.js @@ -0,0 +1,352 @@ +const { describe, it, before } = require('node:test'); +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const lib = require('../wasm/index.cjs'); +const { parse, parseSync, loadModule } = lib; + +const DEEP = 'SELECT 1' + '+1'.repeat(100000); + +// An exception out of the WASM parser skips libpg_query's cleanup. Before the +// wrapper replaced the instance, enough of them made even `SELECT 1` trap. +describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { + before(async () => { + await loadModule(); + }); + + it('stays usable after repeated stack overflows on deeply nested SQL', async () => { + for (let i = 0; i < 60; i++) { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + } + await loadModule(); + 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 () => { + await loadModule(); + const errors = []; + for (let i = 0; i < 60; i++) { + try { + parseSync(DEEP); + } catch (error) { + errors.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); + }); + + it('frees each replaced instance, so WASM memory stays at one instance', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + const deep = 'SELECT 1' + '+1'.repeat(100000); + for (let i = 0; i < 30; i++) { + try { lib.parseSync(deep); } catch {} + await lib.loadModule(); + } + const huge = 'SELECT 1;\\n'.repeat(5000000); + for (let i = 0; i < 2; i++) { + try { await lib.parse(huge); } catch {} + } + await lib.loadModule(); + lib.parseSync('SELECT 1'); + // V8 frees a dropped instance's memory on a later task, not in the same microtask run. + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + process.stdout.write('\\ngrown=' + (process.memoryUsage().external - base)); + })();`; + // A non-zero exit code (libpg_query's exit() leaking out) makes execFileSync throw. + // libpg_query prints a memory dump to stdout on out-of-memory, so read the last line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const grown = Number(output.match(/^grown=(-?\d+)$/m)?.[1]); + assert.ok(grown < 32 * 1024 * 1024, `WASM memory grew by ${grown} bytes after 32 replacements`); + }); + + it('lets a call queued behind a failing one wait for the fresh instance', async () => { + await loadModule(); + const [failed, queued] = await Promise.allSettled([ + parse(DEEP), + parse('SELECT 1'), + ]); + assert.equal(failed.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); + + it('throws an Error and keeps process.exitCode when the memory limit is hit', async () => { + const exitCode = process.exitCode; + await assert.rejects(parse('SELECT 1;\n'.repeat(5000000)), (error) => { + assert.ok(error instanceof Error); + assert.match(error.message, /out of WASM memory/); + return true; + }); + assert.equal(process.exitCode, exitCode); + assert.equal((await parse('SELECT 1')).stmts.length, 1); + }); + + for (const name of ['parsePlPgSQL', 'fingerprint', 'normalize', 'scan']) { + const skip = typeof lib[name] !== 'function' && 'not in this build'; + it(`${name}: rejects a 50 MB input, keeps process.exitCode and recovers`, { skip }, async () => { + const exitCode = process.exitCode; + await assert.rejects(lib[name]('SELECT 1;\n'.repeat(5000000))); + assert.equal(process.exitCode, exitCode); + assert.ok(await lib[name](name === 'parsePlPgSQL' + ? 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql' + : 'SELECT 1')); + assert.ok(lib[`${name}Sync`]('SELECT 1')); + }); + } + + it('retries the load when loading a fresh instance fails', async () => { + let loads = 0; + const isolated = loadIsolated((real, ...args) => + ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + await isolated.loadModule(); + await assert.rejects(isolated.parse(DEEP)); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal((await isolated.parse('SELECT 1')).stmts.length, 1); + assert.equal(loads, 3); + }); + + it('lets loadModule() reject on a failed reload and retry on the next 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 assert.rejects(isolated.loadModule(), /simulated load failure/); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); + }); + + it('keeps sync calls in the same synchronous loop working after a failure', () => { + const queries = Array.from({ length: 200 }, (_, i) => `SELECT ${i}, 'x${i}' FROM t${i} WHERE a = ${i}`); + const expected = queries.map((query) => JSON.stringify(parseSync(query))); + let failed = 0; + const results = []; + for (const [i, query] of queries.entries()) { + if (i === 10) { + try { parseSync(DEEP); } catch { failed++; } + } + results.push(JSON.stringify(parseSync(query))); + } + assert.equal(failed, 1); + assert.deepEqual(results, expected); + }); + + it('retries a failed fresh load on the next failure, for callers that never await', 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); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + }); + + it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + const failing = parse(DEEP); + const ready = loadModule(); + await assert.rejects(failing); + await ready; + assert.equal(parseSync('SELECT 1').stmts.length, 1); + }); + + it('keeps every call queued behind two failures waiting for a working instance', async () => { + await loadModule(); + const [first, second, queued] = await Promise.allSettled([parse(DEEP), parse(DEEP), parse('SELECT 1')]); + assert.equal(first.status, 'rejected'); + assert.equal(second.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); +}); + +// Every entry point, with the next WASM call made to throw: a RangeError +// stands for a stack overflow, a non-Error object for Emscripten's ExitStatus. +describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => { + const PLPGSQL = 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql'; + const entryPoints = [ + ['parse', 'SELECT 1'], + ['parseSync', 'SELECT 1'], + ['parsePlPgSQL', PLPGSQL], + ['parsePlPgSQLSync', PLPGSQL], + ['fingerprint', 'SELECT 1'], + ['fingerprintSync', 'SELECT 1'], + ['normalize', 'SELECT 1'], + ['normalizeSync', 'SELECT 1'], + ['scan', 'SELECT 1'], + ['scanSync', 'SELECT 1'], + ]; + + // Instances are numbered by load; `used` is the one the last WASM call ran on. + let loads = 0; + let used; + let frees = 0; + let throwNext; + const isolated = loadIsolated(async (real, ...args) => { + const id = ++loads; + const module = await real(...args); + const free = module._free; + module._free = (ptr) => { + frees++; + return free(ptr); + }; + for (const name of Object.keys(module).filter((key) => key.startsWith('_wasm_'))) { + const call = module[name]; + module[name] = (...callArgs) => { + used = id; + if (throwNext === undefined) return call(...callArgs); + const thrown = throwNext; + throwNext = undefined; + process.exitCode = 7; + throw thrown; + }; + } + return module; + }); + + async function failOnce(name, query, thrown) { + await isolated.loadModule(); + const loadsBefore = loads; + const exitCode = process.exitCode; + process.exitCode = 0; + const freesBefore = frees; + throwNext = thrown; + let caught; + try { + await isolated[name](query); + } catch (error) { + caught = error; + } finally { + var exitCodeAfter = process.exitCode; + process.exitCode = exitCode; + } + 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(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`); + return caught; + } + + it('starts one fresh instance for many failures in one synchronous loop', async () => { + await isolated.loadModule(); + const loadsBefore = loads; + for (let i = 0; i < 5; i++) { + throwNext = new RangeError('Maximum call stack size exceeded'); + assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + } + assert.equal(loads, loadsBefore + 1); + await isolated.loadModule(); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads); + }); + + it('stops sync calls on a stale instance after 10 failures until the fresh one loads', 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 () => { + 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'); + }); + + it('names a thrown object that has no message', async () => { + class EmscriptenSjLj {} + const caught = await failOnce('parseSync', 'SELECT 1', new EmscriptenSjLj()); + assert.match(caught.message, /libpg_query terminated \(EmscriptenSjLj\)/); + }); + + for (const [name, query] of entryPoints) { + const skip = typeof isolated[name] !== 'function' && 'not in this build'; + + it(`${name}: rethrows the original Error and recovers`, { skip }, async () => { + const thrown = new RangeError('Maximum call stack size exceeded'); + assert.equal(await failOnce(name, query, thrown), thrown); + }); + + it(`${name}: wraps a non-Error throw in an Error and recovers`, { skip }, async () => { + const thrown = { name: 'ExitStatus', message: 'Program terminated with exit(1)', status: 1 }; + 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); + }); + } +}); + +// 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'); + const factory = require.cache[require.resolve('../wasm/libpg-query.js')]; + const shared = require.cache[indexPath]; + const realFactory = factory.exports; + factory.exports = (...args) => wrapFactory(realFactory, ...args); + delete require.cache[indexPath]; + try { + return require(indexPath); + } finally { + factory.exports = realFactory; + require.cache[indexPath] = shared; + } +} diff --git a/versions/15/package.json b/versions/15/package.json index e8ce05b5..a5e9d877 100644 --- a/versions/15/package.json +++ b/versions/15/package.json @@ -28,7 +28,7 @@ "wasm:rebuild": "pnpm wasm:make rebuild", "wasm:clean": "pnpm wasm:make clean", "wasm:clean-cache": "pnpm wasm:make clean-cache", - "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js" + "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js test/wasm-recovery.test.js" }, "author": "Constructive ", "license": "MIT", diff --git a/versions/15/src/index.ts b/versions/15/src/index.ts index 51e15b6b..ea70bb2a 100644 --- a/versions/15/src/index.ts +++ b/versions/15/src/index.ts @@ -131,23 +131,78 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -const initPromise = PgQueryModule().then((module: any) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -197,11 +252,12 @@ export const parse = awaitInit(async (query: string) => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -246,9 +302,11 @@ export const parse = awaitInit(async (query: string) => { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); @@ -266,11 +324,12 @@ export function parseSync(query: string) { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -315,9 +374,11 @@ export function parseSync(query: string) { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/15/test/wasm-recovery.test.js b/versions/15/test/wasm-recovery.test.js new file mode 100644 index 00000000..6d16d379 --- /dev/null +++ b/versions/15/test/wasm-recovery.test.js @@ -0,0 +1,352 @@ +const { describe, it, before } = require('node:test'); +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const lib = require('../wasm/index.cjs'); +const { parse, parseSync, loadModule } = lib; + +const DEEP = 'SELECT 1' + '+1'.repeat(100000); + +// An exception out of the WASM parser skips libpg_query's cleanup. Before the +// wrapper replaced the instance, enough of them made even `SELECT 1` trap. +describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { + before(async () => { + await loadModule(); + }); + + it('stays usable after repeated stack overflows on deeply nested SQL', async () => { + for (let i = 0; i < 60; i++) { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + } + await loadModule(); + 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 () => { + await loadModule(); + const errors = []; + for (let i = 0; i < 60; i++) { + try { + parseSync(DEEP); + } catch (error) { + errors.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); + }); + + it('frees each replaced instance, so WASM memory stays at one instance', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + const deep = 'SELECT 1' + '+1'.repeat(100000); + for (let i = 0; i < 30; i++) { + try { lib.parseSync(deep); } catch {} + await lib.loadModule(); + } + const huge = 'SELECT 1;\\n'.repeat(5000000); + for (let i = 0; i < 2; i++) { + try { await lib.parse(huge); } catch {} + } + await lib.loadModule(); + lib.parseSync('SELECT 1'); + // V8 frees a dropped instance's memory on a later task, not in the same microtask run. + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + process.stdout.write('\\ngrown=' + (process.memoryUsage().external - base)); + })();`; + // A non-zero exit code (libpg_query's exit() leaking out) makes execFileSync throw. + // libpg_query prints a memory dump to stdout on out-of-memory, so read the last line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const grown = Number(output.match(/^grown=(-?\d+)$/m)?.[1]); + assert.ok(grown < 32 * 1024 * 1024, `WASM memory grew by ${grown} bytes after 32 replacements`); + }); + + it('lets a call queued behind a failing one wait for the fresh instance', async () => { + await loadModule(); + const [failed, queued] = await Promise.allSettled([ + parse(DEEP), + parse('SELECT 1'), + ]); + assert.equal(failed.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); + + it('throws an Error and keeps process.exitCode when the memory limit is hit', async () => { + const exitCode = process.exitCode; + await assert.rejects(parse('SELECT 1;\n'.repeat(5000000)), (error) => { + assert.ok(error instanceof Error); + assert.match(error.message, /out of WASM memory/); + return true; + }); + assert.equal(process.exitCode, exitCode); + assert.equal((await parse('SELECT 1')).stmts.length, 1); + }); + + for (const name of ['parsePlPgSQL', 'fingerprint', 'normalize', 'scan']) { + const skip = typeof lib[name] !== 'function' && 'not in this build'; + it(`${name}: rejects a 50 MB input, keeps process.exitCode and recovers`, { skip }, async () => { + const exitCode = process.exitCode; + await assert.rejects(lib[name]('SELECT 1;\n'.repeat(5000000))); + assert.equal(process.exitCode, exitCode); + assert.ok(await lib[name](name === 'parsePlPgSQL' + ? 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql' + : 'SELECT 1')); + assert.ok(lib[`${name}Sync`]('SELECT 1')); + }); + } + + it('retries the load when loading a fresh instance fails', async () => { + let loads = 0; + const isolated = loadIsolated((real, ...args) => + ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + await isolated.loadModule(); + await assert.rejects(isolated.parse(DEEP)); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal((await isolated.parse('SELECT 1')).stmts.length, 1); + assert.equal(loads, 3); + }); + + it('lets loadModule() reject on a failed reload and retry on the next 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 assert.rejects(isolated.loadModule(), /simulated load failure/); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); + }); + + it('keeps sync calls in the same synchronous loop working after a failure', () => { + const queries = Array.from({ length: 200 }, (_, i) => `SELECT ${i}, 'x${i}' FROM t${i} WHERE a = ${i}`); + const expected = queries.map((query) => JSON.stringify(parseSync(query))); + let failed = 0; + const results = []; + for (const [i, query] of queries.entries()) { + if (i === 10) { + try { parseSync(DEEP); } catch { failed++; } + } + results.push(JSON.stringify(parseSync(query))); + } + assert.equal(failed, 1); + assert.deepEqual(results, expected); + }); + + it('retries a failed fresh load on the next failure, for callers that never await', 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); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + }); + + it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + const failing = parse(DEEP); + const ready = loadModule(); + await assert.rejects(failing); + await ready; + assert.equal(parseSync('SELECT 1').stmts.length, 1); + }); + + it('keeps every call queued behind two failures waiting for a working instance', async () => { + await loadModule(); + const [first, second, queued] = await Promise.allSettled([parse(DEEP), parse(DEEP), parse('SELECT 1')]); + assert.equal(first.status, 'rejected'); + assert.equal(second.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); +}); + +// Every entry point, with the next WASM call made to throw: a RangeError +// stands for a stack overflow, a non-Error object for Emscripten's ExitStatus. +describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => { + const PLPGSQL = 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql'; + const entryPoints = [ + ['parse', 'SELECT 1'], + ['parseSync', 'SELECT 1'], + ['parsePlPgSQL', PLPGSQL], + ['parsePlPgSQLSync', PLPGSQL], + ['fingerprint', 'SELECT 1'], + ['fingerprintSync', 'SELECT 1'], + ['normalize', 'SELECT 1'], + ['normalizeSync', 'SELECT 1'], + ['scan', 'SELECT 1'], + ['scanSync', 'SELECT 1'], + ]; + + // Instances are numbered by load; `used` is the one the last WASM call ran on. + let loads = 0; + let used; + let frees = 0; + let throwNext; + const isolated = loadIsolated(async (real, ...args) => { + const id = ++loads; + const module = await real(...args); + const free = module._free; + module._free = (ptr) => { + frees++; + return free(ptr); + }; + for (const name of Object.keys(module).filter((key) => key.startsWith('_wasm_'))) { + const call = module[name]; + module[name] = (...callArgs) => { + used = id; + if (throwNext === undefined) return call(...callArgs); + const thrown = throwNext; + throwNext = undefined; + process.exitCode = 7; + throw thrown; + }; + } + return module; + }); + + async function failOnce(name, query, thrown) { + await isolated.loadModule(); + const loadsBefore = loads; + const exitCode = process.exitCode; + process.exitCode = 0; + const freesBefore = frees; + throwNext = thrown; + let caught; + try { + await isolated[name](query); + } catch (error) { + caught = error; + } finally { + var exitCodeAfter = process.exitCode; + process.exitCode = exitCode; + } + 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(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`); + return caught; + } + + it('starts one fresh instance for many failures in one synchronous loop', async () => { + await isolated.loadModule(); + const loadsBefore = loads; + for (let i = 0; i < 5; i++) { + throwNext = new RangeError('Maximum call stack size exceeded'); + assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + } + assert.equal(loads, loadsBefore + 1); + await isolated.loadModule(); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads); + }); + + it('stops sync calls on a stale instance after 10 failures until the fresh one loads', 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 () => { + 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'); + }); + + it('names a thrown object that has no message', async () => { + class EmscriptenSjLj {} + const caught = await failOnce('parseSync', 'SELECT 1', new EmscriptenSjLj()); + assert.match(caught.message, /libpg_query terminated \(EmscriptenSjLj\)/); + }); + + for (const [name, query] of entryPoints) { + const skip = typeof isolated[name] !== 'function' && 'not in this build'; + + it(`${name}: rethrows the original Error and recovers`, { skip }, async () => { + const thrown = new RangeError('Maximum call stack size exceeded'); + assert.equal(await failOnce(name, query, thrown), thrown); + }); + + it(`${name}: wraps a non-Error throw in an Error and recovers`, { skip }, async () => { + const thrown = { name: 'ExitStatus', message: 'Program terminated with exit(1)', status: 1 }; + 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); + }); + } +}); + +// 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'); + const factory = require.cache[require.resolve('../wasm/libpg-query.js')]; + const shared = require.cache[indexPath]; + const realFactory = factory.exports; + factory.exports = (...args) => wrapFactory(realFactory, ...args); + delete require.cache[indexPath]; + try { + return require(indexPath); + } finally { + factory.exports = realFactory; + require.cache[indexPath] = shared; + } +} diff --git a/versions/16/package.json b/versions/16/package.json index f168fb2b..59404315 100644 --- a/versions/16/package.json +++ b/versions/16/package.json @@ -28,7 +28,7 @@ "wasm:rebuild": "pnpm wasm:make rebuild", "wasm:clean": "pnpm wasm:make clean", "wasm:clean-cache": "pnpm wasm:make clean-cache", - "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js" + "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js test/wasm-recovery.test.js" }, "author": "Constructive ", "license": "MIT", diff --git a/versions/16/src/index.ts b/versions/16/src/index.ts index 51e15b6b..ea70bb2a 100644 --- a/versions/16/src/index.ts +++ b/versions/16/src/index.ts @@ -131,23 +131,78 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -const initPromise = PgQueryModule().then((module: any) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -197,11 +252,12 @@ export const parse = awaitInit(async (query: string) => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -246,9 +302,11 @@ export const parse = awaitInit(async (query: string) => { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); @@ -266,11 +324,12 @@ export function parseSync(query: string) { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -315,9 +374,11 @@ export function parseSync(query: string) { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/16/test/wasm-recovery.test.js b/versions/16/test/wasm-recovery.test.js new file mode 100644 index 00000000..6d16d379 --- /dev/null +++ b/versions/16/test/wasm-recovery.test.js @@ -0,0 +1,352 @@ +const { describe, it, before } = require('node:test'); +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const lib = require('../wasm/index.cjs'); +const { parse, parseSync, loadModule } = lib; + +const DEEP = 'SELECT 1' + '+1'.repeat(100000); + +// An exception out of the WASM parser skips libpg_query's cleanup. Before the +// wrapper replaced the instance, enough of them made even `SELECT 1` trap. +describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { + before(async () => { + await loadModule(); + }); + + it('stays usable after repeated stack overflows on deeply nested SQL', async () => { + for (let i = 0; i < 60; i++) { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + } + await loadModule(); + 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 () => { + await loadModule(); + const errors = []; + for (let i = 0; i < 60; i++) { + try { + parseSync(DEEP); + } catch (error) { + errors.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); + }); + + it('frees each replaced instance, so WASM memory stays at one instance', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + const deep = 'SELECT 1' + '+1'.repeat(100000); + for (let i = 0; i < 30; i++) { + try { lib.parseSync(deep); } catch {} + await lib.loadModule(); + } + const huge = 'SELECT 1;\\n'.repeat(5000000); + for (let i = 0; i < 2; i++) { + try { await lib.parse(huge); } catch {} + } + await lib.loadModule(); + lib.parseSync('SELECT 1'); + // V8 frees a dropped instance's memory on a later task, not in the same microtask run. + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + process.stdout.write('\\ngrown=' + (process.memoryUsage().external - base)); + })();`; + // A non-zero exit code (libpg_query's exit() leaking out) makes execFileSync throw. + // libpg_query prints a memory dump to stdout on out-of-memory, so read the last line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const grown = Number(output.match(/^grown=(-?\d+)$/m)?.[1]); + assert.ok(grown < 32 * 1024 * 1024, `WASM memory grew by ${grown} bytes after 32 replacements`); + }); + + it('lets a call queued behind a failing one wait for the fresh instance', async () => { + await loadModule(); + const [failed, queued] = await Promise.allSettled([ + parse(DEEP), + parse('SELECT 1'), + ]); + assert.equal(failed.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); + + it('throws an Error and keeps process.exitCode when the memory limit is hit', async () => { + const exitCode = process.exitCode; + await assert.rejects(parse('SELECT 1;\n'.repeat(5000000)), (error) => { + assert.ok(error instanceof Error); + assert.match(error.message, /out of WASM memory/); + return true; + }); + assert.equal(process.exitCode, exitCode); + assert.equal((await parse('SELECT 1')).stmts.length, 1); + }); + + for (const name of ['parsePlPgSQL', 'fingerprint', 'normalize', 'scan']) { + const skip = typeof lib[name] !== 'function' && 'not in this build'; + it(`${name}: rejects a 50 MB input, keeps process.exitCode and recovers`, { skip }, async () => { + const exitCode = process.exitCode; + await assert.rejects(lib[name]('SELECT 1;\n'.repeat(5000000))); + assert.equal(process.exitCode, exitCode); + assert.ok(await lib[name](name === 'parsePlPgSQL' + ? 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql' + : 'SELECT 1')); + assert.ok(lib[`${name}Sync`]('SELECT 1')); + }); + } + + it('retries the load when loading a fresh instance fails', async () => { + let loads = 0; + const isolated = loadIsolated((real, ...args) => + ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + await isolated.loadModule(); + await assert.rejects(isolated.parse(DEEP)); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal((await isolated.parse('SELECT 1')).stmts.length, 1); + assert.equal(loads, 3); + }); + + it('lets loadModule() reject on a failed reload and retry on the next 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 assert.rejects(isolated.loadModule(), /simulated load failure/); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); + }); + + it('keeps sync calls in the same synchronous loop working after a failure', () => { + const queries = Array.from({ length: 200 }, (_, i) => `SELECT ${i}, 'x${i}' FROM t${i} WHERE a = ${i}`); + const expected = queries.map((query) => JSON.stringify(parseSync(query))); + let failed = 0; + const results = []; + for (const [i, query] of queries.entries()) { + if (i === 10) { + try { parseSync(DEEP); } catch { failed++; } + } + results.push(JSON.stringify(parseSync(query))); + } + assert.equal(failed, 1); + assert.deepEqual(results, expected); + }); + + it('retries a failed fresh load on the next failure, for callers that never await', 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); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + }); + + it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + const failing = parse(DEEP); + const ready = loadModule(); + await assert.rejects(failing); + await ready; + assert.equal(parseSync('SELECT 1').stmts.length, 1); + }); + + it('keeps every call queued behind two failures waiting for a working instance', async () => { + await loadModule(); + const [first, second, queued] = await Promise.allSettled([parse(DEEP), parse(DEEP), parse('SELECT 1')]); + assert.equal(first.status, 'rejected'); + assert.equal(second.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); +}); + +// Every entry point, with the next WASM call made to throw: a RangeError +// stands for a stack overflow, a non-Error object for Emscripten's ExitStatus. +describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => { + const PLPGSQL = 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql'; + const entryPoints = [ + ['parse', 'SELECT 1'], + ['parseSync', 'SELECT 1'], + ['parsePlPgSQL', PLPGSQL], + ['parsePlPgSQLSync', PLPGSQL], + ['fingerprint', 'SELECT 1'], + ['fingerprintSync', 'SELECT 1'], + ['normalize', 'SELECT 1'], + ['normalizeSync', 'SELECT 1'], + ['scan', 'SELECT 1'], + ['scanSync', 'SELECT 1'], + ]; + + // Instances are numbered by load; `used` is the one the last WASM call ran on. + let loads = 0; + let used; + let frees = 0; + let throwNext; + const isolated = loadIsolated(async (real, ...args) => { + const id = ++loads; + const module = await real(...args); + const free = module._free; + module._free = (ptr) => { + frees++; + return free(ptr); + }; + for (const name of Object.keys(module).filter((key) => key.startsWith('_wasm_'))) { + const call = module[name]; + module[name] = (...callArgs) => { + used = id; + if (throwNext === undefined) return call(...callArgs); + const thrown = throwNext; + throwNext = undefined; + process.exitCode = 7; + throw thrown; + }; + } + return module; + }); + + async function failOnce(name, query, thrown) { + await isolated.loadModule(); + const loadsBefore = loads; + const exitCode = process.exitCode; + process.exitCode = 0; + const freesBefore = frees; + throwNext = thrown; + let caught; + try { + await isolated[name](query); + } catch (error) { + caught = error; + } finally { + var exitCodeAfter = process.exitCode; + process.exitCode = exitCode; + } + 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(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`); + return caught; + } + + it('starts one fresh instance for many failures in one synchronous loop', async () => { + await isolated.loadModule(); + const loadsBefore = loads; + for (let i = 0; i < 5; i++) { + throwNext = new RangeError('Maximum call stack size exceeded'); + assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + } + assert.equal(loads, loadsBefore + 1); + await isolated.loadModule(); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads); + }); + + it('stops sync calls on a stale instance after 10 failures until the fresh one loads', 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 () => { + 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'); + }); + + it('names a thrown object that has no message', async () => { + class EmscriptenSjLj {} + const caught = await failOnce('parseSync', 'SELECT 1', new EmscriptenSjLj()); + assert.match(caught.message, /libpg_query terminated \(EmscriptenSjLj\)/); + }); + + for (const [name, query] of entryPoints) { + const skip = typeof isolated[name] !== 'function' && 'not in this build'; + + it(`${name}: rethrows the original Error and recovers`, { skip }, async () => { + const thrown = new RangeError('Maximum call stack size exceeded'); + assert.equal(await failOnce(name, query, thrown), thrown); + }); + + it(`${name}: wraps a non-Error throw in an Error and recovers`, { skip }, async () => { + const thrown = { name: 'ExitStatus', message: 'Program terminated with exit(1)', status: 1 }; + 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); + }); + } +}); + +// 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'); + const factory = require.cache[require.resolve('../wasm/libpg-query.js')]; + const shared = require.cache[indexPath]; + const realFactory = factory.exports; + factory.exports = (...args) => wrapFactory(realFactory, ...args); + delete require.cache[indexPath]; + try { + return require(indexPath); + } finally { + factory.exports = realFactory; + require.cache[indexPath] = shared; + } +} diff --git a/versions/17/package.json b/versions/17/package.json index 8e6eacb7..c8685ca3 100644 --- a/versions/17/package.json +++ b/versions/17/package.json @@ -28,7 +28,7 @@ "wasm:rebuild": "pnpm wasm:make rebuild", "wasm:clean": "pnpm wasm:make clean", "wasm:clean-cache": "pnpm wasm:make clean-cache", - "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js" + "test": "node --test test/parsing.test.js test/errors.test.js test/input-encoding.test.js test/wasm-recovery.test.js" }, "author": "Constructive ", "license": "MIT", diff --git a/versions/17/src/index.ts b/versions/17/src/index.ts index 51e15b6b..ea70bb2a 100644 --- a/versions/17/src/index.ts +++ b/versions/17/src/index.ts @@ -131,23 +131,78 @@ export function hasSqlDetails(error: any): error is Error & { sqlDetails: SqlErr 'cursorPosition' in (error as any).sqlDetails; } -const initPromise = PgQueryModule().then((module: any) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule() { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit any>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -197,11 +252,12 @@ export const parse = awaitInit(async (query: string) => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -246,9 +302,11 @@ export const parse = awaitInit(async (query: string) => { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); @@ -266,11 +324,12 @@ export function parseSync(query: string) { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { // Call the raw function that returns a struct pointer - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to allocate memory for parse result'); } @@ -315,9 +374,11 @@ export function parseSync(query: string) { return JSON.parse(parseTree); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/17/test/wasm-recovery.test.js b/versions/17/test/wasm-recovery.test.js new file mode 100644 index 00000000..6d16d379 --- /dev/null +++ b/versions/17/test/wasm-recovery.test.js @@ -0,0 +1,352 @@ +const { describe, it, before } = require('node:test'); +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const lib = require('../wasm/index.cjs'); +const { parse, parseSync, loadModule } = lib; + +const DEEP = 'SELECT 1' + '+1'.repeat(100000); + +// An exception out of the WASM parser skips libpg_query's cleanup. Before the +// wrapper replaced the instance, enough of them made even `SELECT 1` trap. +describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { + before(async () => { + await loadModule(); + }); + + it('stays usable after repeated stack overflows on deeply nested SQL', async () => { + for (let i = 0; i < 60; i++) { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + } + await loadModule(); + 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 () => { + await loadModule(); + const errors = []; + for (let i = 0; i < 60; i++) { + try { + parseSync(DEEP); + } catch (error) { + errors.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); + }); + + it('frees each replaced instance, so WASM memory stays at one instance', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + const deep = 'SELECT 1' + '+1'.repeat(100000); + for (let i = 0; i < 30; i++) { + try { lib.parseSync(deep); } catch {} + await lib.loadModule(); + } + const huge = 'SELECT 1;\\n'.repeat(5000000); + for (let i = 0; i < 2; i++) { + try { await lib.parse(huge); } catch {} + } + await lib.loadModule(); + lib.parseSync('SELECT 1'); + // V8 frees a dropped instance's memory on a later task, not in the same microtask run. + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + process.stdout.write('\\ngrown=' + (process.memoryUsage().external - base)); + })();`; + // A non-zero exit code (libpg_query's exit() leaking out) makes execFileSync throw. + // libpg_query prints a memory dump to stdout on out-of-memory, so read the last line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const grown = Number(output.match(/^grown=(-?\d+)$/m)?.[1]); + assert.ok(grown < 32 * 1024 * 1024, `WASM memory grew by ${grown} bytes after 32 replacements`); + }); + + it('lets a call queued behind a failing one wait for the fresh instance', async () => { + await loadModule(); + const [failed, queued] = await Promise.allSettled([ + parse(DEEP), + parse('SELECT 1'), + ]); + assert.equal(failed.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); + + it('throws an Error and keeps process.exitCode when the memory limit is hit', async () => { + const exitCode = process.exitCode; + await assert.rejects(parse('SELECT 1;\n'.repeat(5000000)), (error) => { + assert.ok(error instanceof Error); + assert.match(error.message, /out of WASM memory/); + return true; + }); + assert.equal(process.exitCode, exitCode); + assert.equal((await parse('SELECT 1')).stmts.length, 1); + }); + + for (const name of ['parsePlPgSQL', 'fingerprint', 'normalize', 'scan']) { + const skip = typeof lib[name] !== 'function' && 'not in this build'; + it(`${name}: rejects a 50 MB input, keeps process.exitCode and recovers`, { skip }, async () => { + const exitCode = process.exitCode; + await assert.rejects(lib[name]('SELECT 1;\n'.repeat(5000000))); + assert.equal(process.exitCode, exitCode); + assert.ok(await lib[name](name === 'parsePlPgSQL' + ? 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql' + : 'SELECT 1')); + assert.ok(lib[`${name}Sync`]('SELECT 1')); + }); + } + + it('retries the load when loading a fresh instance fails', async () => { + let loads = 0; + const isolated = loadIsolated((real, ...args) => + ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + await isolated.loadModule(); + await assert.rejects(isolated.parse(DEEP)); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal((await isolated.parse('SELECT 1')).stmts.length, 1); + assert.equal(loads, 3); + }); + + it('lets loadModule() reject on a failed reload and retry on the next 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 assert.rejects(isolated.loadModule(), /simulated load failure/); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); + }); + + it('keeps sync calls in the same synchronous loop working after a failure', () => { + const queries = Array.from({ length: 200 }, (_, i) => `SELECT ${i}, 'x${i}' FROM t${i} WHERE a = ${i}`); + const expected = queries.map((query) => JSON.stringify(parseSync(query))); + let failed = 0; + const results = []; + for (const [i, query] of queries.entries()) { + if (i === 10) { + try { parseSync(DEEP); } catch { failed++; } + } + results.push(JSON.stringify(parseSync(query))); + } + assert.equal(failed, 1); + assert.deepEqual(results, expected); + }); + + it('retries a failed fresh load on the next failure, for callers that never await', 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); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + }); + + it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + const failing = parse(DEEP); + const ready = loadModule(); + await assert.rejects(failing); + await ready; + assert.equal(parseSync('SELECT 1').stmts.length, 1); + }); + + it('keeps every call queued behind two failures waiting for a working instance', async () => { + await loadModule(); + const [first, second, queued] = await Promise.allSettled([parse(DEEP), parse(DEEP), parse('SELECT 1')]); + assert.equal(first.status, 'rejected'); + assert.equal(second.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); +}); + +// Every entry point, with the next WASM call made to throw: a RangeError +// stands for a stack overflow, a non-Error object for Emscripten's ExitStatus. +describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => { + const PLPGSQL = 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql'; + const entryPoints = [ + ['parse', 'SELECT 1'], + ['parseSync', 'SELECT 1'], + ['parsePlPgSQL', PLPGSQL], + ['parsePlPgSQLSync', PLPGSQL], + ['fingerprint', 'SELECT 1'], + ['fingerprintSync', 'SELECT 1'], + ['normalize', 'SELECT 1'], + ['normalizeSync', 'SELECT 1'], + ['scan', 'SELECT 1'], + ['scanSync', 'SELECT 1'], + ]; + + // Instances are numbered by load; `used` is the one the last WASM call ran on. + let loads = 0; + let used; + let frees = 0; + let throwNext; + const isolated = loadIsolated(async (real, ...args) => { + const id = ++loads; + const module = await real(...args); + const free = module._free; + module._free = (ptr) => { + frees++; + return free(ptr); + }; + for (const name of Object.keys(module).filter((key) => key.startsWith('_wasm_'))) { + const call = module[name]; + module[name] = (...callArgs) => { + used = id; + if (throwNext === undefined) return call(...callArgs); + const thrown = throwNext; + throwNext = undefined; + process.exitCode = 7; + throw thrown; + }; + } + return module; + }); + + async function failOnce(name, query, thrown) { + await isolated.loadModule(); + const loadsBefore = loads; + const exitCode = process.exitCode; + process.exitCode = 0; + const freesBefore = frees; + throwNext = thrown; + let caught; + try { + await isolated[name](query); + } catch (error) { + caught = error; + } finally { + var exitCodeAfter = process.exitCode; + process.exitCode = exitCode; + } + 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(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`); + return caught; + } + + it('starts one fresh instance for many failures in one synchronous loop', async () => { + await isolated.loadModule(); + const loadsBefore = loads; + for (let i = 0; i < 5; i++) { + throwNext = new RangeError('Maximum call stack size exceeded'); + assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + } + assert.equal(loads, loadsBefore + 1); + await isolated.loadModule(); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads); + }); + + it('stops sync calls on a stale instance after 10 failures until the fresh one loads', 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 () => { + 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'); + }); + + it('names a thrown object that has no message', async () => { + class EmscriptenSjLj {} + const caught = await failOnce('parseSync', 'SELECT 1', new EmscriptenSjLj()); + assert.match(caught.message, /libpg_query terminated \(EmscriptenSjLj\)/); + }); + + for (const [name, query] of entryPoints) { + const skip = typeof isolated[name] !== 'function' && 'not in this build'; + + it(`${name}: rethrows the original Error and recovers`, { skip }, async () => { + const thrown = new RangeError('Maximum call stack size exceeded'); + assert.equal(await failOnce(name, query, thrown), thrown); + }); + + it(`${name}: wraps a non-Error throw in an Error and recovers`, { skip }, async () => { + const thrown = { name: 'ExitStatus', message: 'Program terminated with exit(1)', status: 1 }; + 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); + }); + } +}); + +// 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'); + const factory = require.cache[require.resolve('../wasm/libpg-query.js')]; + const shared = require.cache[indexPath]; + const realFactory = factory.exports; + factory.exports = (...args) => wrapFactory(realFactory, ...args); + delete require.cache[indexPath]; + try { + return require(indexPath); + } finally { + factory.exports = realFactory; + require.cache[indexPath] = shared; + } +} diff --git a/versions/18/package.json b/versions/18/package.json index 8c9ad17c..45802db3 100644 --- a/versions/18/package.json +++ b/versions/18/package.json @@ -29,7 +29,7 @@ "wasm:rebuild": "pnpm wasm:make rebuild", "wasm:clean": "pnpm wasm:make clean", "wasm:clean-cache": "pnpm wasm:make clean-cache", - "test": "node --test test/parsing.test.js test/pg18.test.js test/fingerprint.test.js test/normalize.test.js test/plpgsql.test.js test/scan.test.js test/errors.test.js test/input-encoding.test.js" + "test": "node --test test/parsing.test.js test/pg18.test.js test/fingerprint.test.js test/normalize.test.js test/plpgsql.test.js test/scan.test.js test/errors.test.js test/input-encoding.test.js test/wasm-recovery.test.js" }, "author": "Constructive ", "license": "MIT", diff --git a/versions/18/src/index.ts b/versions/18/src/index.ts index 30febc6e..e905d6c6 100644 --- a/versions/18/src/index.ts +++ b/versions/18/src/index.ts @@ -143,23 +143,78 @@ interface WasmModule { let wasmModule: WasmModule; -const initPromise = PgQueryModule().then((module: WasmModule) => { - wasmModule = module; -}); +// 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; +let loading: Promise | undefined; +// Counts WASM calls that threw, so cleanup skips an instance that threw mid-call. +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; + +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; + } + ); + loading = promise; + // Rejects only for the callers awaiting it. + promise.catch(() => {}); + return promise; +} + +instantiate(); + +// 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) { + 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 } + ); + } +} function ensureLoaded() { if (!wasmModule) throw new Error("WASM module not initialized. Call `loadModule()` first."); } export async function loadModule(): Promise { - if (!wasmModule) { - await initPromise; - } + while (!wasmModule || stale) await instantiate(); } function awaitInit Promise>(fn: T): T { return (async (...args: Parameters) => { - await initPromise; + // Re-checked after every await: another call may have failed in between. + while (!wasmModule || stale) await instantiate(); return fn(...args); }) as T; } @@ -207,10 +262,11 @@ export const parse = awaitInit(async (query: string): Promise => { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to parse query: memory allocation failed'); } @@ -248,19 +304,22 @@ export const parse = awaitInit(async (query: string): Promise => { const parseTreeStr = wasmModule.UTF8ToString(parseTreePtr); return JSON.parse(parseTreeStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } }); export const parsePlPgSQL = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_plpgsql(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_plpgsql(queryPtr)); const resultStr = ptrToString(resultPtr); // Success is always a JSON object; anything else is an error message @@ -270,19 +329,22 @@ export const parsePlPgSQL = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_fingerprint(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_fingerprint(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -291,19 +353,22 @@ export const fingerprint = awaitInit(async (query: string): Promise => { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } }); export const normalize = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_normalize_query(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_normalize_query(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -312,9 +377,11 @@ export const normalize = awaitInit(async (query: string): Promise => { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } }); @@ -335,10 +402,11 @@ export function parseSync(query: string): ParseResult { } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_query_raw(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_query_raw(queryPtr)); if (!resultPtr) { throw new Error('Failed to parse query: memory allocation failed'); } @@ -376,9 +444,11 @@ export function parseSync(query: string): ParseResult { const parseTreeStr = wasmModule.UTF8ToString(parseTreePtr); return JSON.parse(parseTreeStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_parse_result(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_parse_result(resultPtr); + } } } } @@ -388,10 +458,11 @@ export function parsePlPgSQLSync(query: string): ParseResult { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_parse_plpgsql(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_parse_plpgsql(queryPtr)); const resultStr = ptrToString(resultPtr); // Success is always a JSON object; anything else is an error message @@ -401,9 +472,11 @@ export function parsePlPgSQLSync(query: string): ParseResult { return JSON.parse(resultStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } @@ -413,10 +486,11 @@ export function fingerprintSync(query: string): string { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_fingerprint(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_fingerprint(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -425,9 +499,11 @@ export function fingerprintSync(query: string): string { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } @@ -437,10 +513,11 @@ export function normalizeSync(query: string): string { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_normalize_query(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_normalize_query(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -449,19 +526,22 @@ export function normalizeSync(query: string): string { return resultStr; } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } export const scan = awaitInit(async (query: string): Promise => { const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_scan(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_scan(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -470,9 +550,11 @@ export const scan = awaitInit(async (query: string): Promise => { return JSON.parse(resultStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } }); @@ -482,10 +564,11 @@ export function scanSync(query: string): ScanResult { throw new Error('WASM module not initialized. Call loadModule() first.'); } const queryPtr = stringToPtr(query); + const failuresBefore = failures; let resultPtr = 0; try { - resultPtr = wasmModule._wasm_scan(queryPtr); + resultPtr = callWasm(() => wasmModule._wasm_scan(queryPtr)); const resultStr = ptrToString(resultPtr); if (resultStr.startsWith('syntax error') || resultStr.startsWith('deparse error') || resultStr.startsWith('ERROR')) { @@ -494,9 +577,11 @@ export function scanSync(query: string): ScanResult { return JSON.parse(resultStr); } finally { - wasmModule._free(queryPtr); - if (resultPtr) { - wasmModule._wasm_free_string(resultPtr); + if (failures === failuresBefore) { + wasmModule._free(queryPtr); + if (resultPtr) { + wasmModule._wasm_free_string(resultPtr); + } } } } \ No newline at end of file diff --git a/versions/18/test/wasm-recovery.test.js b/versions/18/test/wasm-recovery.test.js new file mode 100644 index 00000000..6d16d379 --- /dev/null +++ b/versions/18/test/wasm-recovery.test.js @@ -0,0 +1,352 @@ +const { describe, it, before } = require('node:test'); +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const lib = require('../wasm/index.cjs'); +const { parse, parseSync, loadModule } = lib; + +const DEEP = 'SELECT 1' + '+1'.repeat(100000); + +// An exception out of the WASM parser skips libpg_query's cleanup. Before the +// wrapper replaced the instance, enough of them made even `SELECT 1` trap. +describe('Recovery after the WASM parser throws', { timeout: 120000 }, () => { + before(async () => { + await loadModule(); + }); + + it('stays usable after repeated stack overflows on deeply nested SQL', async () => { + for (let i = 0; i < 60; i++) { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + } + await loadModule(); + 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 () => { + await loadModule(); + const errors = []; + for (let i = 0; i < 60; i++) { + try { + parseSync(DEEP); + } catch (error) { + errors.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); + }); + + it('frees each replaced instance, so WASM memory stays at one instance', () => { + const script = `const lib = require(${JSON.stringify(require.resolve('../wasm/index.cjs'))}); + (async () => { + await lib.loadModule(); + lib.parseSync('SELECT 1'); + gc(); gc(); + const base = process.memoryUsage().external; + const deep = 'SELECT 1' + '+1'.repeat(100000); + for (let i = 0; i < 30; i++) { + try { lib.parseSync(deep); } catch {} + await lib.loadModule(); + } + const huge = 'SELECT 1;\\n'.repeat(5000000); + for (let i = 0; i < 2; i++) { + try { await lib.parse(huge); } catch {} + } + await lib.loadModule(); + lib.parseSync('SELECT 1'); + // V8 frees a dropped instance's memory on a later task, not in the same microtask run. + await new Promise((resolve) => setTimeout(resolve, 0)); + gc(); gc(); + process.stdout.write('\\ngrown=' + (process.memoryUsage().external - base)); + })();`; + // A non-zero exit code (libpg_query's exit() leaking out) makes execFileSync throw. + // libpg_query prints a memory dump to stdout on out-of-memory, so read the last line. + const output = execFileSync(process.execPath, ['--expose-gc', '-e', script], { + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'ignore'], + }); + const grown = Number(output.match(/^grown=(-?\d+)$/m)?.[1]); + assert.ok(grown < 32 * 1024 * 1024, `WASM memory grew by ${grown} bytes after 32 replacements`); + }); + + it('lets a call queued behind a failing one wait for the fresh instance', async () => { + await loadModule(); + const [failed, queued] = await Promise.allSettled([ + parse(DEEP), + parse('SELECT 1'), + ]); + assert.equal(failed.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); + + it('throws an Error and keeps process.exitCode when the memory limit is hit', async () => { + const exitCode = process.exitCode; + await assert.rejects(parse('SELECT 1;\n'.repeat(5000000)), (error) => { + assert.ok(error instanceof Error); + assert.match(error.message, /out of WASM memory/); + return true; + }); + assert.equal(process.exitCode, exitCode); + assert.equal((await parse('SELECT 1')).stmts.length, 1); + }); + + for (const name of ['parsePlPgSQL', 'fingerprint', 'normalize', 'scan']) { + const skip = typeof lib[name] !== 'function' && 'not in this build'; + it(`${name}: rejects a 50 MB input, keeps process.exitCode and recovers`, { skip }, async () => { + const exitCode = process.exitCode; + await assert.rejects(lib[name]('SELECT 1;\n'.repeat(5000000))); + assert.equal(process.exitCode, exitCode); + assert.ok(await lib[name](name === 'parsePlPgSQL' + ? 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql' + : 'SELECT 1')); + assert.ok(lib[`${name}Sync`]('SELECT 1')); + }); + } + + it('retries the load when loading a fresh instance fails', async () => { + let loads = 0; + const isolated = loadIsolated((real, ...args) => + ++loads === 2 ? Promise.reject(new Error('simulated load failure')) : real(...args)); + await isolated.loadModule(); + await assert.rejects(isolated.parse(DEEP)); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal((await isolated.parse('SELECT 1')).stmts.length, 1); + assert.equal(loads, 3); + }); + + it('lets loadModule() reject on a failed reload and retry on the next 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 assert.rejects(isolated.loadModule(), /simulated load failure/); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + assert.equal(loads, 3); + }); + + it('keeps sync calls in the same synchronous loop working after a failure', () => { + const queries = Array.from({ length: 200 }, (_, i) => `SELECT ${i}, 'x${i}' FROM t${i} WHERE a = ${i}`); + const expected = queries.map((query) => JSON.stringify(parseSync(query))); + let failed = 0; + const results = []; + for (const [i, query] of queries.entries()) { + if (i === 10) { + try { parseSync(DEEP); } catch { failed++; } + } + results.push(JSON.stringify(parseSync(query))); + } + assert.equal(failed, 1); + assert.deepEqual(results, expected); + }); + + it('retries a failed fresh load on the next failure, for callers that never await', 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); + await isolated.loadModule(); + assert.equal(isolated.parseSync('SELECT 1').stmts.length, 1); + }); + + it('makes loadModule() wait again when a queued call breaks the fresh instance', async () => { + await loadModule(); + assert.throws(() => parseSync(DEEP)); + const failing = parse(DEEP); + const ready = loadModule(); + await assert.rejects(failing); + await ready; + assert.equal(parseSync('SELECT 1').stmts.length, 1); + }); + + it('keeps every call queued behind two failures waiting for a working instance', async () => { + await loadModule(); + const [first, second, queued] = await Promise.allSettled([parse(DEEP), parse(DEEP), parse('SELECT 1')]); + assert.equal(first.status, 'rejected'); + assert.equal(second.status, 'rejected'); + assert.equal(queued.status, 'fulfilled', queued.reason?.message); + assert.equal(queued.value.stmts.length, 1); + }); +}); + +// Every entry point, with the next WASM call made to throw: a RangeError +// stands for a stack overflow, a non-Error object for Emscripten's ExitStatus. +describe('Recovery from a throw in each entry point', { timeout: 120000 }, () => { + const PLPGSQL = 'CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql'; + const entryPoints = [ + ['parse', 'SELECT 1'], + ['parseSync', 'SELECT 1'], + ['parsePlPgSQL', PLPGSQL], + ['parsePlPgSQLSync', PLPGSQL], + ['fingerprint', 'SELECT 1'], + ['fingerprintSync', 'SELECT 1'], + ['normalize', 'SELECT 1'], + ['normalizeSync', 'SELECT 1'], + ['scan', 'SELECT 1'], + ['scanSync', 'SELECT 1'], + ]; + + // Instances are numbered by load; `used` is the one the last WASM call ran on. + let loads = 0; + let used; + let frees = 0; + let throwNext; + const isolated = loadIsolated(async (real, ...args) => { + const id = ++loads; + const module = await real(...args); + const free = module._free; + module._free = (ptr) => { + frees++; + return free(ptr); + }; + for (const name of Object.keys(module).filter((key) => key.startsWith('_wasm_'))) { + const call = module[name]; + module[name] = (...callArgs) => { + used = id; + if (throwNext === undefined) return call(...callArgs); + const thrown = throwNext; + throwNext = undefined; + process.exitCode = 7; + throw thrown; + }; + } + return module; + }); + + async function failOnce(name, query, thrown) { + await isolated.loadModule(); + const loadsBefore = loads; + const exitCode = process.exitCode; + process.exitCode = 0; + const freesBefore = frees; + throwNext = thrown; + let caught; + try { + await isolated[name](query); + } catch (error) { + caught = error; + } finally { + var exitCodeAfter = process.exitCode; + process.exitCode = exitCode; + } + 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(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`); + return caught; + } + + it('starts one fresh instance for many failures in one synchronous loop', async () => { + await isolated.loadModule(); + const loadsBefore = loads; + for (let i = 0; i < 5; i++) { + throwNext = new RangeError('Maximum call stack size exceeded'); + assert.throws(() => isolated.parseSync('SELECT 1'), RangeError); + } + assert.equal(loads, loadsBefore + 1); + await isolated.loadModule(); + assert.ok(isolated.parseSync('SELECT 1')); + assert.equal(used, loads); + }); + + it('stops sync calls on a stale instance after 10 failures until the fresh one loads', 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 () => { + 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'); + }); + + it('names a thrown object that has no message', async () => { + class EmscriptenSjLj {} + const caught = await failOnce('parseSync', 'SELECT 1', new EmscriptenSjLj()); + assert.match(caught.message, /libpg_query terminated \(EmscriptenSjLj\)/); + }); + + for (const [name, query] of entryPoints) { + const skip = typeof isolated[name] !== 'function' && 'not in this build'; + + it(`${name}: rethrows the original Error and recovers`, { skip }, async () => { + const thrown = new RangeError('Maximum call stack size exceeded'); + assert.equal(await failOnce(name, query, thrown), thrown); + }); + + it(`${name}: wraps a non-Error throw in an Error and recovers`, { skip }, async () => { + const thrown = { name: 'ExitStatus', message: 'Program terminated with exit(1)', status: 1 }; + 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); + }); + } +}); + +// 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'); + const factory = require.cache[require.resolve('../wasm/libpg-query.js')]; + const shared = require.cache[indexPath]; + const realFactory = factory.exports; + factory.exports = (...args) => wrapFactory(realFactory, ...args); + delete require.cache[indexPath]; + try { + return require(indexPath); + } finally { + factory.exports = realFactory; + require.cache[indexPath] = shared; + } +}