fix(state-store): preserve ownership and errors through cleanup

Preserve the original contributor histories and apply the exact independently reviewed repair.

Source-Parent: 5785183108
Review-Manifest-SHA256: e223d98c24e357715527b786257901d2ef0bebadb75ac61e5c000fe60da10f9a
This commit is contained in:
affaan-m
2026-09-28 07:29:56 -04:00
parent 5785183108
commit ed4b856bfe
10 changed files with 794 additions and 262 deletions
+17 -5
View File
@@ -6,7 +6,7 @@ const { Worker } = require('worker_threads');
const MAX_ACTIVE_MUTATIONS = 2;
let activeMutations = 0;
/** Bound live workers process-wide; excess mutations are rejected, not queued. */
/** Bound workers in this loaded runner/isolate; reject excess work without queuing. */
function runWorkItemMutation(dbPath, action, args) {
return new Promise((resolve, reject) => {
if (activeMutations >= MAX_ACTIVE_MUTATIONS) {
@@ -27,14 +27,26 @@ function runWorkItemMutation(dbPath, action, args) {
}
let response;
let workerError;
let failed = false;
worker.once('message', message => { response = message; });
worker.once('error', error => { workerError = error; });
worker.once('error', error => { failed = true; workerError = error; });
worker.once('exit', code => {
// Keep the slot until the thread is gone, even after a message/error.
activeMutations -= 1;
if (workerError) reject(workerError);
else if (code !== 0 || !response) reject(new Error(`Work-item worker exited without a successful result (${code})`));
else if (response.ok) resolve(response.result);
if (failed) { reject(workerError); return; }
if (code !== 0 || response === undefined) {
reject(new Error(`Work-item worker exited without a successful result (${code})`));
return;
}
// Worker messages are structured clones from our own helper, but malformed
// results still must reject instead of throwing outside the Promise executor.
if (!response || typeof response !== 'object' || Array.isArray(response)
|| typeof response.ok !== 'boolean'
|| (response.ok === false && (!response.error || typeof response.error !== 'object'
|| typeof response.error.message !== 'string'
|| (response.error.code !== undefined && typeof response.error.code !== 'string')))) {
reject(new Error('Invalid work-item worker response'));
} else if (response.ok) resolve(response.result);
else reject(Object.assign(new Error(response.error.message), { code: response.error.code }));
});
// Let the worker exit naturally after closing the store. Terminating it on
+26 -5
View File
@@ -2,6 +2,7 @@
const { parentPort, workerData } = require('worker_threads');
const { createStateStore } = require('../state-store');
const { attachCleanupError } = require('../state-store/file-lock');
const { claimWorkItem, moveWorkItem } = require('./work-item-mutations');
/** Complete one board mutation away from the HTTP event loop, then close. */
@@ -10,14 +11,34 @@ async function mutate() {
const mutation = action === 'claim' ? claimWorkItem : action === 'move' ? moveWorkItem : null;
if (!mutation) throw new Error('Unknown work-item mutation');
const store = await createStateStore({ dbPath });
try {
return store._database.transaction(() => mutation(store, args))();
} finally {
store.close();
let result;
let failed = false;
let primary;
try { result = store._database.transaction(() => mutation(store, args))(); }
catch (error) { failed = true; primary = error; }
try { store.close(); }
catch (error) {
if (failed) attachCleanupError(primary, 'closeError', error);
else { failed = true; primary = error; }
}
if (failed) throw primary;
return result;
}
function serializeError(error) {
let message = 'Work-item mutation failed';
let code;
try {
if (error && typeof error.message === 'string') message = error.message.slice(0, 1024);
else if (error === null || typeof error !== 'object') message = String(error).slice(0, 1024);
} catch (_error) { /* A throwing conversion must not prevent the failure response. */ }
try {
if (error && typeof error.code === 'string') code = error.code.slice(0, 128);
} catch (_error) { /* Optional diagnostic. */ }
return { message, code };
}
mutate().then(
result => parentPort.postMessage({ ok: true, result }),
error => parentPort.postMessage({ ok: false, error: { message: error.message, code: error.code } })
error => parentPort.postMessage({ ok: false, error: serializeError(error) })
);
+53 -13
View File
@@ -7,22 +7,62 @@ const { performance } = require('perf_hooks');
const WAIT_BUFFER = new Int32Array(new SharedArrayBuffer(4));
const DEFAULT_TIMEOUT_MS = 5000;
// Cleanup must never replace an arbitrary primary throw (including frozen errors,
// primitives, or objects with throwing property traps). Diagnostics are best effort.
function attachCleanupError(primary, name, secondary) {
if (primary === null || (typeof primary !== 'object' && typeof primary !== 'function')) return;
try { Object.defineProperty(primary, name, { value: secondary, configurable: true }); }
catch (_error) { /* Preserve the primary when it cannot accept diagnostics. */ }
}
function hasCode(error, code) {
try { return error !== null && error !== undefined && error.code === code; }
catch (_error) { return false; }
}
function sameIdentity(left, right) {
return left.ino === right.ino
&& (left.dev === right.dev || (process.platform === 'win32' && (!left.dev || !right.dev)));
return left.ino === right.ino && left.dev === right.dev;
}
function lostLock(lockPath) {
return Object.assign(new Error(`State-store lock changed or lost; refusing to remove it: ${lockPath}`),
{ code: 'STATE_STORE_LOCK_LOST' });
}
function releaseOwnedLock(lockPath, descriptor, identity) {
let current;
let failed = false;
let primary;
let pendingDelete = false;
let inspected = false;
let closeFailed = false;
try {
current = fs.lstatSync(lockPath, { bigint: true });
} finally {
fs.closeSync(descriptor);
const current = fs.lstatSync(lockPath, { bigint: true });
inspected = true;
if (!current.isFile() || current.isSymbolicLink() || !sameIdentity(identity, current)) {
throw lostLock(lockPath);
}
// Keep the owned descriptor through validation and unlink. This cooperative
// lock is not an atomic compare-and-unlink against a directory mutator.
fs.unlinkSync(lockPath);
} catch (error) {
failed = true;
primary = hasCode(error, 'ENOENT') ? lostLock(lockPath) : error;
pendingDelete = !inspected && hasCode(error, 'EPERM');
}
if (!current.isFile() || current.isSymbolicLink() || !sameIdentity(identity, current)) {
throw new Error(`State-store lock changed; refusing to remove it: ${lockPath}`);
try {
fs.closeSync(descriptor); // Exactly one attempt, even after a release failure.
} catch (error) {
closeFailed = true;
if (failed) attachCleanupError(primary, 'closeError', error);
else { failed = true; primary = error; }
}
fs.unlinkSync(lockPath);
if (pendingDelete && !closeFailed) {
// Windows can defer deletion until the handle closes. Only a subsequent
// ENOENT proves loss; preserve genuine EPERM and never unlink after close.
try { fs.lstatSync(lockPath, { bigint: true }); }
catch (error) { if (hasCode(error, 'ENOENT')) primary = lostLock(lockPath); }
}
if (failed) throw primary;
}
function acquireLock(dbPath, timeoutMs) {
@@ -33,7 +73,7 @@ function acquireLock(dbPath, timeoutMs) {
try {
descriptor = fs.openSync(lockPath, 'wx', 0o600);
} catch (error) {
if (error.code !== 'EEXIST') throw error;
if (!hasCode(error, '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. `
@@ -54,7 +94,7 @@ function acquireLock(dbPath, timeoutMs) {
if (identity) releaseOwnedLock(lockPath, descriptor, identity);
else fs.closeSync(descriptor);
} catch (releaseError) {
error.releaseError = releaseError;
attachCleanupError(error, 'releaseError', releaseError);
}
throw error;
}
@@ -81,10 +121,10 @@ function withStateStoreLock(dbPath, callback, { timeoutMs = DEFAULT_TIMEOUT_MS }
release();
} catch (releaseError) {
if (!failed) throw releaseError;
if (primaryError instanceof Error) primaryError.releaseError = releaseError;
attachCleanupError(primaryError, 'releaseError', releaseError);
}
if (failed) throw primaryError;
return result;
}
module.exports = { withStateStoreLock };
module.exports = { withStateStoreLock, attachCleanupError };
+18 -6
View File
@@ -5,7 +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 { withStateStoreLock, attachCleanupError } = require('./file-lock');
const { applyMigrations, getAppliedMigrations } = require('./migrations');
const { createQueryApi } = require('./queries');
@@ -218,8 +218,20 @@ function wrapSqlJsDatabase(SQL, dbPath) {
// sql.js can use its input buffer as writable backing storage.
const originalBytes = bytes ? Buffer.from(bytes) : null;
const latest = new SQL.Database(bytes);
latest.run('PRAGMA foreign_keys = ON');
if (rawDb) rawDb.close();
try {
latest.run('PRAGMA foreign_keys = ON');
if (rawDb) {
const previous = rawDb;
rawDb = null;
try { previous.close(); }
catch (error) { closed = true; throw error; }
}
} catch (error) {
// The replacement is not adopted until both initialization and the old
// handle's close succeed. Never reuse an uncertain previous handle.
try { latest.close(); } catch (closeError) { attachCleanupError(error, 'closeError', closeError); }
throw error;
}
rawDb = latest;
snapshotBytes = originalBytes;
}
@@ -312,9 +324,9 @@ function wrapSqlJsDatabase(SQL, dbPath) {
} catch (rollbackError) {
// Never reuse an uncertain transaction, including an in-memory store.
closed = true;
try { rawDb.close(); } catch (_closeError) { /* Preserve the original failure. */ }
try { rawDb.close(); } catch (closeError) { attachCleanupError(error, 'closeError', closeError); }
rawDb = null;
if (error instanceof Error) error.rollbackError = rollbackError;
attachCleanupError(error, 'rollbackError', rollbackError);
}
dirty = previouslyDirty;
throw error;
@@ -386,7 +398,7 @@ async function createStateStore(options = {}) {
try {
appliedMigrations = db.withSnapshot(() => applyMigrations(db));
} catch (error) {
db.close();
try { db.close(); } catch (closeError) { attachCleanupError(error, 'closeError', closeError); }
throw error;
}
const queryApi = createQueryApi(db);