diff --git a/docs/architecture/hud-status-session-control.md b/docs/architecture/hud-status-session-control.md index 0787c0003..f8f1c269d 100644 --- a/docs/architecture/hud-status-session-control.md +++ b/docs/architecture/hud-status-session-control.md @@ -82,9 +82,14 @@ 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. +with a sibling `.ecc-state.lock` file. It checks the current database +under that lock and publishes successful writes before releasing it. A handle +can reuse its last successful read snapshot when the file identity, size, and +modification/change timestamps match. Fixed SELECTs in the query API avoid +reloading or exporting an unchanged database; query results are still computed +afresh. Generic SQL, transactions, exports, and failed operations invalidate +reuse. Changes from another writer trigger a reload on the next operation. +Queries use one snapshot; closing a handle never writes an older snapshot back. All concurrent writers must use this adapter; older ECC versions and external SQLite writers do not participate in this locking protocol. diff --git a/scripts/lib/state-store/index.js b/scripts/lib/state-store/index.js index 6bb4fc256..18a5a4de5 100644 --- a/scripts/lib/state-store/index.js +++ b/scripts/lib/state-store/index.js @@ -27,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; @@ -100,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'); @@ -114,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; @@ -202,7 +211,7 @@ 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(SQL, dbPath) { +function wrapSqlJsDatabase(SQL, dbPath, trustedReads = new WeakSet()) { let rawDb = null; let closed = false; let inSnapshot = false; @@ -210,11 +219,17 @@ function wrapSqlJsDatabase(SQL, dbPath) { let inTransaction = false; let needsSnapshotComparison = false; let snapshotBytes = null; + let readCacheIdentity = null; + let cacheableSnapshot = false; function reload() { + const cachedIdentity = readCacheIdentity; + readCacheIdentity = null; if (dbPath === ':memory:' && rawDb) return; - const bytes = dbPath !== ':memory:' && assertSafeDatabaseFile(dbPath) - ? readDatabaseFile(dbPath) : undefined; + 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); @@ -234,6 +249,7 @@ function wrapSqlJsDatabase(SQL, dbPath) { } rawDb = latest; snapshotBytes = originalBytes; + return snapshot && snapshot.identity; } // Hold one lock from reload through commit. Nested statements and public @@ -242,10 +258,11 @@ function wrapSqlJsDatabase(SQL, dbPath) { if (closed) throw new Error('State store is closed'); if (inSnapshot) return callback(); const execute = () => { - reload(); + const identity = reload(); inSnapshot = true; dirty = false; needsSnapshotComparison = false; + cacheableSnapshot = true; try { const result = callback(); if (result && typeof result.then === 'function') { @@ -262,18 +279,29 @@ function wrapSqlJsDatabase(SQL, dbPath) { 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; } }; - return dbPath === ':memory:' ? execute() : withStateStoreLock(dbPath, execute); + try { + return dbPath === ':memory:' ? execute() : withStateStoreLock(dbPath, execute); + } catch (error) { + readCacheIdentity = null; + throw error; + } } - function query(sql, positionalArgs, firstOnly) { + 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') { @@ -281,7 +309,6 @@ function wrapSqlJsDatabase(SQL, dbPath) { } else if (positionalArgs.length > 1) { stmt.bind(positionalArgs); } - needsSnapshotComparison = true; if (firstOnly) return stmt.step() ? stmt.getAsObject() : null; const rows = []; while (stmt.step()) rows.push(stmt.getAsObject()); @@ -294,6 +321,7 @@ function wrapSqlJsDatabase(SQL, dbPath) { function runStatement(sql, namedParams) { return withSnapshot(() => { + needsSnapshotComparison = true; const stmt = rawDb.prepare(sql); try { if (namedParams && typeof namedParams === 'object' && !Array.isArray(namedParams)) { @@ -310,6 +338,7 @@ function wrapSqlJsDatabase(SQL, dbPath) { 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; @@ -341,6 +370,7 @@ function wrapSqlJsDatabase(SQL, dbPath) { withSnapshot, exec(sql) { return withSnapshot(() => { + needsSnapshotComparison = true; rawDb.run(sql); dirty = true; }); @@ -348,25 +378,26 @@ function wrapSqlJsDatabase(SQL, dbPath) { pragma(pragmaStr) { return withSnapshot(() => { - rawDb.run(`PRAGMA ${pragmaStr}`); needsSnapshotComparison = true; + rawDb.run(`PRAGMA ${pragmaStr}`); }); }, prepare(sql) { - return { + const statement = { all(...positionalArgs) { - return query(sql, positionalArgs, false); + return query(sql, positionalArgs, false, trustedReads.has(statement)); }, get(...positionalArgs) { - return query(sql, positionalArgs, true); + return query(sql, positionalArgs, true, trustedReads.has(statement)); }, run(namedParams) { return runStatement(sql, namedParams); }, }; + return statement; }, transaction(fn) { @@ -377,6 +408,7 @@ function wrapSqlJsDatabase(SQL, dbPath) { if (inSnapshot) throw new Error('Cannot close a state store during an operation'); if (closed) return; closed = true; + readCacheIdentity = null; if (rawDb) rawDb.close(); }, }; @@ -384,18 +416,19 @@ function wrapSqlJsDatabase(SQL, dbPath) { return db; } -function openDatabase(SQL, dbPath) { +function openDatabase(SQL, dbPath, trustedReads) { if (dbPath !== ':memory:') { ensurePrivateDirectory(path.dirname(dbPath)); } - return wrapSqlJsDatabase(SQL, dbPath); + return wrapSqlJsDatabase(SQL, dbPath, trustedReads); } async function createStateStore(options = {}) { const dbPath = resolveStateStorePath(options); const SQL = await initSqlJs(); - const db = openDatabase(SQL, dbPath); + const trustedReads = new WeakSet(); + const db = openDatabase(SQL, dbPath, trustedReads); let appliedMigrations; try { appliedMigrations = db.withSnapshot(() => applyMigrations(db)); @@ -403,7 +436,13 @@ async function createStateStore(options = {}) { try { db.close(); } catch (closeError) { recordCleanupError(error, 'closeError', closeError); } throw error; } - const queryApi = createQueryApi(db); + 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))])); diff --git a/scripts/lib/state-store/queries.js b/scripts/lib/state-store/queries.js index 0225f4761..7074d5a96 100644 --- a/scripts/lib/state-store/queries.js +++ b/scripts/lib/state-store/queries.js @@ -396,35 +396,36 @@ function normalizeWorkItemInput(workItem) { }; } -function createQueryApi(db) { - const listRecentSessionsStatement = db.prepare(` +function createQueryApi(db, { prepareRead = sql => db.prepare(sql) } = {}) { + // Only these fixed SELECTs qualify for snapshot reuse; generic SQL stays conservative. + const listRecentSessionsStatement = prepareRead(` SELECT * FROM sessions ORDER BY COALESCE(started_at, ended_at, '') DESC, id DESC LIMIT ? `); - const countSessionsStatement = db.prepare(` + const countSessionsStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM sessions `); - const getSessionStatement = db.prepare(` + const getSessionStatement = prepareRead(` SELECT * FROM sessions WHERE id = ? `); - const getSessionSkillRunsStatement = db.prepare(` + const getSessionSkillRunsStatement = prepareRead(` SELECT * FROM skill_runs WHERE session_id = ? ORDER BY created_at DESC, id DESC `); - const getSessionDecisionsStatement = db.prepare(` + const getSessionDecisionsStatement = prepareRead(` SELECT * FROM decisions WHERE session_id = ? ORDER BY created_at DESC, id DESC `); - const listActiveSessionsStatement = db.prepare(` + const listActiveSessionsStatement = prepareRead(` SELECT * FROM sessions WHERE ended_at IS NULL @@ -432,61 +433,61 @@ function createQueryApi(db) { ORDER BY COALESCE(started_at, ended_at, '') DESC, id DESC LIMIT ? `); - const countActiveSessionsStatement = db.prepare(` + const countActiveSessionsStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM sessions WHERE ended_at IS NULL AND state IN ('active', 'running', 'idle') `); - const listRecentSkillRunsStatement = db.prepare(` + const listRecentSkillRunsStatement = prepareRead(` SELECT * FROM skill_runs ORDER BY created_at DESC, id DESC LIMIT ? `); - const listInstallStateStatement = db.prepare(` + const listInstallStateStatement = prepareRead(` SELECT * FROM install_state ORDER BY installed_at DESC, target_id ASC `); - const getInstallStateStatement = db.prepare(` + const getInstallStateStatement = prepareRead(` SELECT target_id FROM install_state WHERE target_id = ? AND target_root = ? `); - const countPendingGovernanceStatement = db.prepare(` + const countPendingGovernanceStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM governance_events WHERE resolved_at IS NULL `); - const listPendingGovernanceStatement = db.prepare(` + const listPendingGovernanceStatement = prepareRead(` SELECT * FROM governance_events WHERE resolved_at IS NULL ORDER BY created_at DESC, id DESC LIMIT ? `); - const listWorkItemsStatement = db.prepare(` + const listWorkItemsStatement = prepareRead(` SELECT * FROM work_items ORDER BY updated_at DESC, id DESC LIMIT ? `); - const countWorkItemsStatement = db.prepare(` + const countWorkItemsStatement = prepareRead(` SELECT COUNT(*) AS total_count FROM work_items `); - const listAllWorkItemsStatement = db.prepare(` + const listAllWorkItemsStatement = prepareRead(` SELECT * FROM work_items ORDER BY updated_at DESC, id DESC `); - const getWorkItemStatement = db.prepare(` + const getWorkItemStatement = prepareRead(` SELECT * FROM work_items WHERE id = ? `); - const getSkillVersionStatement = db.prepare(` + const getSkillVersionStatement = prepareRead(` SELECT * FROM skill_versions WHERE skill_id = ? AND version = ? diff --git a/tests/lib/state-store-concurrency.test.js b/tests/lib/state-store-concurrency.test.js index c96a27196..79d137b64 100644 --- a/tests/lib/state-store-concurrency.test.js +++ b/tests/lib/state-store-concurrency.test.js @@ -101,6 +101,8 @@ async function run() { '--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'); diff --git a/tests/lib/state-store-read-cache.test.js b/tests/lib/state-store-read-cache.test.js new file mode 100644 index 000000000..64d6850b4 --- /dev/null +++ b/tests/lib/state-store-read-cache.test.js @@ -0,0 +1,504 @@ +'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) { + const directory = 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; });