diff --git a/docs/architecture/hud-status-session-control.md b/docs/architecture/hud-status-session-control.md index 8b773e20c..f8f1c269d 100644 --- a/docs/architecture/hud-status-session-control.md +++ b/docs/architecture/hud-status-session-control.md @@ -78,3 +78,43 @@ handoffs. The `ecc.hud-status.v1` payload is the common outer contract these surfaces can project into before ECC grows a dedicated full-screen HUD. + +## State Store Concurrency And Recovery + +ECC's file-backed state store serializes each synchronous query or transaction +with a sibling `.ecc-state.lock` file. It checks the current database +under that lock and publishes successful writes before releasing it. A handle +can reuse its last successful read snapshot when the file identity, size, and +modification/change timestamps match. Fixed SELECTs in the query API avoid +reloading or exporting an unchanged database; query results are still computed +afresh. Generic SQL, transactions, exports, and failed operations invalidate +reuse. Changes from another writer trigger a reload on the next operation. +Queries use one snapshot; closing a handle never writes an older snapshot back. +All concurrent writers must use this adapter; older ECC versions and external +SQLite writers do not participate in this locking protocol. + +An operation waits up to five seconds before reporting `STATE_STORE_BUSY`. +Retry when the other operation finishes. Locks are never stolen based on age: +a paused process may still be writing. After an abnormal exit, stop all ECC +processes using that database, inspect the PID and hostname in the lock file, +and remove only the leftover `.ecc-state.lock` file before retrying. Do not +remove the database itself. In-memory stores do not create lock files. + +The control-pane HTTP server runs board claims and moves in one-shot workers, +so waiting for another writer does not stall health checks or snapshots. A +mutation that times out while lock creation reports `EEXIST` returns HTTP 503 +with `code: STATE_STORE_BUSY` and `Retry-After: 1`. On Windows, lock creation +also retries `EPERM` within the same deadline because deletion may still be +pending. If the final attempt still reports `EPERM`, the original permission +error is preserved rather than replaced with busy or leftover-lock recovery +advice. Invalid mutations continue to return HTTP 400. Each worker +closes its store before reporting the result and exits naturally, including +when the requesting browser disconnects, to avoid interrupting a write. + +At most two board-mutation workers may exist in a server process, shared +across database paths and server instances. Admission happens before worker +creation. When both slots are occupied, extra requests receive the same +retryable HTTP 503 immediately; they are not queued or executed later. +A slot remains occupied until the worker actually exits, including after a +result, error, or browser disconnect. Construction failure releases the slot +immediately. This bounds waiting workers even while the database is locked. diff --git a/scripts/lib/control-pane/server.js b/scripts/lib/control-pane/server.js index 8391ac796..cf4d73168 100644 --- a/scripts/lib/control-pane/server.js +++ b/scripts/lib/control-pane/server.js @@ -11,20 +11,7 @@ const { renderControlPaneHtml } = require('./ui'); const { renderProximityVizHtml } = require('./proximity-viz'); const { renderControlPlaneViewHtml } = require('./control-plane-view-ui'); const { createControlPlaneViewSource } = require('./control-plane-view'); -const { claimWorkItem, moveWorkItem } = require('./work-item-mutations'); - -// Run a single write against the local work-item store, then close it. Kept -// thin so the loopback-only server can mutate the JIT board without holding a -// long-lived handle. -async function withStateStore(stateDbPath, fn) { - const { createStateStore } = require('../state-store'); - const store = await createStateStore({ dbPath: stateDbPath }); - try { - return await fn(store); - } finally { - store.close(); - } -} +const { runWorkItemMutation } = require('./work-item-runner'); // Host/Origin gating lives in scripts/lib/loopback-guard.js so every ECC // loopback server shares one hardened implementation; re-exported below to @@ -349,19 +336,25 @@ function createControlPaneServer(options = {}) { const id = decodeURIComponent((claimMatch || moveMatch)[1]); const body = await readRequestJson(req); try { - const result = await withStateStore(resolvedConfig.stateDbPath, store => + const result = await runWorkItemMutation(resolvedConfig.stateDbPath, + claimMatch ? 'claim' : 'move', claimMatch - ? claimWorkItem(store, { + ? { id, owner: body.owner, assigneeKind: body.as || body.assigneeKind, sessionId: body.sessionId - }) - : moveWorkItem(store, { id, lane: body.lane }) + } + : { id, lane: body.lane } ); sendJson(res, 200, { ok: true, ...result }); } catch (mutationError) { - sendJson(res, 400, { ok: false, error: mutationError.message }); + if (mutationError.code === 'STATE_STORE_BUSY') { + res.setHeader('Retry-After', '1'); + sendJson(res, 503, { ok: false, code: mutationError.code, error: mutationError.message }); + } else { + sendJson(res, 400, { ok: false, error: mutationError.message }); + } } return; } diff --git a/scripts/lib/control-pane/work-item-runner.js b/scripts/lib/control-pane/work-item-runner.js new file mode 100644 index 000000000..898f2e1f2 --- /dev/null +++ b/scripts/lib/control-pane/work-item-runner.js @@ -0,0 +1,57 @@ +'use strict'; + +const path = require('path'); +const { Worker } = require('worker_threads'); + +const MAX_ACTIVE_MUTATIONS = 2; +let activeMutations = 0; + +/** 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) { + reject(Object.assign(new Error('Too many work-item mutations are running. Retry after an operation finishes.'), + { code: 'STATE_STORE_BUSY' })); + return; + } + activeMutations += 1; + let worker; + try { + worker = new Worker(path.join(__dirname, 'work-item-worker.js'), { + workerData: { dbPath, action, args } + }); + } catch (error) { + activeMutations -= 1; + reject(error); + return; + } + let response; + let workerError; + let failed = false; + worker.once('message', message => { response = message; }); + 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 (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 + // a client disconnect could interrupt a write and abandon its file lock. + }); +} + +module.exports = { runWorkItemMutation }; diff --git a/scripts/lib/control-pane/work-item-worker.js b/scripts/lib/control-pane/work-item-worker.js new file mode 100644 index 000000000..495b00073 --- /dev/null +++ b/scripts/lib/control-pane/work-item-worker.js @@ -0,0 +1,44 @@ +'use strict'; + +const { parentPort, workerData } = require('worker_threads'); +const { createStateStore } = require('../state-store'); +const { recordCleanupError } = require('../state-store/file-lock'); +const { claimWorkItem, moveWorkItem } = require('./work-item-mutations'); + +/** Complete one board mutation away from the HTTP event loop, then close. */ +async function mutate() { + const { dbPath, action, args } = workerData; + const mutation = action === 'claim' ? claimWorkItem : action === 'move' ? moveWorkItem : null; + if (!mutation) throw new Error('Unknown work-item mutation'); + const store = await createStateStore({ dbPath }); + 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) recordCleanupError(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: serializeError(error) }) +); diff --git a/scripts/lib/state-store/file-lock.js b/scripts/lib/state-store/file-lock.js new file mode 100644 index 000000000..735b9bf23 --- /dev/null +++ b/scripts/lib/state-store/file-lock.js @@ -0,0 +1,147 @@ +'use strict'; + +const fs = require('fs'); +const os = require('os'); +const { performance } = require('perf_hooks'); + +const WAIT_BUFFER = new Int32Array(new SharedArrayBuffer(4)); +const DEFAULT_TIMEOUT_MS = 5000; + +const CLEANUP_ERRORS = new WeakMap(); +const EMPTY_CLEANUP_ERRORS = Object.freeze({}); + +// Preserve the original throw's identity and properties, including frozen errors +// and proxies. Primitive throws cannot own diagnostics and are left unchanged. +function recordCleanupError(primary, name, secondary) { + if (primary === null || (typeof primary !== 'object' && typeof primary !== 'function')) return; + CLEANUP_ERRORS.set(primary, Object.freeze({ ...getCleanupErrors(primary), [name]: secondary })); +} + +// Frozen snapshots are local to this module instance; they do not cross workers. +function getCleanupErrors(primary) { + return CLEANUP_ERRORS.get(primary) || EMPTY_CLEANUP_ERRORS; +} + +function hasCode(error, code) { + try { return error !== null && error !== undefined && error.code === code; } + catch (_error) { return false; } +} + +function sameIdentity(left, right) { + if (left.ino !== right.ino) return false; + // Windows path stats can omit the volume serial (dev = 0) even when fstat + // reports it. Keep device checks strict elsewhere and when both are known. + if (process.platform === 'win32' && (left.dev === 0n || right.dev === 0n)) return true; + return 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 failed = false; + let primary; + let pendingDelete = false; + let inspected = false; + let closeFailed = false; + try { + 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'); + } + try { + fs.closeSync(descriptor); // Exactly one attempt, even after a release failure. + } catch (error) { + closeFailed = true; + if (failed) recordCleanupError(primary, 'closeError', error); + else { failed = true; primary = error; } + } + 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) { + const lockPath = `${dbPath}.ecc-state.lock`; + const deadline = performance.now() + timeoutMs; + for (;;) { + let descriptor; + try { + descriptor = fs.openSync(lockPath, 'wx', 0o600); + } catch (error) { + // Windows may report EPERM while the previous owner's unlinked file is + // pending deletion. Retry exclusive creation within the same deadline. + const retryablePermissionError = process.platform === 'win32' && hasCode(error, 'EPERM'); + if (!hasCode(error, 'EEXIST') && !retryablePermissionError) throw error; + const remaining = deadline - performance.now(); + if (remaining <= 0) { + // EPERM can also mean genuine access denial; preserve its diagnostics + // rather than advising the caller to remove a possibly unrelated lock. + if (retryablePermissionError) throw error; + const busy = new Error(`State store is busy: ${dbPath}. Retry after the other ECC operation finishes. ` + + `If an operation terminated unexpectedly, stop all ECC processes using this database, ` + + `then inspect and remove the leftover lock: ${lockPath}`); + busy.code = 'STATE_STORE_BUSY'; + throw busy; + } + Atomics.wait(WAIT_BUFFER, 0, 0, Math.min(20, remaining)); + continue; + } + let identity; + try { + identity = fs.fstatSync(descriptor, { bigint: true }); + fs.writeFileSync(descriptor, `${JSON.stringify({ pid: process.pid, hostname: os.hostname() })}\n`); + } catch (error) { + try { + if (identity) releaseOwnedLock(lockPath, descriptor, identity); + else fs.closeSync(descriptor); + } catch (releaseError) { + recordCleanupError(error, 'releaseError', releaseError); + } + throw error; + } + return () => releaseOwnedLock(lockPath, descriptor, identity); + } +} + +/** Serialize one synchronous snapshot operation, never a handle's lifetime. + * Locks are not expired or stolen: a paused writer must not lose exclusivity. + * An abnormal process exit can leave a lock requiring the explicit recovery + * described by STATE_STORE_BUSY. No database bytes are changed on timeout. */ +function withStateStoreLock(dbPath, callback, { timeoutMs = DEFAULT_TIMEOUT_MS } = {}) { + const release = acquireLock(dbPath, timeoutMs); + let result; + let primaryError; + let failed = false; + try { + result = callback(); + } catch (error) { + failed = true; + primaryError = error; + } + try { + release(); + } catch (releaseError) { + if (!failed) throw releaseError; + recordCleanupError(primaryError, 'releaseError', releaseError); + } + if (failed) throw primaryError; + return result; +} + +module.exports = { withStateStoreLock, recordCleanupError, getCleanupErrors }; diff --git a/scripts/lib/state-store/index.js b/scripts/lib/state-store/index.js index 49b6204d6..18a5a4de5 100644 --- a/scripts/lib/state-store/index.js +++ b/scripts/lib/state-store/index.js @@ -5,6 +5,7 @@ const crypto = require('crypto'); const os = require('os'); const path = require('path'); const initSqlJs = require('sql.js'); +const { withStateStoreLock, recordCleanupError } = require('./file-lock'); const { applyMigrations, getAppliedMigrations } = require('./migrations'); const { createQueryApi } = require('./queries'); @@ -26,9 +27,9 @@ function stateStorePathError(targetPath, detail) { return new Error(`Unsafe state-store path '${targetPath}': ${detail}`); } -function lstatIfPresent(targetPath) { +function lstatIfPresent(targetPath, options) { try { - return fs.lstatSync(targetPath); + return fs.lstatSync(targetPath, options); } catch (error) { if (error && error.code === 'ENOENT') { return null; @@ -99,8 +100,8 @@ function ensurePrivateDirectory(directoryPath) { return absolutePath; } -function assertSafeDatabaseFile(dbPath) { - const stats = lstatIfPresent(dbPath); +function assertSafeDatabaseFile(dbPath, options) { + const stats = lstatIfPresent(dbPath, options); assertNotSymlink(dbPath, stats); if (stats && !stats.isFile()) { throw stateStorePathError(dbPath, 'database path is not a regular file'); @@ -113,16 +114,25 @@ function readDatabaseFile(dbPath) { const noFollow = fs.constants.O_NOFOLLOW || 0; const fileDescriptor = fs.openSync(dbPath, fs.constants.O_RDONLY | noFollow); try { - const stats = fs.fstatSync(fileDescriptor); + const stats = fs.fstatSync(fileDescriptor, { bigint: true }); if (!stats.isFile()) { throw stateStorePathError(dbPath, 'database path is not a regular file'); } - return fs.readFileSync(fileDescriptor); + return { bytes: fs.readFileSync(fileDescriptor), identity: stats }; } finally { fs.closeSync(fileDescriptor); } } +function sameDatabaseSnapshot(left, right) { + if (!left || !right || left.ino !== right.ino) return false; + // Windows path stats can omit the device reported by the open descriptor. + const sameDevice = left.dev === right.dev + || (process.platform === 'win32' && (left.dev === 0n || right.dev === 0n)); + return sameDevice && left.size === right.size + && left.mtimeNs === right.mtimeNs && left.ctimeNs === right.ctimeNs; +} + function syncDirectory(directoryPath) { if (process.platform === 'win32') { return; @@ -201,142 +211,240 @@ function resolveStateStorePath(options = {}) { * IMPORTANT: sql.js db.export() implicitly ends any active transaction, so * we must defer all disk writes until after the transaction commits. */ -function wrapSqlJsDatabase(rawDb, dbPath) { +function wrapSqlJsDatabase(SQL, dbPath, trustedReads = new WeakSet()) { + let rawDb = null; + let closed = false; + let inSnapshot = false; + let dirty = false; let inTransaction = false; + let needsSnapshotComparison = false; + let snapshotBytes = null; + let readCacheIdentity = null; + let cacheableSnapshot = false; - function saveToDisk() { - if (dbPath === ':memory:' || inTransaction) { - return; + function reload() { + const cachedIdentity = readCacheIdentity; + readCacheIdentity = null; + if (dbPath === ':memory:' && rawDb) return; + const current = dbPath !== ':memory:' ? assertSafeDatabaseFile(dbPath, { bigint: true }) : null; + if (rawDb && sameDatabaseSnapshot(cachedIdentity, current)) return cachedIdentity; + const snapshot = current ? readDatabaseFile(dbPath) : null; + const bytes = snapshot ? snapshot.bytes : undefined; + // sql.js can use its input buffer as writable backing storage. + const originalBytes = bytes ? Buffer.from(bytes) : null; + const latest = new SQL.Database(bytes); + 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) { recordCleanupError(error, 'closeError', closeError); } + throw error; + } + rawDb = latest; + snapshotBytes = originalBytes; + return snapshot && snapshot.identity; + } + + // Hold one lock from reload through commit. Nested statements and public + // query methods reuse the snapshot, including a transaction's reads. + function withSnapshot(callback) { + if (closed) throw new Error('State store is closed'); + if (inSnapshot) return callback(); + const execute = () => { + const identity = reload(); + inSnapshot = true; + dirty = false; + needsSnapshotComparison = false; + cacheableSnapshot = true; + try { + const result = callback(); + if (result && typeof result.then === 'function') { + throw new Error('State-store operations must be synchronous'); + } + if ((dirty || needsSnapshotComparison) && dbPath !== ':memory:') { + // Prepared get/all statements and PRAGMAs can also write. Compare + // database bytes instead of guessing from SQL syntax, while keeping + // read-only queries from rewriting the file. Export only here: + // exporting inside a transaction would implicitly end it before + // our commit/rollback boundary. + const data = Buffer.from(rawDb.export()); + if (dirty || !snapshotBytes || !data.equals(snapshotBytes)) { + writeDatabaseFileAtomic(dbPath, data); + } + } + // Only a successful snapshot of trusted SELECTs can retain its handle. + // Export and generic SQL may reset connection settings or alter state; + // those paths must reload before the next operation, even on failure. + if (cacheableSnapshot && !dirty && !needsSnapshotComparison) readCacheIdentity = identity; + return result; + } finally { + inSnapshot = false; + dirty = false; + needsSnapshotComparison = false; + cacheableSnapshot = false; + } + }; + try { + return dbPath === ':memory:' ? execute() : withStateStoreLock(dbPath, execute); + } catch (error) { + readCacheIdentity = null; + throw error; + } + } + + function query(sql, positionalArgs, firstOnly, trustedRead) { + return withSnapshot(() => { + if (!trustedRead) needsSnapshotComparison = true; + const stmt = rawDb.prepare(sql); + try { + if (positionalArgs.length === 1 && typeof positionalArgs[0] !== 'object') { + stmt.bind([positionalArgs[0]]); + } else if (positionalArgs.length > 1) { + stmt.bind(positionalArgs); + } + if (firstOnly) return stmt.step() ? stmt.getAsObject() : null; + const rows = []; + while (stmt.step()) rows.push(stmt.getAsObject()); + return rows; + } finally { + stmt.free(); + } + }); + } + + function runStatement(sql, namedParams) { + return withSnapshot(() => { + needsSnapshotComparison = true; + const stmt = rawDb.prepare(sql); + try { + if (namedParams && typeof namedParams === 'object' && !Array.isArray(namedParams)) { + stmt.bind(Object.fromEntries(Object.entries(namedParams) + .map(([key, value]) => [`@${key}`, value === undefined ? null : value]))); + } + stmt.step(); + dirty = true; + } finally { + stmt.free(); + } + }); + } + + function transact(fn, args) { + if (inTransaction) throw new Error('Nested state-store transactions are not supported'); + cacheableSnapshot = false; + rawDb.run('BEGIN'); + inTransaction = true; + const previouslyDirty = dirty; + try { + const result = fn(...args); + if (result && typeof result.then === 'function') { + throw new Error('State-store transactions must be synchronous'); + } + rawDb.run('COMMIT'); + return result; + } catch (error) { + try { + rawDb.run('ROLLBACK'); + } catch (rollbackError) { + // Never reuse an uncertain transaction, including an in-memory store. + closed = true; + try { rawDb.close(); } catch (closeError) { recordCleanupError(error, 'closeError', closeError); } + rawDb = null; + recordCleanupError(error, 'rollbackError', rollbackError); + } + dirty = previouslyDirty; + throw error; + } finally { + inTransaction = false; } - const data = rawDb.export(); - const buffer = Buffer.from(data); - writeDatabaseFileAtomic(dbPath, buffer); } const db = { + withSnapshot, exec(sql) { - rawDb.run(sql); - saveToDisk(); + return withSnapshot(() => { + needsSnapshotComparison = true; + rawDb.run(sql); + dirty = true; + }); }, pragma(pragmaStr) { - try { + return withSnapshot(() => { + needsSnapshotComparison = true; rawDb.run(`PRAGMA ${pragmaStr}`); - } catch (_error) { - // Ignore unsupported pragmas (e.g. WAL for in-memory databases). - } + }); }, prepare(sql) { - return { + const statement = { all(...positionalArgs) { - const stmt = rawDb.prepare(sql); - if (positionalArgs.length === 1 && typeof positionalArgs[0] !== 'object') { - stmt.bind([positionalArgs[0]]); - } else if (positionalArgs.length > 1) { - stmt.bind(positionalArgs); - } - - const rows = []; - while (stmt.step()) { - rows.push(stmt.getAsObject()); - } - stmt.free(); - return rows; + return query(sql, positionalArgs, false, trustedReads.has(statement)); }, get(...positionalArgs) { - const stmt = rawDb.prepare(sql); - if (positionalArgs.length === 1 && typeof positionalArgs[0] !== 'object') { - stmt.bind([positionalArgs[0]]); - } else if (positionalArgs.length > 1) { - stmt.bind(positionalArgs); - } - - let row = null; - if (stmt.step()) { - row = stmt.getAsObject(); - } - stmt.free(); - return row; + return query(sql, positionalArgs, true, trustedReads.has(statement)); }, run(namedParams) { - const stmt = rawDb.prepare(sql); - if (namedParams && typeof namedParams === 'object' && !Array.isArray(namedParams)) { - const sqlJsParams = {}; - for (const [key, value] of Object.entries(namedParams)) { - sqlJsParams[`@${key}`] = value === undefined ? null : value; - } - stmt.bind(sqlJsParams); - } - stmt.step(); - stmt.free(); - saveToDisk(); + return runStatement(sql, namedParams); }, }; + return statement; }, transaction(fn) { - return (...args) => { - rawDb.run('BEGIN'); - inTransaction = true; - try { - const result = fn(...args); - rawDb.run('COMMIT'); - inTransaction = false; - saveToDisk(); - return result; - } catch (error) { - try { - rawDb.run('ROLLBACK'); - } catch (_rollbackError) { - // Transaction may already be rolled back. - } - inTransaction = false; - throw error; - } - }; + return (...args) => withSnapshot(() => transact(fn, args)); }, close() { - saveToDisk(); - rawDb.close(); + if (inSnapshot) throw new Error('Cannot close a state store during an operation'); + if (closed) return; + closed = true; + readCacheIdentity = null; + if (rawDb) rawDb.close(); }, }; return db; } -async function openDatabase(SQL, dbPath) { +function openDatabase(SQL, dbPath, trustedReads) { if (dbPath !== ':memory:') { ensurePrivateDirectory(path.dirname(dbPath)); } - let rawDb; - if (dbPath !== ':memory:' && assertSafeDatabaseFile(dbPath)) { - const fileBuffer = readDatabaseFile(dbPath); - rawDb = new SQL.Database(fileBuffer); - } else { - rawDb = new SQL.Database(); - } - - const db = wrapSqlJsDatabase(rawDb, dbPath); - db.pragma('foreign_keys = ON'); - try { - db.pragma('journal_mode = WAL'); - } catch (_error) { - // Some SQLite environments reject WAL for in-memory or readonly contexts. - } - return db; + return wrapSqlJsDatabase(SQL, dbPath, trustedReads); } async function createStateStore(options = {}) { const dbPath = resolveStateStorePath(options); const SQL = await initSqlJs(); - const db = await openDatabase(SQL, dbPath); - const appliedMigrations = applyMigrations(db); - const queryApi = createQueryApi(db); + const trustedReads = new WeakSet(); + const db = openDatabase(SQL, dbPath, trustedReads); + let appliedMigrations; + try { + appliedMigrations = db.withSnapshot(() => applyMigrations(db)); + } catch (error) { + try { db.close(); } catch (closeError) { recordCleanupError(error, 'closeError', closeError); } + throw error; + } + const queryApi = createQueryApi(db, { + prepareRead(sql) { + const statement = db.prepare(sql); + trustedReads.add(statement); + return statement; + }, + }); + const synchronizedQueries = Object.fromEntries(Object.entries(queryApi) + .map(([name, query]) => [name, (...args) => db.withSnapshot(() => query(...args))])); return { dbPath, @@ -344,11 +452,11 @@ async function createStateStore(options = {}) { db.close(); }, getAppliedMigrations() { - return getAppliedMigrations(db); + return db.withSnapshot(() => getAppliedMigrations(db)); }, validateEntity, assertValidEntity, - ...queryApi, + ...synchronizedQueries, _database: db, _migrations: appliedMigrations, }; diff --git a/scripts/lib/state-store/migrations.js b/scripts/lib/state-store/migrations.js index 7716c9925..c88021bc2 100644 --- a/scripts/lib/state-store/migrations.js +++ b/scripts/lib/state-store/migrations.js @@ -147,6 +147,9 @@ const MIGRATIONS = [ ]; function ensureMigrationTable(db) { + if (db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'schema_migrations'").get()) { + return; + } db.exec(` CREATE TABLE IF NOT EXISTS schema_migrations ( version INTEGER PRIMARY KEY, diff --git a/scripts/lib/state-store/queries.js b/scripts/lib/state-store/queries.js index 0225f4761..7074d5a96 100644 --- a/scripts/lib/state-store/queries.js +++ b/scripts/lib/state-store/queries.js @@ -396,35 +396,36 @@ function normalizeWorkItemInput(workItem) { }; } -function createQueryApi(db) { - const listRecentSessionsStatement = db.prepare(` +function createQueryApi(db, { prepareRead = sql => db.prepare(sql) } = {}) { + // Only these fixed SELECTs qualify for snapshot reuse; generic SQL stays conservative. + const listRecentSessionsStatement = prepareRead(` SELECT * FROM sessions ORDER BY COALESCE(started_at, ended_at, '') DESC, id DESC LIMIT ? `); - const countSessionsStatement = db.prepare(` + const countSessionsStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM sessions `); - const getSessionStatement = db.prepare(` + const getSessionStatement = prepareRead(` SELECT * FROM sessions WHERE id = ? `); - const getSessionSkillRunsStatement = db.prepare(` + const getSessionSkillRunsStatement = prepareRead(` SELECT * FROM skill_runs WHERE session_id = ? ORDER BY created_at DESC, id DESC `); - const getSessionDecisionsStatement = db.prepare(` + const getSessionDecisionsStatement = prepareRead(` SELECT * FROM decisions WHERE session_id = ? ORDER BY created_at DESC, id DESC `); - const listActiveSessionsStatement = db.prepare(` + const listActiveSessionsStatement = prepareRead(` SELECT * FROM sessions WHERE ended_at IS NULL @@ -432,61 +433,61 @@ function createQueryApi(db) { ORDER BY COALESCE(started_at, ended_at, '') DESC, id DESC LIMIT ? `); - const countActiveSessionsStatement = db.prepare(` + const countActiveSessionsStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM sessions WHERE ended_at IS NULL AND state IN ('active', 'running', 'idle') `); - const listRecentSkillRunsStatement = db.prepare(` + const listRecentSkillRunsStatement = prepareRead(` SELECT * FROM skill_runs ORDER BY created_at DESC, id DESC LIMIT ? `); - const listInstallStateStatement = db.prepare(` + const listInstallStateStatement = prepareRead(` SELECT * FROM install_state ORDER BY installed_at DESC, target_id ASC `); - const getInstallStateStatement = db.prepare(` + const getInstallStateStatement = prepareRead(` SELECT target_id FROM install_state WHERE target_id = ? AND target_root = ? `); - const countPendingGovernanceStatement = db.prepare(` + const countPendingGovernanceStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM governance_events WHERE resolved_at IS NULL `); - const listPendingGovernanceStatement = db.prepare(` + const listPendingGovernanceStatement = prepareRead(` SELECT * FROM governance_events WHERE resolved_at IS NULL ORDER BY created_at DESC, id DESC LIMIT ? `); - const listWorkItemsStatement = db.prepare(` + const listWorkItemsStatement = prepareRead(` SELECT * FROM work_items ORDER BY updated_at DESC, id DESC LIMIT ? `); - const countWorkItemsStatement = db.prepare(` + const countWorkItemsStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM work_items `); - const listAllWorkItemsStatement = db.prepare(` + const listAllWorkItemsStatement = prepareRead(` SELECT * FROM work_items ORDER BY updated_at DESC, id DESC `); - const getWorkItemStatement = db.prepare(` + const getWorkItemStatement = prepareRead(` SELECT * FROM work_items WHERE id = ? `); - const getSkillVersionStatement = db.prepare(` + const getSkillVersionStatement = prepareRead(` SELECT * FROM skill_versions WHERE skill_id = ? AND version = ? diff --git a/tests/lib/control-pane-contention.test.js b/tests/lib/control-pane-contention.test.js new file mode 100644 index 000000000..fea89463a --- /dev/null +++ b/tests/lib/control-pane-contention.test.js @@ -0,0 +1,309 @@ +'use strict'; + +const assert = require('assert'); +const fs = require('fs'); +const http = require('http'); +const os = require('os'); +const path = require('path'); +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 => { + 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 { finish(false, null, { status: res.statusCode, headers: res.headers, body: JSON.parse(text) }); } + catch (error) { finish(true, error); } + }); + }); + // 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 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(); + client.observe(message => { + if (message && message.type === 'workers') { + metrics = message; + for (const notify of observers) notify(); + } + }); + let ready; + 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 client.waitFor(message => message && message.type === 'mutation-received', 5000); }, + close: () => client.stop(), + }; +} + +function waitForObservation(server, predicate, timeoutMs = 3000) { + return new Promise((resolve, reject) => { + let unsubscribe; + const timer = setTimeout(() => { + unsubscribe(); + reject(new Error(`Timed out waiting for worker lifecycle: ${JSON.stringify(server.metrics)}`)); + }, timeoutMs); + function check() { + if (!predicate()) return; + clearTimeout(timer); + unsubscribe(); + resolve(); + } + unsubscribe = server.observe(check); + check(); + }); +} + +async function holdFixtureLock(dbPath) { + const lockPath = `${dbPath}.ecc-state.lock`; + const fd = fs.openSync(lockPath, 'wx', 0o600); + 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; }); + 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 => { + results.push({ index, ...result }); + if (results.filter(item => item.status === 503).length >= 6) resolveRejected(); + return result; + }, error => ({ transportError: error }))); + } + await waitForObservation(server, () => server.metrics.received >= 8); + await waitForObservation(server, () => server.metrics.started >= 2); + // Response completion is also observed; no sleep guesses how long Worker + // construction or request body parsing needs on the current machine. + let timer; + try { + await Promise.race([ + sixRejected, + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`Excess requests did not promptly return 503; peak workers=${server.metrics.peak}`)), 1500); }) + ]); + } finally { + clearTimeout(timer); + } + assert.strictEqual(server.metrics.started, 2, 'Overflow must not start additional workers'); + assert.strictEqual(server.metrics.peak, 2); + assert.strictEqual(results.length, 6, 'Only the six rejected requests complete while lock is held'); + for (const result of results) { + assert.strictEqual(result.status, 503); + assert.strictEqual(result.body.code, 'STATE_STORE_BUSY'); + assert.ok(Number(result.headers['retry-after']) > 0); + } + assert.strictEqual((await request(`${server.url}/api/health`)).status, 200); + const snapshot = await request(`${server.url}/api/snapshot`, 'GET', undefined, 3000); + assert.strictEqual(snapshot.status, 200); + assert.deepStrictEqual(fs.readFileSync(dbPath), bytesBefore); + const rejected = results.map(result => result.index); + 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 }); + await withCleanup(() => { + for (const index of rejected) assert.strictEqual(store.getWorkItemById(`burst-${index}`).owner, null); + }, [() => 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'); + }, [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; + return withCleanup(async () => { + const store = await createStateStore({ dbPath }); + 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' }); + } + }, [() => store.close()]); + server = await startServer(dbPath, readOnly); + return await fn(server, dbPath); + }, [() => { if (server) return server.close(); }, + () => fs.rmSync(dir, { recursive: true, force: true })]); +} + +async function disconnectScenario(server, dbPath) { + 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 })); + await withCleanup(async () => { + await waitForObservation(server, () => server.metrics.active === 2); + disconnectedRequest.destroy(new Error('Intentional client disconnect')); + await waitForObservation(server, () => server.metrics.disconnected === 1); + const excess = await request(`${server.url}/api/work-items/burst-2/claim`, 'POST', { owner: 'excess-client' }); + 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); + }, [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'); +} + +async function contentionScenario() { + return withServer(false, async (server, dbPath) => { + const bytesBefore = fs.readFileSync(dbPath); + const releaseLock = await holdFixtureLock(dbPath); + let outcome; + let mutation; + await withCleanup(async () => { + const received = server.mutationReceived(); + let settled = false; + 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; + let healthChecks = 0; + let snapshotChecks = 0; + // Probe continuously after actual HTTP receipt, including asynchronous + // database initialization. No timing guess about when lock wait begins. + while (!settled) { + try { + const health = await request(`${server.url}/api/health`); + assert.strictEqual(health.status, 200); + healthChecks += 1; + const snapshot = await request(`${server.url}/api/snapshot`, 'GET', undefined, 3000); + assert.strictEqual(snapshot.status, 200); + assert.ok(snapshot.body.workItems.items.some(item => item.id === 'task')); + snapshotChecks += 1; + } catch (error) { + healthError = error.message; + break; + } + } + const result = await mutation; + if (result.transportError) throw result.transportError; + outcome = { result, healthError, healthChecks, snapshotChecks }; + }, [releaseLock, () => mutation]); + assert.deepStrictEqual(fs.readFileSync(dbPath), bytesBefore, 'Timed-out mutation must not change database bytes'); + const store = await createStateStore({ dbPath }); + await withCleanup(() => { + const item = store.getWorkItemById('task'); + assert.strictEqual(item.status, 'open'); + assert.strictEqual(item.owner, null); + }, [() => store.close()]); + return outcome; + }); +} + +async function run() { + let passed = 0; + let failed = 0; + async function test(name, fn) { + try { + await fn(); + console.log(` PASS ${name}`); + passed += 1; + } catch (error) { + console.log(` FAIL ${name}\n ${error && error.message ? error.message : String(error)}`); + failed += 1; + } + } + console.log('\n=== Testing control-pane mutation contention ===\n'); + const contention = await contentionScenario(); + await test('health and snapshots remain responsive while a board mutation waits for another database writer', () => { + assert.strictEqual(contention.healthError, null, contention.healthError || undefined); + assert.ok(contention.healthChecks > 0); + assert.ok(contention.snapshotChecks > 0); + }); + await test('lock timeout is retryable HTTP 503 with Retry-After', () => { + assert.strictEqual(contention.result.status, 503); + assert.ok(Number(contention.result.headers['retry-after']) > 0, 'Retry-After must give a positive delay'); + assert.strictEqual(contention.result.body.ok, false); + assert.strictEqual(contention.result.body.code, 'STATE_STORE_BUSY'); + }); + await test('claim and move still save their changes, while invalid edits return 400', () => withServer(false, async (server, dbPath) => { + const claim = await request(`${server.url}/api/work-items/task/claim`, 'POST', { owner: 'alice', as: 'human' }, 5000); + assert.strictEqual(claim.status, 200); + assert.strictEqual(claim.body.item.owner, 'alice'); + assert.strictEqual(claim.body.item.status, 'running'); + const move = await request(`${server.url}/api/work-items/task/move`, 'POST', { lane: 'blocked' }, 5000); + assert.strictEqual(move.status, 200); + assert.strictEqual(move.body.item.status, 'blocked'); + const invalid = await request(`${server.url}/api/work-items/task/move`, 'POST', { lane: 'imaginary' }, 5000); + assert.strictEqual(invalid.status, 400); + assert.strictEqual(invalid.body.ok, false); + const store = await createStateStore({ dbPath }); + await withCleanup(() => { + const item = store.getWorkItemById('task'); + assert.strictEqual(item.owner, 'alice'); + assert.strictEqual(item.status, 'blocked'); + assert.strictEqual(item.metadata.assigneeKind, 'human'); + }, [() => store.close()]); + })); + await test('read-only mode rejects board edits before database work', () => withServer(true, async server => { + for (const action of ['claim', 'move']) { + const result = await request(`${server.url}/api/work-items/task/${action}`, 'POST', { owner: 'alice', lane: 'done' }); + assert.strictEqual(result.status, 403); + } + })); + await test('an eight-request burst caps live workers at two and rejects excess work without queuing', () => withServer(false, burstScenario)); + await test('a disconnected HTTP client retains its worker slot until actual exit', () => withServer(false, disconnectScenario)); + 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 new file mode 100644 index 000000000..9f70ba0d2 --- /dev/null +++ b/tests/lib/helpers/control-pane-contention-server.js @@ -0,0 +1,74 @@ +'use strict'; + +const workerThreads = require('worker_threads'); +const OriginalWorker = workerThreads.Worker; +let metrics = { active: 0, peak: 0, started: 0, exited: 0, received: 0, disconnected: 0 }; +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); + metrics = { + ...metrics, + active: metrics.active + 1, + started: metrics.started + 1, + peak: Math.max(metrics.peak, metrics.active + 1) + }; + report(); + this.once('exit', () => { + metrics = { ...metrics, active: metrics.active - 1, exited: metrics.exited + 1 }; + report(); + }); + } +}; +const { createControlPaneServer } = require('../../../scripts/lib/control-pane/server'); + +async function main() { + const app = createControlPaneServer({ + host: '127.0.0.1', port: 0, stateDbPath: process.argv[2], + dbPath: `${process.argv[2]}.ecc2`, allowActions: process.argv[3] !== 'read-only' + }); + app.server.prependListener('request', (req, res) => { + if (req.method === 'POST') { + metrics = { ...metrics, received: metrics.received + 1 }; + report(); + send({ type: 'mutation-received' }); + res.once('close', () => { + if (!res.writableFinished) { + metrics = { ...metrics, disconnected: metrics.disconnected + 1 }; + report(); + } + }); + } + }); + 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; + if (process.connected) process.disconnect(); +}); diff --git a/tests/lib/helpers/state-store-worker.js b/tests/lib/helpers/state-store-worker.js new file mode 100644 index 000000000..184921a9e --- /dev/null +++ b/tests/lib/helpers/state-store-worker.js @@ -0,0 +1,241 @@ +'use strict'; + +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { fork } = require('child_process'); + +// 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; } } + } + 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); + // Absolute fixture paths resolve dependencies from the checkout, even with + // an isolated cwd. No package-private metadata or NODE_PATH is needed. + 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), + 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 ipcClosed = false; + const closedStreams = new Set(); + 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); + }; + const markClosed = () => { + closed = true; + for (const done of [...closeWaiters]) done(true); + }; + // Some Node versions omit the aggregate close event after a parent-initiated + // IPC disconnect. Require public confirmation of every owned resource instead. + const checkClosed = () => { + if (reaped && ipcClosed && closedStreams.size === 2) markClosed(); + }; + child.once('disconnect', () => { ipcClosed = true; checkClosed(); }); + 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); + checkClosed(); + }); + child.once('close', markClosed); + 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.once('close', () => { closedStreams.add(stream); checkClosed(); }); + 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 new file mode 100644 index 000000000..79d137b64 --- /dev/null +++ b/tests/lib/state-store-concurrency.test.js @@ -0,0 +1,177 @@ +'use strict'; + +const assert = require('assert'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { EventEmitter } = require('events'); +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'); + +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; + return { + async request(action, options = {}) { + const id = ++nextId; + 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); + }, + 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'); + return withCleanup(() => fn(dbPath), [() => fs.rmSync(dir, { recursive: true, force: true })]); +} + +async function readItems(dbPath) { + const store = await createStateStore({ dbPath }); + return withCleanup(() => store.listWorkItems({ limit: 1000 }).items, [() => store.close()]); +} + +async function withWorkers(count, fn) { + return withOwnedChildren(count, startWorker, fn); +} + +async function run() { + let passed = 0; + let failed = 0; + async function test(name, fn) { + try { + await withDatabase(fn); + console.log(` PASS ${name}`); + passed += 1; + } catch (error) { + console.log(` FAIL ${name}\n ${error && error.message ? error.message : String(error)}`); + failed += 1; + } + } + + console.log('\n=== Testing state-store concurrency ===\n'); + + await test('owned fixtures wait for exit, both pipes, and IPC when aggregate close is absent', async () => { + const events = ['exit', 'stdout', 'stderr', 'disconnect']; + for (const last of events) { + const child = new EventEmitter(); + child.stdout = new EventEmitter(); + child.stderr = new EventEmitter(); + child.stdout.destroy = () => child.stdout.emit('close'); + child.stderr.destroy = () => child.stderr.emit('close'); + child.connected = true; + child.pid = 12345; + child.disconnect = () => { child.connected = false; }; + const signals = []; + const client = startOwnedChild(WORKER, [], { + fork: () => child, platform: 'linux', kill: (...args) => signals.push(args) + }); + child.emit('spawn'); + const stopping = client.stop(); + const emit = event => { + if (event === 'exit') child.emit('exit', 0, null); + else if (event === 'disconnect') child.emit('disconnect'); + else child[event].emit('close'); + }; + for (const event of events.filter(event => event !== last)) emit(event); + assert.strictEqual(client.diagnostics.closed, false, `Must wait for ${last}`); + emit(last); + await stopping; + assert.strictEqual(client.diagnostics.closed, true); + assert.deepStrictEqual(signals, [], 'Fully closed fixtures need no forced termination'); + } + }); + + await test('closing an old reader preserves a task saved by the real CLI', async dbPath => { + const reader = await createStateStore({ dbPath }); + await withCleanup(async () => { + assert.strictEqual(reader.listWorkItems().items.length, 0); + 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); + assert.ok(reader.getWorkItemById('cli-task'), 'A warm reader must observe the other process\'s commit'); + assert.strictEqual(reader.listWorkItems().totalCount, 1); + }, [() => 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 }); + 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']); + }, [() => first.close(), () => { if (second) second.close(); }]); + assert.deepStrictEqual((await readItems(dbPath)).map(item => item.id).sort(), ['a', 'b', 'c']); + }); + + await test('independent processes retain every task after all handles open before writing', async dbPath => { + const initial = await createStateStore({ dbPath }); + initial.close(); + await withWorkers(3, async workers => { + // IPC acknowledgements form a barrier: every handle exists before any write. + await Promise.all(workers.map(worker => worker.request('open', { dbPath }))); + await Promise.all(workers.map((worker, index) => worker.request('write', { worker: `worker-${index}`, count: 4 }))); + await Promise.all(workers.map(worker => worker.request('close'))); + }); + const actual = (await readItems(dbPath)).map(item => item.id).sort(); + const expected = Array.from({ length: 3 }, (_, worker) => + Array.from({ length: 4 }, (_, item) => `worker-${worker}-${item}`)).flat().sort(); + assert.deepStrictEqual(actual, expected, 'Every acknowledged task must survive all worker exits'); + }); + + await test('read-modify-write transactions from concurrent processes retain every increment', async dbPath => { + const initial = await createStateStore({ dbPath }); + 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 }))); + await Promise.all(workers.map(worker => worker.request('close'))); + }); + const counter = (await readItems(dbPath)).find(item => item.id === 'counter'); + assert.strictEqual(counter.metadata.value, 12, 'No committed increment may be lost'); + }); + + await test('concurrent first opens complete migrations and accept all writers', async dbPath => { + assert.strictEqual(fs.existsSync(dbPath), false); + await withWorkers(3, async workers => { + await Promise.all(workers.map(worker => worker.request('open', { dbPath }))); + await Promise.all(workers.map((worker, index) => worker.request('write', { worker: `fresh-${index}` }))); + await Promise.all(workers.map(worker => worker.request('close'))); + }); + const store = await createStateStore({ dbPath }); + 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']); + }, [() => 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 new file mode 100644 index 000000000..be9f88a7a --- /dev/null +++ b/tests/lib/state-store-persistence.test.js @@ -0,0 +1,779 @@ +'use strict'; + +const assert = require('assert'); +// 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 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, recordCleanupError, getCleanupErrors } = 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', performance = require('perf_hooks').performance) { + 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' }, perf_hooks: { performance } }, undefined, + { platform, pid: 7 }); + return { ...state, calls, getCleanupErrors: lock.getCleanupErrors, + run: (callback, timeoutMs = 0) => lock.withStateStoreLock('synthetic.db', callback, { timeoutMs }) }; +} + +async function runSynthetic(test) { + await test('Windows pending-delete EPERM retries until exclusive creation succeeds', () => { + const denied = Object.assign(new Error('pending deletion'), { code: 'EPERM' }); + let attempts = 0; + let callbacks = 0; + const fixture = lockFixture({ openSync(state, open, file, flags, mode) { + assert.strictEqual(file, 'synthetic.db.ecc-state.lock'); + assert.strictEqual(flags, 'wx'); + assert.strictEqual(mode, 0o600); + attempts += 1; + if (attempts === 1) { state.calls.push('open'); throw denied; } + return open(); + } }, 'win32'); + assert.strictEqual(fixture.run(() => { callbacks += 1; return 'done'; }, 1000), 'done'); + assert.strictEqual(attempts, 2); + assert.strictEqual(callbacks, 1); + assert.deepStrictEqual(fixture.calls, ['open', 'open', 'fstat', 'write', 'lstat', 'unlink', 'close']); + }); + for (const codes of [['EPERM', 'EPERM'], ['EEXIST', 'EPERM'], ['EPERM', 'EEXIST']]) { + await test(`Windows lock timeout preserves the final error classification (${codes.join(' -> ')})`, () => { + const errors = codes.map(code => Object.assign(new Error(`open failed: ${code}`), { code })); + let attempts = 0; + const clock = { now: () => attempts * 10 }; + const fixture = lockFixture({ openSync(state) { + state.calls.push('open'); + throw errors[attempts++]; + } }, 'win32', clock); + const result = thrownBy(() => fixture.run(() => assert.fail('callback must not run'), 20)); + assert.ok(result.failed); + if (codes[1] === 'EPERM') { + assert.strictEqual(result.error, errors[1]); + assert.doesNotMatch(result.error.message, /remove|leftover lock/i); + } else { + assert.strictEqual(result.error.code, 'STATE_STORE_BUSY'); + assert.match(result.error.message, /synthetic\.db\.ecc-state\.lock/); + } + assert.strictEqual(attempts, 2); + assert.deepStrictEqual(fixture.calls, ['open', 'open']); + }); + } + for (const [platform, code] of [['linux', 'EPERM'], ['darwin', 'EPERM'], ['win32', 'EACCES'], ['win32', 'EIO']]) { + await test(`unrelated lock-open errors propagate immediately (${platform}, ${code})`, () => { + const denied = Object.assign(new Error('open denied'), { code }); + const fixture = lockFixture({ openSync(state) { state.calls.push('open'); throw denied; } }, platform); + const result = thrownBy(() => fixture.run(() => assert.fail('callback must not run'), 1000)); + assert.ok(result.failed); + assert.strictEqual(result.error, denied); + assert.deepStrictEqual(fixture.calls, ['open']); + }); + } + 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', 'darwin', 'win32']) { + for (const replacement of [ + { ino: 3n }, { dev: 9n }, { isFile: () => false }, { isSymbolicLink: () => true }, + ...(platform === 'win32' ? [{ dev: 0n, isFile: () => false }, { dev: 0n, isSymbolicLink: () => true }] : []), + ]) { + await test(`observed lock replacement is preserved (${platform}, ${Object.keys(replacement).join('+')})`, () => { + 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); + }); + } + } + const highInode = 9007199254740992n; + const identityCases = [ + { name: 'matching high inode and devices', platform: 'win32', ownedDev: 9n, pathDev: 9n, allowed: true }, + { name: 'missing path device', platform: 'win32', ownedDev: 9n, pathDev: 0n, allowed: true }, + { name: 'missing descriptor device', platform: 'win32', ownedDev: 0n, pathDev: 9n, allowed: true }, + { name: 'both devices unavailable', platform: 'win32', ownedDev: 0n, pathDev: 0n, allowed: true }, + ...['linux', 'darwin'].flatMap(platform => [ + { name: 'missing path device', platform, ownedDev: 9n, pathDev: 0n, allowed: false }, + { name: 'missing descriptor device', platform, ownedDev: 0n, pathDev: 9n, allowed: false }, + ]), + ...[[9n, 9n], [9n, 0n], [0n, 9n], [0n, 0n]].map(([ownedDev, pathDev]) => ({ + name: `distinct high inodes with devices ${ownedDev}/${pathDev}`, + platform: 'win32', ownedDev, pathDev, pathInode: highInode + 1n, allowed: false, + })), + ]; + for (const identityCase of identityCases) { + await test(`lock identity handles ${identityCase.name} (${identityCase.platform})`, () => { + const fixture = lockFixture({ + fstatSync(state, stat, descriptor, options) { + assert.strictEqual(options.bigint, true); + return { ...stat(), dev: identityCase.ownedDev, ino: highInode }; + }, + lstatSync(state, stat, lockPath, options) { + assert.strictEqual(options.bigint, true); + return { ...stat(), dev: identityCase.pathDev, ino: identityCase.pathInode ?? highInode }; + }, + }, identityCase.platform); + const result = thrownBy(() => fixture.run(() => 'done')); + if (identityCase.allowed) { + assert.strictEqual(result.failed, false, result.error && result.error.message); + assert.strictEqual(result.value, 'done'); + assert.deepStrictEqual(fixture.calls, ['open', 'fstat', 'write', 'lstat', 'unlink', 'close']); + } else { + assert.strictEqual(result.failed, true); + assert.strictEqual(result.error.code, 'STATE_STORE_LOCK_LOST'); + assert.deepStrictEqual(fixture.calls, ['open', 'fstat', 'write', 'lstat', 'close']); + } + }); + } + 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(fixture.getCleanupErrors(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 revoked = Proxy.revocable({}, {}); + revoked.revoke(); + 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'); } }), + revoked.proxy]; + 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); + const diagnostics = fixture.getCleanupErrors(primary); + if (primary !== null && typeof primary === 'object') assert.strictEqual(diagnostics.releaseError, cleanup); + else assert.deepStrictEqual(diagnostics, {}); + assert.ok(Object.isFrozen(diagnostics)); + }); + } + 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 cleanupLock = loadPrivate(LOCK); + const api = loadPrivate(STORE, { + 'sql.js': async () => fakeSQL, + './file-lock': cleanupLock, + './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); + const diagnostics = cleanupLock.getCleanupErrors(primary); + if (primary !== null && typeof primary === 'object') { + assert.strictEqual(diagnostics.closeError, secondary); + if (stage === 'rollback') assert.strictEqual(diagnostics.rollbackError, secondary); + } else assert.deepStrictEqual(diagnostics, {}); + assert.ok(Object.isFrozen(diagnostics)); + }); + } + } + await test('cleanup diagnostics preserve caller descriptors without invoking accessors', () => { + let accesses = 0; + const primary = Object.defineProperty(new Error('operation failed'), 'releaseError', { + configurable: true, + enumerable: true, + get() { accesses++; throw new Error('getter must not run'); }, + set() { accesses++; throw new Error('setter must not run'); }, + }); + const before = Object.getOwnPropertyDescriptors(primary); + const cleanup = new Error('unlink failed'); + const fixture = lockFixture({ unlinkSync() { throw cleanup; } }); + const result = thrownBy(() => fixture.run(() => { throw primary; })); + assert.ok(result.failed); + assert.strictEqual(result.error, primary); + assert.deepStrictEqual(Object.getOwnPropertyDescriptors(primary), before); + assert.strictEqual(accesses, 0); + assert.strictEqual(fixture.getCleanupErrors(primary).releaseError, cleanup); + }); + await test('successive cleanup diagnostics are immutable snapshots isolated by primary', () => { + const primary = Object.freeze(new Error('primary')); + const other = () => {}; + const closeError = new Error('close failed'); + const rollbackError = new Error('rollback failed'); + const untouched = getCleanupErrors(primary); + assert.deepStrictEqual(untouched, {}); + recordCleanupError(primary, 'closeError', closeError); + const first = getCleanupErrors(primary); + recordCleanupError(primary, 'rollbackError', rollbackError); + const second = getCleanupErrors(primary); + assert.notStrictEqual(first, second); + assert.deepStrictEqual(untouched, {}); + assert.deepStrictEqual(first, { closeError }); + assert.deepStrictEqual(second, { closeError, rollbackError }); + assert.ok(!Object.isFrozen(closeError)); + assert.ok(!Object.isFrozen(rollbackError)); + assert.ok([untouched, first, second].every(Object.isFrozen)); + assert.throws(() => { first.closeError = rollbackError; }, TypeError); + recordCleanupError(primary, 'closeError', rollbackError); + const third = getCleanupErrors(primary); + assert.notStrictEqual(third, second); + assert.deepStrictEqual(second, { closeError, rollbackError }); + assert.deepStrictEqual(third, { closeError: rollbackError, rollbackError }); + assert.ok(Object.isFrozen(third)); + assert.deepStrictEqual(getCleanupErrors(other), {}); + recordCleanupError(other, 'releaseError', rollbackError); + assert.deepStrictEqual(getCleanupErrors(other), { releaseError: rollbackError }); + assert.strictEqual(getCleanupErrors(primary), third); + assert.deepStrictEqual(Object.keys(primary), []); + }); + 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' }); + +async function run() { + let passed = 0; + let failed = 0; + async function test(name, callback) { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'ecc-state-persistence-')); + try { + await callback(path.join(directory, 'state.db')); + console.log(` PASS ${name}`); + passed += 1; + } catch (error) { + 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')); + initial.close(); + const before = fs.readFileSync(dbPath); + const rename = fs.renameSync; + let writes = 0; + fs.renameSync = (from, to) => { + if (to === dbPath) writes += 1; + return rename(from, to); + }; + try { + const reader = await createStateStore({ dbPath }); + try { + assert.ok(reader.getWorkItemById('saved')); + reader.listWorkItems(); + reader.getStatus(); + assert.strictEqual(reader.getAppliedMigrations().length, 2); + for (const method of ['get', 'all']) { + const read = () => reader._database.prepare('SELECT id FROM work_items')[method](); + assert.deepStrictEqual(read(), method === 'get' ? { id: 'saved' } : [{ id: 'saved' }]); + reader._database.transaction(read)(); + } + } finally { reader.close(); } + } finally { fs.renameSync = rename; } + assert.strictEqual(writes, 0); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + }); + + const returningMutations = [ + { + action: 'INSERT', + sql: "INSERT INTO returning_probe VALUES (3, 'new'), (4, 'new') RETURNING id, value", + returned: [{ id: 3, value: 'new' }, { id: 4, value: 'new' }], + saved: [{ id: 1, value: 'original' }, { id: 2, value: 'original' }, { id: 3, value: 'new' }, { id: 4, value: 'new' }], + }, + { + action: 'UPDATE', + sql: "UPDATE returning_probe SET value = 'updated' RETURNING id, value", + returned: [{ id: 1, value: 'updated' }, { id: 2, value: 'updated' }], + saved: [{ id: 1, value: 'updated' }, { id: 2, value: 'updated' }], + }, + { + action: 'DELETE', + sql: 'DELETE FROM returning_probe RETURNING id, value', + returned: [{ id: 1, value: 'original' }, { id: 2, value: 'original' }], + saved: [], + }, + ]; + for (const method of ['get', 'all']) { + for (const transactional of [false, true]) { + for (const mutation of returningMutations) { + await test(`prepared ${method} persists ${mutation.action} RETURNING (${transactional ? 'query-only transaction' : 'standalone'})`, async dbPath => { + const store = await createStateStore({ dbPath }); + try { + store._database.exec("CREATE TABLE returning_probe (id INTEGER PRIMARY KEY, value TEXT NOT NULL); INSERT INTO returning_probe VALUES (1, 'original'), (2, 'original');"); + const before = fs.readFileSync(dbPath); + const mutate = () => { + const result = store._database.prepare(mutation.sql)[method](); + if (method === 'get') assert.ok(mutation.returned.some(row => row.id === result.id && row.value === result.value)); + else assert.deepStrictEqual(result.sort((a, b) => a.id - b.id), mutation.returned); + if (transactional) assert.deepStrictEqual(fs.readFileSync(dbPath), before, 'RETURNING must not publish before COMMIT'); + }; + if (transactional) store._database.transaction(mutate)(); + else mutate(); + assert.deepStrictEqual(store._database.prepare('SELECT id, value FROM returning_probe ORDER BY id').all(), mutation.saved, + 'the next snapshot must retain the mutation'); + } finally { store.close(); } + const reopened = await createStateStore({ dbPath }); + try { + assert.deepStrictEqual(reopened._database.prepare('SELECT id, value FROM returning_probe ORDER BY id').all(), mutation.saved); + } finally { reopened.close(); } + }); + } + + await test(`prepared ${method} persists no-row PRAGMA writes (${transactional ? 'query-only transaction' : 'standalone'})`, async dbPath => { + const store = await createStateStore({ dbPath }); + try { + const before = fs.readFileSync(dbPath); + const mutate = () => { + for (const pragma of ['user_version = 3252', 'application_id(2468)']) { + const result = store._database.prepare(`PRAGMA ${pragma}`)[method](); + assert.deepStrictEqual(result, method === 'get' ? null : []); + } + if (transactional) assert.deepStrictEqual(fs.readFileSync(dbPath), before, 'prepared PRAGMAs must not publish before COMMIT'); + }; + if (transactional) store._database.transaction(mutate)(); + else mutate(); + assert.strictEqual(store._database.prepare('PRAGMA user_version').get().user_version, 3252); + assert.strictEqual(store._database.prepare('PRAGMA application_id').get().application_id, 2468); + } finally { store.close(); } + const reopened = await createStateStore({ dbPath }); + try { + assert.strictEqual(reopened._database.prepare('PRAGMA user_version').get().user_version, 3252); + assert.strictEqual(reopened._database.prepare('PRAGMA application_id').get().application_id, 2468); + } finally { reopened.close(); } + }); + } + + await test(`prepared ${method} RETURNING and PRAGMA writes roll back without changing disk`, async dbPath => { + const store = await createStateStore({ dbPath }); + try { + store._database.exec("CREATE TABLE returning_probe (id INTEGER PRIMARY KEY, value TEXT NOT NULL); INSERT INTO returning_probe VALUES (1, 'original'), (2, 'original');"); + const before = fs.readFileSync(dbPath); + const failure = new Error('abort prepared writes'); + assert.throws(() => store._database.transaction(() => { + for (const mutation of returningMutations) store._database.prepare(mutation.sql)[method](); + store._database.prepare('PRAGMA user_version = 3252')[method](); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + throw failure; + })(), error => error === failure); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + assert.deepStrictEqual(store._database.prepare('SELECT id, value FROM returning_probe ORDER BY id').all(), + [{ id: 1, value: 'original' }, { id: 2, value: 'original' }]); + assert.strictEqual(store._database.prepare('PRAGMA user_version').get().user_version, 0); + } finally { store.close(); } + }); + } + + await test('persistent PRAGMA changes survive snapshot reload and reopening', async dbPath => { + const store = await createStateStore({ dbPath }); + try { + assert.strictEqual(store._database.pragma('user_version = 3252'), undefined); + assert.strictEqual(store._database.prepare('PRAGMA user_version').get().user_version, 3252); + store._database.pragma('application_id(2468)'); + } finally { store.close(); } + const reopened = await createStateStore({ dbPath }); + try { + assert.strictEqual(reopened._database.prepare('PRAGMA user_version').get().user_version, 3252); + assert.strictEqual(reopened._database.prepare('PRAGMA application_id').get().application_id, 2468); + } finally { reopened.close(); } + }); + + await test('read-only and connection-local PRAGMAs never rewrite database bytes', async dbPath => { + const store = await createStateStore({ dbPath }); + const before = fs.readFileSync(dbPath); + const rename = fs.renameSync; + let writes = 0; + fs.renameSync = (from, to) => { + if (to === dbPath) writes += 1; + return rename(from, to); + }; + try { + store._database.pragma('user_version'); + store._database.pragma('table_info(work_items)'); + store._database.pragma('integrity_check'); + for (const method of ['get', 'all']) { + const read = () => { + for (const pragma of ['user_version', 'table_info(work_items)', 'integrity_check']) { + store._database.prepare(`PRAGMA ${pragma}`)[method](); + } + }; + read(); + store._database.transaction(read)(); + store._database.withSnapshot(() => { + store._database.prepare('PRAGMA cache_size = 512')[method](); + assert.strictEqual(store._database.prepare('PRAGMA cache_size').get().cache_size, 512); + }); + } + store._database.transaction(() => { + store._database.pragma('user_version'); + store._database.pragma('table_info(work_items)'); + })(); + store._database.withSnapshot(() => { + store._database.pragma('cache_size = 256'); + assert.strictEqual(store._database.prepare('PRAGMA cache_size').get().cache_size, 256); + }); + assert.strictEqual(writes, 0); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + } finally { + fs.renameSync = rename; + store.close(); + } + }); + + await test('PRAGMA persistence waits for commit and never commits a rolled-back transaction', async dbPath => { + const store = await createStateStore({ dbPath }); + try { + store._database.transaction(() => { + store.upsertWorkItem(item('committed-pragma')); + store._database.pragma('table_info(work_items)'); + store._database.pragma('user_version = 3252'); + })(); + store._database.transaction(() => store._database.pragma('application_id(2468)'))(); + assert.strictEqual(store._database.prepare('PRAGMA application_id').get().application_id, 2468); + const before = fs.readFileSync(dbPath); + assert.throws(() => store._database.transaction(() => { + store.upsertWorkItem(item('rolled-back-pragma')); + store._database.pragma('user_version = 9999'); + throw new Error('abort PRAGMA transaction'); + })(), /abort PRAGMA transaction/); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + assert.strictEqual(store._database.prepare('PRAGMA user_version').get().user_version, 3252); + assert.deepStrictEqual(store.listWorkItems().items.map(row => row.id), ['committed-pragma']); + } finally { store.close(); } + }); + + await test('failed persistence is discarded, releases the lock and cannot overwrite a later writer', async dbPath => { + const first = await createStateStore({ dbPath }); + const second = await createStateStore({ dbPath }); + const before = fs.readFileSync(dbPath); + const rename = fs.renameSync; + const failure = new Error('simulated disk failure'); + try { + fs.renameSync = (from, to) => { + if (to === dbPath) throw failure; + return rename(from, to); + }; + try { + assert.throws(() => first.upsertWorkItem(item('failed')), error => error === failure); + } finally { fs.renameSync = rename; } + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + assert.strictEqual(fs.existsSync(`${dbPath}.ecc-state.lock`), false); + second.upsertWorkItem(item('other-process')); + first.upsertWorkItem(item('retry')); + assert.deepStrictEqual(first.listWorkItems().items.map(row => row.id).sort(), ['other-process', 'retry']); + } finally { + first.close(); + second.close(); + } + }); + + await test('transaction exceptions roll back, preserve the error and release the writer lock', async dbPath => { + const store = await createStateStore({ dbPath }); + const before = fs.readFileSync(dbPath); + const failure = new Error('cancel this transaction'); + try { + assert.throws(() => store._database.transaction(() => { + store.upsertWorkItem(item('rolled-back')); + throw failure; + })(), error => error === failure); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + assert.strictEqual(fs.existsSync(`${dbPath}.ecc-state.lock`), false); + store.upsertWorkItem(item('next')); + assert.deepStrictEqual(store.listWorkItems().items.map(row => row.id), ['next']); + } finally { store.close(); } + }); + + await test('reloading a snapshot preserves foreign-key enforcement', async dbPath => { + const store = await createStateStore({ dbPath }); + try { + store._database.exec('CREATE TABLE parent (id INTEGER PRIMARY KEY); CREATE TABLE child (parent_id INTEGER REFERENCES parent(id));'); + assert.throws(() => store._database.exec('INSERT INTO child VALUES (42)'), /FOREIGN KEY constraint failed/); + store._database.exec('INSERT INTO parent VALUES (42)'); + store._database.exec('INSERT INTO child VALUES (42)'); + assert.strictEqual(store._database.prepare('SELECT COUNT(*) AS count FROM child').get().count, 1); + } finally { store.close(); } + }); + + await test('in-memory transactions roll back and closed handles cannot reopen silently', async () => { + const store = await createStateStore({ dbPath: ':memory:' }); + store.upsertWorkItem(item('keep')); + assert.throws(() => store._database.transaction(() => { + store.upsertWorkItem(item('discard')); + throw new Error('abort'); + })(), /abort/); + assert.deepStrictEqual(store.listWorkItems().items.map(row => row.id), ['keep']); + store.close(); + store.close(); + assert.throws(() => store.listWorkItems(), /closed/); + }); + + await test('failed rollback keeps the primary error and invalidates an uncertain in-memory handle', async () => { + const store = await createStateStore({ dbPath: ':memory:' }); + const failure = new Error('application error after explicit rollback'); + assert.throws(() => store._database.transaction(() => { + store._database.exec('ROLLBACK'); + throw failure; + })(), error => error === failure && getCleanupErrors(error).rollbackError instanceof Error); + assert.throws(() => store.listWorkItems(), /closed/); + store.close(); + }); + + await test('live and abandoned locks time out without stealing ownership or touching the database', async dbPath => { + const lock = `${dbPath}.ecc-state.lock`; + fs.writeFileSync(dbPath, 'untouched'); + for (const metadata of [{ pid: process.pid }, { pid: 999999999 }, null]) { + const bytes = JSON.stringify(metadata); + fs.writeFileSync(lock, bytes); + fs.utimesSync(lock, new Date(0), new Date(0)); + assert.throws(() => withStateStoreLock(dbPath, () => assert.fail('lock was stolen'), { timeoutMs: 20 }), + error => error.code === 'STATE_STORE_BUSY' && error.message.includes(lock)); + assert.strictEqual(fs.readFileSync(lock, 'utf8'), bytes); + assert.strictEqual(fs.readFileSync(dbPath, 'utf8'), 'untouched'); + fs.unlinkSync(lock); + } + }); + + await test('failed lock metadata writes release the owned file', async dbPath => { + const write = fs.writeFileSync; + const failure = new Error('metadata write failure'); + fs.writeFileSync = () => { throw failure; }; + try { + assert.throws(() => withStateStoreLock(dbPath, () => assert.fail('callback should not run')), error => error === failure); + } finally { fs.writeFileSync = write; } + assert.strictEqual(fs.existsSync(`${dbPath}.ecc-state.lock`), false); + assert.strictEqual(withStateStoreLock(dbPath, () => 'retry'), 'retry'); + }); + + await test('callback failure remains primary even when lock cleanup also fails', async dbPath => { + const unlink = fs.unlinkSync; + const primary = new Error('operation failed'); + const cleanup = new Error('cleanup failed'); + const lock = `${dbPath}.ecc-state.lock`; + fs.unlinkSync = file => { + if (file === lock) throw cleanup; + return unlink(file); + }; + try { + assert.throws(() => withStateStoreLock(dbPath, () => { throw primary; }), + error => error === primary && getCleanupErrors(error).releaseError === cleanup); + } finally { + fs.unlinkSync = unlink; + unlink(lock); + } + let caught = false; + try { withStateStoreLock(dbPath, () => { throw null; }); } + catch (error) { caught = true; assert.strictEqual(error, null); } + assert.ok(caught, 'even falsy thrown values must propagate'); + }); + + 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/state-store-read-cache.test.js b/tests/lib/state-store-read-cache.test.js new file mode 100644 index 000000000..2acd95386 --- /dev/null +++ b/tests/lib/state-store-read-cache.test.js @@ -0,0 +1,505 @@ +'use strict'; + +const assert = require('assert'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const vm = require('vm'); +const { createRequire } = require('module'); +const initSqlJs = require('sql.js'); +const { createStateStore: createWriter } = require('../../scripts/lib/state-store'); + +const STORE = require.resolve('../../scripts/lib/state-store'); +const LOCK = require.resolve('../../scripts/lib/state-store/file-lock'); + +function loadPrivate(filename, overrides, processFacade = process) { + const module = { exports: {} }; + const nativeRequire = createRequire(filename); + const localRequire = id => Object.hasOwn(overrides, id) ? overrides[id] : nativeRequire(id); + vm.compileFunction(fs.readFileSync(filename, 'utf8'), + ['exports', 'require', 'module', '__filename', '__dirname', 'process'], { filename })( + module.exports, localRequire, module, filename, path.dirname(filename), processFacade); + return module.exports; +} + +async function createFixture(dbPath, { platform = process.platform } = {}) { + const processFacade = Object.create(process); + Object.defineProperty(processFacade, 'platform', { value: platform }); + const metrics = { reads: 0, databases: 0, exports: 0, publishes: 0 }; + const descriptors = new Map(); + let publishFailure; + let transformStats = stats => stats; + let lockBlocked = false; + let lockAttempts = 0; + const facade = { + ...fs, + openSync(file, ...args) { + if (file === `${dbPath}.ecc-state.lock`) { + lockAttempts += 1; + if (lockBlocked) throw Object.assign(new Error('held by another writer'), { code: 'EEXIST' }); + } + const descriptor = fs.openSync(file, ...args); + descriptors.set(descriptor, file); + return descriptor; + }, + lstatSync(file, options) { + const stats = fs.lstatSync(file, options); + return file === dbPath ? transformStats(stats, 'path', options) : stats; + }, + fstatSync(descriptor, options) { + const stats = fs.fstatSync(descriptor, options); + return descriptors.get(descriptor) === dbPath ? transformStats(stats, 'descriptor', options) : stats; + }, + readFileSync(file, ...args) { + if ((typeof file === 'number' ? descriptors.get(file) : file) === dbPath) metrics.reads += 1; + return fs.readFileSync(file, ...args); + }, + closeSync(descriptor) { + try { return fs.closeSync(descriptor); } + finally { descriptors.delete(descriptor); } + }, + renameSync(from, to) { + if (to === dbPath) { + if (publishFailure) throw publishFailure; + metrics.publishes += 1; + } + return fs.renameSync(from, to); + }, + }; + const SQL = await initSqlJs(); + class ObservedDatabase extends SQL.Database { + constructor(...args) { super(...args); metrics.databases += 1; } + export() { metrics.exports += 1; return super.export(); } + } + const lock = loadPrivate(LOCK, { fs: facade }, processFacade); + const lockFacade = { + ...lock, + withStateStoreLock(file, callback) { + return lock.withStateStoreLock(file, callback, { timeoutMs: lockBlocked ? 0 : 5000 }); + }, + }; + const { createStateStore } = loadPrivate(STORE, { + fs: facade, + './file-lock': lockFacade, + 'sql.js': async () => ({ ...SQL, Database: ObservedDatabase }), + }, processFacade); + return { + createStateStore, + metrics, + reset() { for (const key of Object.keys(metrics)) metrics[key] = 0; }, + failPublication(error) { publishFailure = error; }, + setStatTransform(transform) { transformStats = transform; }, + blockLock(blocked) { lockBlocked = blocked; }, + get lockAttempts() { return lockAttempts; }, + }; +} + +function item(id, title = id) { + return { id, title, source: 'manual', status: 'open', metadata: { nested: { value: 'saved' } } }; +} + +function readBoard(store) { + return { + item: store.getWorkItemById('task'), + list: store.listWorkItems(), + status: store.getStatus(), + }; +} + +function assertNoSnapshotWork(fixture) { + assert.deepStrictEqual(fixture.metrics, { reads: 0, databases: 0, exports: 0, publishes: 0 }); +} + +async function run() { + let passed = 0; + let failed = 0; + async function test(name, callback, options) { + // Resolve host aliases before a fixture simulates another platform. + const directory = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), 'ecc-state-read-cache-'))); + try { + const dbPath = path.join(directory, 'state.db'); + await callback(dbPath, await createFixture(dbPath, options)); + console.log(` PASS ${name}`); + passed += 1; + } catch (error) { + console.log(` FAIL ${name}\n ${error && error.stack ? error.stack : String(error)}`); + failed += 1; + } finally { fs.rmSync(directory, { recursive: true, force: true }); } + } + + await test('unchanged public reads reuse the database without reading, constructing or exporting', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + fixture.reset(); + for (let index = 0; index < 3; index += 1) { + const result = readBoard(store); + assert.strictEqual(result.item.title, 'task'); + assert.strictEqual(result.list.totalCount, 1); + assert.strictEqual(result.status.workItems.items.length, 1); + } + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + await test('atomic replacement by another handle refreshes once and then reuses the new snapshot', async (dbPath, fixture) => { + const reader = await fixture.createStateStore({ dbPath }); + const writer = await createWriter({ dbPath }); + try { + writer.upsertWorkItem(item('task', 'before')); + readBoard(reader); + writer.upsertWorkItem(item('task', 'after')); + writer.upsertWorkItem(item('second')); + fixture.reset(); + const result = readBoard(reader); + assert.strictEqual(result.item.title, 'after'); + assert.strictEqual(result.list.totalCount, 2); + assert.strictEqual(result.status.workItems.items.length, 2); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + fixture.reset(); + readBoard(reader); + assertNoSnapshotWork(fixture); + } finally { reader.close(); writer.close(); } + }); + + await test('cached reads return fresh objects and respect each call\'s limits', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + store.upsertWorkItem(item('second')); + const first = readBoard(store); + first.item.title = 'caller mutation'; + first.item.metadata.nested.value = 'caller mutation'; + first.list.items.length = 0; + first.status.workItems.items[0].metadata.nested.value = 'caller mutation'; + fixture.reset(); + const next = readBoard(store); + assert.notStrictEqual(next.item, first.item); + assert.notStrictEqual(next.item.metadata, first.item.metadata); + assert.notStrictEqual(next.list, first.list); + assert.notStrictEqual(next.status, first.status); + assert.strictEqual(next.item.title, 'task'); + assert.strictEqual(next.item.metadata.nested.value, 'saved'); + assert.strictEqual(next.list.items.length, 2); + assert.ok(next.status.workItems.items.every(row => row.metadata.nested.value === 'saved')); + assert.strictEqual(store.listWorkItems({ limit: 1 }).items.length, 1); + assert.strictEqual(store.getStatus({ workItemLimit: 1 }).workItems.items.length, 1); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + for (const method of ['get', 'all']) { + await test(`generic ${method} RETURNING and PRAGMA writes persist and invalidate public reads`, async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + for (const transactional of [false, true]) { + readBoard(store); + const title = transactional ? 'transactional' : 'standalone'; + const version = transactional ? 202 : 101; + const mutate = () => { + const result = store._database.prepare(`UPDATE work_items SET title = '${title}' WHERE id = 'task' RETURNING id, title`)[method](); + assert.deepStrictEqual(result, method === 'get' ? { id: 'task', title } : [{ id: 'task', title }]); + assert.deepStrictEqual(store._database.prepare(`PRAGMA user_version = ${version}`)[method](), method === 'get' ? null : []); + }; + if (transactional) store._database.transaction(mutate)(); + else mutate(); + const result = readBoard(store); + assert.strictEqual(result.item.title, title); + assert.strictEqual(result.list.items[0].title, title); + assert.strictEqual(result.status.workItems.items[0].title, title); + const reopened = await createWriter({ dbPath }); + try { + assert.strictEqual(reopened.getWorkItemById('task').title, title); + assert.strictEqual(reopened._database.prepare('PRAGMA user_version').get().user_version, version); + } finally { reopened.close(); } + readBoard(store); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } + } finally { store.close(); } + }); + } + + for (const method of ['get', 'all']) { + await test(`a limit getter can reenter generic ${method} RETURNING and invalidate the outer read cache`, async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task', 'before')); + readBoard(store); + let calls = 0; + const result = store.listWorkItems({ + get limit() { + calls += 1; + const changed = store._database.prepare("UPDATE work_items SET title = 'getter write' WHERE id = 'task' RETURNING id, title")[method](); + assert.deepStrictEqual(changed, method === 'get' + ? { id: 'task', title: 'getter write' } : [{ id: 'task', title: 'getter write' }]); + return 1; + }, + }); + assert.strictEqual(calls, 1); + assert.strictEqual(result.items[0].title, 'getter write'); + fixture.reset(); + assert.strictEqual(readBoard(store).item.title, 'getter write'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + const reopened = await createWriter({ dbPath }); + try { assert.strictEqual(reopened.getWorkItemById('task').title, 'getter write'); } + finally { reopened.close(); } + } finally { store.close(); } + }); + } + + await test('public reads inside a write transaction see its writes without publishing early', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task', 'before')); + readBoard(store); + const before = fs.readFileSync(dbPath); + store._database.transaction(() => { + assert.strictEqual(readBoard(store).item.title, 'before'); + store.upsertWorkItem(item('task', 'committed')); + assert.strictEqual(readBoard(store).item.title, 'committed'); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + })(); + assert.strictEqual(readBoard(store).item.title, 'committed'); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + await test('rollback discards changes observed through public reads inside a transaction', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task', 'before')); + readBoard(store); + const before = fs.readFileSync(dbPath); + const failure = new Error('rollback cached reads'); + assert.throws(() => store._database.transaction(() => { + store.upsertWorkItem(item('task', 'discarded')); + assert.strictEqual(readBoard(store).item.title, 'discarded'); + throw failure; + })(), error => error === failure); + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + assert.strictEqual(readBoard(store).item.title, 'before'); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + await test('failed publication cannot contaminate cached reads or overwrite a later writer', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + const writer = await createWriter({ dbPath }); + try { + store.upsertWorkItem(item('task', 'before')); + readBoard(store); + const before = fs.readFileSync(dbPath); + const failure = new Error('publication failed'); + fixture.failPublication(failure); + try { assert.throws(() => store.upsertWorkItem(item('task', 'discarded')), error => error === failure); } + finally { fixture.failPublication(undefined); } + assert.deepStrictEqual(fs.readFileSync(dbPath), before); + assert.strictEqual(readBoard(store).item.title, 'before'); + writer.upsertWorkItem(item('task', 'other writer')); + assert.strictEqual(readBoard(store).item.title, 'other writer'); + store.upsertWorkItem(item('second')); + assert.strictEqual(readBoard(store).item.title, 'other writer'); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } finally { store.close(); writer.close(); } + }); + + await test('closing a cached file-backed handle does not publish and prevents later reads', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + fixture.reset(); + store.close(); + store.close(); + for (const read of [() => store.getWorkItemById('task'), () => store.listWorkItems(), () => store.getStatus()]) { + assert.throws(read, /closed/); + } + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + await test('in-memory reads retain writes and rollback without snapshot IO', async (_dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath: ':memory:' }); + try { + store.upsertWorkItem(item('task', 'before')); + readBoard(store); + fixture.reset(); + readBoard(store); + store.upsertWorkItem(item('task', 'after')); + assert.strictEqual(readBoard(store).item.title, 'after'); + assert.throws(() => store._database.transaction(() => { + store.upsertWorkItem(item('task', 'discarded')); + throw new Error('rollback memory'); + })(), /rollback memory/); + assert.strictEqual(readBoard(store).item.title, 'after'); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + await test('a warm cache still acquires the writer lock and reports contention', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + fixture.reset(); + const attempts = fixture.lockAttempts; + fixture.blockLock(true); + try { assert.throws(() => store.getWorkItemById('task'), error => error.code === 'STATE_STORE_BUSY'); } + finally { fixture.blockLock(false); } + assert.strictEqual(fixture.lockAttempts, attempts + 1); + assertNoSnapshotWork(fixture); + assert.strictEqual(readBoard(store).item.title, 'task'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + } finally { store.close(); } + }); + + for (const kind of ['symlink', 'directory']) { + await test(`a warm cache rejects a database path replaced by a ${kind}`, async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + // Simulate the path metadata so Windows does not require symlink privileges. + fixture.setStatTransform(stats => Object.assign(Object.create(stats), { + isSymbolicLink: () => kind === 'symlink', + isFile: () => false, + isDirectory: () => kind === 'directory', + })); + fixture.reset(); + assert.throws(() => store.getWorkItemById('task'), kind === 'symlink' ? /symlink/ : /not a regular file/); + assertNoSnapshotWork(fixture); + fixture.setStatTransform(stats => stats); + assert.strictEqual(readBoard(store).item.title, 'task'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + } finally { store.close(); } + }); + } + + await test('a throwing read callback invalidates the previous cached snapshot', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + const failure = new Error('invalid read options'); + fixture.reset(); + assert.throws(() => store.listWorkItems({ get limit() { throw failure; } }), error => error === failure); + assertNoSnapshotWork(fixture); + assert.strictEqual(readBoard(store).item.title, 'task'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + await test('read-only transactions do not export but cannot seed the next read cache', async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + fixture.reset(); + store._database.transaction(() => assert.strictEqual(readBoard(store).item.title, 'task'))(); + assertNoSnapshotWork(fixture); + assert.strictEqual(readBoard(store).item.title, 'task'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + for (const method of ['exec', 'run', 'get', 'all']) { + await test(`a caught partial ${method} failure inside read options publishes changes and invalidates cache`, async (dbPath, fixture) => { + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task', 'before')); + readBoard(store); + const targetId = method === 'exec' ? 'task' : 'partial'; + let sqlFailure; + fixture.reset(); + const result = store.listWorkItems({ + get limit() { + try { + if (method === 'exec') { + store._database.exec("UPDATE work_items SET title = 'partial write' WHERE id = 'task'; INVALID SQL;"); + } else { + // OR FAIL preserves the first row when the duplicate primary key fails. + store._database.prepare(` + INSERT OR FAIL INTO work_items (id, source, title, status, metadata, created_at, updated_at) + VALUES ('partial', 'manual', 'partial write', 'open', '{}', '2026-01-01', '2026-01-01'), + ('partial', 'manual', 'duplicate', 'open', '{}', '2026-01-01', '2026-01-01') + `)[method](); + } + } catch (error) { sqlFailure = error; } + return 20; + }, + }); + assert.ok(sqlFailure, 'SQL must fail after a partial write'); + assert.match(sqlFailure.message, method === 'exec' ? /syntax error/ : /UNIQUE constraint failed/); + assert.strictEqual(result.items.find(row => row.id === targetId).title, 'partial write'); + assert.deepStrictEqual(fixture.metrics, { reads: 0, databases: 0, exports: 1, publishes: 1 }); + fixture.reset(); + assert.strictEqual(readBoard(store).list.items.find(row => row.id === targetId).title, 'partial write'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + const reopened = await createWriter({ dbPath }); + try { assert.strictEqual(reopened.getWorkItemById(targetId).title, 'partial write'); } + finally { reopened.close(); } + } finally { store.close(); } + }); + } + + for (const missingDevice of ['path', 'descriptor']) { + await test(`Windows cached reads accept an unavailable ${missingDevice} device`, async (dbPath, fixture) => { + fixture.setStatTransform((stats, source, options) => !options || !options.bigint ? stats + : Object.assign(Object.create(stats), { dev: source === missingDevice ? 0n : 42n })); + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + fixture.reset(); + assert.strictEqual(readBoard(store).item.title, 'task'); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }, { platform: 'win32' }); + } + + await test('adjacent high bigint inodes invalidate cache even if Number would round them together', async (dbPath, fixture) => { + let inode = 9007199254740992n; + fixture.setStatTransform((stats, _source, options) => !options || !options.bigint ? stats + : Object.assign(Object.create(stats), { dev: 42n, ino: inode })); + const store = await fixture.createStateStore({ dbPath }); + try { + store.upsertWorkItem(item('task')); + readBoard(store); + fixture.reset(); + inode += 1n; + assert.strictEqual(readBoard(store).item.title, 'task'); + assert.deepStrictEqual(fixture.metrics, { reads: 1, databases: 1, exports: 0, publishes: 0 }); + fixture.reset(); + readBoard(store); + assertNoSnapshotWork(fixture); + } finally { store.close(); } + }); + + 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/work-item-runner.test.js b/tests/lib/work-item-runner.test.js new file mode 100644 index 000000000..dfed8bcf2 --- /dev/null +++ b/tests/lib/work-item-runner.test.js @@ -0,0 +1,257 @@ +'use strict'; + +const assert = require('assert'); +const { EventEmitter } = require('events'); +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 workers = []; + const requests = []; + let constructorFailures = 0; + let constructionAttempts = 0; + class FakeWorker extends EventEmitter { + constructor(filename, options) { + super(); + constructionAttempts += 1; + if (constructorFailures > 0) { + constructorFailures -= 1; + throw new Error('worker construction failed'); + } + this.filename = filename; + this.options = options; + this.exited = false; + workers.push(this); + } + + exit(code = 0) { + assert.strictEqual(this.exited, false, 'a worker can only exit once'); + this.exited = true; + this.emit('exit', code); + } + + finish(result = 'done') { + this.emit('message', { ok: true, result }); + this.exit(); + } + } + + 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; }, + failNextConstructor() { constructorFailures += 1; }, + request(dbPath = 'synthetic-state.db', action = 'claim') { + const request = { settled: false }; + request.result = new Promise(resolve => { + const finish = result => { + request.settled = true; + clearTimeout(timer); + resolve(result); + }; + const timer = setTimeout(() => finish({ ok: false, error: new Error('Synthetic worker request never settled') }), 2000); + Promise.resolve().then(() => runWorkItemMutation(dbPath, action, { id: 'synthetic-task' })) + .then(value => finish({ ok: true, value }), error => finish({ ok: false, error })); + }); + requests.push(request); + return request; + } + }; + try { + await callback(context); + } finally { + // Failed assertions still drain every synthetic worker and rejection. + await nextTurn(); + for (const worker of workers) { + if (!worker.exited) worker.finish(); + } + await Promise.all(requests.map(request => request.result)); + } +} + +async function assertBusy(request) { + await nextTurn(); + assert.strictEqual(request.settled, true, 'over-capacity requests must reject without waiting or queuing'); + const result = await request.result; + assert.strictEqual(result.ok, false); + assert.strictEqual(result.error.code, 'STATE_STORE_BUSY'); +} + +async function run() { + let passed = 0; + let failed = 0; + async function test(name, callback) { + try { + await withFakeWorkers(callback); + console.log(` PASS ${name}`); + passed += 1; + } catch (error) { + console.log(` FAIL ${name}\n ${error && error.stack ? error.stack : String(error)}`); + failed += 1; + } + } + + await test('burst requests share a process-wide two-worker limit without queuing', async context => { + const requests = Array.from({ length: 20 }, (_, index) => context.request(`state-${index}.db`, index % 2 ? 'move' : 'claim')); + await nextTurn(); + assert.strictEqual(context.constructionAttempts, 2, 'capacity must be checked before constructing another worker'); + assert.strictEqual(context.workers.length, 2); + await Promise.all(requests.slice(2).map(assertBusy)); + context.workers[0].finish('first'); + context.workers[1].finish('second'); + assert.deepStrictEqual(await Promise.all(requests.slice(0, 2).map(request => request.result)), [ + { ok: true, value: 'first' }, { ok: true, value: 'second' } + ]); + await nextTurn(); + assert.strictEqual(context.constructionAttempts, 2, 'rejected work must not start later'); + const replacement = context.request(); + await nextTurn(); + context.workers[2].finish('replacement'); + assert.deepStrictEqual(await replacement.result, { ok: true, value: 'replacement' }); + }); + + await test('a result message cannot settle the request or release capacity before exit', async context => { + const first = context.request(); + context.request(); + await nextTurn(); + context.workers[0].emit('message', { ok: true, result: 'acknowledged' }); + await nextTurn(); + assert.strictEqual(first.settled, false, 'a worker that has not exited still owns its slot'); + await assertBusy(context.request()); + assert.strictEqual(context.constructionAttempts, 2); + context.workers[0].exit(); + assert.deepStrictEqual(await first.result, { ok: true, value: 'acknowledged' }); + const replacement = context.request(); + await nextTurn(); + assert.strictEqual(context.workers.length, 3); + context.workers[2].finish(); + assert.strictEqual((await replacement.result).ok, true); + }); + + await test('worker errors retain capacity until exit and preserve the original failure', async context => { + const first = context.request(); + context.request(); + await nextTurn(); + const failure = new Error('worker execution failed'); + context.workers[0].emit('error', failure); + await nextTurn(); + assert.strictEqual(first.settled, false); + await assertBusy(context.request()); + assert.strictEqual(context.constructionAttempts, 2); + context.workers[0].exit(1); + const result = await first.result; + assert.strictEqual(result.ok, false); + assert.strictEqual(result.error, failure); + const replacement = context.request(); + await nextTurn(); + context.workers[2].finish(); + assert.strictEqual((await replacement.result).ok, true); + }); + + await test('reported mutation failures reject after exit and then release capacity', async context => { + const first = context.request(); + context.request(); + await nextTurn(); + context.workers[0].emit('message', { ok: false, error: { message: 'database is busy', code: 'STATE_STORE_BUSY' } }); + await nextTurn(); + assert.strictEqual(first.settled, false); + await assertBusy(context.request()); + context.workers[0].exit(); + const result = await first.result; + assert.strictEqual(result.ok, false); + assert.strictEqual(result.error.message, 'database is busy'); + assert.strictEqual(result.error.code, 'STATE_STORE_BUSY'); + const replacement = context.request(); + await nextTurn(); + context.workers[2].finish(); + assert.strictEqual((await replacement.result).ok, true); + }); + + await test('constructor failures release capacity immediately', async context => { + context.failNextConstructor(); + const failed = await context.request().result; + assert.strictEqual(failed.ok, false); + assert.match(failed.error.message, /worker construction failed/); + const first = context.request(); + const second = context.request(); + await nextTurn(); + assert.strictEqual(context.workers.length, 2); + assert.strictEqual(context.constructionAttempts, 3); + await assertBusy(context.request()); + assert.strictEqual(context.constructionAttempts, 3); + context.workers.forEach(worker => worker.finish()); + assert.ok((await first.result).ok && (await second.result).ok); + }); + + await test('exiting without a message rejects and releases capacity', async context => { + const first = context.request(); + context.request(); + await nextTurn(); + context.workers[0].exit(0); + const result = await first.result; + assert.strictEqual(result.ok, false); + assert.match(result.error.message, /exit.*without.*result/i); + const replacement = context.request(); + await nextTurn(); + assert.strictEqual(context.workers.length, 3); + await assertBusy(context.request()); + context.workers[2].finish(); + assert.strictEqual((await replacement.result).ok, true); + }); + + await test('a nonzero exit cannot turn a premature success message into success', async context => { + const request = context.request(); + await nextTurn(); + context.workers[0].emit('message', { ok: true, result: 'premature' }); + context.workers[0].exit(1); + const result = await request.result; + assert.strictEqual(result.ok, false); + 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; }); diff --git a/tests/scripts/npm-publish-surface.test.js b/tests/scripts/npm-publish-surface.test.js index 7b81957e0..b911cb9d2 100644 --- a/tests/scripts/npm-publish-surface.test.js +++ b/tests/scripts/npm-publish-surface.test.js @@ -248,6 +248,9 @@ function main() { "docs/design/context-carriers.md", "docs/design/context-profile-delivery.md", "scripts/control-pane.js", + "scripts/lib/control-pane/work-item-runner.js", + "scripts/lib/control-pane/work-item-worker.js", + "scripts/lib/state-store/file-lock.js", "scripts/feedback.js", "scripts/ito.js", "scripts/memory.js",