mirror of
https://github.com/affaan-m/ECC.git
synced 2026-10-21 07:21:46 +02:00
fix(state-store): preserve concurrent writes across processes
This commit is contained in:
Yann Roberto
@@ -78,3 +78,19 @@ 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 reloads the latest database
|
||||
under that lock and publishes successful writes before releasing it. 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.
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
'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;
|
||||
|
||||
function sameIdentity(left, right) {
|
||||
return left.ino === right.ino
|
||||
&& (left.dev === right.dev || (process.platform === 'win32' && (!left.dev || !right.dev)));
|
||||
}
|
||||
|
||||
function releaseOwnedLock(lockPath, descriptor, identity) {
|
||||
let current;
|
||||
try {
|
||||
current = fs.lstatSync(lockPath, { bigint: true });
|
||||
} finally {
|
||||
fs.closeSync(descriptor);
|
||||
}
|
||||
if (!current.isFile() || current.isSymbolicLink() || !sameIdentity(identity, current)) {
|
||||
throw new Error(`State-store lock changed; refusing to remove it: ${lockPath}`);
|
||||
}
|
||||
fs.unlinkSync(lockPath);
|
||||
}
|
||||
|
||||
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) {
|
||||
if (error.code !== 'EEXIST') throw error;
|
||||
const remaining = deadline - performance.now();
|
||||
if (remaining <= 0) {
|
||||
const busy = new Error(`State store is busy: ${dbPath}. Retry after the other ECC operation finishes. `
|
||||
+ `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) {
|
||||
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;
|
||||
if (primaryError instanceof Error) primaryError.releaseError = releaseError;
|
||||
}
|
||||
if (failed) throw primaryError;
|
||||
return result;
|
||||
}
|
||||
|
||||
module.exports = { withStateStoreLock };
|
||||
@@ -5,6 +5,7 @@ const crypto = require('crypto');
|
||||
const os = require('os');
|
||||
const path = require('path');
|
||||
const initSqlJs = require('sql.js');
|
||||
const { withStateStoreLock } = require('./file-lock');
|
||||
|
||||
const { applyMigrations, getAppliedMigrations } = require('./migrations');
|
||||
const { createQueryApi } = require('./queries');
|
||||
@@ -201,142 +202,179 @@ 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) {
|
||||
let rawDb = null;
|
||||
let closed = false;
|
||||
let inSnapshot = false;
|
||||
let dirty = false;
|
||||
let inTransaction = false;
|
||||
|
||||
function saveToDisk() {
|
||||
if (dbPath === ':memory:' || inTransaction) {
|
||||
return;
|
||||
function reload() {
|
||||
if (dbPath === ':memory:' && rawDb) return;
|
||||
const bytes = dbPath !== ':memory:' && assertSafeDatabaseFile(dbPath)
|
||||
? readDatabaseFile(dbPath) : undefined;
|
||||
const latest = new SQL.Database(bytes);
|
||||
latest.run('PRAGMA foreign_keys = ON');
|
||||
if (rawDb) rawDb.close();
|
||||
rawDb = latest;
|
||||
}
|
||||
|
||||
// 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 = () => {
|
||||
reload();
|
||||
inSnapshot = true;
|
||||
dirty = false;
|
||||
try {
|
||||
const result = callback();
|
||||
if (result && typeof result.then === 'function') {
|
||||
throw new Error('State-store operations must be synchronous');
|
||||
}
|
||||
if (dirty && dbPath !== ':memory:') {
|
||||
writeDatabaseFileAtomic(dbPath, Buffer.from(rawDb.export()));
|
||||
}
|
||||
return result;
|
||||
} finally {
|
||||
inSnapshot = false;
|
||||
dirty = false;
|
||||
}
|
||||
};
|
||||
return dbPath === ':memory:' ? execute() : withStateStoreLock(dbPath, execute);
|
||||
}
|
||||
|
||||
function query(sql, positionalArgs, firstOnly) {
|
||||
return withSnapshot(() => {
|
||||
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(() => {
|
||||
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');
|
||||
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) { /* Preserve the original failure. */ }
|
||||
rawDb = null;
|
||||
if (error instanceof Error) 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(() => {
|
||||
rawDb.run(sql);
|
||||
dirty = true;
|
||||
});
|
||||
},
|
||||
|
||||
pragma(pragmaStr) {
|
||||
try {
|
||||
rawDb.run(`PRAGMA ${pragmaStr}`);
|
||||
} catch (_error) {
|
||||
// Ignore unsupported pragmas (e.g. WAL for in-memory databases).
|
||||
}
|
||||
return withSnapshot(() => rawDb.run(`PRAGMA ${pragmaStr}`));
|
||||
},
|
||||
|
||||
prepare(sql) {
|
||||
return {
|
||||
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);
|
||||
},
|
||||
|
||||
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);
|
||||
},
|
||||
|
||||
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);
|
||||
},
|
||||
};
|
||||
},
|
||||
|
||||
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;
|
||||
if (rawDb) rawDb.close();
|
||||
},
|
||||
};
|
||||
|
||||
return db;
|
||||
}
|
||||
|
||||
async function openDatabase(SQL, dbPath) {
|
||||
function openDatabase(SQL, dbPath) {
|
||||
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);
|
||||
}
|
||||
|
||||
async function createStateStore(options = {}) {
|
||||
const dbPath = resolveStateStorePath(options);
|
||||
const SQL = await initSqlJs();
|
||||
const db = await openDatabase(SQL, dbPath);
|
||||
const appliedMigrations = applyMigrations(db);
|
||||
const db = openDatabase(SQL, dbPath);
|
||||
let appliedMigrations;
|
||||
try {
|
||||
appliedMigrations = db.withSnapshot(() => applyMigrations(db));
|
||||
} catch (error) {
|
||||
db.close();
|
||||
throw error;
|
||||
}
|
||||
const queryApi = createQueryApi(db);
|
||||
const synchronizedQueries = Object.fromEntries(Object.entries(queryApi)
|
||||
.map(([name, query]) => [name, (...args) => db.withSnapshot(() => query(...args))]));
|
||||
|
||||
return {
|
||||
dbPath,
|
||||
@@ -344,11 +382,11 @@ async function createStateStore(options = {}) {
|
||||
db.close();
|
||||
},
|
||||
getAppliedMigrations() {
|
||||
return getAppliedMigrations(db);
|
||||
return db.withSnapshot(() => getAppliedMigrations(db));
|
||||
},
|
||||
validateEntity,
|
||||
assertValidEntity,
|
||||
...queryApi,
|
||||
...synchronizedQueries,
|
||||
_database: db,
|
||||
_migrations: appliedMigrations,
|
||||
};
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
'use strict';
|
||||
|
||||
const { createStateStore } = require('../../../scripts/lib/state-store');
|
||||
|
||||
let store;
|
||||
process.on('message', async ({ id, action, dbPath, worker, count = 1 }) => {
|
||||
try {
|
||||
if (action === 'open') {
|
||||
store = await createStateStore({ dbPath });
|
||||
} else if (action === 'write') {
|
||||
for (let index = 0; index < count; index += 1) {
|
||||
store.upsertWorkItem({
|
||||
id: `${worker}-${index}`, source: 'manual', title: `Task ${worker}-${index}`, status: 'open'
|
||||
});
|
||||
}
|
||||
} else if (action === 'increment') {
|
||||
for (let index = 0; index < count; index += 1) {
|
||||
store._database.transaction(() => {
|
||||
const item = store.getWorkItemById('counter');
|
||||
// Widen real transaction overlap without assuming which worker wins.
|
||||
Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 15);
|
||||
store.upsertWorkItem({ ...item, metadata: { value: item.metadata.value + 1 } });
|
||||
})();
|
||||
}
|
||||
} else if (action === 'close') {
|
||||
store.close();
|
||||
store = null;
|
||||
} else {
|
||||
throw new Error(`Unknown fixture action: ${action}`);
|
||||
}
|
||||
process.send({ id, ok: true });
|
||||
} catch (error) {
|
||||
process.send({ id, ok: false, error: error.stack });
|
||||
}
|
||||
});
|
||||
process.send({ ready: true });
|
||||
@@ -0,0 +1,184 @@
|
||||
'use strict';
|
||||
|
||||
const assert = require('assert');
|
||||
const fs = require('fs');
|
||||
const os = require('os');
|
||||
const path = require('path');
|
||||
const { fork, spawnSync } = require('child_process');
|
||||
const { createStateStore } = require('../../scripts/lib/state-store');
|
||||
|
||||
const WORKER = path.join(__dirname, 'helpers', 'state-store-worker.js');
|
||||
const WORK_ITEMS = path.join(__dirname, '..', '..', 'scripts', 'work-items.js');
|
||||
|
||||
function startWorker() {
|
||||
const child = fork(WORKER, [], { stdio: ['ignore', 'pipe', 'pipe', 'ipc'] });
|
||||
const pending = new Map();
|
||||
let nextId = 0;
|
||||
let stderr = '';
|
||||
child.stderr.on('data', data => { stderr += data; });
|
||||
child.stdout.resume();
|
||||
child.on('message', message => {
|
||||
const request = pending.get(message.id);
|
||||
if (!request) return;
|
||||
clearTimeout(request.timer);
|
||||
pending.delete(message.id);
|
||||
if (message.ok) request.resolve();
|
||||
else request.reject(new Error(message.error));
|
||||
});
|
||||
child.on('exit', code => {
|
||||
for (const request of pending.values()) {
|
||||
clearTimeout(request.timer);
|
||||
request.reject(new Error(`Worker exited ${code}: ${stderr}`));
|
||||
}
|
||||
pending.clear();
|
||||
});
|
||||
return {
|
||||
request(action, options = {}) {
|
||||
const id = ++nextId;
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
pending.delete(id);
|
||||
reject(new Error(`Worker timed out during ${action}: ${stderr}`));
|
||||
}, 30000);
|
||||
pending.set(id, { resolve, reject, timer });
|
||||
child.send({ id, action, ...options });
|
||||
});
|
||||
},
|
||||
async stop() {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
await new Promise(resolve => {
|
||||
child.once('exit', resolve);
|
||||
child.kill();
|
||||
});
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
async function withDatabase(fn) {
|
||||
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ecc-store-concurrency-'));
|
||||
const dbPath = path.join(dir, 'state.db');
|
||||
try {
|
||||
await fn(dbPath);
|
||||
} finally {
|
||||
fs.rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
async function readItems(dbPath) {
|
||||
const store = await createStateStore({ dbPath });
|
||||
try {
|
||||
return store.listWorkItems({ limit: 1000 }).items;
|
||||
} finally {
|
||||
store.close();
|
||||
}
|
||||
}
|
||||
|
||||
async function withWorkers(count, fn) {
|
||||
const workers = Array.from({ length: count }, () => startWorker());
|
||||
try {
|
||||
await fn(workers);
|
||||
} finally {
|
||||
await Promise.all(workers.map(worker => worker.stop()));
|
||||
}
|
||||
}
|
||||
|
||||
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.message}`);
|
||||
failed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
console.log('\n=== Testing state-store concurrency ===\n');
|
||||
|
||||
await test('closing an old reader preserves a task saved by the real CLI', async dbPath => {
|
||||
const reader = await createStateStore({ dbPath });
|
||||
try {
|
||||
assert.strictEqual(reader.listWorkItems().items.length, 0);
|
||||
const result = spawnSync(process.execPath, [WORK_ITEMS, 'upsert', 'cli-task',
|
||||
'--title', 'Task saved while dashboard is open', '--db', dbPath, '--json'],
|
||||
{ encoding: 'utf8', timeout: 30000 });
|
||||
assert.strictEqual(result.status, 0, result.stderr);
|
||||
} finally {
|
||||
reader.close();
|
||||
}
|
||||
assert.ok((await readItems(dbPath)).some(item => item.id === 'cli-task'),
|
||||
'Closing the older reader must not remove the CLI task');
|
||||
});
|
||||
|
||||
await test('long-lived handles preserve each other\'s writes and return current data', async dbPath => {
|
||||
const first = await createStateStore({ dbPath });
|
||||
const second = await createStateStore({ dbPath });
|
||||
try {
|
||||
first.upsertWorkItem({ id: 'a', source: 'manual', title: 'First task', status: 'open' });
|
||||
second.upsertWorkItem({ id: 'b', source: 'manual', title: 'Second task', status: 'open' });
|
||||
first.upsertWorkItem({ id: 'c', source: 'manual', title: 'Third task', status: 'open' });
|
||||
assert.deepStrictEqual(second.listWorkItems().items.map(item => item.id).sort(), ['a', 'b', 'c']);
|
||||
} finally {
|
||||
first.close();
|
||||
second.close();
|
||||
}
|
||||
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 });
|
||||
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 });
|
||||
try {
|
||||
const migrations = store.getAppliedMigrations();
|
||||
assert.ok(migrations.length > 0);
|
||||
assert.strictEqual(new Set(migrations.map(migration => migration.version)).size, migrations.length);
|
||||
assert.deepStrictEqual(store.listWorkItems().items.map(item => item.id).sort(), ['fresh-0-0', 'fresh-1-0', 'fresh-2-0']);
|
||||
} finally {
|
||||
store.close();
|
||||
}
|
||||
});
|
||||
|
||||
console.log(`\nResults: Passed: ${passed}, Failed: ${failed}`);
|
||||
process.exitCode = failed > 0 ? 1 : 0;
|
||||
}
|
||||
|
||||
run().catch(error => {
|
||||
console.error(error);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
@@ -0,0 +1,181 @@
|
||||
'use strict';
|
||||
|
||||
const assert = require('assert');
|
||||
const fs = require('fs');
|
||||
const os = require('os');
|
||||
const path = require('path');
|
||||
const { createStateStore } = require('../../scripts/lib/state-store');
|
||||
const { withStateStoreLock } = require('../../scripts/lib/state-store/file-lock');
|
||||
|
||||
const 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.stack}`);
|
||||
failed += 1;
|
||||
} finally {
|
||||
fs.rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
} finally { reader.close(); }
|
||||
} finally { fs.renameSync = rename; }
|
||||
assert.strictEqual(writes, 0);
|
||||
assert.deepStrictEqual(fs.readFileSync(dbPath), before);
|
||||
});
|
||||
|
||||
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 && 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 && 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;
|
||||
}
|
||||
|
||||
run().catch(error => { console.error(error); process.exitCode = 1; });
|
||||
Reference in New Issue
Block a user