mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-10-27 07:51:52 +00:00
Add scheduler lock retries for backend failures
This commit is contained in:
1 parent
70c458250e
commit
9c069f75a6
3 files changed
+183
-11
No files matched your search
@@ -92,6 +92,14 @@ function normalizeStringList(value, fallback = []) {
|
||||
return value.filter((item) => typeof item === 'string' && item.trim()).map((item) => item.trim());
|
||||
}
|
||||
|
||||
function parseNonNegativeInt(value, fallback) {
|
||||
const parsed = Number.parseInt(String(value ?? ''), 10);
|
||||
if (Number.isFinite(parsed) && parsed >= 0) {
|
||||
return parsed;
|
||||
}
|
||||
return fallback;
|
||||
}
|
||||
|
||||
function redactSensitive(text) {
|
||||
if (!text) return '';
|
||||
return String(text)
|
||||
@@ -279,6 +287,19 @@ async function getSchedulerConfig(cadence, { isInteractive }) {
|
||||
const missingHandoffCommandForMode = !isInteractive && !handoffCommand;
|
||||
const memoryPolicyRaw = scheduler.memoryPolicyByCadence?.[cadence] || {};
|
||||
const mode = memoryPolicyRaw.mode === 'required' ? 'required' : 'optional';
|
||||
const lockRetryRaw = scheduler.lockRetry || {};
|
||||
const maxRetries = parseNonNegativeInt(
|
||||
process.env.SCHEDULER_LOCK_MAX_RETRIES,
|
||||
parseNonNegativeInt(lockRetryRaw.maxRetries, 2),
|
||||
);
|
||||
const backoffMs = parseNonNegativeInt(
|
||||
process.env.SCHEDULER_LOCK_BACKOFF_MS,
|
||||
parseNonNegativeInt(lockRetryRaw.backoffMs, 250),
|
||||
);
|
||||
const jitterMs = parseNonNegativeInt(
|
||||
process.env.SCHEDULER_LOCK_JITTER_MS,
|
||||
parseNonNegativeInt(lockRetryRaw.jitterMs, 75),
|
||||
);
|
||||
|
||||
return {
|
||||
firstPrompt: scheduler.firstPromptByCadence?.[cadence] || null,
|
||||
@@ -287,6 +308,11 @@ async function getSchedulerConfig(cadence, { isInteractive }) {
|
||||
validationCommands: Array.isArray(scheduler.validationCommandsByCadence?.[cadence])
|
||||
? scheduler.validationCommandsByCadence[cadence].filter((cmd) => typeof cmd === 'string' && cmd.trim())
|
||||
: ['npm', 'run', 'lint'],
|
||||
lockRetry: {
|
||||
maxRetries,
|
||||
backoffMs,
|
||||
jitterMs,
|
||||
},
|
||||
memoryPolicy: {
|
||||
mode,
|
||||
retrieveCommand: typeof memoryPolicyRaw.retrieveCommand === 'string' && memoryPolicyRaw.retrieveCommand.trim()
|
||||
@@ -303,6 +329,54 @@ async function getSchedulerConfig(cadence, { isInteractive }) {
|
||||
};
|
||||
}
|
||||
|
||||
function sleep(ms) {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetry }) {
|
||||
const lockCommandArgs = ['run', 'lock:lock', '--', '--agent', selectedAgent, '--cadence', cadence];
|
||||
const backoffScheduleMs = [];
|
||||
let attempts = 0;
|
||||
|
||||
while (true) {
|
||||
attempts += 1;
|
||||
const result = await runCommand('npm', lockCommandArgs, { env: { AGENT_PLATFORM: platform } });
|
||||
|
||||
if (result.code !== 2) {
|
||||
return { result, attempts, backoffScheduleMs };
|
||||
}
|
||||
|
||||
if (attempts > lockRetry.maxRetries) {
|
||||
return {
|
||||
result,
|
||||
attempts,
|
||||
backoffScheduleMs,
|
||||
finalBackendCategory: classifyLockBackendError(`${result.stderr}\n${result.stdout}`),
|
||||
};
|
||||
}
|
||||
|
||||
const exponentialBase = lockRetry.backoffMs * (2 ** (attempts - 1));
|
||||
const jitter = lockRetry.jitterMs > 0
|
||||
? Math.floor(Math.random() * (lockRetry.jitterMs + 1))
|
||||
: 0;
|
||||
const delayMs = exponentialBase + jitter;
|
||||
backoffScheduleMs.push(delayMs);
|
||||
|
||||
console.log(JSON.stringify({
|
||||
event: 'scheduler.lock.retry',
|
||||
attempt: attempts,
|
||||
max_retries: lockRetry.maxRetries,
|
||||
delay_ms: delayMs,
|
||||
selected_agent: selectedAgent,
|
||||
cadence,
|
||||
}));
|
||||
|
||||
if (delayMs > 0) {
|
||||
await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function toYamlScalar(value) {
|
||||
const str = String(value ?? '');
|
||||
return `'${str.replace(/'/g, "''")}'`;
|
||||
@@ -403,11 +477,13 @@ async function main() {
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const lockResult = await runCommand(
|
||||
'npm',
|
||||
['run', 'lock:lock', '--', '--agent', selectedAgent, '--cadence', cadence],
|
||||
{ env: { AGENT_PLATFORM: platform } },
|
||||
);
|
||||
const lockAttempt = await acquireLockWithRetry({
|
||||
selectedAgent,
|
||||
cadence,
|
||||
platform,
|
||||
lockRetry: schedulerConfig.lockRetry,
|
||||
});
|
||||
const lockResult = lockAttempt.result;
|
||||
|
||||
if (lockResult.code === 3) {
|
||||
continue;
|
||||
@@ -415,17 +491,20 @@ async function main() {
|
||||
|
||||
if (lockResult.code === 2) {
|
||||
const combinedLockOutput = `${lockResult.stderr}\n${lockResult.stdout}`;
|
||||
const backendCategory = classifyLockBackendError(combinedLockOutput);
|
||||
const backendCategory = lockAttempt.finalBackendCategory || classifyLockBackendError(combinedLockOutput);
|
||||
const lockCommand = `AGENT_PLATFORM=${platform} npm run lock:lock -- --agent ${selectedAgent} --cadence ${cadence}`;
|
||||
const stderrExcerpt = excerptText(lockResult.stderr);
|
||||
const stdoutExcerpt = excerptText(lockResult.stdout);
|
||||
const backoffSchedule = lockAttempt.backoffScheduleMs.join(', ');
|
||||
await writeLog({
|
||||
cadence,
|
||||
agent: selectedAgent,
|
||||
status: 'failed',
|
||||
reason: 'Lock backend error',
|
||||
detail: `Lock backend error (${backendCategory}). Retry ${lockCommand} after verifying relay connectivity, relay URLs, and DNS/network status.`,
|
||||
detail: `Lock backend error (${backendCategory}) after ${lockAttempt.attempts} attempt(s). Retry ${lockCommand} after verifying relay connectivity, relay URLs, and DNS/network status.`,
|
||||
metadata: {
|
||||
lock_attempts_total: lockAttempt.attempts,
|
||||
lock_backoff_schedule_ms: backoffSchedule || '(none)',
|
||||
backend_category: backendCategory,
|
||||
lock_command: lockCommand,
|
||||
lock_stderr_excerpt: stderrExcerpt || '(empty)',
|
||||
|
||||
@@ -32,7 +32,7 @@ async function runNode(scriptPath, args, { cwd, env }) {
|
||||
});
|
||||
}
|
||||
|
||||
async function setupFixture({ memoryPolicy, lockShellBody = '' }) {
|
||||
async function setupFixture({ memoryPolicy, lockShellBody = '', roster = ['agent-a'] }) {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'scheduler-memory-policy-'));
|
||||
const scriptsDir = path.join(root, 'scripts', 'agent');
|
||||
const binDir = path.join(root, 'bin');
|
||||
@@ -49,8 +49,10 @@ async function setupFixture({ memoryPolicy, lockShellBody = '' }) {
|
||||
'utf8',
|
||||
);
|
||||
|
||||
await fs.writeFile(path.join(root, 'src', 'prompts', 'roster.json'), JSON.stringify({ daily: ['agent-a'] }, null, 2));
|
||||
await fs.writeFile(path.join(root, 'src', 'prompts', 'daily', 'agent-a.md'), '# agent-a\n', 'utf8');
|
||||
await fs.writeFile(path.join(root, 'src', 'prompts', 'roster.json'), JSON.stringify({ daily: roster }, null, 2));
|
||||
for (const agent of roster) {
|
||||
await fs.writeFile(path.join(root, 'src', 'prompts', 'daily', `${agent}.md`), `# ${agent}\n`, 'utf8');
|
||||
}
|
||||
|
||||
await fs.writeFile(
|
||||
path.join(root, 'torch-config.json'),
|
||||
@@ -179,3 +181,94 @@ test('records backend failure metadata when lock command exits with code 2', { c
|
||||
assert.match(failedBody, /lock_stdout_excerpt: 'websocket: connection refused'/);
|
||||
assert.match(failedBody, /Retry AGENT_PLATFORM=codex npm run lock:lock -- --agent agent-a --cadence daily/);
|
||||
});
|
||||
|
||||
test('retries lock acquisition for backend error and succeeds on a later attempt', { concurrency: false }, async () => {
|
||||
const fixture = await setupFixture({
|
||||
memoryPolicy: { mode: 'optional' },
|
||||
lockShellBody: ` count_file="$PWD/.lock-count"
|
||||
count=0
|
||||
if [[ -f "$count_file" ]]; then
|
||||
count="$(cat "$count_file")"
|
||||
fi
|
||||
count=$((count + 1))
|
||||
echo "$count" > "$count_file"
|
||||
if [[ "$count" -lt 2 ]]; then
|
||||
echo 'relay query timeout' >&2
|
||||
exit 2
|
||||
fi
|
||||
:`,
|
||||
});
|
||||
|
||||
const result = await runNode(fixture.scriptPath, ['--cadence', 'daily'], {
|
||||
cwd: fixture.root,
|
||||
env: {
|
||||
...fixture.env,
|
||||
SCHEDULER_LOCK_MAX_RETRIES: '2',
|
||||
SCHEDULER_LOCK_BACKOFF_MS: '0',
|
||||
SCHEDULER_LOCK_JITTER_MS: '0',
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(result.code, 0, `stdout: ${result.stdout}\nstderr: ${result.stderr}`);
|
||||
assert.match(result.stdout, /"event":"scheduler.lock.retry"/);
|
||||
const logs = await fs.readdir(path.join(fixture.root, 'task-logs', 'daily'));
|
||||
assert.equal(logs.filter((name) => name.endsWith('__failed.md')).length, 0);
|
||||
});
|
||||
|
||||
test('stops retrying after configured lock backend retries are exhausted', { concurrency: false }, async () => {
|
||||
const fixture = await setupFixture({
|
||||
memoryPolicy: { mode: 'optional' },
|
||||
lockShellBody: ` echo 'publish failed to all relays' >&2
|
||||
exit 2`,
|
||||
});
|
||||
|
||||
const result = await runNode(fixture.scriptPath, ['--cadence', 'daily'], {
|
||||
cwd: fixture.root,
|
||||
env: {
|
||||
...fixture.env,
|
||||
SCHEDULER_LOCK_MAX_RETRIES: '1',
|
||||
SCHEDULER_LOCK_BACKOFF_MS: '0',
|
||||
SCHEDULER_LOCK_JITTER_MS: '0',
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(result.code, 2, `stdout: ${result.stdout}\nstderr: ${result.stderr}`);
|
||||
const logs = await fs.readdir(path.join(fixture.root, 'task-logs', 'daily'));
|
||||
const failedLog = logs.find((name) => name.endsWith('__failed.md'));
|
||||
assert.ok(failedLog);
|
||||
const failedBody = await fs.readFile(path.join(fixture.root, 'task-logs', 'daily', failedLog), 'utf8');
|
||||
|
||||
assert.match(failedBody, /lock_attempts_total: '2'/);
|
||||
assert.match(failedBody, /lock_backoff_schedule_ms: '0'/);
|
||||
assert.match(failedBody, /backend_category: 'publish failed to all relays'/);
|
||||
});
|
||||
|
||||
test('does not use backend retry flow for exit code 3 lock conflicts', { concurrency: false }, async () => {
|
||||
const fixture = await setupFixture({
|
||||
memoryPolicy: { mode: 'optional' },
|
||||
lockShellBody: ` count_file="$PWD/.lock-count"
|
||||
count=0
|
||||
if [[ -f "$count_file" ]]; then
|
||||
count="$(cat "$count_file")"
|
||||
fi
|
||||
count=$((count + 1))
|
||||
echo "$count" > "$count_file"
|
||||
if [[ "$count" -eq 1 ]]; then
|
||||
exit 3
|
||||
fi
|
||||
:`,
|
||||
});
|
||||
|
||||
const result = await runNode(fixture.scriptPath, ['--cadence', 'daily'], {
|
||||
cwd: fixture.root,
|
||||
env: {
|
||||
...fixture.env,
|
||||
SCHEDULER_LOCK_MAX_RETRIES: '3',
|
||||
SCHEDULER_LOCK_BACKOFF_MS: '0',
|
||||
SCHEDULER_LOCK_JITTER_MS: '0',
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(result.code, 0, `stdout: ${result.stdout}\nstderr: ${result.stderr}`);
|
||||
assert.doesNotMatch(result.stdout, /"event":"scheduler.lock.retry"/);
|
||||
});
|
||||
@@ -129,7 +129,7 @@ describe('Scheduler cycle ordering guarantees', () => {
|
||||
|
||||
const order = [
|
||||
"const checkResult = await runCommand('npm', ['run', `lock:check:${cadence}`]);",
|
||||
"const lockResult = await runCommand(",
|
||||
"const lockAttempt = await acquireLockWithRetry({",
|
||||
"if (schedulerConfig.handoffCommand)",
|
||||
"const artifactCheck = await runCommand('node', [",
|
||||
"for (const validation of schedulerConfig.validationCommands)",
|
||||
|
||||
Reference in new issue
Block a user