From 46198a500e648a38bcd6e427d34af076ffeab571 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Sat, 14 Feb 2026 19:30:00 -0500 Subject: [PATCH] Improve lock publish diagnostics and scheduler failure summaries --- scripts/agent/run-scheduler-cycle.mjs | 57 +++++++++- src/lock-ops.mjs | 107 +++++++++++++++++- test/lock-ops.test.mjs | 92 ++++++++++++--- ...run-scheduler-cycle-memory-policy.test.mjs | 3 + 4 files changed, 238 insertions(+), 21 deletions(-) diff --git a/scripts/agent/run-scheduler-cycle.mjs b/scripts/agent/run-scheduler-cycle.mjs index f0e0a29..6bfe19b 100644 --- a/scripts/agent/run-scheduler-cycle.mjs +++ b/scripts/agent/run-scheduler-cycle.mjs @@ -50,6 +50,23 @@ function parseJsonFromOutput(text) { return null; } +function parseJsonEventsFromOutput(text) { + const events = []; + for (const line of String(text || '').split(/\r?\n/)) { + const trimmed = line.trim(); + if (!trimmed.startsWith('{')) continue; + try { + const parsed = JSON.parse(trimmed); + if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) { + events.push(parsed); + } + } catch { + // Ignore non-JSON lines. + } + } + return events; +} + async function runCommand(command, args = [], options = {}) { return new Promise((resolve) => { const child = spawn(command, args, { @@ -428,6 +445,7 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr const lockCommandArgs = ['run', 'lock:lock', '--', '--agent', selectedAgent, '--cadence', cadence]; const backoffScheduleMs = []; let attempts = 0; + const correlationId = idempotencyKey || `${cadence}:${selectedAgent}:${getRunDateKey()}`; while (true) { attempts += 1; @@ -435,11 +453,13 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr env: { AGENT_PLATFORM: platform, ...(idempotencyKey ? { SCHEDULER_LOCK_IDEMPOTENCY_KEY: idempotencyKey } : {}), + SCHEDULER_LOCK_CORRELATION_ID: correlationId, + SCHEDULER_LOCK_ATTEMPT_ID: String(attempts), }, }); if (result.code !== 2) { - return { result, attempts, backoffScheduleMs }; + return { result, attempts, backoffScheduleMs, correlationId }; } if (attempts > lockRetry.maxRetries) { @@ -447,6 +467,7 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr result, attempts, backoffScheduleMs, + correlationId, finalBackendCategory: classifyLockBackendError(`${result.stderr}\n${result.stdout}`), }; } @@ -463,6 +484,7 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr attempt: attempts, max_retries: lockRetry.maxRetries, delay_ms: delayMs, + correlation_id: correlationId, selected_agent: selectedAgent, cadence, })); @@ -473,6 +495,34 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr } } +function summarizeLockFailureReasons(outputText) { + const events = parseJsonEventsFromOutput(outputText); + const quorumFailure = [...events].reverse().find((entry) => entry.event === 'lock_publish_quorum_failed'); + if (quorumFailure && quorumFailure.reasonDistribution && typeof quorumFailure.reasonDistribution === 'object') { + return { + reasonDistribution: quorumFailure.reasonDistribution, + attemptId: quorumFailure.attemptId || null, + correlationId: quorumFailure.correlationId || null, + totalElapsedMs: quorumFailure.totalElapsedMs ?? null, + }; + } + + const distribution = {}; + for (const entry of events) { + if (entry.event !== 'lock_publish_failure') continue; + const reason = typeof entry.reason === 'string' && entry.reason.trim() + ? entry.reason.trim() + : 'unknown'; + distribution[reason] = (distribution[reason] || 0) + 1; + } + return { + reasonDistribution: distribution, + attemptId: null, + correlationId: null, + totalElapsedMs: null, + }; +} + function toYamlScalar(value) { const str = String(value ?? ''); return `'${str.replace(/'/g, "''")}'`; @@ -622,6 +672,7 @@ async function main() { if (lockResult.code === 2) { const combinedLockOutput = `${lockResult.stderr}\n${lockResult.stdout}`; + const diagnosticsSummary = summarizeLockFailureReasons(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); @@ -678,6 +729,10 @@ async function main() { failure_class: 'backend_unavailable', lock_attempts_total: lockAttempt.attempts, lock_backoff_schedule_ms: backoffSchedule || '(none)', + lock_correlation_id: diagnosticsSummary.correlationId || lockAttempt.correlationId, + lock_attempt_id: diagnosticsSummary.attemptId || String(lockAttempt.attempts), + lock_total_retry_timeline_ms: diagnosticsSummary.totalElapsedMs, + lock_failure_reason_distribution: JSON.stringify(diagnosticsSummary.reasonDistribution), backend_category: backendCategory, deferral_attempt_count: deferralAttemptCount, deferral_first_failure_timestamp: firstFailureAt, diff --git a/src/lock-ops.mjs b/src/lock-ops.mjs index 112be88..0f89217 100644 --- a/src/lock-ops.mjs +++ b/src/lock-ops.mjs @@ -69,6 +69,39 @@ function relayListLabel(relays) { function classifyPublishError(message) { const normalized = String(message || '').toLowerCase(); + if (normalized.includes('publish timed out after') || normalized.includes('publish timeout')) { + return 'publish_timeout'; + } + if ( + normalized.includes('enotfound') + || normalized.includes('eai_again') + || normalized.includes('getaddrinfo') + || (normalized.includes('dns') && normalized.includes('websocket')) + ) { + return 'dns_resolution'; + } + if ( + normalized.includes('connect etimedout') + || normalized.includes('tcp connect timed out') + || normalized.includes('connect timeout') + ) { + return 'tcp_connect_timeout'; + } + if ( + normalized.includes('tls') + || normalized.includes('ssl') + || normalized.includes('certificate') + || normalized.includes('handshake') + ) { + return 'tls_handshake'; + } + if ( + normalized.includes('websocket') + || normalized.includes('bad response') + || normalized.includes('unexpected server response') + ) { + return 'websocket_open_failure'; + } if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) { return 'network_timeout'; } @@ -89,7 +122,16 @@ function classifyPublishError(message) { } function isTransientPublishCategory(category) { - return category === 'network_timeout' || category === 'connection_reset' || category === 'relay_unavailable'; + return [ + 'publish_timeout', + 'dns_resolution', + 'tcp_connect_timeout', + 'tls_handshake', + 'websocket_open_failure', + 'network_timeout', + 'connection_reset', + 'relay_unavailable', + ].includes(category); } function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = Math.random) { @@ -366,8 +408,12 @@ export async function publishLock(relays, event, deps = {}) { randomFn = Math.random, telemetryLogger = console.error, healthLogger = console.error, + diagnostics = {}, } = deps; + const correlationId = diagnostics.correlationId || process.env.SCHEDULER_LOCK_CORRELATION_ID || 'none'; + const attemptId = diagnostics.attemptId || process.env.SCHEDULER_LOCK_ATTEMPT_ID || '1'; + const pool = poolFactory(); const publishTimeoutMs = getPublishTimeoutMsFn(); const minSuccesses = getMinSuccessfulRelayPublishesFn(); @@ -423,11 +469,32 @@ export async function publishLock(relays, event, deps = {}) { try { maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:periodic'); let lastAttemptState = null; + const retryTimeline = []; + const overallStartedAt = Date.now(); for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) { const startedAtMs = Date.now(); lastAttemptState = await publishOnce(); + const elapsedMs = Date.now() - startedAtMs; + retryTimeline.push({ + publishAttempt: attemptNumber, + successCount: lastAttemptState.successCount, + relayAttemptedCount: lastAttemptState.attempted.size, + elapsedMs, + }); if (lastAttemptState.successCount >= minSuccesses) { + telemetryLogger(JSON.stringify({ + event: 'lock_publish_quorum_met', + correlationId, + attemptId, + publishAttempt: attemptNumber, + successCount: lastAttemptState.successCount, + relayAttemptedCount: lastAttemptState.attempted.size, + requiredSuccesses: minSuccesses, + timeoutMs: publishTimeoutMs, + retryTimeline, + totalElapsedMs: Date.now() - overallStartedAt, + })); console.error( ` Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays ` + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`, @@ -443,13 +510,14 @@ export async function publishLock(relays, event, deps = {}) { break; } - const elapsedMs = Date.now() - startedAtMs; const nextDelayMs = calculateBackoffDelayMs(attemptNumber, retryBaseDelayMs, retryCapDelayMs, randomFn); for (const failure of lastAttemptState.failures) { if (!isTransientPublishCategory(failure.category)) continue; telemetryLogger(JSON.stringify({ event: 'lock_publish_retry', - attempt: attemptNumber, + correlationId, + attemptId, + publishAttempt: attemptNumber, relayUrl: failure.relay, errorCategory: failure.category, elapsedMs, @@ -460,13 +528,42 @@ export async function publishLock(relays, event, deps = {}) { await sleepFn(nextDelayMs); } + const reasonDistribution = {}; const failureLines = lastAttemptState.failures - .map((result) => `${result.relay} (${result.phase}, category=${result.category}): ${result.message}`); + .map((result) => { + reasonDistribution[result.category] = (reasonDistribution[result.category] || 0) + 1; + telemetryLogger(JSON.stringify({ + event: 'lock_publish_failure', + correlationId, + attemptId, + relayUrl: result.relay, + phase: result.phase, + reason: result.category, + message: result.message, + })); + return `${result.relay} (${result.phase}, reason=${result.category}): ${result.message}`; + }); + + telemetryLogger(JSON.stringify({ + event: 'lock_publish_quorum_failed', + correlationId, + attemptId, + successCount: lastAttemptState.successCount, + relayAttemptedCount: lastAttemptState.attempted.size, + requiredSuccesses: minSuccesses, + timeoutMs: publishTimeoutMs, + attempts: maxAttempts, + reasonDistribution, + retryTimeline, + totalElapsedMs: Date.now() - overallStartedAt, + })); maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:failure', true); throw new Error( `Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful ` - + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts})\n ${failureLines.join('\n ')}`, + + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts}, attempt_id=${attemptId}, correlation_id=${correlationId}, total_retry_timeline_ms=${Date.now() - overallStartedAt})\n` + + ` retry timeline: ${retryTimeline.map((item) => `#${item.publishAttempt}:${item.elapsedMs}ms`).join(', ')}\n` + + ` ${failureLines.join('\n ')}`, ); } finally { pool.close(allRelays); diff --git a/test/lock-ops.test.mjs b/test/lock-ops.test.mjs index 50d3d73..2064ab6 100644 --- a/test/lock-ops.test.mjs +++ b/test/lock-ops.test.mjs @@ -109,30 +109,47 @@ describe('lock-ops', () => { telemetry.push(JSON.parse(line)); }, healthLogger: () => {}, + diagnostics: { + attemptId: 'attempt-1', + correlationId: 'corr-1', + }, }); assert.strictEqual(result, event); assert.strictEqual(publishCalls, 2); assert.deepStrictEqual(sleepCalls, [250]); - assert.strictEqual(telemetry.length, 2); - assert.deepStrictEqual(telemetry[0], { + + const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); + assert.strictEqual(retryEvents.length, 2); + assert.deepStrictEqual(retryEvents[0], { event: 'lock_publish_retry', - attempt: 1, + correlationId: 'corr-1', + attemptId: 'attempt-1', + publishAttempt: 1, relayUrl: 'wss://primary-a', errorCategory: 'connection_reset', - elapsedMs: telemetry[0].elapsedMs, + elapsedMs: retryEvents[0].elapsedMs, nextDelayMs: 250, }); - assert.deepStrictEqual(telemetry[1], { + assert.deepStrictEqual(retryEvents[1], { event: 'lock_publish_retry', - attempt: 1, + correlationId: 'corr-1', + attemptId: 'attempt-1', + publishAttempt: 1, relayUrl: 'wss://primary-b', errorCategory: 'connection_reset', - elapsedMs: telemetry[1].elapsedMs, + elapsedMs: retryEvents[1].elapsedMs, nextDelayMs: 250, }); - assert.ok(Number.isInteger(telemetry[0].elapsedMs)); - assert.ok(Number.isInteger(telemetry[1].elapsedMs)); + assert.ok(Number.isInteger(retryEvents[0].elapsedMs)); + assert.ok(Number.isInteger(retryEvents[1].elapsedMs)); + + const metEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met'); + assert.ok(metEvent); + assert.equal(metEvent.correlationId, 'corr-1'); + assert.equal(metEvent.attemptId, 'attempt-1'); + assert.equal(metEvent.publishAttempt, 2); + assert.ok(Array.isArray(metEvent.retryTimeline)); }); it('fails after retry budget is exhausted for persistent transient failures', async () => { @@ -162,7 +179,7 @@ describe('lock-ops', () => { telemetryLogger: () => {}, healthLogger: () => {}, }), - /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=3\)/, + /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=3,/, ); assert.strictEqual(publishCalls, 3); @@ -190,17 +207,63 @@ describe('lock-ops', () => { sleepFn: async () => { throw new Error('should not sleep for permanent failures'); }, - telemetryLogger: () => { - throw new Error('should not emit retry telemetry for permanent failures'); - }, + telemetryLogger: () => {}, healthLogger: () => {}, }), - /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4\)/, + /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4,/, ); assert.strictEqual(publishCalls, 1); }); + it('includes per-relay reason categories and retry metadata in failure diagnostics', async () => { + const telemetry = []; + const mockPool = { + publish: (relays) => relays.map((relay) => { + if (relay.includes('dns')) return Promise.reject(new Error('getaddrinfo ENOTFOUND relay.example')); + if (relay.includes('tcp')) return Promise.reject(new Error('connect ETIMEDOUT 1.2.3.4:443')); + if (relay.includes('tls')) return Promise.reject(new Error('TLS handshake failure')); + if (relay.includes('ws')) return Promise.reject(new Error('websocket: unexpected server response')); + return Promise.reject(new Error('[publish:primary] Publish timed out after 200ms')); + }), + close: () => {}, + }; + + await assert.rejects( + () => publishLock(['wss://dns', 'wss://tcp', 'wss://tls', 'wss://ws', 'relay-timeout'], { id: 'evt-diag' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + telemetryLogger: (line) => telemetry.push(JSON.parse(line)), + healthLogger: () => {}, + diagnostics: { attemptId: 'attempt-42', correlationId: 'corr-42' }, + }), + ); + + const summaryEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); + assert.ok(summaryEvent); + const failureEvents = telemetry.filter((entry) => entry.event === 'lock_publish_failure'); + const failureText = failureEvents.map((entry) => `${entry.relayUrl}:${entry.reason}`).join('\n'); + + assert.match(failureText, /wss:\/\/dns:dns_resolution/); + assert.match(failureText, /wss:\/\/tcp:tcp_connect_timeout/); + assert.match(failureText, /wss:\/\/tls:tls_handshake/); + assert.match(failureText, /wss:\/\/ws:websocket_open_failure/); + assert.match(failureText, /relay-timeout:publish_timeout/); + + + assert.deepStrictEqual(summaryEvent.reasonDistribution, { + dns_resolution: 1, + tcp_connect_timeout: 1, + tls_handshake: 1, + websocket_open_failure: 1, + publish_timeout: 1, + }); + }); + it('returns success when mixed relay outcomes still satisfy quorum', async () => { let publishCalls = 0; const mockPool = { @@ -269,7 +332,6 @@ describe('lock-ops', () => { const event = { id: 'evt-health' }; - // Warm up failure history so bad relay enters quarantine. for (let i = 0; i < 3; i += 1) { await publishLock(['wss://bad', 'wss://healthy'], event, { poolFactory: () => mockPool, diff --git a/test/run-scheduler-cycle-memory-policy.test.mjs b/test/run-scheduler-cycle-memory-policy.test.mjs index 2cf78df..fb8876e 100644 --- a/test/run-scheduler-cycle-memory-policy.test.mjs +++ b/test/run-scheduler-cycle-memory-policy.test.mjs @@ -193,6 +193,9 @@ test('records backend failure metadata when lock command exits with code 2', { c assert.match(failedBody, /lock_command: 'AGENT_PLATFORM=codex npm run lock:lock -- --agent agent-a --cadence daily'/); assert.match(failedBody, /lock_stderr_excerpt: 'publish failed to all relays token=\[REDACTED\] SECRET_KEY=\[REDACTED\]'/); assert.match(failedBody, /lock_stdout_excerpt: 'websocket: connection refused'/); + assert.match(failedBody, /lock_failure_reason_distribution: '\{\}'/); + assert.match(failedBody, /lock_attempt_id: '3'/); + assert.match(failedBody, /lock_correlation_id: 'daily:agent-a:/); assert.match(failedBody, /Retry AGENT_PLATFORM=codex npm run lock:lock -- --agent agent-a --cadence daily/); assert.match(failedBody, /platform: 'codex'/); });