From 521de010bcbd8f17696980c0ba7185cb3fdb901b Mon Sep 17 00:00:00 2001 From: Yann Roberto <1922498827@qq.com> Date: Mon, 28 Sep 2026 16:31:09 +0800 Subject: [PATCH] fix(state-store): preserve concurrent writes across processes --- .../hud-status-session-control.md | 16 ++ scripts/lib/state-store/file-lock.js | 90 +++++++ scripts/lib/state-store/index.js | 224 ++++++++++-------- scripts/lib/state-store/migrations.js | 3 + tests/lib/helpers/state-store-worker.js | 36 +++ tests/lib/state-store-concurrency.test.js | 184 ++++++++++++++ tests/lib/state-store-persistence.test.js | 181 ++++++++++++++ 7 files changed, 641 insertions(+), 93 deletions(-) create mode 100644 scripts/lib/state-store/file-lock.js create mode 100644 tests/lib/helpers/state-store-worker.js create mode 100644 tests/lib/state-store-concurrency.test.js create mode 100644 tests/lib/state-store-persistence.test.js diff --git a/docs/architecture/hud-status-session-control.md b/docs/architecture/hud-status-session-control.md index 8b773e20c..7225018d4 100644 --- a/docs/architecture/hud-status-session-control.md +++ b/docs/architecture/hud-status-session-control.md @@ -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 `.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. diff --git a/scripts/lib/state-store/file-lock.js b/scripts/lib/state-store/file-lock.js new file mode 100644 index 000000000..3bc3edf37 --- /dev/null +++ b/scripts/lib/state-store/file-lock.js @@ -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 }; diff --git a/scripts/lib/state-store/index.js b/scripts/lib/state-store/index.js index 49b6204d6..3d60fc579 100644 --- a/scripts/lib/state-store/index.js +++ b/scripts/lib/state-store/index.js @@ -5,6 +5,7 @@ const crypto = require('crypto'); const os = require('os'); const path = require('path'); const initSqlJs = require('sql.js'); +const { withStateStoreLock } = 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, }; diff --git a/scripts/lib/state-store/migrations.js b/scripts/lib/state-store/migrations.js index 7716c9925..c88021bc2 100644 --- a/scripts/lib/state-store/migrations.js +++ b/scripts/lib/state-store/migrations.js @@ -147,6 +147,9 @@ const MIGRATIONS = [ ]; function ensureMigrationTable(db) { + if (db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'schema_migrations'").get()) { + return; + } db.exec(` CREATE TABLE IF NOT EXISTS schema_migrations ( version INTEGER PRIMARY KEY, diff --git a/tests/lib/helpers/state-store-worker.js b/tests/lib/helpers/state-store-worker.js new file mode 100644 index 000000000..045424e57 --- /dev/null +++ b/tests/lib/helpers/state-store-worker.js @@ -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 }); diff --git a/tests/lib/state-store-concurrency.test.js b/tests/lib/state-store-concurrency.test.js new file mode 100644 index 000000000..6a5f499a8 --- /dev/null +++ b/tests/lib/state-store-concurrency.test.js @@ -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; +}); diff --git a/tests/lib/state-store-persistence.test.js b/tests/lib/state-store-persistence.test.js new file mode 100644 index 000000000..00edf7ab4 --- /dev/null +++ b/tests/lib/state-store-persistence.test.js @@ -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; });