diff --git a/src/apps/ops-sentinel.mjs b/src/apps/ops-sentinel.mjs index 2198dc5..2e37038 100644 --- a/src/apps/ops-sentinel.mjs +++ b/src/apps/ops-sentinel.mjs @@ -14,6 +14,7 @@ import { import { listDashboardServices } from '../core/operator-dashboard.mjs'; import { ageMs, + buildExecutorRuntimeAlerts, buildMakerCompetitivenessRuntimeAlerts, buildRuntimeAlert, createRuntimeHealthThresholds, @@ -433,6 +434,12 @@ function buildDeterministicRuntimeAlerts({ servicesByName, now, previousRuntimeE }, })); } + alerts.push(...buildExecutorRuntimeAlerts({ + executor, + thresholds, + now, + pair: config.activePair, + })); const writer = servicesByName['history-writer']; const writerState = writer?.state || {}; diff --git a/src/apps/trade-executor.mjs b/src/apps/trade-executor.mjs index a554c00..69a9554 100644 --- a/src/apps/trade-executor.mjs +++ b/src/apps/trade-executor.mjs @@ -5,6 +5,7 @@ import { createConsumer } from '../bus/kafka/consumer.mjs'; import { createProducer } from '../bus/kafka/producer.mjs'; import { createArmedStateStore } from '../core/armed-state-store.mjs'; import { startControlApi } from '../core/control-api.mjs'; +import { createExecutorQueueDelayTracker } from '../core/executor-queue-delay.mjs'; import { classifyExecuteCommandExpiry } from '../core/executor-command-expiry.mjs'; import { buildEventEnvelope, parseEventMessage } from '../core/event-envelope.mjs'; import { createExecutorStateStore } from '../core/executor-state-store.mjs'; @@ -107,6 +108,7 @@ const relayClient = await startSolverRelayWs({ }, }); const stateStore = createExecutorStateStore({ stateDir: config.executorStateDir }); +const executorQueueDelayTracker = createExecutorQueueDelayTracker(); const armedStateStore = createArmedStateStore({ stateDir: config.executorStateDir, fileName: 'trade-executor-control.json', @@ -124,6 +126,9 @@ const state = { last_error: null, in_flight_count: 0, submitted_count: 0, + salt_unavailable_reject_count: 0, + last_salt_unavailable_at: null, + last_salt_unavailable_state: null, request_creation: { last_preflight: null, last_submission_result: null, @@ -189,6 +194,7 @@ await consumer.run({ async function handleCommand(event) { const payload = event.payload; const timing = startExecutorTiming(event); + recordExecutorQueueDelay(timing, payload); state.last_command = payload; const existing = stateStore.get(payload.command_id); @@ -254,14 +260,32 @@ async function handleCommand(event) { try { const saltStartMs = performance.now(); let currentSaltHex; - try { - const salt = await verifierSaltCache.getFreshSalt(); - currentSaltHex = salt.currentSaltHex; - timing.current_salt_source = salt.source; - timing.current_salt_age_ms = roundTimingMs(salt.ageMs); - } finally { - recordExecutorTiming(timing, 'current_salt_ms', saltStartMs); + const salt = verifierSaltCache.getCachedFreshSalt(); + recordExecutorTiming(timing, 'current_salt_ms', saltStartMs); + timing.current_salt_source = salt.source; + timing.current_salt_age_ms = roundTimingMs(salt.ageMs); + timing.current_salt_unavailable_reason = salt.available ? null : salt.reason; + timing.verifier_salt_cache_fresh = salt.state?.salt_fresh ?? null; + if (!salt.available) { + recordSaltUnavailable(salt); + stateStore.markFailed(payload.command_id, { + quote_id: payload.quote_id, + error: { + name: 'VerifierSaltUnavailable', + message: salt.reason || 'fresh verifier salt unavailable', + }, + }); + await publishResult(payload, withExecutorTiming({ + status: 'rejected', + result_code: 'verifier_salt_unavailable', + failure_category: 'salt_unavailable', + note: 'fresh verifier salt unavailable; command was not signed or relayed', + verifier_salt_unavailable_reason: salt.reason || 'unavailable', + verifier_salt_cache: salt.state, + }, timing)); + return; } + currentSaltHex = salt.currentSaltHex; const signStartMs = performance.now(); let submission; @@ -336,6 +360,7 @@ function startExecutorTiming(event) { received_at: receivedAt.toISOString(), started_at_ms: performance.now(), maker_timing: makerTiming, + command_to_executor_ms: makerTiming.command_to_executor_ms, quote_age_at_executor_receipt_ms: makerTiming.quote_age_at_executor_receipt_ms, command_event_age_ms: Number.isFinite(commandEventAtMs) ? roundTimingMs(receivedAt.getTime() - commandEventAtMs) @@ -374,11 +399,14 @@ function finishExecutorTiming(timing) { executor_result_at: executorResultAt, relay_result_at: relayResultAt, command_event_age_ms: timing.command_event_age_ms, + command_to_executor_ms: timing.command_to_executor_ms ?? null, quote_age_at_executor_receipt_ms: timing.quote_age_at_executor_receipt_ms ?? null, quote_age_at_relay_result_ms: makerTiming.quote_age_at_relay_result_ms ?? null, current_salt_ms: timing.current_salt_ms ?? null, current_salt_source: timing.current_salt_source ?? null, current_salt_age_ms: timing.current_salt_age_ms ?? null, + current_salt_unavailable_reason: timing.current_salt_unavailable_reason ?? null, + verifier_salt_cache_fresh: timing.verifier_salt_cache_fresh ?? null, sign_ms: timing.sign_ms ?? null, relay_response_ms: timing.relay_response_ms ?? null, executor_total_ms: roundTimingMs(performance.now() - timing.started_at_ms), @@ -391,6 +419,22 @@ function roundTimingMs(value) { return Number.isFinite(number) ? Math.round(number * 1000) / 1000 : null; } +function recordExecutorQueueDelay(timing, payload) { + executorQueueDelayTracker.record({ + commandToExecutorMs: timing.command_to_executor_ms, + commandId: payload?.command_id || null, + quoteId: payload?.quote_id || null, + pair: payload?.pair || null, + at: timing.received_at, + }); +} + +function recordSaltUnavailable(salt) { + state.salt_unavailable_reject_count += 1; + state.last_salt_unavailable_at = new Date().toISOString(); + state.last_salt_unavailable_state = salt.state || null; +} + async function publishResult(command, extraPayload) { const event = buildEventEnvelope({ source: 'trade-executor', @@ -439,6 +483,7 @@ const controlApi = startControlApi({ signer_registered: signerRegistered, relay: relayClient.getState(), verifier_salt_cache: verifierSaltCache.getState(), + executor_queue_delay: executorQueueDelayTracker.getState(), trading_config: tradingConfigStore.getState(), ...state, durable_control_state: armedStateStore.getState(), @@ -450,6 +495,8 @@ const controlApi = startControlApi({ getHealth() { const relay = relayClient.getState(); const freshnessAgeMs = ageMs(relay.last_message_at); + const saltState = verifierSaltCache.getState(); + const queueDelay = executorQueueDelayTracker.getState(); return { ok: relay.connected && tradingConfigStore.getState().ok === true @@ -459,6 +506,14 @@ const controlApi = startControlApi({ trading_config_block_reason: tradingConfigStore.getState().block_reason, relay_last_message_at: relay.last_message_at, relay_freshness_age_ms: freshnessAgeMs, + verifier_salt_cache: saltState, + verifier_salt_fresh: saltState.salt_fresh, + verifier_salt_age_ms: saltState.salt_age_ms, + verifier_salt_refresh_in_flight: saltState.refresh_in_flight, + verifier_salt_last_refresh_error: saltState.last_refresh_error, + executor_queue_delay: queueDelay, + executor_queue_delay_status: queueDelay.status, + executor_queue_delay_warning_reason: queueDelay.warning_reason, paused: state.paused, armed: state.armed, reason: diff --git a/src/core/executor-queue-delay.mjs b/src/core/executor-queue-delay.mjs new file mode 100644 index 0000000..e9aaafa --- /dev/null +++ b/src/core/executor-queue-delay.mjs @@ -0,0 +1,127 @@ +export const DEFAULT_EXECUTOR_QUEUE_DELAY_WARNING_MS = 1_000; +export const DEFAULT_EXECUTOR_QUEUE_DELAY_CRITICAL_MS = 5_000; +export const DEFAULT_EXECUTOR_QUEUE_DELAY_SAMPLE_LIMIT = 50; + +export function createExecutorQueueDelayTracker({ + sampleLimit = DEFAULT_EXECUTOR_QUEUE_DELAY_SAMPLE_LIMIT, + warningMs = DEFAULT_EXECUTOR_QUEUE_DELAY_WARNING_MS, + criticalMs = DEFAULT_EXECUTOR_QUEUE_DELAY_CRITICAL_MS, + now = () => Date.now(), +} = {}) { + let samples = []; + + function record({ + commandToExecutorMs, + commandId = null, + quoteId = null, + pair = null, + at = now(), + } = {}) { + const value = roundTimingMs(commandToExecutorMs); + if (value == null) return getState(); + + samples.push({ + at: toIsoTimestamp(at), + command_id: commandId, + quote_id: quoteId, + pair, + command_to_executor_ms: value, + }); + samples = samples.slice(-positiveInteger(sampleLimit, DEFAULT_EXECUTOR_QUEUE_DELAY_SAMPLE_LIMIT)); + return getState(); + } + + function getState() { + return summarizeExecutorQueueDelaySamples(samples, { + warningMs, + criticalMs, + sampleLimit, + }); + } + + return { + record, + getState, + }; +} + +export function summarizeExecutorQueueDelaySamples(samples = [], { + warningMs = DEFAULT_EXECUTOR_QUEUE_DELAY_WARNING_MS, + criticalMs = DEFAULT_EXECUTOR_QUEUE_DELAY_CRITICAL_MS, + sampleLimit = DEFAULT_EXECUTOR_QUEUE_DELAY_SAMPLE_LIMIT, +} = {}) { + const normalizedSamples = (samples || []) + .map((sample) => ({ + at: toIsoTimestamp(sample.at), + command_id: sample.command_id || null, + quote_id: sample.quote_id || null, + pair: sample.pair || null, + command_to_executor_ms: roundTimingMs(sample.command_to_executor_ms), + })) + .filter((sample) => sample.command_to_executor_ms != null) + .slice(-positiveInteger(sampleLimit, DEFAULT_EXECUTOR_QUEUE_DELAY_SAMPLE_LIMIT)); + const values = normalizedSamples + .map((sample) => sample.command_to_executor_ms) + .sort((left, right) => left - right); + const latest = normalizedSamples.at(-1) || null; + const maxMs = values.length ? values.at(-1) : null; + const status = maxMs == null + ? 'unknown' + : maxMs >= criticalMs + ? 'critical' + : maxMs >= warningMs + ? 'warning' + : 'healthy'; + const thresholdMs = status === 'critical' ? criticalMs : warningMs; + + return { + status, + warning: status === 'warning' || status === 'critical', + critical: status === 'critical', + warning_reason: status === 'warning' || status === 'critical' + ? `recent command_to_executor_ms max ${maxMs}ms exceeds ${thresholdMs}ms` + : null, + warning_after_ms: warningMs, + critical_after_ms: criticalMs, + sample_count: normalizedSamples.length, + latest_ms: latest?.command_to_executor_ms ?? null, + latest_at: latest?.at || null, + latest_command_id: latest?.command_id || null, + latest_quote_id: latest?.quote_id || null, + latest_pair: latest?.pair || null, + max_ms: maxMs, + p50_ms: percentile(values, 0.5), + p90_ms: percentile(values, 0.9), + p99_ms: percentile(values, 0.99), + recent_samples: normalizedSamples.slice(-10), + }; +} + +function percentile(sortedValues, ratio) { + if (!sortedValues.length) return null; + const index = Math.min( + sortedValues.length - 1, + Math.max(0, Math.ceil(sortedValues.length * ratio) - 1), + ); + return sortedValues[index]; +} + +function roundTimingMs(value) { + const number = Number(value); + return Number.isFinite(number) ? Math.round(number * 1000) / 1000 : null; +} + +function positiveInteger(value, fallback) { + const parsed = Number(value); + return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; +} + +function toIsoTimestamp(value) { + if (!value) return null; + if (typeof value === 'string') { + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? new Date(parsed).toISOString() : null; + } + const date = new Date(value); + return Number.isNaN(date.getTime()) ? null : date.toISOString(); +} diff --git a/src/core/operator-dashboard.mjs b/src/core/operator-dashboard.mjs index fe7ab03..d4af30c 100644 --- a/src/core/operator-dashboard.mjs +++ b/src/core/operator-dashboard.mjs @@ -2093,9 +2093,13 @@ function buildServiceSummary(service, state) { return { in_flight_count: state.in_flight_count ?? 0, submitted_count: state.submitted_count ?? state.completed_count ?? 0, + salt_unavailable_reject_count: state.salt_unavailable_reject_count ?? 0, + last_salt_unavailable_at: state.last_salt_unavailable_at || null, signer_registered: state.signer_registered ?? null, relay_connected: state.relay?.connected ?? null, relay_last_message_at: state.relay?.last_message_at || null, + verifier_salt_cache: state.verifier_salt_cache || null, + executor_queue_delay: state.executor_queue_delay || null, }; case 'operator-dashboard': return { diff --git a/src/core/runtime-health.mjs b/src/core/runtime-health.mjs index a98ebb6..a2199d3 100644 --- a/src/core/runtime-health.mjs +++ b/src/core/runtime-health.mjs @@ -1,4 +1,6 @@ export const SERVICE_HEALTH_LEVELS = ['healthy', 'warning', 'critical', 'offline', 'paused']; +export const DEFAULT_EXECUTOR_QUEUE_DELAY_WARNING_MS = 1_000; +export const DEFAULT_EXECUTOR_QUEUE_DELAY_CRITICAL_MS = 5_000; const HEALTH_RANK = { healthy: 0, @@ -27,6 +29,12 @@ export function createRuntimeHealthThresholds(config = {}) { anomalyQuoteRateCollapseRatio: Number(config.opsSentinelAnomalyQuoteRateCollapseRatio || 0.25), anomalyReconnectSpikeMultiplier: Number(config.opsSentinelAnomalyReconnectSpikeMultiplier || 2), containmentCooldownMs: Number(config.opsSentinelContainmentCooldownMs || 60_000), + executorQueueDelayWarningMs: Number( + config.executorQueueDelayWarningMs || DEFAULT_EXECUTOR_QUEUE_DELAY_WARNING_MS, + ), + executorQueueDelayCriticalMs: Number( + config.executorQueueDelayCriticalMs || DEFAULT_EXECUTOR_QUEUE_DELAY_CRITICAL_MS, + ), }; } @@ -129,6 +137,31 @@ export function deriveServiceHealth({ } } + if (service === 'trade-executor') { + const saltState = state.verifier_salt_cache || health.verifier_salt_cache || null; + const queueDelay = state.executor_queue_delay || health.executor_queue_delay || null; + + if (saltState?.configured && saltState.salt_fresh !== true) { + status = escalateHealth(status, 'warning'); + if (label === 'healthy') label = 'salt unavailable'; + reasons.push('verifier salt cache is stale or unavailable'); + } + if (saltState?.last_refresh_error) { + status = escalateHealth(status, 'warning'); + if (label === 'healthy') label = 'salt refresh failed'; + reasons.push('verifier salt refresh failed'); + } + if (queueDelay?.status === 'critical') { + status = escalateHealth(status, 'critical'); + label = 'queue delayed'; + reasons.push(queueDelay.warning_reason || 'executor queue delay is critical'); + } else if (queueDelay?.status === 'warning') { + status = escalateHealth(status, 'warning'); + if (label === 'healthy') label = 'queue delayed'; + reasons.push(queueDelay.warning_reason || 'executor queue delay is high'); + } + } + if (service === 'history-writer') { if (state.database_connectivity === false) { status = escalateHealth(status, 'critical'); @@ -379,6 +412,90 @@ export function buildMakerCompetitivenessRuntimeAlerts({ }); } +export function buildExecutorRuntimeAlerts({ + executor, + thresholds = createRuntimeHealthThresholds(), + now = new Date().toISOString(), + pair = null, +} = {}) { + void now; + const alerts = []; + const state = executor?.state || {}; + const health = executor?.health || {}; + const saltState = state.verifier_salt_cache || health.verifier_salt_cache || null; + const queueDelay = state.executor_queue_delay || health.executor_queue_delay || null; + + if (saltState?.configured && saltState.salt_fresh !== true) { + alerts.push(buildRuntimeAlert({ + alert_code: 'trade_executor_verifier_salt_unavailable', + severity: 'warning', + reason: saltState.has_cached_salt + ? 'trade-executor verifier salt cache is stale' + : 'trade-executor verifier salt cache is unavailable', + service_scope: 'trade-executor', + pair, + details: { + has_cached_salt: saltState.has_cached_salt ?? null, + salt_fresh: saltState.salt_fresh ?? saltState.fresh ?? null, + salt_age_ms: saltState.salt_age_ms ?? saltState.cached_age_ms ?? null, + max_age_ms: saltState.max_age_ms ?? null, + refresh_in_flight: saltState.refresh_in_flight ?? null, + last_refresh_started_at: saltState.last_refresh_started_at || null, + last_refresh_completed_at: saltState.last_refresh_completed_at || null, + last_refresh_error: saltState.last_refresh_error || null, + }, + })); + } + + if (saltState?.last_refresh_error) { + alerts.push(buildRuntimeAlert({ + alert_code: 'trade_executor_verifier_salt_refresh_failed', + severity: 'warning', + reason: 'trade-executor verifier salt refresh failed', + service_scope: 'trade-executor', + pair, + details: { + last_refresh_error: saltState.last_refresh_error, + last_refresh_started_at: saltState.last_refresh_started_at || null, + last_refresh_completed_at: saltState.last_refresh_completed_at || null, + last_refresh_duration_ms: saltState.last_refresh_duration_ms ?? null, + refresh_in_flight: saltState.refresh_in_flight ?? null, + has_cached_salt: saltState.has_cached_salt ?? null, + salt_fresh: saltState.salt_fresh ?? saltState.fresh ?? null, + }, + })); + } + + const queueMaxMs = Number(queueDelay?.max_ms); + if (Number.isFinite(queueMaxMs) && queueMaxMs >= thresholds.executorQueueDelayWarningMs) { + const critical = queueMaxMs >= thresholds.executorQueueDelayCriticalMs; + alerts.push(buildRuntimeAlert({ + alert_code: 'trade_executor_command_queue_delay_high', + severity: critical ? 'critical' : 'warning', + reason: `trade-executor command queue delay ${queueMaxMs}ms exceeds ${ + critical ? thresholds.executorQueueDelayCriticalMs : thresholds.executorQueueDelayWarningMs + }ms`, + service_scope: 'trade-executor', + pair, + details: { + status: queueDelay.status || (critical ? 'critical' : 'warning'), + latest_ms: queueDelay.latest_ms ?? null, + max_ms: queueDelay.max_ms ?? null, + p50_ms: queueDelay.p50_ms ?? null, + p90_ms: queueDelay.p90_ms ?? null, + p99_ms: queueDelay.p99_ms ?? null, + sample_count: queueDelay.sample_count ?? null, + warning_after_ms: thresholds.executorQueueDelayWarningMs, + critical_after_ms: thresholds.executorQueueDelayCriticalMs, + latest_command_id: queueDelay.latest_command_id || null, + latest_quote_id: queueDelay.latest_quote_id || null, + }, + })); + } + + return alerts; +} + export function shouldRaiseIngestPublishStale({ lastMatchingQuoteAt = null, lastPublishedAt = null, diff --git a/src/core/service-snapshot-summary.mjs b/src/core/service-snapshot-summary.mjs index 236f0d8..f88ed85 100644 --- a/src/core/service-snapshot-summary.mjs +++ b/src/core/service-snapshot-summary.mjs @@ -109,9 +109,14 @@ export function summarizeServiceState(service, state) { 'armed', 'last_error', 'relay', + 'verifier_salt_cache', + 'executor_queue_delay', 'last_result', 'last_quote_status', 'submitted_count', + 'salt_unavailable_reject_count', + 'last_salt_unavailable_at', + 'last_salt_unavailable_state', 'failed_count', 'blocked_count', ]); diff --git a/src/core/verifier-salt-cache.mjs b/src/core/verifier-salt-cache.mjs index 2de7e87..6f48a71 100644 --- a/src/core/verifier-salt-cache.mjs +++ b/src/core/verifier-salt-cache.mjs @@ -18,30 +18,47 @@ export function createVerifierSaltCache({ let cached = null; let refreshInFlight = null; let timer = null; + let lastRefreshStartedAtMs = null; + let lastRefreshStartedAt = null; + let lastRefreshCompletedAt = null; + let lastRefreshDurationMs = null; + let lastRefreshReason = null; + let refreshInFlightReason = null; let lastRefreshError = null; let lastRefreshErrorLoggedAtMs = 0; - async function refresh({ reason = 'manual' } = {}) { + async function refreshNow({ reason = 'manual' } = {}) { if (refreshInFlight) return refreshInFlight; + const startedAtMs = now(); + lastRefreshStartedAtMs = startedAtMs; + lastRefreshStartedAt = toIsoTimestamp(startedAtMs); + lastRefreshReason = reason; + refreshInFlightReason = reason; + refreshInFlight = Promise.resolve() .then(() => loadSalt()) .then((salt) => { + const refreshedAtMs = now(); cached = { currentSaltHex: normalizeSalt(salt), - refreshedAtMs: now(), + refreshedAtMs, + refreshedAt: toIsoTimestamp(refreshedAtMs), refreshedReason: reason, }; lastRefreshError = null; + recordRefreshCompleted(); return buildSaltResult(cached, 'refresh'); }) .catch((error) => { logRefreshError(error, reason); lastRefreshError = serializeSaltError(error); + recordRefreshCompleted(); throw error; }) .finally(() => { refreshInFlight = null; + refreshInFlightReason = null; }); return refreshInFlight; @@ -50,18 +67,55 @@ export function createVerifierSaltCache({ async function getFreshSalt() { if (isFresh(cached)) return buildSaltResult(cached, 'cache'); - await refresh({ reason: cached ? 'stale_cache' : 'empty_cache' }); + await refreshNow({ reason: cached ? 'stale_cache' : 'empty_cache' }); if (!isFresh(cached)) { throw new Error('verifier salt cache refresh did not produce a fresh salt'); } return buildSaltResult(cached, 'refresh'); } + function getCachedFreshSalt(options = {}) { + const readNowMs = options.now == null ? now() : timestampMs(options.now); + const readMaxAgeMs = positiveNumber(options.maxAgeMs, maxAgeMs); + const state = getState({ now: readNowMs, maxAgeMs: readMaxAgeMs }); + + if (!cached) { + return { + available: false, + source: 'unavailable', + reason: 'empty_cache', + ageMs: null, + state, + }; + } + + const ageMs = readNowMs - cached.refreshedAtMs; + if (!isFresh(cached, { nowMs: readNowMs, maxAgeMs: readMaxAgeMs })) { + return { + available: false, + source: 'unavailable', + reason: ageMs < 0 ? 'clock_skew' : 'stale_cache', + ageMs: Number.isFinite(ageMs) ? Math.max(0, ageMs) : null, + state, + }; + } + + return { + available: true, + currentSaltHex: cached.currentSaltHex, + source: 'cache', + ageMs: Math.max(0, ageMs), + refreshedAt: cached.refreshedAt, + refreshedReason: cached.refreshedReason, + state, + }; + } + function start() { if (timer) return; - void refresh({ reason: 'startup' }).catch(() => {}); + void refreshNow({ reason: 'startup' }).catch(() => {}); timer = setIntervalFn(() => { - void refresh({ reason: 'prefetch' }).catch(() => {}); + void refreshNow({ reason: 'prefetch' }).catch(() => {}); }, refreshIntervalMs); timer?.unref?.(); } @@ -72,25 +126,37 @@ export function createVerifierSaltCache({ timer = null; } - function getState() { - const ageMs = cached ? now() - cached.refreshedAtMs : null; + function getState(options = {}) { + const stateNowMs = options.now == null ? now() : timestampMs(options.now); + const stateMaxAgeMs = positiveNumber(options.maxAgeMs, maxAgeMs); + const ageMs = cached ? stateNowMs - cached.refreshedAtMs : null; + const fresh = isFresh(cached, { nowMs: stateNowMs, maxAgeMs: stateMaxAgeMs }); + const roundedAgeMs = Number.isFinite(ageMs) ? Math.max(0, Math.round(ageMs)) : null; return { configured: true, has_cached_salt: Boolean(cached), - fresh: isFresh(cached), - max_age_ms: maxAgeMs, + fresh, + salt_fresh: fresh, + salt_age_ms: roundedAgeMs, + max_age_ms: stateMaxAgeMs, refresh_interval_ms: refreshIntervalMs, - cached_age_ms: ageMs == null ? null : Math.max(0, Math.round(ageMs)), + cached_age_ms: roundedAgeMs, + cached_refreshed_at: cached?.refreshedAt || null, refreshed_reason: cached?.refreshedReason || null, refresh_in_flight: Boolean(refreshInFlight), + refresh_in_flight_reason: refreshInFlightReason, + last_refresh_started_at: lastRefreshStartedAt, + last_refresh_completed_at: lastRefreshCompletedAt, + last_refresh_duration_ms: lastRefreshDurationMs, + last_refresh_reason: lastRefreshReason, last_refresh_error: lastRefreshError, }; } - function isFresh(entry) { + function isFresh(entry, { nowMs = now(), maxAgeMs: readMaxAgeMs = maxAgeMs } = {}) { if (!entry) return false; - const ageMs = now() - entry.refreshedAtMs; - return ageMs >= 0 && ageMs <= maxAgeMs; + const ageMs = nowMs - entry.refreshedAtMs; + return Number.isFinite(ageMs) && ageMs >= 0 && ageMs <= readMaxAgeMs; } function buildSaltResult(entry, source) { @@ -102,6 +168,14 @@ export function createVerifierSaltCache({ }; } + function recordRefreshCompleted() { + const completedAtMs = now(); + lastRefreshCompletedAt = toIsoTimestamp(completedAtMs); + lastRefreshDurationMs = lastRefreshStartedAtMs == null + ? null + : Math.max(0, Math.round((completedAtMs - lastRefreshStartedAtMs) * 1000) / 1000); + } + function logRefreshError(error, reason) { if (!logger?.warn) return; const serialized = serializeSaltError(error); @@ -119,9 +193,11 @@ export function createVerifierSaltCache({ } return { + getCachedFreshSalt, getFreshSalt, getState, - refresh, + refresh: refreshNow, + refreshNow, start, stop, }; @@ -141,3 +217,20 @@ function serializeSaltError(error) { message: error?.message || String(error), }; } + +function positiveNumber(value, fallback) { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +function timestampMs(value) { + if (value == null) return Number.NaN; + if (typeof value === 'number') return value; + if (value instanceof Date) return value.getTime(); + return Date.parse(value); +} + +function toIsoTimestamp(value) { + const parsed = timestampMs(value); + return Number.isFinite(parsed) ? new Date(parsed).toISOString() : null; +} diff --git a/src/operator-dashboard/static/components/ServiceCard.jsx b/src/operator-dashboard/static/components/ServiceCard.jsx index c1967fd..1c09ec0 100644 --- a/src/operator-dashboard/static/components/ServiceCard.jsx +++ b/src/operator-dashboard/static/components/ServiceCard.jsx @@ -17,6 +17,9 @@ function useNow(intervalMs = 1000) { export default function ServiceCard({ service }) { const healthLabel = service.health_label || service.health_status || (service.reachable ? 'online' : 'offline'); const now = useNow(); + const summary = service.summary || {}; + const saltCache = summary.verifier_salt_cache || null; + const queueDelay = summary.executor_queue_delay || null; const freshnessAge = service.freshness_at ? formatAgeFromTimestamp(service.freshness_at, now) : formatAge(service.freshness_age_ms); @@ -42,9 +45,43 @@ export default function ServiceCard({ service }) { ) : null} ) : null} + {saltCache ? ( + <> +
{`Verifier salt ${formatSaltFreshness(saltCache)}`}
+
{`Salt age ${formatAge(saltCache.salt_age_ms ?? saltCache.cached_age_ms)}`}
+
{`Salt refresh ${saltCache.refresh_in_flight ? 'in flight' : 'idle'}`}
+
{`Salt refreshed ${formatTimestamp(saltCache.last_refresh_completed_at || saltCache.cached_refreshed_at)}`}
+
{`Salt refresh duration ${formatAge(saltCache.last_refresh_duration_ms)}`}
+ {saltCache.last_refresh_error ? ( +
{`Salt error ${formatError(saltCache.last_refresh_error)}`}
+ ) : null} +
{`Salt rejects ${summary.salt_unavailable_reject_count ?? 0}`}
+ + ) : null} + {queueDelay ? ( + <> +
{`Queue delay ${queueDelay.status || 'unknown'}`}
+
{`Queue latest ${formatAge(queueDelay.latest_ms)}`}
+
{`Queue p90 ${formatAge(queueDelay.p90_ms)}`}
+
{`Queue p99 ${formatAge(queueDelay.p99_ms)}`}
+
{`Queue max ${formatAge(queueDelay.max_ms)}`}
+ {queueDelay.warning_reason ?
{queueDelay.warning_reason}
: null} + + ) : null}
{service.base_url}
{service.last_error ?
{JSON.stringify(service.last_error)}
: null} ); } + +function formatSaltFreshness(saltCache) { + if (saltCache.salt_fresh === true || saltCache.fresh === true) return 'fresh'; + if (saltCache.has_cached_salt) return 'stale'; + return 'unavailable'; +} + +function formatError(error) { + if (!error) return 'Unavailable'; + return error.message || JSON.stringify(error); +} diff --git a/test/executor-queue-delay.test.mjs b/test/executor-queue-delay.test.mjs new file mode 100644 index 0000000..ec132f8 --- /dev/null +++ b/test/executor-queue-delay.test.mjs @@ -0,0 +1,36 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { createExecutorQueueDelayTracker } from '../src/core/executor-queue-delay.mjs'; + +test('executor queue delay tracker records recent percentiles and warning state', () => { + let nowMs = Date.parse('2026-06-12T10:00:00.000Z'); + const tracker = createExecutorQueueDelayTracker({ + sampleLimit: 5, + warningMs: 1_000, + criticalMs: 5_000, + now: () => nowMs, + }); + + for (const value of [10, 25, 100, 1_250, 7_500]) { + tracker.record({ + commandToExecutorMs: value, + commandId: `cmd-${value}`, + quoteId: `quote-${value}`, + pair: 'nbtc->eure', + }); + nowMs += 1_000; + } + + const state = tracker.getState(); + assert.equal(state.status, 'critical'); + assert.equal(state.warning, true); + assert.equal(state.critical, true); + assert.equal(state.sample_count, 5); + assert.equal(state.latest_ms, 7_500); + assert.equal(state.max_ms, 7_500); + assert.equal(state.p50_ms, 100); + assert.equal(state.p90_ms, 7_500); + assert.equal(state.p99_ms, 7_500); + assert.match(state.warning_reason, /command_to_executor_ms/); +}); diff --git a/test/operator-dashboard-ui-static.test.mjs b/test/operator-dashboard-ui-static.test.mjs index 06c4aa3..30153a6 100644 --- a/test/operator-dashboard-ui-static.test.mjs +++ b/test/operator-dashboard-ui-static.test.mjs @@ -98,6 +98,19 @@ test('dashboard freshness surfaces show age and exact timestamp evidence', () => assert.match(stylesSource, /@keyframes quote-row-flash-cell/); }); +test('service cards expose executor verifier salt and queue delay state', () => { + assert.match(serviceCardSource, /verifier_salt_cache/); + assert.match(serviceCardSource, /executor_queue_delay/); + assert.match(serviceCardSource, /Verifier salt/); + assert.match(serviceCardSource, /Salt age/); + assert.match(serviceCardSource, /Salt refresh/); + assert.match(serviceCardSource, /Salt error/); + assert.match(serviceCardSource, /Salt rejects/); + assert.match(serviceCardSource, /Queue delay/); + assert.match(serviceCardSource, /Queue p90/); + assert.match(serviceCardSource, /Queue p99/); +}); + test('mobile status bar uses normal document flow instead of sticky viewport positioning', () => { assert.match( stylesSource, diff --git a/test/operator-dashboard.test.mjs b/test/operator-dashboard.test.mjs index ff7afbc..df9ca56 100644 --- a/test/operator-dashboard.test.mjs +++ b/test/operator-dashboard.test.mjs @@ -2086,6 +2086,65 @@ test('dashboard does not pause relay services for unrelated destination-chain in assert.equal(services['trade-executor'].upstream_status.unrelated_incident_count, 1); }); +test('dashboard service summary exposes executor salt cache and queue delay truth', () => { + const config = buildConfig(); + const dashboard = buildDashboardBootstrap({ + config, + auth: { authenticated: true }, + portfolioMetric: null, + inventorySnapshot: null, + marketPrice: null, + recentQuotes: [], + submissionPage: { page: 1, page_size: 20, total: 0, total_pages: 1, items: [] }, + submissionSummary: { total: 0, last_submission_at: null }, + fundingObservations: [], + recentDepositStatuses: [], + recentTradeDecisions: [], + recentExecuteTradeCommands: [], + recentExecutionResults: [], + recentQuoteOutcomes: [], + recentIntentRequests: [], + recentAlertTransitions: [], + serviceSnapshots: [ + { + service: 'trade-executor', + label: 'Trade Executor', + base_url: 'http://trade-executor', + reachable: true, + health: { ok: true }, + state: { + relay: { connected: true, last_message_at: '2026-06-12T10:00:00.000Z' }, + verifier_salt_cache: { + configured: true, + has_cached_salt: true, + salt_fresh: true, + salt_age_ms: 42, + last_refresh_completed_at: '2026-06-12T09:59:59.950Z', + last_refresh_duration_ms: 25, + refresh_in_flight: false, + }, + executor_queue_delay: { + status: 'warning', + latest_ms: 25, + max_ms: 1_250, + p90_ms: 1_250, + p99_ms: 1_250, + }, + salt_unavailable_reject_count: 1, + last_salt_unavailable_at: '2026-06-12T09:59:00.000Z', + }, + }, + ], + }); + + const executor = dashboard.system.service_health.find((service) => service.service === 'trade-executor'); + assert.equal(executor.summary.verifier_salt_cache.salt_fresh, true); + assert.equal(executor.summary.verifier_salt_cache.salt_age_ms, 42); + assert.equal(executor.summary.executor_queue_delay.status, 'warning'); + assert.equal(executor.summary.executor_queue_delay.max_ms, 1_250); + assert.equal(executor.summary.salt_unavailable_reject_count, 1); +}); + test('bootstrap exposes deduped environment status history as environmental conditions', () => { const config = buildConfig(); diff --git a/test/ops-sentinel-static.test.mjs b/test/ops-sentinel-static.test.mjs index ff3b7fe..9838dbc 100644 --- a/test/ops-sentinel-static.test.mjs +++ b/test/ops-sentinel-static.test.mjs @@ -42,3 +42,9 @@ test('ops sentinel turns quote lifecycle retention health into storage alerts', assert.match(source, /writerState\.retention_mode/); assert.match(source, /writerState\.retention_summary/); }); + +test('ops sentinel turns executor salt and queue state into runtime alerts', () => { + assert.match(source, /buildExecutorRuntimeAlerts/); + assert.match(source, /alerts\.push\(\.\.\.buildExecutorRuntimeAlerts/); + assert.match(source, /executor,\s*thresholds,\s*now,\s*pair: config\.activePair/); +}); diff --git a/test/runtime-health.test.mjs b/test/runtime-health.test.mjs index efe731e..88f5bb3 100644 --- a/test/runtime-health.test.mjs +++ b/test/runtime-health.test.mjs @@ -2,6 +2,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { + buildExecutorRuntimeAlerts, buildMakerCompetitivenessRuntimeAlerts, deriveServiceHealth, shouldContainExecutorForAlerts, @@ -158,3 +159,83 @@ test('history writer retention pressure and stale rollups surface as runtime war assert.equal(stale.label, 'rollup stale'); assert.ok(stale.reasons.some((reason) => /fail-closed/.test(reason))); }); + +test('trade executor salt and queue state surface as runtime health warnings', () => { + const health = deriveServiceHealth({ + service: 'trade-executor', + snapshot: { + reachable: true, + state: { + armed: true, + verifier_salt_cache: { + configured: true, + has_cached_salt: false, + salt_fresh: false, + }, + executor_queue_delay: { + status: 'warning', + warning_reason: 'recent command_to_executor_ms max 1500ms exceeds 1000ms', + }, + }, + health: { ok: true }, + }, + now: '2026-06-12T10:00:00.000Z', + }); + + assert.equal(health.status, 'warning'); + assert.ok(health.reasons.includes('verifier salt cache is stale or unavailable')); + assert.ok(health.reasons.includes('recent command_to_executor_ms max 1500ms exceeds 1000ms')); +}); + +test('executor runtime alerts cover stale salt, refresh failure, and queue delay', () => { + const alerts = buildExecutorRuntimeAlerts({ + executor: { + reachable: true, + state: { + verifier_salt_cache: { + configured: true, + has_cached_salt: true, + salt_fresh: false, + salt_age_ms: 750, + max_age_ms: 500, + refresh_in_flight: false, + last_refresh_started_at: '2026-06-12T09:59:59.000Z', + last_refresh_completed_at: '2026-06-12T09:59:59.100Z', + last_refresh_duration_ms: 100, + last_refresh_error: { + name: 'Error', + message: 'rpc unavailable', + }, + }, + executor_queue_delay: { + status: 'critical', + latest_ms: 200, + max_ms: 7_500, + p90_ms: 5_200, + p99_ms: 7_500, + sample_count: 10, + latest_command_id: 'cmd-1', + latest_quote_id: 'quote-1', + }, + }, + }, + thresholds: { + executorQueueDelayWarningMs: 1_000, + executorQueueDelayCriticalMs: 5_000, + }, + pair: 'nbtc->eure', + }); + + assert.deepEqual( + alerts.map((alert) => alert.alert_code), + [ + 'trade_executor_verifier_salt_unavailable', + 'trade_executor_verifier_salt_refresh_failed', + 'trade_executor_command_queue_delay_high', + ], + ); + assert.equal(alerts[0].severity, 'warning'); + assert.equal(alerts[1].details.last_refresh_error.message, 'rpc unavailable'); + assert.equal(alerts[2].severity, 'critical'); + assert.equal(alerts[2].details.p90_ms, 5_200); +}); diff --git a/test/service-snapshot-summary.test.mjs b/test/service-snapshot-summary.test.mjs index 32d43db..e793f79 100644 --- a/test/service-snapshot-summary.test.mjs +++ b/test/service-snapshot-summary.test.mjs @@ -19,8 +19,20 @@ test('sentinel snapshot summary keeps trade-executor health fields and drops ret paused: false, armed: true, relay: { connected: true, last_message_at: '2026-05-06T15:00:00.000Z' }, + verifier_salt_cache: { + salt_fresh: true, + salt_age_ms: 42, + refresh_in_flight: false, + }, + executor_queue_delay: { + status: 'warning', + max_ms: 1500, + p90_ms: 1200, + }, last_result: { status: 'submitted' }, submitted_count: 100, + salt_unavailable_reject_count: 2, + last_salt_unavailable_at: '2026-05-06T15:00:01.000Z', processed_idempotency_keys: Object.fromEntries(Array.from({ length: 100 }, (_, index) => [`quote:${index}`, true])), submitted_commands: { 'cmd-1': { quote_id: 'quote-1', raw_response: { large: true } }, @@ -30,6 +42,9 @@ test('sentinel snapshot summary keeps trade-executor health fields and drops ret assert.equal(summary.state.armed, true); assert.deepEqual(summary.state.relay, { connected: true, last_message_at: '2026-05-06T15:00:00.000Z' }); + assert.equal(summary.state.verifier_salt_cache.salt_fresh, true); + assert.equal(summary.state.executor_queue_delay.status, 'warning'); + assert.equal(summary.state.salt_unavailable_reject_count, 2); assert.equal(summary.state.processed_idempotency_keys, undefined); assert.equal(summary.state.submitted_commands, undefined); }); diff --git a/test/trade-executor-static.test.mjs b/test/trade-executor-static.test.mjs index e865e86..57168f5 100644 --- a/test/trade-executor-static.test.mjs +++ b/test/trade-executor-static.test.mjs @@ -30,6 +30,7 @@ test('trade executor records hot path timing in result payloads', () => { assert.match(source, /node:perf_hooks/); assert.match(source, /startExecutorTiming\(event\)/); assert.match(source, /executor_timing/); + assert.match(source, /command_to_executor_ms/); assert.match(source, /current_salt_ms/); assert.match(source, /sign_ms/); assert.match(source, /relay_response_ms/); @@ -40,10 +41,31 @@ test('trade executor records hot path timing in result payloads', () => { test('trade executor uses bounded verifier salt cache instead of per-command salt RPC', () => { assert.match(source, /createVerifierSaltCache/); assert.match(source, /verifierSaltCache\.start\(\)/); - assert.match(source, /verifierSaltCache\.getFreshSalt\(\)/); + assert.match(source, /verifierSaltCache\.getCachedFreshSalt\(\)/); + assert.match(source, /verifier_salt_unavailable/); + assert.match(source, /fresh verifier salt unavailable; command was not signed or relayed/); assert.match(source, /current_salt_source/); assert.match(source, /current_salt_age_ms/); + assert.match(source, /current_salt_unavailable_reason/); assert.match(source, /verifier_salt_cache: verifierSaltCache\.getState\(\)/); assert.match(source, /verifierSaltCache\.stop\(\)/); + assert.doesNotMatch(source, /verifierSaltCache\.getFreshSalt\(\)/); assert.doesNotMatch(source, /await verifierClient\.currentSalt\(\);/); + + const expiryIndex = source.indexOf('const expiry = classifyExecuteCommandExpiry(event);'); + const saltLookupIndex = source.indexOf('verifierSaltCache.getCachedFreshSalt()'); + const saltUnavailableIndex = source.indexOf("result_code: 'verifier_salt_unavailable'"); + const signIndex = source.indexOf('buildQuoteResponseSubmission({'); + const relayIndex = source.indexOf("relayClient.request('quote_response'"); + assert.ok(expiryIndex > 0 && expiryIndex < saltLookupIndex); + assert.ok(saltUnavailableIndex > 0 && saltUnavailableIndex < signIndex); + assert.ok(saltUnavailableIndex < relayIndex); +}); + +test('trade executor exposes queue delay and salt rejection state', () => { + assert.match(source, /createExecutorQueueDelayTracker/); + assert.match(source, /recordExecutorQueueDelay\(timing, payload\)/); + assert.match(source, /executor_queue_delay: executorQueueDelayTracker\.getState\(\)/); + assert.match(source, /salt_unavailable_reject_count/); + assert.match(source, /last_salt_unavailable_at/); }); diff --git a/test/verifier-salt-cache.test.mjs b/test/verifier-salt-cache.test.mjs index b9b7f26..f7ed55f 100644 --- a/test/verifier-salt-cache.test.mjs +++ b/test/verifier-salt-cache.test.mjs @@ -3,7 +3,7 @@ import assert from 'node:assert/strict'; import { createVerifierSaltCache } from '../src/core/verifier-salt-cache.mjs'; -test('verifier salt cache reuses a fresh salt without calling the verifier hot path', async () => { +test('verifier salt cache cached fresh read returns without calling the verifier loader', async () => { let nowMs = 1_000; let calls = 0; const cache = createVerifierSaltCache({ @@ -15,37 +15,101 @@ test('verifier salt cache reuses a fresh salt without calling the verifier hot p now: () => nowMs, }); - const refreshed = await cache.getFreshSalt(); + const refreshed = await cache.refreshNow(); nowMs += 100; - const cached = await cache.getFreshSalt(); + const cached = cache.getCachedFreshSalt(); assert.equal(calls, 1); assert.equal(refreshed.source, 'refresh'); assert.equal(cached.source, 'cache'); + assert.equal(cached.available, true); assert.equal(cached.currentSaltHex, '252812b3'); assert.equal(cached.ageMs, 100); + assert.equal(cache.getState().salt_fresh, true); }); -test('verifier salt cache refreshes a stale salt before returning it', async () => { +test('verifier salt cache missing or stale cached read returns unavailable without loader call', async () => { let nowMs = 1_000; - const salts = ['252812b3', '252812b4']; + let calls = 0; const cache = createVerifierSaltCache({ - loadSalt: async () => salts.shift(), + loadSalt: async () => { + calls += 1; + return '252812b3'; + }, maxAgeMs: 500, now: () => nowMs, }); - const first = await cache.getFreshSalt(); - nowMs += 501; - const second = await cache.getFreshSalt(); + const missing = cache.getCachedFreshSalt(); + assert.equal(missing.available, false); + assert.equal(missing.reason, 'empty_cache'); + assert.equal(calls, 0); - assert.equal(first.currentSaltHex, '252812b3'); - assert.equal(second.currentSaltHex, '252812b4'); - assert.equal(second.source, 'refresh'); - assert.equal(second.ageMs, 0); + await cache.refreshNow(); + nowMs += 501; + const stale = cache.getCachedFreshSalt(); + + assert.equal(stale.available, false); + assert.equal(stale.reason, 'stale_cache'); + assert.equal(stale.ageMs, 501); + assert.equal(calls, 1); + assert.equal(cache.getState().salt_fresh, false); }); -test('verifier salt cache fails closed when stale salt refresh fails', async () => { +test('verifier salt cache cached read does not wait for slow background current_salt refresh', async () => { + let calls = 0; + let resolveSalt; + const slowSalt = new Promise((resolve) => { + resolveSalt = resolve; + }); + const cache = createVerifierSaltCache({ + loadSalt: async () => { + calls += 1; + return slowSalt; + }, + setIntervalFn: () => ({ unref() {} }), + clearIntervalFn: () => {}, + }); + + cache.start(); + await Promise.resolve(); + + const cached = cache.getCachedFreshSalt(); + assert.equal(cached.available, false); + assert.equal(cached.reason, 'empty_cache'); + assert.equal(cached.state.refresh_in_flight, true); + assert.equal(calls, 1); + + resolveSalt('252812b3'); + await cache.refreshNow(); + assert.equal(cache.getCachedFreshSalt().available, true); +}); + +test('verifier salt cache background refresh records success duration and state', async () => { + let nowMs = 1_000; + const cache = createVerifierSaltCache({ + loadSalt: async () => { + nowMs += 37; + return '252812b3'; + }, + maxAgeMs: 500, + now: () => nowMs, + }); + + const refreshed = await cache.refreshNow({ reason: 'unit_test' }); + const state = cache.getState(); + + assert.equal(refreshed.currentSaltHex, '252812b3'); + assert.equal(state.salt_fresh, true); + assert.equal(state.last_refresh_started_at, '1970-01-01T00:00:01.000Z'); + assert.equal(state.last_refresh_completed_at, '1970-01-01T00:00:01.037Z'); + assert.equal(state.last_refresh_duration_ms, 37); + assert.equal(state.last_refresh_reason, 'unit_test'); + assert.equal(state.refresh_in_flight, false); + assert.equal(state.last_refresh_error, null); +}); + +test('verifier salt cache refresh failure preserves still-fresh cached salt', async () => { let nowMs = 1_000; let fail = false; const cache = createVerifierSaltCache({ @@ -57,15 +121,19 @@ test('verifier salt cache fails closed when stale salt refresh fails', async () now: () => nowMs, }); - await cache.getFreshSalt(); - nowMs += 501; + await cache.refreshNow(); + nowMs += 100; fail = true; await assert.rejects( - () => cache.getFreshSalt(), + () => cache.refreshNow({ reason: 'prefetch' }), /rpc unavailable/, ); - assert.equal(cache.getState().fresh, false); + const cached = cache.getCachedFreshSalt(); + assert.equal(cached.available, true); + assert.equal(cached.currentSaltHex, '252812b3'); + assert.equal(cached.ageMs, 100); + assert.equal(cache.getState().fresh, true); assert.equal(cache.getState().last_refresh_error.message, 'rpc unavailable'); }); @@ -75,8 +143,9 @@ test('verifier salt cache rejects malformed salts before signing can use them', }); await assert.rejects( - () => cache.getFreshSalt(), + () => cache.refreshNow(), /current_salt must be 4 bytes in hex/, ); assert.equal(cache.getState().has_cached_salt, false); + assert.equal(cache.getCachedFreshSalt().available, false); });