diff --git a/scripts/lib/control-pane/work-item-runner.js b/scripts/lib/control-pane/work-item-runner.js index d3ee361ff..898f2e1f2 100644 --- a/scripts/lib/control-pane/work-item-runner.js +++ b/scripts/lib/control-pane/work-item-runner.js @@ -6,7 +6,7 @@ const { Worker } = require('worker_threads'); const MAX_ACTIVE_MUTATIONS = 2; let activeMutations = 0; -/** Bound live workers process-wide; excess mutations are rejected, not queued. */ +/** Bound workers in this loaded runner/isolate; reject excess work without queuing. */ function runWorkItemMutation(dbPath, action, args) { return new Promise((resolve, reject) => { if (activeMutations >= MAX_ACTIVE_MUTATIONS) { @@ -27,14 +27,26 @@ function runWorkItemMutation(dbPath, action, args) { } let response; let workerError; + let failed = false; worker.once('message', message => { response = message; }); - worker.once('error', error => { workerError = error; }); + worker.once('error', error => { failed = true; workerError = error; }); worker.once('exit', code => { // Keep the slot until the thread is gone, even after a message/error. activeMutations -= 1; - if (workerError) reject(workerError); - else if (code !== 0 || !response) reject(new Error(`Work-item worker exited without a successful result (${code})`)); - else if (response.ok) resolve(response.result); + if (failed) { reject(workerError); return; } + if (code !== 0 || response === undefined) { + reject(new Error(`Work-item worker exited without a successful result (${code})`)); + return; + } + // Worker messages are structured clones from our own helper, but malformed + // results still must reject instead of throwing outside the Promise executor. + if (!response || typeof response !== 'object' || Array.isArray(response) + || typeof response.ok !== 'boolean' + || (response.ok === false && (!response.error || typeof response.error !== 'object' + || typeof response.error.message !== 'string' + || (response.error.code !== undefined && typeof response.error.code !== 'string')))) { + reject(new Error('Invalid work-item worker response')); + } else if (response.ok) resolve(response.result); else reject(Object.assign(new Error(response.error.message), { code: response.error.code })); }); // Let the worker exit naturally after closing the store. Terminating it on diff --git a/scripts/lib/control-pane/work-item-worker.js b/scripts/lib/control-pane/work-item-worker.js index 72c808eb5..0518a3734 100644 --- a/scripts/lib/control-pane/work-item-worker.js +++ b/scripts/lib/control-pane/work-item-worker.js @@ -2,6 +2,7 @@ const { parentPort, workerData } = require('worker_threads'); const { createStateStore } = require('../state-store'); +const { attachCleanupError } = require('../state-store/file-lock'); const { claimWorkItem, moveWorkItem } = require('./work-item-mutations'); /** Complete one board mutation away from the HTTP event loop, then close. */ @@ -10,14 +11,34 @@ async function mutate() { const mutation = action === 'claim' ? claimWorkItem : action === 'move' ? moveWorkItem : null; if (!mutation) throw new Error('Unknown work-item mutation'); const store = await createStateStore({ dbPath }); - try { - return store._database.transaction(() => mutation(store, args))(); - } finally { - store.close(); + let result; + let failed = false; + let primary; + try { result = store._database.transaction(() => mutation(store, args))(); } + catch (error) { failed = true; primary = error; } + try { store.close(); } + catch (error) { + if (failed) attachCleanupError(primary, 'closeError', error); + else { failed = true; primary = error; } } + if (failed) throw primary; + return result; +} + +function serializeError(error) { + let message = 'Work-item mutation failed'; + let code; + try { + if (error && typeof error.message === 'string') message = error.message.slice(0, 1024); + else if (error === null || typeof error !== 'object') message = String(error).slice(0, 1024); + } catch (_error) { /* A throwing conversion must not prevent the failure response. */ } + try { + if (error && typeof error.code === 'string') code = error.code.slice(0, 128); + } catch (_error) { /* Optional diagnostic. */ } + return { message, code }; } mutate().then( result => parentPort.postMessage({ ok: true, result }), - error => parentPort.postMessage({ ok: false, error: { message: error.message, code: error.code } }) + error => parentPort.postMessage({ ok: false, error: serializeError(error) }) ); diff --git a/scripts/lib/state-store/file-lock.js b/scripts/lib/state-store/file-lock.js index 3bc3edf37..200622299 100644 --- a/scripts/lib/state-store/file-lock.js +++ b/scripts/lib/state-store/file-lock.js @@ -7,22 +7,62 @@ const { performance } = require('perf_hooks'); const WAIT_BUFFER = new Int32Array(new SharedArrayBuffer(4)); const DEFAULT_TIMEOUT_MS = 5000; +// Cleanup must never replace an arbitrary primary throw (including frozen errors, +// primitives, or objects with throwing property traps). Diagnostics are best effort. +function attachCleanupError(primary, name, secondary) { + if (primary === null || (typeof primary !== 'object' && typeof primary !== 'function')) return; + try { Object.defineProperty(primary, name, { value: secondary, configurable: true }); } + catch (_error) { /* Preserve the primary when it cannot accept diagnostics. */ } +} + +function hasCode(error, code) { + try { return error !== null && error !== undefined && error.code === code; } + catch (_error) { return false; } +} + function sameIdentity(left, right) { - return left.ino === right.ino - && (left.dev === right.dev || (process.platform === 'win32' && (!left.dev || !right.dev))); + return left.ino === right.ino && left.dev === right.dev; +} + +function lostLock(lockPath) { + return Object.assign(new Error(`State-store lock changed or lost; refusing to remove it: ${lockPath}`), + { code: 'STATE_STORE_LOCK_LOST' }); } function releaseOwnedLock(lockPath, descriptor, identity) { - let current; + let failed = false; + let primary; + let pendingDelete = false; + let inspected = false; + let closeFailed = false; try { - current = fs.lstatSync(lockPath, { bigint: true }); - } finally { - fs.closeSync(descriptor); + const current = fs.lstatSync(lockPath, { bigint: true }); + inspected = true; + if (!current.isFile() || current.isSymbolicLink() || !sameIdentity(identity, current)) { + throw lostLock(lockPath); + } + // Keep the owned descriptor through validation and unlink. This cooperative + // lock is not an atomic compare-and-unlink against a directory mutator. + fs.unlinkSync(lockPath); + } catch (error) { + failed = true; + primary = hasCode(error, 'ENOENT') ? lostLock(lockPath) : error; + pendingDelete = !inspected && hasCode(error, 'EPERM'); } - if (!current.isFile() || current.isSymbolicLink() || !sameIdentity(identity, current)) { - throw new Error(`State-store lock changed; refusing to remove it: ${lockPath}`); + try { + fs.closeSync(descriptor); // Exactly one attempt, even after a release failure. + } catch (error) { + closeFailed = true; + if (failed) attachCleanupError(primary, 'closeError', error); + else { failed = true; primary = error; } } - fs.unlinkSync(lockPath); + if (pendingDelete && !closeFailed) { + // Windows can defer deletion until the handle closes. Only a subsequent + // ENOENT proves loss; preserve genuine EPERM and never unlink after close. + try { fs.lstatSync(lockPath, { bigint: true }); } + catch (error) { if (hasCode(error, 'ENOENT')) primary = lostLock(lockPath); } + } + if (failed) throw primary; } function acquireLock(dbPath, timeoutMs) { @@ -33,7 +73,7 @@ function acquireLock(dbPath, timeoutMs) { try { descriptor = fs.openSync(lockPath, 'wx', 0o600); } catch (error) { - if (error.code !== 'EEXIST') throw error; + if (!hasCode(error, 'EEXIST')) throw error; const remaining = deadline - performance.now(); if (remaining <= 0) { const busy = new Error(`State store is busy: ${dbPath}. Retry after the other ECC operation finishes. ` @@ -54,7 +94,7 @@ function acquireLock(dbPath, timeoutMs) { if (identity) releaseOwnedLock(lockPath, descriptor, identity); else fs.closeSync(descriptor); } catch (releaseError) { - error.releaseError = releaseError; + attachCleanupError(error, 'releaseError', releaseError); } throw error; } @@ -81,10 +121,10 @@ function withStateStoreLock(dbPath, callback, { timeoutMs = DEFAULT_TIMEOUT_MS } release(); } catch (releaseError) { if (!failed) throw releaseError; - if (primaryError instanceof Error) primaryError.releaseError = releaseError; + attachCleanupError(primaryError, 'releaseError', releaseError); } if (failed) throw primaryError; return result; } -module.exports = { withStateStoreLock }; +module.exports = { withStateStoreLock, attachCleanupError }; diff --git a/scripts/lib/state-store/index.js b/scripts/lib/state-store/index.js index 3b4d8f33d..93ceed4ca 100644 --- a/scripts/lib/state-store/index.js +++ b/scripts/lib/state-store/index.js @@ -5,7 +5,7 @@ const crypto = require('crypto'); const os = require('os'); const path = require('path'); const initSqlJs = require('sql.js'); -const { withStateStoreLock } = require('./file-lock'); +const { withStateStoreLock, attachCleanupError } = require('./file-lock'); const { applyMigrations, getAppliedMigrations } = require('./migrations'); const { createQueryApi } = require('./queries'); @@ -218,8 +218,20 @@ function wrapSqlJsDatabase(SQL, dbPath) { // sql.js can use its input buffer as writable backing storage. const originalBytes = bytes ? Buffer.from(bytes) : null; const latest = new SQL.Database(bytes); - latest.run('PRAGMA foreign_keys = ON'); - if (rawDb) rawDb.close(); + try { + latest.run('PRAGMA foreign_keys = ON'); + if (rawDb) { + const previous = rawDb; + rawDb = null; + try { previous.close(); } + catch (error) { closed = true; throw error; } + } + } catch (error) { + // The replacement is not adopted until both initialization and the old + // handle's close succeed. Never reuse an uncertain previous handle. + try { latest.close(); } catch (closeError) { attachCleanupError(error, 'closeError', closeError); } + throw error; + } rawDb = latest; snapshotBytes = originalBytes; } @@ -312,9 +324,9 @@ function wrapSqlJsDatabase(SQL, dbPath) { } catch (rollbackError) { // Never reuse an uncertain transaction, including an in-memory store. closed = true; - try { rawDb.close(); } catch (_closeError) { /* Preserve the original failure. */ } + try { rawDb.close(); } catch (closeError) { attachCleanupError(error, 'closeError', closeError); } rawDb = null; - if (error instanceof Error) error.rollbackError = rollbackError; + attachCleanupError(error, 'rollbackError', rollbackError); } dirty = previouslyDirty; throw error; @@ -386,7 +398,7 @@ async function createStateStore(options = {}) { try { appliedMigrations = db.withSnapshot(() => applyMigrations(db)); } catch (error) { - db.close(); + try { db.close(); } catch (closeError) { attachCleanupError(error, 'closeError', closeError); } throw error; } const queryApi = createQueryApi(db); diff --git a/tests/lib/control-pane-contention.test.js b/tests/lib/control-pane-contention.test.js index 7308e1bf3..fea89463a 100644 --- a/tests/lib/control-pane-contention.test.js +++ b/tests/lib/control-pane-contention.test.js @@ -5,83 +5,73 @@ const fs = require('fs'); const http = require('http'); const os = require('os'); const path = require('path'); -const { fork } = require('child_process'); +const { startOwnedChild, withCleanup } = require('./helpers/state-store-worker'); const { createStateStore } = require('../../scripts/lib/state-store'); const FIXTURE = path.join(__dirname, 'helpers', 'control-pane-contention-server.js'); function request(url, method = 'GET', body, timeoutMs = 1000, onRequest = () => {}) { return new Promise((resolve, reject) => { + let settled = false; + let timer; + let response; + const finish = (failed, error, value) => { + if (settled) return; + settled = true; + clearTimeout(timer); + if (failed) { + try { if (response) response.destroy(); } catch (_cleanupError) { /* Primary wins. */ } + try { req.destroy(); } catch (_cleanupError) { /* Primary wins. */ } + reject(error); + } + else resolve(value); + }; const req = http.request(url, { method, agent: false, headers: { 'Content-Type': 'application/json' } }, res => { + if (settled) { res.destroy(); return; } + response = res; let text = ''; + let bytes = 0; res.setEncoding('utf8'); - res.on('data', data => { text += data; }); + res.on('data', data => { + bytes += Buffer.byteLength(data); + if (bytes > 128 * 1024) finish(true, new Error('HTTP fixture response exceeded 128 KiB')); + else text += data; + }); + res.on('error', error => finish(true, error)); + res.on('close', () => { if (!res.complete) finish(true, new Error('HTTP fixture response closed before completion')); }); + res.on('aborted', () => finish(true, new Error('HTTP fixture response aborted'))); res.on('end', () => { - try { - resolve({ status: res.statusCode, headers: res.headers, body: JSON.parse(text) }); - } catch (error) { - reject(error); - } + try { finish(false, null, { status: res.statusCode, headers: res.headers, body: JSON.parse(text) }); } + catch (error) { finish(true, error); } }); }); - req.setTimeout(timeoutMs, () => req.destroy(new Error(`HTTP ${method} timed out after ${timeoutMs}ms`))); - req.on('error', reject); - onRequest(req); - req.end(body === undefined ? undefined : JSON.stringify(body)); + // A total deadline covers headers and body, even if bytes keep arriving. + timer = setTimeout(() => finish(true, new Error(`HTTP ${method} timed out after ${timeoutMs}ms`)), timeoutMs); + req.on('error', error => finish(true, error)); + try { onRequest(req); req.end(body === undefined ? undefined : JSON.stringify(body)); } + catch (error) { finish(true, error); } }); } async function startServer(dbPath, readOnly = false) { - const child = fork(FIXTURE, [dbPath, readOnly ? 'read-only' : 'editable'], { - stdio: ['ignore', 'pipe', 'pipe', 'ipc'] - }); - const closed = new Promise(resolve => child.once('close', resolve)); - child.stdout.resume(); - let stderr = ''; + const client = startOwnedChild(FIXTURE, [dbPath, readOnly ? 'read-only' : 'editable']); let metrics = { active: 0, peak: 0, started: 0, exited: 0, received: 0, disconnected: 0 }; const observers = new Set(); - child.on('message', message => { - if (message.type === 'workers') { + client.observe(message => { + if (message && message.type === 'workers') { metrics = message; for (const notify of observers) notify(); } }); - child.stderr.on('data', data => { stderr += data; }); let ready; - try { - ready = await new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error(`Server startup timed out: ${stderr}`)), 10000); - child.once('error', error => { clearTimeout(timer); reject(error); }); - child.on('message', message => { - if (message.type === 'ready') { clearTimeout(timer); resolve(message); } - }); - child.once('exit', code => { clearTimeout(timer); reject(new Error(`Server exited ${code}: ${stderr}`)); }); - }); - } catch (error) { - child.kill(); - await closed; - throw error; - } + try { ready = await client.waitFor(message => message && message.type === 'ready'); } + catch (error) { return withCleanup(() => { throw error; }, [() => client.stop()]); } return { url: ready.url, get metrics() { return metrics; }, observe(fn) { observers.add(fn); return () => observers.delete(fn); }, - mutationReceived() { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { child.off('message', onMessage); reject(new Error('Mutation never reached server')); }, 5000); - function onMessage(message) { - if (message.type !== 'mutation-received') return; - clearTimeout(timer); - child.off('message', onMessage); - resolve(); - } - child.on('message', onMessage); - }); - }, - async close() { - if (child.exitCode === null && child.signalCode === null) child.kill(); - await closed; - } + mutationReceived() { return client.waitFor(message => message && message.type === 'mutation-received', 5000); }, + close: () => client.stop(), }; } @@ -103,17 +93,28 @@ function waitForObservation(server, predicate, timeoutMs = 3000) { }); } -async function burstScenario(server, dbPath) { - const bytesBefore = fs.readFileSync(dbPath); +async function holdFixtureLock(dbPath) { const lockPath = `${dbPath}.ecc-state.lock`; const fd = fs.openSync(lockPath, 'wx', 0o600); - fs.writeFileSync(fd, JSON.stringify({ pid: process.pid, hostname: os.hostname() })); + let released = false; + const release = () => { + if (released) return; + released = true; + return withCleanup(() => fs.unlinkSync(lockPath), [() => fs.closeSync(fd)]); + }; + try { fs.writeFileSync(fd, JSON.stringify({ pid: process.pid, hostname: os.hostname() })); } + catch (error) { return withCleanup(() => { throw error; }, [release]); } + return release; +} + +async function burstScenario(server, dbPath) { + const bytesBefore = fs.readFileSync(dbPath); + const releaseLock = await holdFixtureLock(dbPath); const results = []; const requests = []; let resolveRejected; const sixRejected = new Promise(resolve => { resolveRejected = resolve; }); - let released = false; - try { + return withCleanup(async () => { for (let index = 0; index < 8; index += 1) { requests.push(request(`${server.url}/api/work-items/burst-${index}/claim`, 'POST', { owner: `owner-${index}` }, 12000) .then(result => { @@ -148,58 +149,47 @@ async function burstScenario(server, dbPath) { assert.strictEqual(snapshot.status, 200); assert.deepStrictEqual(fs.readFileSync(dbPath), bytesBefore); const rejected = results.map(result => result.index); - fs.closeSync(fd); - fs.unlinkSync(lockPath); - released = true; + await releaseLock(); await Promise.all(requests); await waitForObservation(server, () => server.metrics.active === 0); assert.strictEqual(results.filter(result => result.status === 200).length, 2); assert.strictEqual(server.metrics.started, 2, 'Rejected requests must never execute later'); const store = await createStateStore({ dbPath }); - try { + await withCleanup(() => { for (const index of rejected) assert.strictEqual(store.getWorkItemById(`burst-${index}`).owner, null); - } finally { store.close(); } + }, [() => store.close()]); const retry = await request(`${server.url}/api/work-items/burst-${rejected[0]}/claim`, 'POST', { owner: 'retry-owner' }, 5000); assert.strictEqual(retry.status, 200); assert.strictEqual(retry.body.item.owner, 'retry-owner'); - } finally { - if (!released) { fs.closeSync(fd); fs.unlinkSync(lockPath); } - await Promise.all(requests); - } + }, [releaseLock, () => Promise.all(requests)]); } async function withServer(readOnly, fn) { const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ecc-http-contention-')); const dbPath = path.join(dir, 'state.db'); let server; - try { + return withCleanup(async () => { const store = await createStateStore({ dbPath }); - try { + await withCleanup(() => { store.upsertWorkItem({ id: 'task', title: 'Synthetic board task', source: 'manual', status: 'open' }); for (let index = 0; index < 8; index += 1) { store.upsertWorkItem({ id: `burst-${index}`, title: `Burst task ${index}`, source: 'manual', status: 'open' }); } - } finally { - store.close(); - } + }, [() => store.close()]); server = await startServer(dbPath, readOnly); return await fn(server, dbPath); - } finally { - if (server) await server.close(); - fs.rmSync(dir, { recursive: true, force: true }); - } + }, [() => { if (server) return server.close(); }, + () => fs.rmSync(dir, { recursive: true, force: true })]); } async function disconnectScenario(server, dbPath) { - const lockPath = `${dbPath}.ecc-state.lock`; - const fd = fs.openSync(lockPath, 'wx', 0o600); - fs.writeFileSync(fd, JSON.stringify({ pid: process.pid, hostname: os.hostname() })); + const releaseLock = await holdFixtureLock(dbPath); let disconnectedRequest; const abandoned = request(`${server.url}/api/work-items/burst-0/claim`, 'POST', { owner: 'abandoned-client' }, 12000, req => { disconnectedRequest = req; }).catch(error => ({ error })); const accepted = request(`${server.url}/api/work-items/burst-1/claim`, 'POST', { owner: 'connected-client' }, 12000) .catch(error => ({ error })); - try { + await withCleanup(async () => { await waitForObservation(server, () => server.metrics.active === 2); disconnectedRequest.destroy(new Error('Intentional client disconnect')); await waitForObservation(server, () => server.metrics.disconnected === 1); @@ -207,12 +197,7 @@ async function disconnectScenario(server, dbPath) { assert.strictEqual(excess.status, 503, 'Disconnect must not release a slot while its worker is still alive'); assert.strictEqual(server.metrics.started, 2); assert.strictEqual(server.metrics.active, 2); - } finally { - fs.closeSync(fd); - fs.unlinkSync(lockPath); - await abandoned; - await accepted; - } + }, [releaseLock, () => abandoned, () => accepted]); await waitForObservation(server, () => server.metrics.active === 0); const retry = await request(`${server.url}/api/work-items/burst-2/claim`, 'POST', { owner: 'retry-client' }, 5000); assert.strictEqual(retry.status, 200, 'Actual worker exits must restore capacity'); @@ -221,14 +206,13 @@ async function disconnectScenario(server, dbPath) { async function contentionScenario() { return withServer(false, async (server, dbPath) => { const bytesBefore = fs.readFileSync(dbPath); - const lockPath = `${dbPath}.ecc-state.lock`; - const fd = fs.openSync(lockPath, 'wx', 0o600); - fs.writeFileSync(fd, JSON.stringify({ pid: process.pid, hostname: os.hostname() })); + const releaseLock = await holdFixtureLock(dbPath); let outcome; - try { + let mutation; + await withCleanup(async () => { const received = server.mutationReceived(); let settled = false; - const mutation = request(`${server.url}/api/work-items/task/claim`, 'POST', { owner: 'alice' }, 12000) + mutation = request(`${server.url}/api/work-items/task/claim`, 'POST', { owner: 'alice' }, 12000) .then(result => { settled = true; return result; }, error => { settled = true; return { transportError: error }; }); await received; let healthError = null; @@ -253,19 +237,14 @@ async function contentionScenario() { const result = await mutation; if (result.transportError) throw result.transportError; outcome = { result, healthError, healthChecks, snapshotChecks }; - } finally { - fs.closeSync(fd); - fs.unlinkSync(lockPath); - } + }, [releaseLock, () => mutation]); assert.deepStrictEqual(fs.readFileSync(dbPath), bytesBefore, 'Timed-out mutation must not change database bytes'); const store = await createStateStore({ dbPath }); - try { + await withCleanup(() => { const item = store.getWorkItemById('task'); assert.strictEqual(item.status, 'open'); assert.strictEqual(item.owner, null); - } finally { - store.close(); - } + }, [() => store.close()]); return outcome; }); } @@ -279,7 +258,7 @@ async function run() { console.log(` PASS ${name}`); passed += 1; } catch (error) { - console.log(` FAIL ${name}\n ${error.message}`); + console.log(` FAIL ${name}\n ${error && error.message ? error.message : String(error)}`); failed += 1; } } @@ -308,14 +287,12 @@ async function run() { assert.strictEqual(invalid.status, 400); assert.strictEqual(invalid.body.ok, false); const store = await createStateStore({ dbPath }); - try { + await withCleanup(() => { const item = store.getWorkItemById('task'); assert.strictEqual(item.owner, 'alice'); assert.strictEqual(item.status, 'blocked'); assert.strictEqual(item.metadata.assigneeKind, 'human'); - } finally { - store.close(); - } + }, [() => store.close()]); })); await test('read-only mode rejects board edits before database work', () => withServer(true, async server => { for (const action of ['claim', 'move']) { @@ -328,4 +305,5 @@ async function run() { console.log(`\nResults: Passed: ${passed}, Failed: ${failed}`); process.exitCode = failed ? 1 : 0; } +process.exitCode = 1; run().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/tests/lib/helpers/control-pane-contention-server.js b/tests/lib/helpers/control-pane-contention-server.js index c485ac23a..516db59ad 100644 --- a/tests/lib/helpers/control-pane-contention-server.js +++ b/tests/lib/helpers/control-pane-contention-server.js @@ -3,7 +3,10 @@ const workerThreads = require('worker_threads'); const OriginalWorker = workerThreads.Worker; const metrics = { active: 0, peak: 0, started: 0, exited: 0, received: 0, disconnected: 0 }; -function report() { process.send({ type: 'workers', ...metrics }); } +function send(message) { + if (process.connected) process.send(message, error => { if (error) process.exitCode = 1; }); +} +function report() { send({ type: 'workers', ...metrics }); } workerThreads.Worker = class ObservedWorker extends OriginalWorker { constructor(...args) { super(...args); @@ -25,21 +28,36 @@ async function main() { if (req.method === 'POST') { metrics.received += 1; report(); - process.send({ type: 'mutation-received' }); + send({ type: 'mutation-received' }); res.once('close', () => { if (!res.writableFinished) { metrics.disconnected += 1; report(); } }); } }); - await app.listen(); - process.send({ type: 'ready', url: app.url }); - process.on('message', async message => { - if (message === 'close') { - await app.close(); - process.disconnect(); - } - }); + let closing; + let startupFailed = false; + const close = () => { + if (!closing) closing = Promise.resolve().then(() => app.close()); + return closing; + }; + const finish = async () => { + try { await close(); if (!startupFailed) process.exitCode = 0; } + catch (error) { console.error(error); process.exitCode = 1; } + if (process.connected) process.disconnect(); + }; + process.on('disconnect', finish); + process.on('message', message => { if (message === 'close') finish(); }); + try { + await app.listen(); + send({ type: 'ready', url: app.url }); + } catch (error) { + startupFailed = true; + try { await close(); } catch (_cleanupError) { /* Preserve startup error. */ } + throw error; + } + } +process.exitCode = 1; main().catch(error => { console.error(error); process.exitCode = 1; diff --git a/tests/lib/helpers/state-store-worker.js b/tests/lib/helpers/state-store-worker.js index 045424e57..ccb6294f3 100644 --- a/tests/lib/helpers/state-store-worker.js +++ b/tests/lib/helpers/state-store-worker.js @@ -1,36 +1,229 @@ 'use strict'; -const { createStateStore } = require('../../../scripts/lib/state-store'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { fork } = require('child_process'); -let store; -process.on('message', async ({ id, action, dbPath, worker, count = 1 }) => { - try { - if (action === 'open') { - store = await createStateStore({ dbPath }); - } else if (action === 'write') { - for (let index = 0; index < count; index += 1) { - store.upsertWorkItem({ - id: `${worker}-${index}`, source: 'manual', title: `Task ${worker}-${index}`, status: 'open' - }); - } - } else if (action === 'increment') { - for (let index = 0; index < count; index += 1) { - store._database.transaction(() => { - const item = store.getWorkItemById('counter'); - // Widen real transaction overlap without assuming which worker wins. - Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 15); - store.upsertWorkItem({ ...item, metadata: { value: item.metadata.value + 1 } }); - })(); - } - } else if (action === 'close') { - store.close(); - store = null; - } else { - throw new Error(`Unknown fixture action: ${action}`); - } - process.send({ id, ok: true }); - } catch (error) { - process.send({ id, ok: false, error: error.stack }); +// These utilities belong only to the two process-based regression fixtures. +async function withCleanup(callback, cleanups) { + let failed = false; + let primary; + let value; + try { value = await callback(); } + catch (error) { failed = true; primary = error; } + for (const cleanup of cleanups) { + try { await cleanup(); } + catch (error) { if (!failed) { failed = true; primary = error; } } } -}); -process.send({ ready: true }); + if (failed) throw primary; + return value; +} + +function startOwnedChild(script, args = [], dependencies = {}) { + const forkChild = dependencies.fork || fork; + const platform = dependencies.platform || process.platform; + const kill = dependencies.kill || process.kill.bind(process); + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ecc-state-child-')); + const home = path.join(root, 'home'); + const tmp = path.join(root, 'tmp'); + let child; + try { + fs.mkdirSync(home); fs.mkdirSync(tmp); + const dependencyRoot = path.resolve(path.dirname(require.resolve('sql.js/package.json')), '..'); + child = forkChild(script, args, { + execPath: process.execPath, execArgv: [], shell: false, detached: platform !== 'win32', + cwd: root, stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + env: { PATH: path.dirname(process.execPath), NODE_PATH: dependencyRoot, + HOME: home, USERPROFILE: home, APPDATA: home, LOCALAPPDATA: home, + TMPDIR: tmp, TMP: tmp, TEMP: tmp, NODE_USE_ENV_PROXY: '0' }, + }); + } catch (error) { + try { fs.rmSync(root, { recursive: true, force: true }); } catch (_cleanupError) { /* Primary wins. */ } + throw error; + } + const observers = new Set(); + const waiters = new Set(); + const closeWaiters = new Set(); + let ownedPid = null; + let closed = false; + let reaped = false; + let exitCode = null; + let exitSignal = null; + let failed = false; + let failure; + let stderr = ''; + let captured = 0; + let stopPromise; + const fail = error => { + if (!failed) { failed = true; failure = error; } + for (const waiter of [...waiters]) waiter.reject(error); + }; + child.once('spawn', () => { if (Number.isInteger(child.pid) && child.pid > 0) ownedPid = child.pid; }); + child.on('error', fail); + child.once('exit', (code, signal) => { + reaped = true; exitCode = code; exitSignal = signal; + const error = new Error(`Fixture exited (${code}, ${signal || 'no signal'}): ${stderr}`); + for (const waiter of [...waiters]) waiter.reject(error); + }); + child.once('close', () => { + closed = true; + for (const done of [...closeWaiters]) done(true); + }); + child.on('message', message => { + for (const observer of [...observers]) { + try { observer(message); } catch (error) { fail(error); } + } + for (const waiter of [...waiters]) waiter.check(message); + }); + for (const stream of [child.stdout, child.stderr]) { + stream.on('data', data => { + captured += data.length; + if (captured > 128 * 1024) { fail(new Error('Fixture output exceeded 128 KiB')); return; } + if (stream === child.stderr) stderr += data.toString(); + }); + stream.on('error', fail); + } + function waitClosed(milliseconds) { + if (closed) return Promise.resolve(true); + return new Promise(resolve => { + let timer; + const done = value => { clearTimeout(timer); closeWaiters.delete(done); resolve(value); }; + closeWaiters.add(done); + timer = setTimeout(() => done(false), milliseconds); + }); + } + function signalOwned(signal) { + if (ownedPid === null) return; + try { + if (platform === 'win32') { if (!reaped) child.kill(signal); } + else kill(-ownedPid, signal); + } catch (error) { if (!error || error.code !== 'ESRCH') throw error; } + } + const client = { + child, + get diagnostics() { return { closed, reaped, captured, failed }; }, + observe(callback) { observers.add(callback); return () => observers.delete(callback); }, + waitFor(predicate, timeoutMs = 10000) { + if (failed || reaped) return Promise.reject(failed ? failure : new Error('Fixture already exited')); + return new Promise((resolve, reject) => { + let timer; + const finish = (fn, value) => { clearTimeout(timer); waiters.delete(waiter); fn(value); }; + const waiter = { + check(message) { + try { if (predicate(message)) finish(resolve, message); } + catch (error) { finish(reject, error); } + }, + reject: error => finish(reject, error), + }; + waiters.add(waiter); + timer = setTimeout(() => waiter.reject(new Error(`Fixture response timed out: ${stderr}`)), timeoutMs); + }); + }, + async completion() { + if (!await waitClosed(10000)) throw new Error('Fixture completion timed out'); + if (failed) throw failure; + return { status: exitCode, signal: exitSignal, stderr }; + }, + send(message) { + try { + if (!child.connected) throw new Error('Fixture IPC disconnected'); + child.send(message, error => { if (error) fail(error); }); + } catch (error) { fail(error); } + }, + stop() { + if (stopPromise) return stopPromise; + stopPromise = withCleanup(async () => { + // Disconnect asks the fixture to close normally. If it cannot, escalate + // only this newly spawned child/group; retain the deadline through close. + if (child.connected) { + try { child.disconnect(); } catch (error) { fail(error); } + } + if (await waitClosed(300)) { + if (reaped && (exitCode !== 0 || exitSignal !== null)) { + throw new Error(`Fixture closed unsuccessfully (${exitCode}, ${exitSignal || 'no signal'}): ${stderr}`); + } + return; + } + let signalFailed = false; + let signalError; + try { signalOwned('SIGTERM'); } catch (error) { signalFailed = true; signalError = error; } + await new Promise(resolve => setTimeout(resolve, 300)); + try { signalOwned('SIGKILL'); } catch (error) { if (!signalFailed) { signalFailed = true; signalError = error; } } + if (!await waitClosed(1000)) { + return withCleanup(() => { + if (signalFailed) throw signalError; + throw new Error('Fixture did not close after termination; pipes forcibly closed'); + }, [() => child.stdout.destroy(), () => child.stderr.destroy(), () => waitClosed(100)]); + } + if (signalFailed) throw signalError; + }, [() => { + for (const waiter of [...waiters]) waiter.reject(new Error('Fixture stopped')); + observers.clear(); + fs.rmSync(root, { recursive: true, force: true }); + }]).then(() => { + // A successful close does not erase an earlier IPC/stream/disconnect + // failure. Report it only after every owned cleanup stage was attempted. + if (failed) throw failure; + }, cleanupError => { + if (failed) throw failure; + throw cleanupError; + }); + return stopPromise; + }, + }; + return client; +} + +async function withOwnedChildren(count, start, callback) { + const children = []; + return withCleanup(async () => { + for (let index = 0; index < count; index += 1) children.push(await start(index)); + return callback(children); + }, [() => withCleanup(() => {}, children.map(child => () => child.stop()))]); +} + +module.exports = { startOwnedChild, withOwnedChildren, withCleanup }; + +function runWorker() { + const { createStateStore } = require('../../../scripts/lib/state-store'); + process.exitCode = 1; + let store; + process.on('message', async ({ id, action, dbPath, worker, count = 1 }) => { + try { + if (action === 'open') { + store = await createStateStore({ dbPath }); + } else if (action === 'write') { + for (let index = 0; index < count; index += 1) { + store.upsertWorkItem({ + id: `${worker}-${index}`, source: 'manual', title: `Task ${worker}-${index}`, status: 'open' + }); + } + } else if (action === 'increment') { + for (let index = 0; index < count; index += 1) { + store._database.transaction(() => { + const item = store.getWorkItemById('counter'); + // Widen real transaction overlap without assuming which worker wins. + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 15); + store.upsertWorkItem({ ...item, metadata: { value: item.metadata.value + 1 } }); + })(); + } + } else if (action === 'close') { + store.close(); + store = null; + } else { + throw new Error(`Unknown fixture action: ${action}`); + } + if (process.connected) process.send({ id, ok: true }); + } catch (error) { + if (process.connected) process.send({ id, ok: false, error: error && error.stack ? String(error.stack).slice(0, 4096) : String(error).slice(0, 4096) }); + } + }); + process.on('disconnect', () => { + try { if (store) store.close(); process.exitCode = 0; } + catch (error) { console.error(error); process.exitCode = 1; } + }); + process.send({ ready: true }); +} + +if (require.main === module) runWorker(); diff --git a/tests/lib/state-store-concurrency.test.js b/tests/lib/state-store-concurrency.test.js index 6a5f499a8..d9dfbb082 100644 --- a/tests/lib/state-store-concurrency.test.js +++ b/tests/lib/state-store-concurrency.test.js @@ -4,82 +4,45 @@ const assert = require('assert'); const fs = require('fs'); const os = require('os'); const path = require('path'); -const { fork, spawnSync } = require('child_process'); +const { startOwnedChild, withOwnedChildren, withCleanup } = require('./helpers/state-store-worker'); const { createStateStore } = require('../../scripts/lib/state-store'); const WORKER = path.join(__dirname, 'helpers', 'state-store-worker.js'); const WORK_ITEMS = path.join(__dirname, '..', '..', 'scripts', 'work-items.js'); -function startWorker() { - const child = fork(WORKER, [], { stdio: ['ignore', 'pipe', 'pipe', 'ipc'] }); - const pending = new Map(); +async function startWorker() { + const client = startOwnedChild(WORKER); + try { + await client.waitFor(message => message && message.ready === true); + } catch (error) { + return withCleanup(() => { throw error; }, [() => client.stop()]); + } let nextId = 0; - let stderr = ''; - child.stderr.on('data', data => { stderr += data; }); - child.stdout.resume(); - child.on('message', message => { - const request = pending.get(message.id); - if (!request) return; - clearTimeout(request.timer); - pending.delete(message.id); - if (message.ok) request.resolve(); - else request.reject(new Error(message.error)); - }); - child.on('exit', code => { - for (const request of pending.values()) { - clearTimeout(request.timer); - request.reject(new Error(`Worker exited ${code}: ${stderr}`)); - } - pending.clear(); - }); return { - request(action, options = {}) { + async request(action, options = {}) { const id = ++nextId; - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - pending.delete(id); - reject(new Error(`Worker timed out during ${action}: ${stderr}`)); - }, 30000); - pending.set(id, { resolve, reject, timer }); - child.send({ id, action, ...options }); - }); + const response = client.waitFor(message => message && message.id === id); + client.send({ id, action, ...options }); + const message = await response; + if (!message.ok) throw new Error(message.error); }, - async stop() { - if (child.exitCode !== null || child.signalCode !== null) return; - await new Promise(resolve => { - child.once('exit', resolve); - child.kill(); - }); - } + stop: () => client.stop(), }; } async function withDatabase(fn) { const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ecc-store-concurrency-')); const dbPath = path.join(dir, 'state.db'); - try { - await fn(dbPath); - } finally { - fs.rmSync(dir, { recursive: true, force: true }); - } + return withCleanup(() => fn(dbPath), [() => fs.rmSync(dir, { recursive: true, force: true })]); } async function readItems(dbPath) { const store = await createStateStore({ dbPath }); - try { - return store.listWorkItems({ limit: 1000 }).items; - } finally { - store.close(); - } + return withCleanup(() => store.listWorkItems({ limit: 1000 }).items, [() => store.close()]); } async function withWorkers(count, fn) { - const workers = Array.from({ length: count }, () => startWorker()); - try { - await fn(workers); - } finally { - await Promise.all(workers.map(worker => worker.stop())); - } + return withOwnedChildren(count, startWorker, fn); } async function run() { @@ -91,7 +54,7 @@ async function run() { console.log(` PASS ${name}`); passed += 1; } catch (error) { - console.log(` FAIL ${name}\n ${error.message}`); + console.log(` FAIL ${name}\n ${error && error.message ? error.message : String(error)}`); failed += 1; } } @@ -100,31 +63,27 @@ async function run() { await test('closing an old reader preserves a task saved by the real CLI', async dbPath => { const reader = await createStateStore({ dbPath }); - try { + await withCleanup(async () => { assert.strictEqual(reader.listWorkItems().items.length, 0); - const result = spawnSync(process.execPath, [WORK_ITEMS, 'upsert', 'cli-task', - '--title', 'Task saved while dashboard is open', '--db', dbPath, '--json'], - { encoding: 'utf8', timeout: 30000 }); + const cli = startOwnedChild(WORK_ITEMS, ['upsert', 'cli-task', + '--title', 'Task saved while dashboard is open', '--db', dbPath, '--json']); + const result = await withCleanup(() => cli.completion(), [() => cli.stop()]); assert.strictEqual(result.status, 0, result.stderr); - } finally { - reader.close(); - } + }, [() => reader.close()]); assert.ok((await readItems(dbPath)).some(item => item.id === 'cli-task'), 'Closing the older reader must not remove the CLI task'); }); await test('long-lived handles preserve each other\'s writes and return current data', async dbPath => { const first = await createStateStore({ dbPath }); - const second = await createStateStore({ dbPath }); - try { + let second; + await withCleanup(async () => { + second = await createStateStore({ dbPath }); first.upsertWorkItem({ id: 'a', source: 'manual', title: 'First task', status: 'open' }); second.upsertWorkItem({ id: 'b', source: 'manual', title: 'Second task', status: 'open' }); first.upsertWorkItem({ id: 'c', source: 'manual', title: 'Third task', status: 'open' }); assert.deepStrictEqual(second.listWorkItems().items.map(item => item.id).sort(), ['a', 'b', 'c']); - } finally { - first.close(); - second.close(); - } + }, [() => first.close(), () => { if (second) second.close(); }]); assert.deepStrictEqual((await readItems(dbPath)).map(item => item.id).sort(), ['a', 'b', 'c']); }); @@ -145,8 +104,9 @@ async function run() { await test('read-modify-write transactions from concurrent processes retain every increment', async dbPath => { const initial = await createStateStore({ dbPath }); - initial.upsertWorkItem({ id: 'counter', source: 'manual', title: 'Completed jobs', status: 'open', metadata: { value: 0 } }); - initial.close(); + await withCleanup(() => { + initial.upsertWorkItem({ id: 'counter', source: 'manual', title: 'Completed jobs', status: 'open', metadata: { value: 0 } }); + }, [() => initial.close()]); await withWorkers(3, async workers => { await Promise.all(workers.map(worker => worker.request('open', { dbPath }))); await Promise.all(workers.map(worker => worker.request('increment', { count: 4 }))); @@ -164,20 +124,19 @@ async function run() { await Promise.all(workers.map(worker => worker.request('close'))); }); const store = await createStateStore({ dbPath }); - try { + await withCleanup(() => { const migrations = store.getAppliedMigrations(); assert.ok(migrations.length > 0); assert.strictEqual(new Set(migrations.map(migration => migration.version)).size, migrations.length); assert.deepStrictEqual(store.listWorkItems().items.map(item => item.id).sort(), ['fresh-0-0', 'fresh-1-0', 'fresh-2-0']); - } finally { - store.close(); - } + }, [() => store.close()]); }); console.log(`\nResults: Passed: ${passed}, Failed: ${failed}`); process.exitCode = failed > 0 ? 1 : 0; } +process.exitCode = 1; run().catch(error => { console.error(error); process.exitCode = 1; diff --git a/tests/lib/state-store-persistence.test.js b/tests/lib/state-store-persistence.test.js index ce7c702ca..2a8077f66 100644 --- a/tests/lib/state-store-persistence.test.js +++ b/tests/lib/state-store-persistence.test.js @@ -1,11 +1,272 @@ 'use strict'; const assert = require('assert'); -const fs = require('fs'); +// Private facade: fault injection must not mutate shared fs or require.cache. +const fs = { ...require('fs') }; +const vm = require('vm'); +const { createRequire } = require('module'); const os = require('os'); const path = require('path'); -const { createStateStore } = require('../../scripts/lib/state-store'); -const { withStateStoreLock } = require('../../scripts/lib/state-store/file-lock'); +const STORE = require.resolve('../../scripts/lib/state-store'); +const LOCK = require.resolve('../../scripts/lib/state-store/file-lock'); +function loadPrivate(filename, overrides = {}, source = fs.readFileSync(filename, 'utf8'), processFacade = process) { + const module = { exports: {} }; + const nativeRequire = createRequire(filename); + const localRequire = id => Object.hasOwn(overrides, id) ? overrides[id] : nativeRequire(id); + vm.compileFunction(source, ['exports', 'require', 'module', '__filename', '__dirname', 'process'], + { filename })(module.exports, localRequire, module, filename, path.dirname(filename), processFacade); + return module.exports; +} +const lockModule = loadPrivate(LOCK, { fs }); +const { withStateStoreLock } = lockModule; +// sql.js is only initialized by the original integration cases, never the pure mode. +const { createStateStore } = loadPrivate(STORE, { fs, './file-lock': lockModule }); + +function thrownBy(fn) { + try { return { failed: false, value: fn() }; } + catch (error) { return { failed: true, error }; } +} + +function lockFixture(overrides = {}, platform = 'linux') { + const calls = []; + const identity = { dev: 1n, ino: 2n, isFile: () => true, isSymbolicLink: () => false }; + let present = true; + let closed = false; + const facade = { + openSync() { calls.push('open'); return 3; }, + fstatSync() { calls.push('fstat'); return identity; }, + writeFileSync() { calls.push('write'); }, + lstatSync() { calls.push('lstat'); if (!present) throw Object.assign(new Error('missing'), { code: 'ENOENT' }); return identity; }, + unlinkSync() { calls.push('unlink'); present = false; }, + closeSync() { calls.push('close'); closed = true; }, + }; + const state = { calls, identity, get present() { return present; }, set present(value) { present = value; }, + get closed() { return closed; } }; + for (const [name, fn] of Object.entries(overrides)) { + const original = facade[name]; + facade[name] = (...args) => fn(state, original, ...args); + } + const lock = loadPrivate(LOCK, { fs: facade, os: { hostname: () => 'fixture' } }, undefined, + { platform, pid: 7 }); + return { ...state, calls, run: callback => lock.withStateStoreLock('synthetic.db', callback, { timeoutMs: 0 }) }; +} + +async function runSynthetic(test) { + await test('owned lock unlinks before its only close', () => { + const fixture = lockFixture(); + assert.strictEqual(fixture.run(() => 'done'), 'done'); + assert.deepStrictEqual(fixture.calls, ['open', 'fstat', 'write', 'lstat', 'unlink', 'close']); + }); + await test('a replacement created during close is never unlinked', () => { + let replacement = false; + const fixture = lockFixture({ + closeSync(state, close) { close(); replacement = true; state.present = true; }, + unlinkSync(state, unlink) { assert.ok(!replacement, 'must not unlink after releasing the descriptor'); unlink(); }, + }); + fixture.run(() => {}); + assert.ok(replacement); + assert.strictEqual(fixture.calls.filter(call => call === 'close').length, 1); + }); + for (const platform of ['linux', 'win32']) { + for (const replacement of [{ ino: 3n }, { dev: 9n }, { dev: 0n }, { isFile: () => false }, { isSymbolicLink: () => true }]) { + await test(`observed lock replacement is preserved (${platform}, ${Object.keys(replacement)[0]})`, () => { + const fixture = lockFixture({ lstatSync(state) { state.calls.push('lstat'); return { ...state.identity, ...replacement }; } }, platform); + const result = thrownBy(() => fixture.run(() => {})); + assert.ok(result.failed); + assert.match(result.error.message, /lock.*(changed|lost)/i); + assert.ok(!fixture.calls.includes('unlink')); + assert.strictEqual(fixture.calls.filter(call => call === 'close').length, 1); + }); + } + } + for (const afterClose of ['missing', 'permission', 'replacement']) { + await test(`initial EPERM performs only a post-close read probe (${afterClose})`, () => { + const denied = Object.assign(new Error('denied'), { code: 'EPERM' }); + const fixture = lockFixture({ lstatSync(state) { + state.calls.push('lstat'); + if (!state.closed || afterClose === 'permission') throw denied; + if (afterClose === 'missing') throw Object.assign(new Error('gone'), { code: 'ENOENT' }); + return { ...state.identity, ino: 10n }; + } }, 'win32'); + const result = thrownBy(() => fixture.run(() => {})); + assert.ok(result.failed); + if (afterClose === 'missing') assert.strictEqual(result.error.code, 'STATE_STORE_LOCK_LOST'); + else assert.strictEqual(result.error, denied); + assert.deepStrictEqual(fixture.calls.slice(3), ['lstat', 'close', 'lstat']); + }); + } + await test('close failure stops the pending-delete probe and preserves initial EPERM', () => { + const denied = Object.assign(new Error('denied'), { code: 'EPERM' }); + const cleanup = new Error('close failed'); + const fixture = lockFixture({ + lstatSync(state) { state.calls.push('lstat'); throw denied; }, + closeSync(state) { state.calls.push('close'); throw cleanup; }, + }); + const result = thrownBy(() => fixture.run(() => {})); + assert.ok(result.failed); + assert.strictEqual(result.error, denied); + assert.strictEqual(result.error.closeError, cleanup); + assert.deepStrictEqual(fixture.calls.slice(3), ['lstat', 'close']); + }); + await test('initial missing lock is loss after descriptor closure', () => { + const fixture = lockFixture({ lstatSync(state) { state.calls.push('lstat'); throw Object.assign(new Error('gone'), { code: 'ENOENT' }); } }); + const result = thrownBy(() => fixture.run(() => {})); + assert.ok(result.failed); + assert.strictEqual(result.error.code, 'STATE_STORE_LOCK_LOST'); + assert.deepStrictEqual(fixture.calls.slice(3), ['lstat', 'close']); + }); + await test('unlink EPERM is not mistaken for an initial pending-delete lookup', () => { + const denied = Object.assign(new Error('unlink denied'), { code: 'EPERM' }); + const fixture = lockFixture({ unlinkSync(state) { state.calls.push('unlink'); throw denied; } }); + const result = thrownBy(() => fixture.run(() => {})); + assert.ok(result.failed); + assert.strictEqual(result.error, denied); + assert.deepStrictEqual(fixture.calls.slice(3), ['lstat', 'unlink', 'close']); + }); + const primaries = [null, undefined, false, 0, '', 'primitive', Object.freeze(new Error('frozen')), + Object.defineProperty(new Error('setter'), 'releaseError', { set() { throw new Error('setter must not mask'); } })]; + for (const [index, primary] of primaries.entries()) { + for (const stage of ['callback', 'metadata', 'identity']) { + await test(`arbitrary primary survives lock cleanup (${index}, ${stage})`, () => { + const cleanup = new Error('cleanup failed'); + const overrides = stage === 'callback' + ? { unlinkSync() { throw cleanup; } } + : stage === 'metadata' + ? { writeFileSync() { throw primary; }, unlinkSync() { throw cleanup; } } + : { fstatSync() { throw primary; }, closeSync() { throw cleanup; } }; + const fixture = lockFixture(overrides); + const result = thrownBy(() => fixture.run(() => { if (stage === 'callback') throw primary; assert.fail('callback must not run'); })); + assert.ok(result.failed); + assert.strictEqual(result.error, primary); + }); + } + for (const stage of ['migration', 'rollback']) { + await test(`arbitrary primary survives state-store ${stage} cleanup (${index})`, async () => { + const secondary = new Error('cleanup failed'); + let closes = 0; + const fakeSQL = { Database: class { + run(sql) { if (sql === 'ROLLBACK') throw secondary; } + close() { closes++; throw secondary; } + } }; + const api = loadPrivate(STORE, { + 'sql.js': async () => fakeSQL, + './migrations': { applyMigrations() { if (stage === 'migration') throw primary; return []; } }, + './queries': { createQueryApi: () => ({}) }, + }); + let failed = false; + let caught; + try { + const store = await api.createStateStore({ dbPath: ':memory:' }); + store._database.transaction(() => { throw primary; })(); + } catch (error) { failed = true; caught = error; } + assert.ok(failed); + assert.strictEqual(caught, primary); + assert.strictEqual(closes, 1); + }); + } + } + for (const stage of ['initialize', 'previous-close']) { + for (const [index, primary] of [null, undefined, Object.freeze(new Error('reload primary'))].entries()) { + await test(`reload closes its unadopted handle and preserves ${stage} failure (${index})`, () => { + const handles = []; + const SQL = { Database: class { + constructor() { this.id = handles.length; this.closes = 0; handles.push(this); } + run() { if (stage === 'initialize') throw primary; } + close() { + this.closes++; + if (stage === 'previous-close' && this.id === 0) throw primary; + throw new Error('unadopted cleanup'); + } + } }; + // Export only the existing private wrapper in this isolated exact-source module. + const source = `${fs.readFileSync(STORE, 'utf8')}\nmodule.exports.testWrap = wrapSqlJsDatabase;`; + const api = loadPrivate(STORE, { + fs: { lstatSync() { throw Object.assign(new Error('absent'), { code: 'ENOENT' }); } }, + './file-lock': { ...lockModule, withStateStoreLock: (_path, callback) => callback() }, + }, source); + const db = api.testWrap(SQL, 'synthetic.db'); + if (stage === 'previous-close') db.withSnapshot(() => {}); + const result = thrownBy(() => db.withSnapshot(() => assert.fail('must not adopt failed handle'))); + assert.ok(result.failed); + assert.strictEqual(result.error, primary); + assert.strictEqual(handles.at(-1).closes, 1); + if (stage === 'previous-close') { + assert.strictEqual(handles[0].closes, 1); + assert.throws(() => db.withSnapshot(() => {}), /closed/); + } + db.close(); + assert.ok(handles.every(handle => handle.closes === 1)); + }); + } + } + await test('reload adopts a ready replacement and closes each owned handle once', () => { + const calls = []; + let next = 0; + const SQL = { Database: class { + constructor() { this.id = next++; } + run() { calls.push(`ready:${this.id}`); } + close() { calls.push(`close:${this.id}`); } + } }; + const source = `${fs.readFileSync(STORE, 'utf8')}\nmodule.exports.testWrap = wrapSqlJsDatabase;`; + const api = loadPrivate(STORE, { + fs: { lstatSync() { throw Object.assign(new Error('absent'), { code: 'ENOENT' }); } }, + './file-lock': { ...lockModule, withStateStoreLock: (_path, callback) => callback() }, + }, source); + const db = api.testWrap(SQL, 'synthetic.db'); + db.withSnapshot(() => {}); + db.withSnapshot(() => {}); + db.close(); db.close(); + assert.deepStrictEqual(calls, ['ready:0', 'ready:1', 'close:0', 'close:1']); + }); + for (const [index, primary] of [null, undefined, false, 0, '', 'primitive', + Object.freeze(new Error('frozen worker')), { get message() { throw new Error('getter'); } }].entries()) { + await test(`worker reports bounded arbitrary mutation failure after close (${index})`, async () => { + const filename = require.resolve('../../scripts/lib/control-pane/work-item-worker'); + const source = fs.readFileSync(filename, 'utf8'); + assert.strictEqual(source.split('mutate().then(').length, 2); + const captured = source.replace('mutate().then(', 'module.exports = mutate().then('); + const messages = []; + let closes = 0; + await loadPrivate(filename, { + worker_threads: { workerData: { dbPath: ':memory:', action: 'claim', args: {} }, + parentPort: { postMessage(message) { messages.push(message); } } }, + '../state-store': { createStateStore: async () => ({ + _database: { transaction: fn => fn }, close() { closes++; throw new Error('close failed'); }, + }) }, + './work-item-mutations': { claimWorkItem() { throw primary; } }, + }, captured); + assert.strictEqual(closes, 1); + assert.strictEqual(messages.length, 1); + assert.strictEqual(messages[0].ok, false); + assert.strictEqual(typeof messages[0].error.message, 'string'); + assert.ok(messages[0].error.message.length <= 1024); + if (index < 6) assert.strictEqual(messages[0].error.message, String(primary)); + if (index === 6) assert.strictEqual(messages[0].error.message, primary.message); + }); + } + await test('worker rejects close-only failure after successful mutation', async () => { + const filename = require.resolve('../../scripts/lib/control-pane/work-item-worker'); + const source = fs.readFileSync(filename, 'utf8').replace('mutate().then(', 'module.exports = mutate().then('); + const messages = []; + await loadPrivate(filename, { + worker_threads: { workerData: { dbPath: ':memory:', action: 'claim', args: {} }, + parentPort: { postMessage(message) { messages.push(message); } } }, + '../state-store': { createStateStore: async () => ({ + _database: { transaction: fn => fn }, close() { throw new Error('close-only'); }, + }) }, + './work-item-mutations': { claimWorkItem() { return 'committed'; } }, + }, source); + assert.deepStrictEqual(messages, [{ ok: false, error: { message: 'close-only', code: undefined } }]); + }); + await test('cleanup-only lock failure still rejects a successful operation', () => { + const cleanup = new Error('close failed'); + const fixture = lockFixture({ closeSync() { throw cleanup; } }); + const result = thrownBy(() => fixture.run(() => 'committed')); + assert.ok(result.failed); + assert.strictEqual(result.error, cleanup); + }); +} + const item = id => ({ id, source: 'manual', title: id, status: 'open' }); @@ -19,13 +280,20 @@ async function run() { console.log(` PASS ${name}`); passed += 1; } catch (error) { - console.log(` FAIL ${name}\n ${error.stack}`); + console.log(` FAIL ${name}\n ${error && error.stack ? error.stack : String(error)}`); failed += 1; } finally { fs.rmSync(directory, { recursive: true, force: true }); } } + await runSynthetic(test); + if (process.argv.includes('--synthetic-only')) { + console.log(`\nResults: Passed: ${passed}, Failed: ${failed}`); + process.exitCode = failed ? 1 : 0; + return; + } + await test('queries, migration inspection and closing never rewrite an existing database', async dbPath => { const initial = await createStateStore({ dbPath }); initial.upsertWorkItem(item('saved')); @@ -243,4 +511,5 @@ async function run() { process.exitCode = failed ? 1 : 0; } +process.exitCode = 1; run().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/tests/lib/work-item-runner.test.js b/tests/lib/work-item-runner.test.js index a38a148e4..dfed8bcf2 100644 --- a/tests/lib/work-item-runner.test.js +++ b/tests/lib/work-item-runner.test.js @@ -2,13 +2,15 @@ const assert = require('assert'); const { EventEmitter } = require('events'); -const workerThreads = require('worker_threads'); +const fs = require('fs'); +const path = require('path'); +const vm = require('vm'); +const { createRequire } = require('module'); const RUNNER = require.resolve('../../scripts/lib/control-pane/work-item-runner'); const nextTurn = () => new Promise(resolve => setImmediate(resolve)); async function withFakeWorkers(callback) { - const OriginalWorker = workerThreads.Worker; const workers = []; const requests = []; let constructorFailures = 0; @@ -39,14 +41,13 @@ async function withFakeWorkers(callback) { } } - delete require.cache[RUNNER]; - workerThreads.Worker = FakeWorker; - let runWorkItemMutation; - try { - ({ runWorkItemMutation } = require(RUNNER)); - } finally { - workerThreads.Worker = OriginalWorker; - } + const module = { exports: {} }; + const nativeRequire = createRequire(RUNNER); + const localRequire = id => id === 'worker_threads' ? { Worker: FakeWorker } : nativeRequire(id); + vm.compileFunction(fs.readFileSync(RUNNER, 'utf8'), + ['exports', 'require', 'module', '__filename', '__dirname'], { filename: RUNNER })( + module.exports, localRequire, module, RUNNER, path.dirname(RUNNER)); + const { runWorkItemMutation } = module.exports; const context = { workers, get constructionAttempts() { return constructionAttempts; }, @@ -76,7 +77,6 @@ async function withFakeWorkers(callback) { if (!worker.exited) worker.finish(); } await Promise.all(requests.map(request => request.result)); - delete require.cache[RUNNER]; } } @@ -97,7 +97,7 @@ async function run() { console.log(` PASS ${name}`); passed += 1; } catch (error) { - console.log(` FAIL ${name}\n ${error.stack}`); + console.log(` FAIL ${name}\n ${error && error.stack ? error.stack : String(error)}`); failed += 1; } } @@ -220,8 +220,38 @@ async function run() { assert.match(result.error.message, /exit/i); }); + + for (const response of [true, 'bad', {}, { ok: 1, result: 'wrong' }, { ok: false }, + { ok: false, error: null }, { ok: false, error: { message: 7 } }, + { ok: false, error: { message: 'bad', code: {} } }]) { + await test(`malformed worker response rejects after exit (${JSON.stringify(response)})`, async context => { + const request = context.request(); + await nextTurn(); + context.workers[0].emit('message', response); + await nextTurn(); + assert.strictEqual(request.settled, false); + assert.doesNotThrow(() => context.workers[0].exit()); + const result = await request.result; + assert.strictEqual(result.ok, false); + assert.match(result.error.message, /invalid.*response/i); + }); + } + await test('disconnect and a never-exiting worker keep their slots occupied', async context => { + const first = context.request(); + const second = context.request(); + await nextTurn(); + context.workers[0].emit('disconnect'); + context.workers[1].emit('message', { ok: true, result: 'not exited' }); + await assertBusy(context.request()); + assert.strictEqual(first.settled, false); + assert.strictEqual(second.settled, false); + assert.strictEqual(context.constructionAttempts, 2); + // Fake workers are explicitly drained by the harness, never terminated by production. + }); + console.log(`\nResults: Passed: ${passed}, Failed: ${failed}`); process.exitCode = failed ? 1 : 0; } +process.exitCode = 1; run().catch(error => { console.error(error); process.exitCode = 1; });