fix(node): 4-second submit budget, compare-and-advance chain state under a per-agent lock
Claude-Session: https://claude.ai/code/session_01QoUjdc7HGKTsgj5GhG8HvF
This commit is contained in:
1 parent
85c2d2d491
commit
2101cb77e3
2 files changed
+161
-4
No files matched your search
@@ -39,7 +39,7 @@ const CHAIN_STATE_PATH = path.join(ELYDORA_DIR, AGENT_ID, 'chain-state.json');
|
||||
const ERROR_LOG_PATH = path.join(ELYDORA_DIR, AGENT_ID, 'error.log');
|
||||
const ZERO_CHAIN_HASH = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA';
|
||||
const MAX_CHAIN_ATTEMPTS = 5;
|
||||
const SUBMIT_BUDGET_MS = 6000;
|
||||
const SUBMIT_BUDGET_MS = 4000;
|
||||
|
||||
${PROTECTED_RUNTIME_READER}
|
||||
|
||||
@@ -167,6 +167,81 @@ function writeChainState(chainHash) {
|
||||
writeProtectedJson(CHAIN_STATE_PATH, 'Chain state', { prev_chain_hash: chainHash });
|
||||
}
|
||||
|
||||
function lockOwnerAlive(lockPath) {
|
||||
let owner;
|
||||
try {
|
||||
owner = Number.parseInt(fs.readFileSync(lockPath, 'utf-8'), 10);
|
||||
} catch (error) {
|
||||
if (!hasRuntimeErrorCode(error, 'ENOENT')) throw error;
|
||||
return false;
|
||||
}
|
||||
if (!Number.isInteger(owner) || owner <= 0) return false;
|
||||
try {
|
||||
process.kill(owner, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (hasRuntimeErrorCode(error, 'ESRCH')) return false;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
function withChainLock(update) {
|
||||
const lockPath = CHAIN_STATE_PATH + '.lock';
|
||||
const deadline = performance.now() + 1000;
|
||||
for (;;) {
|
||||
let descriptor;
|
||||
try {
|
||||
descriptor = fs.openSync(lockPath, 'wx', 0o600);
|
||||
fs.writeSync(descriptor, String(process.pid));
|
||||
} catch (error) {
|
||||
if (!hasRuntimeErrorCode(error, 'EEXIST')) throw error;
|
||||
if (lockOwnerAlive(lockPath)) {
|
||||
if (performance.now() > deadline) throw new Error('Chain state lock timed out: ' + lockPath);
|
||||
Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 10);
|
||||
continue;
|
||||
}
|
||||
let age = 0;
|
||||
try { age = Date.now() - fs.statSync(lockPath).mtimeMs; } catch (statError) {
|
||||
if (!hasRuntimeErrorCode(statError, 'ENOENT')) throw statError;
|
||||
}
|
||||
if (age > 5000) {
|
||||
const reclaimed = lockPath + '.' + process.pid + '.stale';
|
||||
try {
|
||||
fs.renameSync(lockPath, reclaimed);
|
||||
fs.unlinkSync(reclaimed);
|
||||
} catch (reclaimError) {
|
||||
if (!hasRuntimeErrorCode(reclaimError, 'ENOENT')) throw reclaimError;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (performance.now() > deadline) throw new Error('Chain state lock timed out: ' + lockPath);
|
||||
Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 10);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
return update();
|
||||
} finally {
|
||||
let current = null;
|
||||
try {
|
||||
current = fs.lstatSync(lockPath);
|
||||
} catch (statError) {
|
||||
if (!hasRuntimeErrorCode(statError, 'ENOENT')) throw statError;
|
||||
}
|
||||
const own = fs.fstatSync(descriptor);
|
||||
fs.closeSync(descriptor);
|
||||
if (current && current.ino === own.ino && current.dev === own.dev) fs.unlinkSync(lockPath);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function advanceChainState(fromHash, toHash) {
|
||||
return withChainLock(() => {
|
||||
if (readChainState() !== fromHash) return false;
|
||||
writeChainState(toHash);
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
function readConfigAndKey() {
|
||||
const raw = readProtectedFile(
|
||||
CONFIG_PATH,
|
||||
@@ -306,15 +381,15 @@ async function submitOperation(hookData, runtime) {
|
||||
const built = buildOperation(runtime, payload, payloadHash, toolName, sessionId, previousHash);
|
||||
const result = await postOperation(built.operation, runtime, deadline);
|
||||
if (result.accepted) {
|
||||
writeChainState(built.chainHash);
|
||||
advanceChainState(previousHash, built.chainHash);
|
||||
return;
|
||||
}
|
||||
writeChainState(result.expected);
|
||||
advanceChainState(previousHash, result.expected);
|
||||
logError(new Error('Chain hash resynced to server: ' + result.expected));
|
||||
if (attempt >= MAX_CHAIN_ATTEMPTS) {
|
||||
throw new Error('Audit API rejected prev_chain_hash ' + attempt + ' times');
|
||||
}
|
||||
previousHash = result.expected;
|
||||
previousHash = readChainState();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import { existsSync, utimesSync, writeFileSync } from 'node:fs';
|
||||
import { readFile } from 'node:fs/promises';
|
||||
import http from 'node:http';
|
||||
import path from 'node:path';
|
||||
@@ -256,3 +257,84 @@ test('Claude audit stops retrying when the submit budget is spent', async () =>
|
||||
await api.close();
|
||||
}
|
||||
});
|
||||
|
||||
test('Claude audit does not regress a chain head advanced by a concurrent hook', async () => {
|
||||
const newerHead = 'B'.repeat(43);
|
||||
let statePath;
|
||||
const api = await mismatchServer(() => {
|
||||
writeFileSync(statePath, JSON.stringify({ prev_chain_hash: newerHead }), { mode: 0o600 });
|
||||
return { status: 202, body: { receipt: { seq_no: 1 } } };
|
||||
});
|
||||
const fixture = await createFixture({ baseUrl: api.baseUrl });
|
||||
try {
|
||||
assert.equal((await fixture.install()).code, 0);
|
||||
statePath = path.join(fixture.agentDir, 'chain-state.json');
|
||||
const { settings } = await readSettings(fixture.settingsPath);
|
||||
const audit = managedHandler(settings, 'PostToolUse');
|
||||
const result = await runClaudeHook(
|
||||
audit,
|
||||
JSON.stringify(officialPayload('PostToolUse', { tool_response: { stdout: 'ok' } })),
|
||||
fixture,
|
||||
);
|
||||
assert.equal(result.code, 0, result.stderr);
|
||||
assert.equal(api.submissions[0].prev_chain_hash, 'A'.repeat(43));
|
||||
const state = JSON.parse(await readFile(statePath, 'utf-8'));
|
||||
assert.equal(state.prev_chain_hash, newerHead);
|
||||
} finally {
|
||||
await fixture.close();
|
||||
await api.close();
|
||||
}
|
||||
});
|
||||
|
||||
test('Claude audit clears a stale chain-state lock and proceeds', async () => {
|
||||
const api = await mismatchServer(() => ({ status: 202, body: { receipt: { seq_no: 1 } } }));
|
||||
const fixture = await createFixture({ baseUrl: api.baseUrl });
|
||||
try {
|
||||
assert.equal((await fixture.install()).code, 0);
|
||||
const lockPath = path.join(fixture.agentDir, 'chain-state.json.lock');
|
||||
writeFileSync(lockPath, '', { mode: 0o600 });
|
||||
const stale = new Date(Date.now() - 10_000);
|
||||
utimesSync(lockPath, stale, stale);
|
||||
const { settings } = await readSettings(fixture.settingsPath);
|
||||
const audit = managedHandler(settings, 'PostToolUse');
|
||||
const result = await runClaudeHook(
|
||||
audit,
|
||||
JSON.stringify(officialPayload('PostToolUse', { tool_response: { stdout: 'ok' } })),
|
||||
fixture,
|
||||
);
|
||||
assert.equal(result.code, 0, result.stderr);
|
||||
assert.equal(api.submissions.length, 1);
|
||||
assert.equal(existsSync(lockPath), false);
|
||||
const state = JSON.parse(await readFile(path.join(fixture.agentDir, 'chain-state.json'), 'utf-8'));
|
||||
assert.equal(state.prev_chain_hash, api.submissions[0].chain_hash);
|
||||
} finally {
|
||||
await fixture.close();
|
||||
await api.close();
|
||||
}
|
||||
});
|
||||
|
||||
test('Claude audit does not reclaim a chain-state lock whose owner is alive', async () => {
|
||||
const api = await mismatchServer(() => ({ status: 202, body: { receipt: { seq_no: 1 } } }));
|
||||
const fixture = await createFixture({ baseUrl: api.baseUrl });
|
||||
try {
|
||||
assert.equal((await fixture.install()).code, 0);
|
||||
const lockPath = path.join(fixture.agentDir, 'chain-state.json.lock');
|
||||
writeFileSync(lockPath, String(process.pid), { mode: 0o600 });
|
||||
const stale = new Date(Date.now() - 10_000);
|
||||
utimesSync(lockPath, stale, stale);
|
||||
const { settings } = await readSettings(fixture.settingsPath);
|
||||
const audit = managedHandler(settings, 'PostToolUse');
|
||||
const result = await runClaudeHook(
|
||||
audit,
|
||||
JSON.stringify(officialPayload('PostToolUse', { tool_response: { stdout: 'ok' } })),
|
||||
fixture,
|
||||
);
|
||||
assert.equal(result.code, 0, result.stderr);
|
||||
assert.equal(api.submissions.length, 1);
|
||||
assert.equal(existsSync(lockPath), true);
|
||||
assert.match(await readFile(path.join(fixture.agentDir, 'error.log'), 'utf-8'), /lock timed out/);
|
||||
} finally {
|
||||
await fixture.close();
|
||||
await api.close();
|
||||
}
|
||||
});
|
||||
Reference in new issue
Block a user