merge: preserve serialized store source 3252 with portable cache fixtures

This commit is contained in:
haelyra committed 2026-10-09 16:28:34 -04:00
commit 67c89375c3
16 files changed
+2875 -137

No files matched your search

@@ -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 `<database>.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.
+12 -19
View File
@@ -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;
}
@@ -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 };
@@ -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) })
);
+147
View File
@@ -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 };
+208 -100
View File
@@ -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,
};
+3
View File
@@ -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,
+19 -18
View File
@@ -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 = ?
+309
View File
@@ -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; });
@@ -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();
});
+241
View File
@@ -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();
+177
View File
@@ -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;
});
+779
View File
@@ -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; });
+505
View File
@@ -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; });
+257
View File
@@ -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; });
@@ -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",