mirror of
https://github.com/affaan-m/ECC.git
synced 2026-10-14 12:15:29 +02:00
perf(state-store): reuse unchanged read snapshots
Keep the operation lock while validating bigint file identity and reuse only snapshots produced by successful trusted static SELECTs. Avoid reloading and exporting unchanged databases on repeated public reads. Retain conservative comparison for generic SQL, including RETURNING and PRAGMA writes, and invalidate reuse after transactions, exports or errors. Cover cache invalidation, cross-process visibility, Windows device IDs, reentrant partial writes, lock contention and independent result objects. Validate 188 focused tests per environment on WSL Node 22.22.3 and native Windows Node 22.15.0/24.16.0, plus 38 related WSL tests. Paired warm-read benchmarks show 7.6-18.1x point-query speedup on 160 KiB/1.28 MiB databases.
This commit is contained in:
@@ -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))]));
|
||||
|
||||
|
||||
@@ -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 = ?
|
||||
|
||||
Reference in New Issue
Block a user