From 2d329c9b2f00f3cb3de3992dc3cddff9d59fa30d Mon Sep 17 00:00:00 2001 From: philipp Date: Sat, 6 Jun 2026 15:50:29 +0200 Subject: [PATCH] Add quote lifecycle retention rollups Proof: DB-backed quote lifecycle retention policy, rollup-before-prune maintenance, dashboard storage visibility, and regression tests for preservation and fail-closed behavior. Assumptions: Completed quote outcome attribution remains the current durable successful-trade evidence; venue-native terminal fill ids and fee-complete PnL remain unavailable. Still fake: Historical non-success detail already pruned cannot be recovered; rollups are aggregate analytics, not per-quote settled trade proof. --- scripts/ops/quote_lifecycle_retention.mjs | 86 ++ src/apps/history-writer.mjs | 150 ++- src/apps/operator-dashboard.mjs | 18 + src/apps/ops-sentinel.mjs | 42 + src/core/operator-dashboard.mjs | 41 + src/core/quote-lifecycle-retention.mjs | 492 ++++++++ src/core/runtime-health.mjs | 14 + src/lib/postgres.mjs | 1124 ++++++++++++++++- src/operator-dashboard/static/lib/format.js | 14 + .../static/pages/StrategyPage.jsx | 60 +- .../static/pages/SystemPage.jsx | 109 +- src/operator-dashboard/static/styles.css | 4 + test/history-writer-static.test.mjs | 24 +- test/inventory-and-history.test.mjs | 2 +- test/operator-dashboard-ui-static.test.mjs | 12 + test/operator-dashboard.test.mjs | 73 ++ test/ops-sentinel-static.test.mjs | 8 + test/quote-lifecycle-retention.test.mjs | 299 +++++ test/runtime-health.test.mjs | 35 + test/strategy-threshold-config.test.mjs | 3 + 20 files changed, 2527 insertions(+), 83 deletions(-) create mode 100755 scripts/ops/quote_lifecycle_retention.mjs create mode 100644 src/core/quote-lifecycle-retention.mjs create mode 100644 test/quote-lifecycle-retention.test.mjs diff --git a/scripts/ops/quote_lifecycle_retention.mjs b/scripts/ops/quote_lifecycle_retention.mjs new file mode 100755 index 0000000..8b4f87e --- /dev/null +++ b/scripts/ops/quote_lifecycle_retention.mjs @@ -0,0 +1,86 @@ +#!/usr/bin/env node + +import process from 'node:process'; + +import { loadConfig } from '../../src/lib/config.mjs'; +import { + countQuoteLifecyclePruneCandidates, + createPostgresPool, + ensureHistorySchema, + loadQuoteLifecycleRetentionPolicy, + loadQuoteLifecycleRetentionSummary, + runQuoteLifecycleRetentionMaintenance, +} from '../../src/lib/postgres.mjs'; + +function parseArgs(argv) { + const args = new Set(argv); + const reasonIndex = argv.indexOf('--reason'); + return { + execute: args.has('--execute'), + emergency: args.has('--emergency'), + reason: reasonIndex >= 0 ? argv[reasonIndex + 1] || '' : '', + }; +} + +const args = parseArgs(process.argv.slice(2)); +const config = loadConfig(); +const pool = createPostgresPool({ + connectionString: config.postgresUrl, +}); + +try { + await ensureHistorySchema(pool); + const now = new Date().toISOString(); + const policy = await loadQuoteLifecycleRetentionPolicy(pool); + if (!policy.ok) { + console.error(JSON.stringify({ + ok: false, + dry_run: !args.execute, + refused: true, + reason: policy.block_reason, + }, null, 2)); + process.exitCode = 2; + } else { + const summary = await loadQuoteLifecycleRetentionSummary(pool, { now }); + const mode = summary.retention_mode === 'pressure' ? 'pressure' : 'normal'; + const candidates = await countQuoteLifecyclePruneCandidates(pool, { + policy: policy.policy, + mode, + now, + }); + const preflight = { + ok: true, + dry_run: !args.execute, + emergency: args.emergency, + emergency_reason: args.reason || null, + retention_mode: summary.retention_mode, + policy: policy.policy, + preserved_successful_quote_count: summary.preserved_successful_quote_count, + rollup_watermark: summary.rollup_watermark, + rollup_required_until: summary.rollup_required_until, + candidate_counts: candidates, + }; + + if (!args.execute) { + console.log(JSON.stringify(preflight, null, 2)); + } else if (args.emergency && !args.reason) { + console.error(JSON.stringify({ + ...preflight, + ok: false, + refused: true, + reason: 'emergency execution requires --reason', + }, null, 2)); + process.exitCode = 2; + } else { + const result = await runQuoteLifecycleRetentionMaintenance(pool, { now }); + console.log(JSON.stringify({ + ...preflight, + dry_run: false, + result, + }, null, 2)); + if (result.status !== 'success') process.exitCode = 1; + } + } +} finally { + await pool.end().catch(() => {}); +} diff --git a/src/apps/history-writer.mjs b/src/apps/history-writer.mjs index bc0c889..263674f 100644 --- a/src/apps/history-writer.mjs +++ b/src/apps/history-writer.mjs @@ -24,12 +24,12 @@ import { insertEnvironmentStatusChange, finishNotificationDelivery, insertHistoryEvents, + loadQuoteLifecycleRetentionSummary, loadLatestPortfolioMetric, loadPortfolioMetricInputs, - pruneNonSuccessfulQuoteLifecycleHistory, - pruneRawNearIntentsQuoteHistory, refreshIntentRequestOutcomes, refreshQuoteOutcomes, + runQuoteLifecycleRetentionMaintenance, seedTradingConfig, upsertPortfolioMetric, } from '../lib/postgres.mjs'; @@ -122,14 +122,7 @@ const topics = [ const rawQuoteTopics = [ config.kafkaTopicRawNearIntentsQuote, ]; -const rawQuoteHistoryPruneIntervalMs = 60 * 1000; -const rawQuoteHistoryRetainRecentMs = 30 * 60 * 1000; -const rawQuoteHistoryPruneBatchSize = 500_000; -const rawQuoteHistoryPruneMaxBatches = 20; -const quoteLifecycleHistoryPruneIntervalMs = 60 * 1000; -const quoteLifecycleHistoryRetainRecentMs = 30 * 60 * 1000; -const quoteLifecycleHistoryPruneBatchSize = 100_000; -const quoteLifecycleHistoryPruneMaxBatches = 10; +const quoteLifecycleRetentionMaintenanceIntervalMs = 60 * 1000; const liveEvidenceTopics = [ config.kafkaTopicNormSwapDemand, config.kafkaTopicDecisionTradeDecision, @@ -205,6 +198,15 @@ const state = { last_quote_lifecycle_prune_at: null, quote_lifecycle_prune_deleted_count: 0, quote_lifecycle_prune_error: null, + last_quote_lifecycle_rollup_at: null, + quote_lifecycle_rollup_count: 0, + quote_lifecycle_rollup_error: null, + last_retention_maintenance_at: null, + retention_maintenance: null, + retention_summary: null, + retention_mode: null, + retention_error: null, + storage_pressure: false, derived_refresh_skipped_count: 0, last_derived_refresh_skipped_at: null, last_derived_refresh_skipped_topic: null, @@ -223,6 +225,9 @@ await refreshQuoteOutcomeAttributions().catch((error) => { await refreshIntentRequestOutcomeAttributions().catch((error) => { state.intent_request_outcomes_error = serializeError(error); }); +await refreshQuoteLifecycleRetentionState().catch((error) => { + state.retention_error = serializeError(error); +}); for (const historyConsumer of durableConsumers) { await runHistoryConsumer(historyConsumer); @@ -292,10 +297,7 @@ async function runHistoryConsumer(historyConsumer) { await heartbeat(); } - if (batch.topic === config.kafkaTopicRawNearIntentsQuote) { - await maybePruneRawQuoteHistory(); - } - await maybePruneQuoteLifecycleHistory(); + await maybeRunQuoteLifecycleRetentionMaintenance(); if (state.draining) { setTimeout(() => shutdown(), 0); @@ -304,86 +306,92 @@ async function runHistoryConsumer(historyConsumer) { }); } -async function maybePruneRawQuoteHistory({ force = false } = {}) { +async function maybeRunQuoteLifecycleRetentionMaintenance({ force = false } = {}) { const nowMs = Date.now(); - const lastPruneMs = state.last_raw_quote_prune_at - ? Date.parse(state.last_raw_quote_prune_at) + const lastMaintenanceMs = state.last_retention_maintenance_at + ? Date.parse(state.last_retention_maintenance_at) : 0; if ( !force - && Number.isFinite(lastPruneMs) - && nowMs - lastPruneMs < rawQuoteHistoryPruneIntervalMs + && Number.isFinite(lastMaintenanceMs) + && nowMs - lastMaintenanceMs < quoteLifecycleRetentionMaintenanceIntervalMs ) { return null; } try { - const result = await pruneRawNearIntentsQuoteHistory(pool, { + const result = await runQuoteLifecycleRetentionMaintenance(pool, { now: new Date(nowMs).toISOString(), - retainRecentMs: rawQuoteHistoryRetainRecentMs, - batchSize: rawQuoteHistoryPruneBatchSize, - maxBatches: rawQuoteHistoryPruneMaxBatches, }); - state.last_raw_quote_prune_at = new Date(nowMs).toISOString(); - state.raw_quote_prune_deleted_count += result.deletedCount; - state.raw_quote_prune_error = null; - if (result.deletedCount > 0) { - logger.info('raw_near_intents_quote_history_pruned', { + applyRetentionMaintenanceResult(result); + await refreshQuoteLifecycleRetentionState(); + if (result.status === 'success' && Number(result.pruned_counts?.deletedCount || 0) > 0) { + logger.info('quote_lifecycle_retention_maintenance_completed', { + details: result, + }); + } else if (result.status !== 'success') { + logger.warn('quote_lifecycle_retention_maintenance_skipped', { details: result, }); } return result; } catch (error) { - state.raw_quote_prune_error = serializeError(error); - logger.error('raw_near_intents_quote_history_prune_failed', { + state.retention_error = serializeError(error); + state.quote_lifecycle_prune_error = state.retention_error; + logger.error('quote_lifecycle_retention_maintenance_failed', { details: { - error: state.raw_quote_prune_error, + error: state.retention_error, }, }); return null; } } -async function maybePruneQuoteLifecycleHistory({ force = false } = {}) { - const nowMs = Date.now(); - const lastPruneMs = state.last_quote_lifecycle_prune_at - ? Date.parse(state.last_quote_lifecycle_prune_at) - : 0; - if ( - !force - && Number.isFinite(lastPruneMs) - && nowMs - lastPruneMs < quoteLifecycleHistoryPruneIntervalMs - ) { - return null; +function applyRetentionMaintenanceResult(result) { + const completedAt = result.completed_at || new Date().toISOString(); + state.last_retention_maintenance_at = completedAt; + state.retention_maintenance = result; + state.retention_mode = result.mode || null; + state.retention_error = result.error || null; + state.storage_pressure = result.mode === 'pressure'; + + if (result.rollup_counts?.covered_until) { + state.last_quote_lifecycle_rollup_at = completedAt; + state.quote_lifecycle_rollup_count += Number(result.rollup_counts.rollup_count || 0); + state.quote_lifecycle_rollup_error = null; } - try { - const result = await pruneNonSuccessfulQuoteLifecycleHistory(pool, { - now: new Date(nowMs).toISOString(), - retainRecentMs: quoteLifecycleHistoryRetainRecentMs, - batchSize: quoteLifecycleHistoryPruneBatchSize, - maxBatches: quoteLifecycleHistoryPruneMaxBatches, - }); - state.last_quote_lifecycle_prune_at = new Date(nowMs).toISOString(); - state.quote_lifecycle_prune_deleted_count += result.deletedCount; + const pruned = result.pruned_counts || {}; + if (pruned.ok === true) { + state.last_quote_lifecycle_prune_at = completedAt; + state.quote_lifecycle_prune_deleted_count += Number(pruned.deletedCount || 0); state.quote_lifecycle_prune_error = null; - if (result.deletedCount > 0) { - logger.info('non_successful_quote_lifecycle_history_pruned', { - details: result, - }); + const rawTable = (pruned.tables || []).find((entry) => entry.table === 'raw_near_intents_quotes'); + if (rawTable) { + state.last_raw_quote_prune_at = completedAt; + state.raw_quote_prune_deleted_count += Number(rawTable.deletedCount || 0); + state.raw_quote_prune_error = null; } - return result; - } catch (error) { - state.quote_lifecycle_prune_error = serializeError(error); - logger.error('non_successful_quote_lifecycle_history_prune_failed', { - details: { - error: state.quote_lifecycle_prune_error, - }, - }); - return null; + } else if (result.mode === 'blocked_rollup_stale' || result.mode === 'blocked_invalid_policy') { + state.quote_lifecycle_prune_error = result.error || { reason: result.mode }; } } +async function refreshQuoteLifecycleRetentionState() { + const summary = await loadQuoteLifecycleRetentionSummary(pool); + state.retention_summary = summary; + state.retention_mode = summary.retention_mode; + state.storage_pressure = summary.retention_mode === 'pressure'; + state.retention_maintenance = summary.latest_run || state.retention_maintenance; + state.last_retention_maintenance_at = + summary.latest_run?.completed_at + || state.last_retention_maintenance_at; + state.retention_error = summary.policy_ok === false ? { + reason: summary.policy_block_reason, + } : null; + return summary; +} + async function handleWrittenHistoryEvent({ topic, partition, @@ -520,6 +528,20 @@ const controlApi = startControlApi({ last_environment_status_duplicate_at: state.last_environment_status_duplicate_at, last_environment_status_fingerprint: state.last_environment_status_fingerprint, last_metrics_at: state.last_metrics_at, + retention_mode: state.retention_mode, + retention_summary: state.retention_summary, + last_retention_maintenance_at: state.last_retention_maintenance_at, + retention_error: state.retention_error, + storage_pressure: state.storage_pressure, + last_quote_lifecycle_rollup_at: state.last_quote_lifecycle_rollup_at, + quote_lifecycle_rollup_count: state.quote_lifecycle_rollup_count, + quote_lifecycle_rollup_error: state.quote_lifecycle_rollup_error, + last_quote_lifecycle_prune_at: state.last_quote_lifecycle_prune_at, + quote_lifecycle_prune_deleted_count: state.quote_lifecycle_prune_deleted_count, + quote_lifecycle_prune_error: state.quote_lifecycle_prune_error, + last_raw_quote_prune_at: state.last_raw_quote_prune_at, + raw_quote_prune_deleted_count: state.raw_quote_prune_deleted_count, + raw_quote_prune_error: state.raw_quote_prune_error, freshness_age_ms: Number.isFinite(freshnessAgeMs) ? Math.max(0, freshnessAgeMs) : null, database_connectivity: connectivity, trading_config_ok: tradingConfig.ok, diff --git a/src/apps/operator-dashboard.mjs b/src/apps/operator-dashboard.mjs index bbb8da0..371ac01 100644 --- a/src/apps/operator-dashboard.mjs +++ b/src/apps/operator-dashboard.mjs @@ -42,12 +42,14 @@ import { loadLatestMarketPrice, loadLatestPortfolioMetric, loadPairConfigSummary, + loadQuoteLifecycleRetentionSummary, loadRecentAlertTransitions, loadRecentDepositStatuses, loadRecentEnvironmentStatuses, loadRecentExecuteTradeCommands, loadRecentExecutionResults, loadRecentIntentRequests, + loadRecentQuoteLifecycleRollups, loadRecentQuoteOutcomes, loadRecentTradeDecisions, loadRecentQuotes, @@ -444,6 +446,8 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { recentExecuteTradeCommands, recentExecutionResults, recentQuoteOutcomes, + quoteLifecycleRetention, + quoteLifecycleRollups, recentIntentRequests, recentAlertTransitions, recentEnvironmentStatuses, @@ -515,6 +519,18 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { [], sourceErrors, ), + safeSourceLoad( + 'quote_lifecycle_retention', + () => loadQuoteLifecycleRetentionSummary(pool), + null, + sourceErrors, + ), + safeSourceLoad( + 'quote_lifecycle_rollups', + () => loadRecentQuoteLifecycleRollups(pool, { limit: 40 }), + [], + sourceErrors, + ), safeSourceLoad( 'recent_intent_requests', () => loadRecentIntentRequests(pool, { @@ -559,6 +575,8 @@ async function loadBootstrapPayload({ auth, page, pageSize }) { recentExecuteTradeCommands, recentExecutionResults, recentQuoteOutcomes, + quoteLifecycleRetention, + quoteLifecycleRollups, recentIntentRequests, recentAlertTransitions, recentEnvironmentStatuses, diff --git a/src/apps/ops-sentinel.mjs b/src/apps/ops-sentinel.mjs index 85189f7..2198dc5 100644 --- a/src/apps/ops-sentinel.mjs +++ b/src/apps/ops-sentinel.mjs @@ -475,6 +475,48 @@ function buildDeterministicRuntimeAlerts({ servicesByName, now, previousRuntimeE })); } + if (writerState.retention_mode === 'blocked_invalid_policy') { + alerts.push(buildRuntimeAlert({ + alert_code: 'quote_lifecycle_retention_policy_invalid', + severity: 'warning', + reason: 'quote lifecycle retention policy is invalid or missing, so pruning is fail-closed', + service_scope: 'history-writer', + pair: null, + details: { + retention_mode: writerState.retention_mode, + retention_error: writerState.retention_error || writerState.retention_summary?.policy_block_reason || null, + policy: writerState.retention_summary?.policy || null, + }, + })); + } else if (writerState.retention_mode === 'blocked_rollup_stale') { + alerts.push(buildRuntimeAlert({ + alert_code: 'quote_lifecycle_rollup_stale', + severity: 'warning', + reason: 'quote lifecycle rollup watermark is stale, so detail pruning is fail-closed', + service_scope: 'history-writer', + pair: null, + details: { + retention_mode: writerState.retention_mode, + rollup_watermark: writerState.retention_summary?.rollup_watermark || null, + rollup_required_until: writerState.retention_summary?.rollup_required_until || null, + latest_run: writerState.retention_summary?.latest_run || writerState.retention_maintenance || null, + }, + })); + } else if (writerState.retention_mode === 'pressure' || writerState.storage_pressure === true) { + alerts.push(buildRuntimeAlert({ + alert_code: 'quote_lifecycle_storage_pressure', + severity: 'warning', + reason: 'quote lifecycle table or database size is above the DB-backed pressure threshold', + service_scope: 'history-writer', + pair: null, + details: { + retention_mode: writerState.retention_mode, + storage: writerState.retention_summary?.storage || null, + detail_retention_ms: writerState.retention_summary?.detail_retention_ms || null, + }, + })); + } + const dashboard = servicesByName['operator-dashboard']; const dashboardState = dashboard?.state || {}; alerts.push(...buildMakerCompetitivenessRuntimeAlerts({ diff --git a/src/core/operator-dashboard.mjs b/src/core/operator-dashboard.mjs index 5a75f37..36dc57f 100644 --- a/src/core/operator-dashboard.mjs +++ b/src/core/operator-dashboard.mjs @@ -524,6 +524,8 @@ export function buildDashboardBootstrap({ recentExecuteTradeCommands, recentExecutionResults, recentQuoteOutcomes = [], + quoteLifecycleRetention = null, + quoteLifecycleRollups = [], recentIntentRequests = [], recentAlertTransitions, recentEnvironmentStatuses = [], @@ -618,6 +620,7 @@ export function buildDashboardBootstrap({ recentExecuteTradeCommands, recentExecutionResults, recentQuoteOutcomes, + quoteLifecycleRollups, }), system: buildSystemSummary({ servicesByName, @@ -625,6 +628,8 @@ export function buildDashboardBootstrap({ recentAlerts, nearIntentsStatus: effectiveNearIntentsStatus, environmentStatus, + quoteLifecycleRetention, + quoteLifecycleRollups, }), }; } @@ -1577,6 +1582,7 @@ function buildStrategySummary({ recentExecuteTradeCommands = [], recentExecutionResults = [], recentQuoteOutcomes = [], + quoteLifecycleRollups = [], }) { const strategyState = servicesByName['strategy-engine']?.state || {}; const executorState = servicesByName['trade-executor']?.state || {}; @@ -1625,6 +1631,8 @@ function buildStrategySummary({ const makerCompetitiveness = buildMakerCompetitivenessSummary({ lifecycleRows: allLifecycleRows, }); + makerCompetitiveness.persisted_rollups = quoteLifecycleRollups || []; + makerCompetitiveness.persisted_rollup_count = quoteLifecycleRollups?.length || 0; return { strategy_state: { @@ -1854,6 +1862,8 @@ function buildSystemSummary({ recentAlerts, nearIntentsStatus = null, environmentStatus = { current: null, recent_changes: [] }, + quoteLifecycleRetention = null, + quoteLifecycleRollups = [], }) { const historyWriterState = servicesByName['history-writer']?.state || {}; void activeAlerts; @@ -1887,6 +1897,35 @@ function buildSystemSummary({ offsets: historyWriterState.offsets || {}, metrics_error: historyWriterState.metrics_error || null, quote_outcomes_error: historyWriterState.quote_outcomes_error || null, + retention_mode: + quoteLifecycleRetention?.retention_mode + || historyWriterState.retention_mode + || null, + retention_summary: + quoteLifecycleRetention + || historyWriterState.retention_summary + || null, + quote_lifecycle_rollups: quoteLifecycleRollups || [], + last_retention_maintenance_at: + quoteLifecycleRetention?.latest_run?.completed_at + || historyWriterState.last_retention_maintenance_at + || null, + last_quote_lifecycle_rollup_at: + quoteLifecycleRetention?.rollups?.latest_rollup_at + || historyWriterState.last_quote_lifecycle_rollup_at + || null, + last_quote_lifecycle_prune_at: + quoteLifecycleRetention?.latest_run?.completed_at + || historyWriterState.last_quote_lifecycle_prune_at + || null, + quote_lifecycle_prune_deleted_count: + historyWriterState.quote_lifecycle_prune_deleted_count ?? null, + raw_quote_prune_deleted_count: + historyWriterState.raw_quote_prune_deleted_count ?? null, + preserved_successful_quote_count: + quoteLifecycleRetention?.preserved_successful_quote_count + ?? historyWriterState.retention_summary?.preserved_successful_quote_count + ?? null, }, controls: listDashboardControls({ page: 'system' }), }; @@ -1988,6 +2027,8 @@ function buildServiceSummary(service, state) { last_alert_write_at: state.last_alert_write_at || null, last_quote_outcomes_at: state.last_quote_outcomes_at || null, database_connectivity: state.database_connectivity ?? null, + retention_mode: state.retention_mode || null, + last_retention_maintenance_at: state.last_retention_maintenance_at || null, }; case 'ops-sentinel': return { diff --git a/src/core/quote-lifecycle-retention.mjs b/src/core/quote-lifecycle-retention.mjs new file mode 100644 index 0000000..926b1c8 --- /dev/null +++ b/src/core/quote-lifecycle-retention.mjs @@ -0,0 +1,492 @@ +import { createHash } from 'node:crypto'; + +import { + normalizeCompetitivenessEntry, + notionalBucket, + quoteAgeBucket, +} from './maker-competitiveness.mjs'; +import { TERMINAL_SETTLEMENT_ATTRIBUTION_STATUSES } from './quote-outcomes.mjs'; + +export const QUOTE_LIFECYCLE_RETENTION_POLICY_KEY = 'quote_lifecycle_default'; + +export const QUOTE_LIFECYCLE_DETAIL_TABLES = [ + 'raw_near_intents_quotes', + 'swap_demand_events', + 'trade_decisions', + 'execute_trade_commands', + 'trade_execution_results', + 'quote_outcome_attributions', +]; + +export const QUOTE_LIFECYCLE_DATA_CLASSES = Object.freeze({ + successfulTradeEvidence: 'successful_trade_evidence', + inFlightDetail: 'in_flight_detail', + nonSuccessDetail: 'non_success_detail', + rawDebugFirehose: 'raw_debug_firehose', + aggregateRollup: 'aggregate_rollup', +}); + +export const SUCCESSFUL_SETTLEMENT_ATTRIBUTION_STATUSES = new Set([ + ...TERMINAL_SETTLEMENT_ATTRIBUTION_STATUSES, + 'exact_match', + 'attributed', +]); + +export const DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY = Object.freeze({ + policy_key: QUOTE_LIFECYCLE_RETENTION_POLICY_KEY, + enabled: true, + normal_detail_retention_ms: 30 * 60 * 1000, + pressure_detail_retention_ms: 10 * 60 * 1000, + raw_retention_ms: 30 * 60 * 1000, + rollup_granularity_ms: 5 * 60 * 1000, + rollup_retention_ms: 90 * 24 * 60 * 60 * 1000, + rollup_stale_after_ms: 15 * 60 * 1000, + preserve_successful_evidence: true, + max_delete_rows_per_pass: 100_000, + pressure_table_bytes: 1_500_000_000, + pressure_database_bytes: 3_500_000_000, + config_version: 1, +}); + +export function isSuccessfulQuoteOutcome(row = {}) { + const payload = row.payload || row.outcome || {}; + const outcomeStatus = normalizeToken(row.outcome_status || payload.outcome_status); + const attributionStatus = normalizeToken(row.attribution_status || payload.attribution_status); + return outcomeStatus === 'completed' + || SUCCESSFUL_SETTLEMENT_ATTRIBUTION_STATUSES.has(attributionStatus); +} + +export function classifyQuoteLifecycleDataClass({ + table = null, + row = {}, + now = new Date().toISOString(), + detailRetentionMs = DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.normal_detail_retention_ms, +} = {}) { + if (table === 'quote_lifecycle_rollups' || row.data_class === QUOTE_LIFECYCLE_DATA_CLASSES.aggregateRollup) { + return QUOTE_LIFECYCLE_DATA_CLASSES.aggregateRollup; + } + + if (isSuccessfulQuoteOutcome(row)) { + return QUOTE_LIFECYCLE_DATA_CLASSES.successfulTradeEvidence; + } + + if (table === 'raw_near_intents_quotes' && !resolveQuoteId(row)) { + return QUOTE_LIFECYCLE_DATA_CLASSES.rawDebugFirehose; + } + + const nowMs = timestampMs(now); + const rowMs = timestampMs( + row.outcome_observed_at + || row.computed_at + || row.observed_at + || row.ingested_at + || row.payload?.observed_at + || row.payload?.ingested_at, + ); + if ( + Number.isFinite(nowMs) + && Number.isFinite(rowMs) + && Number.isInteger(detailRetentionMs) + && nowMs - rowMs <= detailRetentionMs + ) { + return QUOTE_LIFECYCLE_DATA_CLASSES.inFlightDetail; + } + + return QUOTE_LIFECYCLE_DATA_CLASSES.nonSuccessDetail; +} + +export function normalizeQuoteLifecycleRetentionPolicy(row = null) { + if (!row) { + return { + ok: false, + block_reason: 'retention_policy_missing', + policy: null, + }; + } + + const policy = { + policy_key: String(row.policy_key || '').trim(), + enabled: row.enabled === true, + normal_detail_retention_ms: positiveInteger(row.normal_detail_retention_ms), + pressure_detail_retention_ms: positiveInteger(row.pressure_detail_retention_ms), + raw_retention_ms: positiveInteger(row.raw_retention_ms), + rollup_granularity_ms: positiveInteger(row.rollup_granularity_ms), + rollup_retention_ms: positiveInteger(row.rollup_retention_ms), + rollup_stale_after_ms: positiveInteger(row.rollup_stale_after_ms), + preserve_successful_evidence: row.preserve_successful_evidence === true, + max_delete_rows_per_pass: positiveInteger(row.max_delete_rows_per_pass), + pressure_table_bytes: positiveInteger(row.pressure_table_bytes), + pressure_database_bytes: positiveInteger(row.pressure_database_bytes), + updated_at: toIsoTimestamp(row.updated_at), + config_version: positiveInteger(row.config_version), + }; + + const invalidFields = Object.entries(policy) + .filter(([field, value]) => ( + [ + 'normal_detail_retention_ms', + 'pressure_detail_retention_ms', + 'raw_retention_ms', + 'rollup_granularity_ms', + 'rollup_retention_ms', + 'rollup_stale_after_ms', + 'max_delete_rows_per_pass', + 'pressure_table_bytes', + 'pressure_database_bytes', + 'config_version', + ].includes(field) + && value == null + )) + .map(([field]) => field); + + if (!policy.policy_key) invalidFields.push('policy_key'); + if (policy.enabled !== true) invalidFields.push('enabled'); + if (policy.preserve_successful_evidence !== true) invalidFields.push('preserve_successful_evidence'); + if ( + policy.pressure_detail_retention_ms != null + && policy.normal_detail_retention_ms != null + && policy.pressure_detail_retention_ms > policy.normal_detail_retention_ms + ) { + invalidFields.push('pressure_detail_retention_ms'); + } + + if (invalidFields.length) { + return { + ok: false, + block_reason: `retention_policy_invalid:${invalidFields.join(',')}`, + policy: null, + }; + } + + return { + ok: true, + block_reason: null, + policy, + }; +} + +export function decideQuoteLifecycleRetentionMode({ + policy, + storageMetrics = null, +} = {}) { + const normalized = policy?.policy ? policy : normalizeQuoteLifecycleRetentionPolicy(policy); + if (!normalized.ok) { + return { + mode: 'blocked_invalid_policy', + pressure: false, + reason: normalized.block_reason, + detail_retention_ms: null, + }; + } + + const activePolicy = normalized.policy; + const pressureTables = (storageMetrics?.tables || []) + .filter((table) => Number(table.total_bytes || 0) >= activePolicy.pressure_table_bytes) + .map((table) => table.table); + const databasePressure = Number(storageMetrics?.database?.total_bytes || 0) >= activePolicy.pressure_database_bytes; + const pressure = pressureTables.length > 0 || databasePressure; + + return { + mode: pressure ? 'pressure' : 'normal', + pressure, + reason: pressure + ? [ + pressureTables.length ? `tables:${pressureTables.join(',')}` : null, + databasePressure ? 'database_bytes' : null, + ].filter(Boolean).join(';') + : null, + detail_retention_ms: pressure + ? activePolicy.pressure_detail_retention_ms + : activePolicy.normal_detail_retention_ms, + }; +} + +export function buildQuoteLifecycleRollups({ + rows = [], + rollupGranularityMs = DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.rollup_granularity_ms, + computedAt = new Date().toISOString(), + now = computedAt, +} = {}) { + const groups = new Map(); + for (const row of rows || []) { + const normalized = normalizeRollupInput({ ...row, now }); + const activityAt = normalized.activity_at; + const windowStart = floorTimestamp(activityAt, rollupGranularityMs); + if (!windowStart) continue; + const windowEnd = new Date(Date.parse(windowStart) + rollupGranularityMs).toISOString(); + const timing = normalized.entry.maker_timing || {}; + const sourceDataClass = normalized.source_data_class; + const keyParts = [ + windowStart, + windowEnd, + sourceDataClass, + normalized.entry.pair || 'unknown', + normalized.entry.direction || 'unknown', + normalized.entry.request_kind || 'unknown', + normalized.entry.edge_bps ?? 'unknown', + normalized.notional_asset || 'unknown', + normalized.entry.notional_bucket || 'unavailable', + normalized.entry.quote_age_bucket || 'unavailable', + normalized.entry.result_code || 'no_result', + normalized.entry.failure_category || 'none', + normalized.entry.outcome_status || 'unknown', + ]; + const key = JSON.stringify(keyParts); + const existing = groups.get(key) || { + rollup_id: buildRollupId(keyParts), + window_start: windowStart, + window_end: windowEnd, + data_class: QUOTE_LIFECYCLE_DATA_CLASSES.aggregateRollup, + source_data_class: sourceDataClass, + pair: normalized.entry.pair || null, + direction: normalized.entry.direction || 'unknown', + request_kind: normalized.entry.request_kind || 'unknown', + edge_bps: normalized.entry.edge_bps == null ? null : String(normalized.entry.edge_bps), + notional_asset: normalized.notional_asset, + notional_bucket: normalized.entry.notional_bucket || notionalBucket(null, normalized.notional_symbol), + quote_age_bucket: normalized.entry.quote_age_bucket || quoteAgeBucket(null), + result_code: normalized.entry.result_code || 'no_result', + failure_category: normalized.entry.failure_category || null, + outcome_status: normalized.entry.outcome_status || 'unknown', + quote_count: 0, + decision_count: 0, + command_count: 0, + executor_result_count: 0, + relay_accepted_count: 0, + relay_failed_count: 0, + not_filled_count: 0, + completed_count: 0, + timing_bucket_counts: {}, + min_observed_at: normalized.activity_at, + max_observed_at: normalized.activity_at, + source_watermark_at: normalized.activity_at, + computed_at: computedAt, + }; + + existing.quote_count += 1; + existing.decision_count += normalized.has_decision ? 1 : 0; + existing.command_count += normalized.has_command ? 1 : 0; + existing.executor_result_count += normalized.has_execution ? 1 : 0; + existing.relay_accepted_count += normalized.entry.accepted ? 1 : 0; + existing.relay_failed_count += normalized.entry.relay_failed ? 1 : 0; + existing.not_filled_count += normalized.entry.outcome_status === 'not_filled' ? 1 : 0; + existing.completed_count += isSuccessfulQuoteOutcome(normalized.outcome || { outcome_status: normalized.entry.outcome_status }) + ? 1 + : 0; + existing.min_observed_at = earlierTimestamp(existing.min_observed_at, normalized.activity_at); + existing.max_observed_at = laterTimestamp(existing.max_observed_at, normalized.activity_at); + existing.source_watermark_at = laterTimestamp(existing.source_watermark_at, normalized.activity_at); + addTimingBuckets(existing.timing_bucket_counts, timing); + groups.set(key, existing); + } + + return [...groups.values()].sort((left, right) => ( + String(left.window_start).localeCompare(String(right.window_start)) + || String(left.pair || '').localeCompare(String(right.pair || '')) + || String(left.result_code || '').localeCompare(String(right.result_code || '')) + )); +} + +export function normalizeQuoteLifecycleRollupRow(row = {}) { + return { + rollup_id: row.rollup_id, + window_start: toIsoTimestamp(row.window_start), + window_end: toIsoTimestamp(row.window_end), + data_class: row.data_class || QUOTE_LIFECYCLE_DATA_CLASSES.aggregateRollup, + source_data_class: row.source_data_class || null, + pair: row.pair || null, + direction: row.direction || 'unknown', + request_kind: row.request_kind || 'unknown', + edge_bps: row.edge_bps == null ? null : String(row.edge_bps), + notional_asset: row.notional_asset || null, + notional_bucket: row.notional_bucket || 'unavailable', + quote_age_bucket: row.quote_age_bucket || 'unavailable', + result_code: row.result_code || 'no_result', + failure_category: row.failure_category || null, + outcome_status: row.outcome_status || 'unknown', + quote_count: Number(row.quote_count || 0), + decision_count: Number(row.decision_count || 0), + command_count: Number(row.command_count || 0), + executor_result_count: Number(row.executor_result_count || 0), + relay_accepted_count: Number(row.relay_accepted_count || 0), + relay_failed_count: Number(row.relay_failed_count || 0), + not_filled_count: Number(row.not_filled_count || 0), + completed_count: Number(row.completed_count || 0), + timing_bucket_counts: row.timing_bucket_counts || {}, + min_observed_at: toIsoTimestamp(row.min_observed_at), + max_observed_at: toIsoTimestamp(row.max_observed_at), + source_watermark_at: toIsoTimestamp(row.source_watermark_at), + computed_at: toIsoTimestamp(row.computed_at), + }; +} + +function normalizeRollupInput(row = {}) { + const decision = row.decision || row.decision_payload || null; + const command = row.command || row.command_payload || null; + const execution = row.execution || row.execution_payload || null; + const outcome = row.outcome || row.outcome_payload || null; + const quote = row.quote || row.quote_payload || row.demand || row.demand_payload || {}; + const makerTiming = + outcome?.maker_timing + || execution?.maker_timing + || command?.maker_timing + || decision?.maker_timing + || quote?.maker_timing + || row.maker_timing + || {}; + const entry = normalizeCompetitivenessEntry({ + quote_id: row.quote_id || quote.quote_id || decision?.quote_id || command?.quote_id || execution?.quote_id || outcome?.quote_id || null, + pair: row.pair || quote.pair || decision?.pair || command?.pair || execution?.pair || outcome?.pair || null, + pair_id: row.pair_id || decision?.pair_id || command?.pair_id || execution?.pair_id || null, + direction: row.direction || decision?.direction || command?.direction || outcome?.direction || null, + request_kind: row.request_kind || quote.request_kind || decision?.request_kind || command?.request_kind || outcome?.request_kind || null, + edge_bps: row.edge_bps ?? decision?.edge_bps ?? command?.edge_bps ?? outcome?.edge_bps ?? null, + notional: row.notional ?? decision?.notional ?? command?.notional ?? outcome?.notional ?? quote?.notional ?? null, + notional_symbol: + row.notional_symbol + || decision?.notional_symbol + || command?.notional_symbol + || outcome?.notional_symbol + || null, + maker_timing: makerTiming, + lifecycle_state: row.lifecycle_state || null, + outcome_status: outcome?.outcome_status || row.outcome_status || execution?.outcome_status || null, + execution, + decision, + command, + }); + const sourceTable = row.source_table || row.table || null; + const activityAt = firstTimestamp([ + row.activity_at, + outcome?.outcome_observed_at, + row.outcome_observed_at, + execution?.result_at, + row.execution_result_at, + command?.command_at, + row.command_at, + decision?.decision_at, + row.decision_at, + quote?.observed_at, + row.quote_observed_at, + row.observed_at, + row.ingested_at, + row.computed_at, + ]); + const sourceDataClass = classifyQuoteLifecycleDataClass({ + table: sourceTable, + row: { + ...row, + payload: outcome || execution || decision || quote || {}, + outcome_status: outcome?.outcome_status || row.outcome_status, + attribution_status: outcome?.attribution_status || row.attribution_status, + observed_at: activityAt, + }, + now: row.now || new Date().toISOString(), + }); + + return { + entry, + outcome, + source_data_class: sourceDataClass, + source_table: sourceTable, + notional_asset: + row.notional_asset + || row.notional_asset_id + || decision?.notional_asset_id + || command?.notional_asset_id + || outcome?.notional_asset_id + || null, + notional_symbol: + row.notional_symbol + || decision?.notional_symbol + || command?.notional_symbol + || outcome?.notional_symbol + || null, + activity_at: activityAt, + has_decision: Boolean(decision?.decision_id || decision?.decision || row.decision_id), + has_command: Boolean(command?.command_id || row.command_id), + has_execution: Boolean(execution?.command_id || execution?.status || row.execution_result_status), + }; +} + +function addTimingBuckets(target, timing = {}) { + for (const [field, value] of Object.entries(timing || {})) { + if (!field.endsWith('_ms')) continue; + const bucket = fixedTimingBucket(value); + target[field] ||= {}; + target[field][bucket] = Number(target[field][bucket] || 0) + 1; + } +} + +function fixedTimingBucket(value) { + const number = Number(value); + if (!Number.isFinite(number)) return 'unavailable'; + if (number < 50) return '<50ms'; + if (number < 100) return '50-100ms'; + if (number < 250) return '100-250ms'; + if (number < 500) return '250-500ms'; + if (number < 1000) return '500-1000ms'; + if (number < 5000) return '1000-5000ms'; + return '>=5000ms'; +} + +function buildRollupId(parts) { + return `quote-lifecycle-rollup:${createHash('sha256').update(JSON.stringify(parts)).digest('hex').slice(0, 32)}`; +} + +function resolveQuoteId(row = {}) { + return row.quote_id || row.payload?.quote_id || row.outcome?.quote_id || null; +} + +function positiveInteger(value) { + const number = Number(value); + return Number.isInteger(number) && number > 0 ? number : null; +} + +function normalizeToken(value) { + return String(value || '').trim().toLowerCase(); +} + +function timestampMs(value) { + if (!value) return NaN; + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : NaN; +} + +function toIsoTimestamp(value) { + if (!value) return null; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? null : date.toISOString(); +} + +function firstTimestamp(values = []) { + for (const value of values) { + const iso = toIsoTimestamp(value); + if (iso) return iso; + } + return null; +} + +function floorTimestamp(value, granularityMs) { + const ms = timestampMs(value); + if (!Number.isFinite(ms)) return null; + const granularity = positiveInteger(granularityMs); + if (!granularity) return null; + return new Date(Math.floor(ms / granularity) * granularity).toISOString(); +} + +function earlierTimestamp(left, right) { + const leftMs = timestampMs(left); + const rightMs = timestampMs(right); + if (!Number.isFinite(leftMs)) return toIsoTimestamp(right); + if (!Number.isFinite(rightMs)) return toIsoTimestamp(left); + return new Date(Math.min(leftMs, rightMs)).toISOString(); +} + +function laterTimestamp(left, right) { + const leftMs = timestampMs(left); + const rightMs = timestampMs(right); + if (!Number.isFinite(leftMs)) return toIsoTimestamp(right); + if (!Number.isFinite(rightMs)) return toIsoTimestamp(left); + return new Date(Math.max(leftMs, rightMs)).toISOString(); +} diff --git a/src/core/runtime-health.mjs b/src/core/runtime-health.mjs index 2b13887..a98ebb6 100644 --- a/src/core/runtime-health.mjs +++ b/src/core/runtime-health.mjs @@ -139,6 +139,20 @@ export function deriveServiceHealth({ label = 'writer lagging'; reasons.push('writer freshness degraded'); } + + if (state.retention_mode === 'blocked_invalid_policy') { + status = escalateHealth(status, 'warning'); + label = 'retention blocked'; + reasons.push('quote lifecycle retention policy is invalid or missing'); + } else if (state.retention_mode === 'blocked_rollup_stale') { + status = escalateHealth(status, 'warning'); + label = 'rollup stale'; + reasons.push('quote lifecycle rollup is stale, pruning is fail-closed'); + } else if (state.retention_mode === 'pressure') { + status = escalateHealth(status, 'warning'); + label = 'storage pressure'; + reasons.push('quote lifecycle storage pressure retention mode is active'); + } } if (service === 'operator-dashboard') { diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index ed3f2dc..90734cd 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -3,6 +3,16 @@ import { Pool } from 'pg'; import { deriveIntentRequestOutcomeRecords } from '../core/intent-request-outcomes.mjs'; import { buildCashEquivalentValuationAssets } from '../core/portfolio-metrics.mjs'; import { deriveQuoteOutcomeRecords } from '../core/quote-outcomes.mjs'; +import { + DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + QUOTE_LIFECYCLE_DATA_CLASSES, + QUOTE_LIFECYCLE_DETAIL_TABLES, + QUOTE_LIFECYCLE_RETENTION_POLICY_KEY, + buildQuoteLifecycleRollups, + decideQuoteLifecycleRetentionMode, + normalizeQuoteLifecycleRetentionPolicy, + normalizeQuoteLifecycleRollupRow, +} from '../core/quote-lifecycle-retention.mjs'; import { CURRENT_NBTC_ASSET_ID, CURRENT_USDC_ASSET_ID, @@ -39,6 +49,10 @@ const TABLES = [ const PORTFOLIO_METRICS_TABLE = 'portfolio_metrics_snapshots'; const QUOTE_OUTCOMES_TABLE = 'quote_outcome_attributions'; const INTENT_REQUEST_OUTCOMES_TABLE = 'intent_request_outcomes'; +const RETENTION_POLICIES_TABLE = 'retention_policies'; +const QUOTE_LIFECYCLE_ROLLUPS_TABLE = 'quote_lifecycle_rollups'; +const QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE = 'quote_lifecycle_rollup_watermarks'; +const QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE = 'quote_lifecycle_retention_runs'; const SUPPORTED_ASSET_IMPORT_RUNS_TABLE = 'supported_asset_import_runs'; const TRADING_ASSETS_TABLE = 'trading_assets'; const TRADING_PAIRS_TABLE = 'trading_pairs'; @@ -238,6 +252,8 @@ export async function ensureHistorySchema(pool) { ON ${QUOTE_OUTCOMES_TABLE} (outcome_status) `); + await ensureQuoteLifecycleRetentionSchema(pool); + await ensureExpressionIndex(pool, { name: 'intent_request_preflights_request_id_idx', table: 'intent_request_preflights', @@ -289,6 +305,170 @@ export async function ensureHistorySchema(pool) { `); } +export async function ensureQuoteLifecycleRetentionSchema(pool) { + await pool.query(` + CREATE TABLE IF NOT EXISTS ${RETENTION_POLICIES_TABLE} ( + policy_key TEXT PRIMARY KEY, + enabled BOOLEAN NOT NULL, + normal_detail_retention_ms BIGINT NOT NULL, + pressure_detail_retention_ms BIGINT NOT NULL, + raw_retention_ms BIGINT NOT NULL, + rollup_granularity_ms BIGINT NOT NULL, + rollup_retention_ms BIGINT NOT NULL, + rollup_stale_after_ms BIGINT NOT NULL, + preserve_successful_evidence BOOLEAN NOT NULL, + max_delete_rows_per_pass INTEGER NOT NULL, + pressure_table_bytes BIGINT NOT NULL, + pressure_database_bytes BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + config_version INTEGER NOT NULL + ) + `); + + await pool.query(` + CREATE TABLE IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} ( + rollup_id TEXT PRIMARY KEY, + window_start TIMESTAMPTZ NOT NULL, + window_end TIMESTAMPTZ NOT NULL, + data_class TEXT NOT NULL, + source_data_class TEXT NOT NULL, + pair TEXT, + direction TEXT NOT NULL, + request_kind TEXT NOT NULL, + edge_bps TEXT, + notional_asset TEXT, + notional_bucket TEXT NOT NULL, + quote_age_bucket TEXT NOT NULL, + result_code TEXT NOT NULL, + failure_category TEXT, + outcome_status TEXT NOT NULL, + quote_count INTEGER NOT NULL DEFAULT 0, + decision_count INTEGER NOT NULL DEFAULT 0, + command_count INTEGER NOT NULL DEFAULT 0, + executor_result_count INTEGER NOT NULL DEFAULT 0, + relay_accepted_count INTEGER NOT NULL DEFAULT 0, + relay_failed_count INTEGER NOT NULL DEFAULT 0, + not_filled_count INTEGER NOT NULL DEFAULT 0, + completed_count INTEGER NOT NULL DEFAULT 0, + timing_bucket_counts JSONB NOT NULL DEFAULT '{}'::jsonb, + min_observed_at TIMESTAMPTZ, + max_observed_at TIMESTAMPTZ, + source_watermark_at TIMESTAMPTZ, + computed_at TIMESTAMPTZ NOT NULL + ) + `); + await pool.query(` + CREATE UNIQUE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUPS_TABLE}_dimension_key_idx + ON ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} ( + window_start, + window_end, + source_data_class, + COALESCE(pair, ''), + direction, + request_kind, + COALESCE(edge_bps, ''), + COALESCE(notional_asset, ''), + notional_bucket, + quote_age_bucket, + result_code, + COALESCE(failure_category, ''), + outcome_status + ) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUPS_TABLE}_window_idx + ON ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} (window_end DESC, window_start DESC) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUPS_TABLE}_pair_idx + ON ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} (pair, window_end DESC) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUPS_TABLE}_source_class_idx + ON ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} (source_data_class, window_end DESC) + `); + + await pool.query(` + CREATE TABLE IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE} ( + rollup_key TEXT PRIMARY KEY, + covered_until TIMESTAMPTZ, + computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + status TEXT NOT NULL, + error JSONB, + row_count INTEGER NOT NULL DEFAULT 0, + rollup_count INTEGER NOT NULL DEFAULT 0 + ) + `); + + await pool.query(` + CREATE TABLE IF NOT EXISTS ${QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE} ( + run_id TEXT PRIMARY KEY, + started_at TIMESTAMPTZ NOT NULL, + completed_at TIMESTAMPTZ NOT NULL, + mode TEXT NOT NULL, + status TEXT NOT NULL, + policy_key TEXT, + detail_cutoff TIMESTAMPTZ, + raw_cutoff TIMESTAMPTZ, + rollup_covered_until TIMESTAMPTZ, + preserved_success_count INTEGER NOT NULL DEFAULT 0, + pruned_counts JSONB NOT NULL DEFAULT '{}'::jsonb, + rollup_counts JSONB NOT NULL DEFAULT '{}'::jsonb, + storage_metrics JSONB NOT NULL DEFAULT '{}'::jsonb, + error JSONB + ) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE}_completed_at_idx + ON ${QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE} (completed_at DESC) + `); + + await seedQuoteLifecycleRetentionPolicy(pool); +} + +export async function seedQuoteLifecycleRetentionPolicy(pool, { + now = new Date().toISOString(), + policy = DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, +} = {}) { + await pool.query( + ` + INSERT INTO ${RETENTION_POLICIES_TABLE} ( + policy_key, + enabled, + normal_detail_retention_ms, + pressure_detail_retention_ms, + raw_retention_ms, + rollup_granularity_ms, + rollup_retention_ms, + rollup_stale_after_ms, + preserve_successful_evidence, + max_delete_rows_per_pass, + pressure_table_bytes, + pressure_database_bytes, + updated_at, + config_version + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) + ON CONFLICT (policy_key) DO NOTHING + `, + [ + policy.policy_key, + policy.enabled, + policy.normal_detail_retention_ms, + policy.pressure_detail_retention_ms, + policy.raw_retention_ms, + policy.rollup_granularity_ms, + policy.rollup_retention_ms, + policy.rollup_stale_after_ms, + policy.preserve_successful_evidence, + policy.max_delete_rows_per_pass, + policy.pressure_table_bytes, + policy.pressure_database_bytes, + now, + policy.config_version, + ], + ); +} + export async function ensureTradingConfigSchema(pool) { await pool.query(` CREATE TABLE IF NOT EXISTS ${SUPPORTED_ASSET_IMPORT_RUNS_TABLE} ( @@ -2125,6 +2305,931 @@ export async function insertHistoryEvents(pool, entries = []) { return { insertedEventIds }; } +export async function loadQuoteLifecycleRetentionPolicy(pool, { + policyKey = QUOTE_LIFECYCLE_RETENTION_POLICY_KEY, +} = {}) { + const result = await pool.query( + ` + SELECT * + FROM ${RETENTION_POLICIES_TABLE} + WHERE policy_key = $1 + LIMIT 1 + `, + [policyKey], + ); + const normalized = normalizeQuoteLifecycleRetentionPolicy(result.rows[0] || null); + return { + ...normalized, + policy_key: policyKey, + }; +} + +export async function loadQuoteLifecycleStorageMetrics(pool) { + const tables = [ + ...QUOTE_LIFECYCLE_DETAIL_TABLES, + QUOTE_LIFECYCLE_ROLLUPS_TABLE, + QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE, + ]; + const [tableResult, databaseResult] = await Promise.all([ + pool.query( + ` + WITH table_names AS ( + SELECT UNNEST($1::text[]) AS table_name + ) + SELECT + table_names.table_name AS table, + COALESCE(pg_total_relation_size(to_regclass(table_names.table_name)), 0)::BIGINT AS total_bytes, + COALESCE(pg_relation_size(to_regclass(table_names.table_name)), 0)::BIGINT AS heap_bytes, + GREATEST(COALESCE(pg_total_relation_size(to_regclass(table_names.table_name)), 0)::BIGINT - COALESCE(pg_relation_size(to_regclass(table_names.table_name)), 0)::BIGINT, 0)::BIGINT AS index_bytes, + COALESCE(pg_stat_user_tables.n_live_tup, 0)::BIGINT AS approximate_rows + FROM table_names + LEFT JOIN pg_stat_user_tables + ON pg_stat_user_tables.relname = table_names.table_name + ORDER BY total_bytes DESC, table_names.table_name ASC + `, + [tables], + ), + pool.query(` + SELECT pg_database_size(current_database())::BIGINT AS total_bytes + `), + ]); + + return { + captured_at: new Date().toISOString(), + database: { + total_bytes: Number(databaseResult.rows[0]?.total_bytes || 0), + }, + tables: tableResult.rows.map((row) => ({ + table: row.table, + total_bytes: Number(row.total_bytes || 0), + heap_bytes: Number(row.heap_bytes || 0), + index_bytes: Number(row.index_bytes || 0), + approximate_rows: Number(row.approximate_rows || 0), + })), + }; +} + +export async function loadQuoteLifecycleRetentionSummary(pool, { + now = new Date().toISOString(), +} = {}) { + const [ + policyResult, + storageMetrics, + watermark, + latestRun, + rollupAggregate, + preservedSuccessCount, + ] = await Promise.all([ + loadQuoteLifecycleRetentionPolicy(pool), + loadQuoteLifecycleStorageMetrics(pool).catch((error) => ({ + captured_at: now, + database: { total_bytes: 0 }, + tables: [], + error: error.message, + })), + loadQuoteLifecycleRollupWatermark(pool), + loadLatestQuoteLifecycleRetentionRun(pool), + loadQuoteLifecycleRollupAggregate(pool), + countSuccessfulQuoteEvidence(pool), + ]); + + const modeDecision = decideQuoteLifecycleRetentionMode({ + policy: policyResult, + storageMetrics, + }); + const policy = policyResult.policy || null; + const detailCutoff = policy && modeDecision.detail_retention_ms + ? floorTimestamp( + new Date(Date.parse(now) - modeDecision.detail_retention_ms).toISOString(), + policy.rollup_granularity_ms, + ) + : null; + const rollupRequiredUntil = policy && detailCutoff + ? detailCutoff + : null; + const rollupCoveredUntil = watermark?.covered_until || null; + const rollupFresh = !policy + ? false + : rollupRequiredUntil == null + ? true + : timestampValue(rollupCoveredUntil) >= timestampValue(rollupRequiredUntil); + const rollupFreshnessAgeMs = watermark?.computed_at + ? Math.max(0, Date.parse(now) - timestampValue(watermark.computed_at)) + : null; + const retentionMode = modeDecision.mode === 'blocked_invalid_policy' + ? modeDecision.mode + : rollupFresh + ? modeDecision.mode + : 'blocked_rollup_stale'; + + return { + generated_at: now, + retention_mode: retentionMode, + mode_reason: retentionMode === 'blocked_rollup_stale' + ? 'rollup watermark does not cover the current prune cutoff' + : modeDecision.reason, + policy_ok: policyResult.ok, + policy_block_reason: policyResult.block_reason, + policy, + detail_retention_ms: modeDecision.detail_retention_ms, + raw_retention_ms: policy?.raw_retention_ms ?? null, + detail_cutoff: detailCutoff, + rollup_required_until: rollupRequiredUntil, + rollup_fresh: rollupFresh, + rollup_freshness_age_ms: Number.isFinite(rollupFreshnessAgeMs) ? rollupFreshnessAgeMs : null, + rollup_watermark: watermark, + latest_run: latestRun, + storage: storageMetrics, + rollups: rollupAggregate, + preserved_successful_quote_count: preservedSuccessCount, + }; +} + +export async function runQuoteLifecycleRetentionMaintenance(pool, { + now = new Date().toISOString(), +} = {}) { + const startedAt = new Date().toISOString(); + const policyResult = await loadQuoteLifecycleRetentionPolicy(pool); + let mode = 'blocked_invalid_policy'; + let status = 'skipped'; + let policy = policyResult.policy || null; + let storageMetrics = null; + let modeDecision = { mode, detail_retention_ms: null, reason: policyResult.block_reason }; + let detailCutoff = null; + let rawCutoff = null; + let rollupCutoff = null; + let rollupResult = null; + let pruneResult = null; + let preservedSuccessCount = 0; + let errorPayload = null; + + try { + storageMetrics = await loadQuoteLifecycleStorageMetrics(pool); + preservedSuccessCount = await countSuccessfulQuoteEvidence(pool); + + if (!policyResult.ok) { + errorPayload = { reason: policyResult.block_reason }; + return await finishRetentionRun(); + } + + modeDecision = decideQuoteLifecycleRetentionMode({ + policy: policyResult, + storageMetrics, + }); + mode = modeDecision.mode; + policy = policyResult.policy; + detailCutoff = floorTimestamp( + new Date(Date.parse(now) - modeDecision.detail_retention_ms).toISOString(), + policy.rollup_granularity_ms, + ); + rawCutoff = floorTimestamp( + new Date(Date.parse(now) - policy.raw_retention_ms).toISOString(), + policy.rollup_granularity_ms, + ); + rollupCutoff = laterTimestamp(detailCutoff, rawCutoff); + + rollupResult = await rollupQuoteLifecycleDetail(pool, { + now, + policy, + windowEnd: rollupCutoff, + }); + + const watermark = await loadQuoteLifecycleRollupWatermark(pool); + if (!watermark || timestampValue(watermark.covered_until) < timestampValue(rollupCutoff)) { + mode = 'blocked_rollup_stale'; + status = 'skipped'; + errorPayload = { + reason: 'rollup watermark does not cover prune cutoff', + rollup_cutoff: rollupCutoff, + covered_until: watermark?.covered_until || null, + }; + return await finishRetentionRun(); + } + + pruneResult = await pruneQuoteLifecycleDetailWithPolicy(pool, { + policy, + mode: modeDecision.mode, + now, + detailCutoff, + rawCutoff, + rollupCoveredUntil: watermark.covered_until, + }); + status = 'success'; + return await finishRetentionRun(); + } catch (error) { + mode = mode === 'blocked_invalid_policy' ? mode : 'blocked_rollup_stale'; + status = 'skipped'; + errorPayload = { message: error.message, name: error.name }; + return await finishRetentionRun(); + } + + async function finishRetentionRun() { + const completedAt = new Date().toISOString(); + const run = { + run_id: `quote-lifecycle-retention:${Date.parse(startedAt)}:${Math.random().toString(16).slice(2, 10)}`, + started_at: startedAt, + completed_at: completedAt, + mode, + status, + policy_key: policy?.policy_key || policyResult.policy_key || null, + detail_cutoff: detailCutoff, + raw_cutoff: rawCutoff, + rollup_covered_until: rollupResult?.covered_until || null, + preserved_success_count: preservedSuccessCount, + pruned_counts: pruneResult || {}, + rollup_counts: rollupResult || {}, + storage_metrics: storageMetrics || {}, + error: errorPayload, + }; + await insertQuoteLifecycleRetentionRun(pool, run); + return { + ...run, + policy_ok: policyResult.ok, + policy_block_reason: policyResult.block_reason, + detail_retention_ms: modeDecision.detail_retention_ms, + pressure: modeDecision.pressure === true, + }; + } +} + +export async function countQuoteLifecyclePruneCandidates(pool, { + policy, + mode = 'normal', + now = new Date().toISOString(), + detailCutoff = null, + rawCutoff = null, +} = {}) { + const normalized = normalizeQuoteLifecycleRetentionPolicy(policy); + if (!normalized.ok) { + return { + ok: false, + block_reason: normalized.block_reason, + tables: [], + total: 0, + }; + } + const activePolicy = normalized.policy; + const effectiveDetailCutoff = detailCutoff || floorTimestamp( + new Date(Date.parse(now) - ( + mode === 'pressure' + ? activePolicy.pressure_detail_retention_ms + : activePolicy.normal_detail_retention_ms + )).toISOString(), + activePolicy.rollup_granularity_ms, + ); + const effectiveRawCutoff = rawCutoff || floorTimestamp( + new Date(Date.parse(now) - activePolicy.raw_retention_ms).toISOString(), + activePolicy.rollup_granularity_ms, + ); + const tables = []; + let total = 0; + for (const spec of quoteLifecyclePruneTableSpecs()) { + const cutoff = spec.table === 'raw_near_intents_quotes' ? effectiveRawCutoff : effectiveDetailCutoff; + const result = await pool.query( + ` + SELECT COUNT(*)::INT AS count + FROM ${spec.table} target + WHERE COALESCE(target.${spec.timestampColumn}, 'epoch'::timestamptz) < $1::timestamptz + AND NOT ${successfulQuoteEvidencePredicate(spec)} + `, + [cutoff], + ); + const count = Number(result.rows[0]?.count || 0); + total += count; + tables.push({ + table: spec.table, + cutoff, + candidate_count: count, + }); + } + return { + ok: true, + total, + tables, + detail_cutoff: effectiveDetailCutoff, + raw_cutoff: effectiveRawCutoff, + }; +} + +async function rollupQuoteLifecycleDetail(pool, { + now, + policy, + windowEnd, +} = {}) { + const currentWatermark = await loadQuoteLifecycleRollupWatermark(pool); + const windowStart = currentWatermark?.covered_until + || await loadEarliestQuoteLifecycleTimestamp(pool); + const alignedWindowStart = windowStart + ? floorTimestamp(windowStart, policy.rollup_granularity_ms) + : windowEnd; + if (!alignedWindowStart || timestampValue(alignedWindowStart) >= timestampValue(windowEnd)) { + await upsertQuoteLifecycleRollupWatermark(pool, { + coveredUntil: windowEnd, + computedAt: now, + status: 'success', + rowCount: 0, + rollupCount: 0, + }); + return { + ok: true, + row_count: 0, + rollup_count: 0, + window_start: alignedWindowStart, + covered_until: windowEnd, + skipped: true, + }; + } + + const rows = await loadQuoteLifecycleRollupInputRows(pool, { + windowStart: alignedWindowStart, + windowEnd, + }); + const rollups = buildQuoteLifecycleRollups({ + rows, + rollupGranularityMs: policy.rollup_granularity_ms, + computedAt: now, + now, + }); + await upsertQuoteLifecycleRollups(pool, rollups); + await upsertQuoteLifecycleRollupWatermark(pool, { + coveredUntil: windowEnd, + computedAt: now, + status: 'success', + rowCount: rows.length, + rollupCount: rollups.length, + }); + return { + ok: true, + row_count: rows.length, + rollup_count: rollups.length, + window_start: alignedWindowStart, + covered_until: windowEnd, + }; +} + +async function pruneQuoteLifecycleDetailWithPolicy(pool, { + policy, + mode, + detailCutoff, + rawCutoff, + rollupCoveredUntil, +} = {}) { + const rollupCutoff = laterTimestamp(detailCutoff, rawCutoff); + if (timestampValue(rollupCoveredUntil) < timestampValue(rollupCutoff)) { + return { + ok: false, + blocked: true, + block_reason: 'rollup_watermark_stale', + deletedCount: 0, + tables: [], + }; + } + + const boundedBatchSize = Math.max( + 1, + Math.min(500_000, Math.floor(Number(policy.max_delete_rows_per_pass) || 0)), + ); + const tables = []; + let deletedCount = 0; + + for (const spec of quoteLifecyclePruneTableSpecs()) { + const cutoff = spec.table === 'raw_near_intents_quotes' ? rawCutoff : detailCutoff; + const result = await pool.query( + ` + WITH stale_rows AS ( + SELECT target.ctid + FROM ${spec.table} target + WHERE COALESCE(target.${spec.timestampColumn}, 'epoch'::timestamptz) < $1::timestamptz + AND NOT ${successfulQuoteEvidencePredicate(spec)} + ORDER BY target.${spec.timestampColumn} ASC NULLS FIRST + LIMIT $2 + ) + DELETE FROM ${spec.table} target + USING stale_rows + WHERE target.ctid = stale_rows.ctid + `, + [cutoff, boundedBatchSize], + ); + const tableDeletedCount = Number(result.rowCount || 0); + deletedCount += tableDeletedCount; + tables.push({ + table: spec.table, + cutoff, + deletedCount: tableDeletedCount, + }); + } + + return { + ok: true, + mode, + deletedCount, + batchSize: boundedBatchSize, + detail_cutoff: detailCutoff, + raw_cutoff: rawCutoff, + tables, + }; +} + +async function loadQuoteLifecycleRollupInputRows(pool, { + windowStart, + windowEnd, +} = {}) { + const linkedResult = await pool.query( + ` + WITH subject_events AS ( + SELECT quote_id, COALESCE(observed_at, ingested_at) AS activity_at + FROM raw_near_intents_quotes + WHERE quote_id IS NOT NULL + AND COALESCE(observed_at, ingested_at) >= $1::timestamptz + AND COALESCE(observed_at, ingested_at) < $2::timestamptz + UNION ALL + SELECT quote_id, COALESCE(observed_at, ingested_at) AS activity_at + FROM swap_demand_events + WHERE quote_id IS NOT NULL + AND COALESCE(observed_at, ingested_at) >= $1::timestamptz + AND COALESCE(observed_at, ingested_at) < $2::timestamptz + UNION ALL + SELECT quote_id, COALESCE(observed_at, ingested_at) AS activity_at + FROM trade_decisions + WHERE quote_id IS NOT NULL + AND COALESCE(observed_at, ingested_at) >= $1::timestamptz + AND COALESCE(observed_at, ingested_at) < $2::timestamptz + UNION ALL + SELECT quote_id, COALESCE(observed_at, ingested_at) AS activity_at + FROM execute_trade_commands + WHERE quote_id IS NOT NULL + AND COALESCE(observed_at, ingested_at) >= $1::timestamptz + AND COALESCE(observed_at, ingested_at) < $2::timestamptz + UNION ALL + SELECT quote_id, COALESCE(observed_at, ingested_at) AS activity_at + FROM trade_execution_results + WHERE quote_id IS NOT NULL + AND COALESCE(observed_at, ingested_at) >= $1::timestamptz + AND COALESCE(observed_at, ingested_at) < $2::timestamptz + UNION ALL + SELECT quote_id, COALESCE(outcome_observed_at, computed_at) AS activity_at + FROM quote_outcome_attributions + WHERE quote_id IS NOT NULL + AND COALESCE(outcome_observed_at, computed_at) >= $1::timestamptz + AND COALESCE(outcome_observed_at, computed_at) < $2::timestamptz + ), quote_subjects AS ( + SELECT quote_id, MIN(activity_at) AS activity_at + FROM subject_events + GROUP BY quote_id + ), latest_raw AS ( + SELECT DISTINCT ON (quote_id) + quote_id, pair, observed_at, ingested_at, payload + FROM raw_near_intents_quotes + WHERE quote_id IN (SELECT quote_id FROM quote_subjects) + ORDER BY quote_id, COALESCE(observed_at, ingested_at) DESC + ), latest_demand AS ( + SELECT DISTINCT ON (quote_id) + quote_id, pair, observed_at, ingested_at, payload + FROM swap_demand_events + WHERE quote_id IN (SELECT quote_id FROM quote_subjects) + ORDER BY quote_id, COALESCE(observed_at, ingested_at) DESC + ), latest_decision AS ( + SELECT DISTINCT ON (quote_id) + quote_id, pair, observed_at, ingested_at, payload + FROM trade_decisions + WHERE quote_id IN (SELECT quote_id FROM quote_subjects) + ORDER BY quote_id, COALESCE(observed_at, ingested_at) DESC + ), latest_command AS ( + SELECT DISTINCT ON (quote_id) + quote_id, pair, observed_at, ingested_at, payload + FROM execute_trade_commands + WHERE quote_id IN (SELECT quote_id FROM quote_subjects) + ORDER BY quote_id, COALESCE(observed_at, ingested_at) DESC + ), latest_execution AS ( + SELECT DISTINCT ON (quote_id) + quote_id, pair, observed_at, ingested_at, payload + FROM trade_execution_results + WHERE quote_id IN (SELECT quote_id FROM quote_subjects) + ORDER BY quote_id, COALESCE(observed_at, ingested_at) DESC + ) + SELECT + quote_subjects.quote_id, + quote_subjects.activity_at, + COALESCE(latest_demand.pair, latest_decision.pair, latest_command.pair, latest_execution.pair, latest_raw.pair) AS pair, + latest_demand.payload AS quote_payload, + latest_demand.observed_at AS quote_observed_at, + latest_demand.ingested_at AS quote_ingested_at, + latest_decision.payload AS decision_payload, + latest_decision.observed_at AS decision_at, + latest_command.payload AS command_payload, + latest_command.observed_at AS command_at, + latest_execution.payload AS execution_payload, + latest_execution.observed_at AS execution_result_at, + quote_outcome_attributions.payload AS outcome_payload, + quote_outcome_attributions.outcome_observed_at, + quote_outcome_attributions.computed_at + FROM quote_subjects + LEFT JOIN latest_raw ON latest_raw.quote_id = quote_subjects.quote_id + LEFT JOIN latest_demand ON latest_demand.quote_id = quote_subjects.quote_id + LEFT JOIN latest_decision ON latest_decision.quote_id = quote_subjects.quote_id + LEFT JOIN latest_command ON latest_command.quote_id = quote_subjects.quote_id + LEFT JOIN latest_execution ON latest_execution.quote_id = quote_subjects.quote_id + LEFT JOIN quote_outcome_attributions + ON quote_outcome_attributions.quote_id = quote_subjects.quote_id + ORDER BY quote_subjects.activity_at ASC + `, + [windowStart, windowEnd], + ); + + const rawDebugResult = await pool.query( + ` + SELECT + event_id, + NULL::text AS quote_id, + COALESCE(observed_at, ingested_at) AS activity_at, + pair, + payload AS quote_payload, + observed_at AS quote_observed_at, + ingested_at AS quote_ingested_at + FROM raw_near_intents_quotes + WHERE quote_id IS NULL + AND COALESCE(observed_at, ingested_at) >= $1::timestamptz + AND COALESCE(observed_at, ingested_at) < $2::timestamptz + ORDER BY COALESCE(observed_at, ingested_at) ASC + `, + [windowStart, windowEnd], + ); + + return [ + ...linkedResult.rows.map(normalizeQuoteLifecycleRollupInputRow), + ...rawDebugResult.rows.map((row) => ({ + ...normalizeQuoteLifecycleRollupInputRow(row), + source_table: 'raw_near_intents_quotes', + })), + ]; +} + +function normalizeQuoteLifecycleRollupInputRow(row) { + return { + quote_id: row.quote_id || null, + pair: row.pair || null, + activity_at: toIsoTimestamp(row.activity_at), + quote: withEventTimestamps(row.quote_payload, { + observed_at: row.quote_observed_at, + ingested_at: row.quote_ingested_at, + }), + decision: withEventTimestamps(row.decision_payload, { + observed_at: row.decision_at, + decision_at: row.decision_at, + }), + command: withEventTimestamps(row.command_payload, { + observed_at: row.command_at, + command_at: row.command_at, + }), + execution: withEventTimestamps(row.execution_payload, { + observed_at: row.execution_result_at, + result_at: row.execution_result_at, + }), + outcome: row.outcome_payload || null, + outcome_observed_at: toIsoTimestamp(row.outcome_observed_at), + computed_at: toIsoTimestamp(row.computed_at), + }; +} + +function withEventTimestamps(payload, timestamps = {}) { + if (!payload) return null; + const next = { ...payload }; + for (const [key, value] of Object.entries(timestamps)) { + if (value && !next[key]) next[key] = toIsoTimestamp(value); + } + return next; +} + +async function upsertQuoteLifecycleRollups(pool, rollups = []) { + for (const rollup of rollups || []) { + await pool.query( + ` + INSERT INTO ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} ( + rollup_id, + window_start, + window_end, + data_class, + source_data_class, + pair, + direction, + request_kind, + edge_bps, + notional_asset, + notional_bucket, + quote_age_bucket, + result_code, + failure_category, + outcome_status, + quote_count, + decision_count, + command_count, + executor_result_count, + relay_accepted_count, + relay_failed_count, + not_filled_count, + completed_count, + timing_bucket_counts, + min_observed_at, + max_observed_at, + source_watermark_at, + computed_at + ) VALUES ( + $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24::jsonb,$25,$26,$27,$28 + ) + ON CONFLICT (rollup_id) DO UPDATE SET + quote_count = EXCLUDED.quote_count, + decision_count = EXCLUDED.decision_count, + command_count = EXCLUDED.command_count, + executor_result_count = EXCLUDED.executor_result_count, + relay_accepted_count = EXCLUDED.relay_accepted_count, + relay_failed_count = EXCLUDED.relay_failed_count, + not_filled_count = EXCLUDED.not_filled_count, + completed_count = EXCLUDED.completed_count, + timing_bucket_counts = EXCLUDED.timing_bucket_counts, + min_observed_at = EXCLUDED.min_observed_at, + max_observed_at = EXCLUDED.max_observed_at, + source_watermark_at = EXCLUDED.source_watermark_at, + computed_at = EXCLUDED.computed_at + `, + [ + rollup.rollup_id, + rollup.window_start, + rollup.window_end, + rollup.data_class, + rollup.source_data_class, + rollup.pair, + rollup.direction, + rollup.request_kind, + rollup.edge_bps, + rollup.notional_asset, + rollup.notional_bucket, + rollup.quote_age_bucket, + rollup.result_code, + rollup.failure_category, + rollup.outcome_status, + rollup.quote_count, + rollup.decision_count, + rollup.command_count, + rollup.executor_result_count, + rollup.relay_accepted_count, + rollup.relay_failed_count, + rollup.not_filled_count, + rollup.completed_count, + JSON.stringify(rollup.timing_bucket_counts || {}), + rollup.min_observed_at, + rollup.max_observed_at, + rollup.source_watermark_at, + rollup.computed_at, + ], + ); + } +} + +async function loadQuoteLifecycleRollupWatermark(pool) { + const result = await pool.query( + ` + SELECT * + FROM ${QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE} + WHERE rollup_key = 'quote_lifecycle' + LIMIT 1 + `, + ); + const row = result.rows[0]; + if (!row) return null; + return { + rollup_key: row.rollup_key, + covered_until: toIsoTimestamp(row.covered_until), + computed_at: toIsoTimestamp(row.computed_at), + status: row.status, + error: row.error || null, + row_count: Number(row.row_count || 0), + rollup_count: Number(row.rollup_count || 0), + }; +} + +async function upsertQuoteLifecycleRollupWatermark(pool, { + coveredUntil, + computedAt, + status, + error = null, + rowCount = 0, + rollupCount = 0, +}) { + await pool.query( + ` + INSERT INTO ${QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE} ( + rollup_key, + covered_until, + computed_at, + status, + error, + row_count, + rollup_count + ) VALUES ('quote_lifecycle', $1, $2, $3, $4::jsonb, $5, $6) + ON CONFLICT (rollup_key) DO UPDATE SET + covered_until = GREATEST( + COALESCE(${QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE}.covered_until, 'epoch'::timestamptz), + EXCLUDED.covered_until + ), + computed_at = EXCLUDED.computed_at, + status = EXCLUDED.status, + error = EXCLUDED.error, + row_count = EXCLUDED.row_count, + rollup_count = EXCLUDED.rollup_count + `, + [ + coveredUntil, + computedAt, + status, + error ? JSON.stringify(error) : null, + rowCount, + rollupCount, + ], + ); +} + +async function loadEarliestQuoteLifecycleTimestamp(pool) { + const result = await pool.query(` + SELECT MIN(activity_at) AS earliest_at + FROM ( + SELECT COALESCE(observed_at, ingested_at) AS activity_at FROM raw_near_intents_quotes + UNION ALL + SELECT COALESCE(observed_at, ingested_at) AS activity_at FROM swap_demand_events + UNION ALL + SELECT COALESCE(observed_at, ingested_at) AS activity_at FROM trade_decisions + UNION ALL + SELECT COALESCE(observed_at, ingested_at) AS activity_at FROM execute_trade_commands + UNION ALL + SELECT COALESCE(observed_at, ingested_at) AS activity_at FROM trade_execution_results + UNION ALL + SELECT COALESCE(outcome_observed_at, computed_at) AS activity_at FROM quote_outcome_attributions + ) lifecycle_times + `); + return toIsoTimestamp(result.rows[0]?.earliest_at); +} + +async function loadLatestQuoteLifecycleRetentionRun(pool) { + const result = await pool.query(` + SELECT * + FROM ${QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE} + ORDER BY completed_at DESC + LIMIT 1 + `); + const row = result.rows[0]; + if (!row) return null; + return { + run_id: row.run_id, + started_at: toIsoTimestamp(row.started_at), + completed_at: toIsoTimestamp(row.completed_at), + mode: row.mode, + status: row.status, + policy_key: row.policy_key || null, + detail_cutoff: toIsoTimestamp(row.detail_cutoff), + raw_cutoff: toIsoTimestamp(row.raw_cutoff), + rollup_covered_until: toIsoTimestamp(row.rollup_covered_until), + preserved_success_count: Number(row.preserved_success_count || 0), + pruned_counts: row.pruned_counts || {}, + rollup_counts: row.rollup_counts || {}, + storage_metrics: row.storage_metrics || {}, + error: row.error || null, + }; +} + +async function loadQuoteLifecycleRollupAggregate(pool) { + const result = await pool.query(` + SELECT + COUNT(*)::INT AS rollup_row_count, + MAX(computed_at) AS latest_rollup_at, + MAX(window_end) AS latest_window_end, + COALESCE(SUM(quote_count), 0)::BIGINT AS quote_count, + COALESCE(SUM(decision_count), 0)::BIGINT AS decision_count, + COALESCE(SUM(command_count), 0)::BIGINT AS command_count, + COALESCE(SUM(executor_result_count), 0)::BIGINT AS executor_result_count, + COALESCE(SUM(relay_accepted_count), 0)::BIGINT AS relay_accepted_count, + COALESCE(SUM(relay_failed_count), 0)::BIGINT AS relay_failed_count, + COALESCE(SUM(not_filled_count), 0)::BIGINT AS not_filled_count, + COALESCE(SUM(completed_count), 0)::BIGINT AS completed_count + FROM ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} + `); + const row = result.rows[0] || {}; + return { + rollup_row_count: Number(row.rollup_row_count || 0), + latest_rollup_at: toIsoTimestamp(row.latest_rollup_at), + latest_window_end: toIsoTimestamp(row.latest_window_end), + quote_count: Number(row.quote_count || 0), + decision_count: Number(row.decision_count || 0), + command_count: Number(row.command_count || 0), + executor_result_count: Number(row.executor_result_count || 0), + relay_accepted_count: Number(row.relay_accepted_count || 0), + relay_failed_count: Number(row.relay_failed_count || 0), + not_filled_count: Number(row.not_filled_count || 0), + completed_count: Number(row.completed_count || 0), + }; +} + +export async function loadRecentQuoteLifecycleRollups(pool, { limit = 50 } = {}) { + const result = await pool.query( + ` + SELECT * + FROM ${QUOTE_LIFECYCLE_ROLLUPS_TABLE} + ORDER BY window_end DESC, quote_count DESC + LIMIT $1 + `, + [Math.max(1, Number(limit) || 50)], + ); + return result.rows.map(normalizeQuoteLifecycleRollupRow); +} + +async function countSuccessfulQuoteEvidence(pool) { + const result = await pool.query(` + SELECT COUNT(*)::INT AS count + FROM quote_outcome_attributions + WHERE ${successfulQuoteEvidenceSelfPredicate()} + `); + return Number(result.rows[0]?.count || 0); +} + +async function insertQuoteLifecycleRetentionRun(pool, run) { + await pool.query( + ` + INSERT INTO ${QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE} ( + run_id, + started_at, + completed_at, + mode, + status, + policy_key, + detail_cutoff, + raw_cutoff, + rollup_covered_until, + preserved_success_count, + pruned_counts, + rollup_counts, + storage_metrics, + error + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::jsonb,$12::jsonb,$13::jsonb,$14::jsonb) + `, + [ + run.run_id, + run.started_at, + run.completed_at, + run.mode, + run.status, + run.policy_key, + run.detail_cutoff, + run.raw_cutoff, + run.rollup_covered_until, + run.preserved_success_count, + JSON.stringify(run.pruned_counts || {}), + JSON.stringify(run.rollup_counts || {}), + JSON.stringify(run.storage_metrics || {}), + run.error ? JSON.stringify(run.error) : null, + ], + ); +} + +function quoteLifecyclePruneTableSpecs() { + return [ + { table: 'raw_near_intents_quotes', timestampColumn: 'ingested_at' }, + { table: 'swap_demand_events', timestampColumn: 'ingested_at' }, + { table: 'trade_decisions', timestampColumn: 'ingested_at' }, + { table: 'execute_trade_commands', timestampColumn: 'ingested_at' }, + { table: 'trade_execution_results', timestampColumn: 'ingested_at' }, + { table: 'quote_outcome_attributions', timestampColumn: 'computed_at', selfOutcome: true }, + ]; +} + +function successfulQuoteEvidencePredicate(spec) { + if (spec.selfOutcome) return successfulQuoteEvidenceTargetPredicate(); + return ` + EXISTS ( + SELECT 1 + FROM quote_outcome_attributions outcome + WHERE outcome.quote_id = target.quote_id + AND ${successfulQuoteEvidenceSelfPredicate('outcome')} + ) + `; +} + +function successfulQuoteEvidenceSelfPredicate(alias = null) { + const prefix = alias ? `${alias}.` : ''; + return ` + ( + ${prefix}outcome_status = 'completed' + OR ${prefix}attribution_status IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') + ) + `; +} + +function successfulQuoteEvidenceTargetPredicate() { + return ` + ( + target.outcome_status = 'completed' + OR target.attribution_status IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') + ) + `; +} + export async function pruneRawNearIntentsQuoteHistory(pool, { now = new Date().toISOString(), retainRecentMs = 30 * 60 * 1000, @@ -2223,7 +3328,7 @@ export async function pruneNonSuccessfulQuoteLifecycleHistory(pool, { ? ` ( target.outcome_status = 'completed' - OR target.attribution_status IN ('heuristic_match', 'exact_match', 'attributed') + OR target.attribution_status IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') ) ` : ` @@ -2233,7 +3338,7 @@ export async function pruneNonSuccessfulQuoteLifecycleHistory(pool, { WHERE outcome.quote_id = target.quote_id AND ( outcome.outcome_status = 'completed' - OR outcome.attribution_status IN ('heuristic_match', 'exact_match', 'attributed') + OR outcome.attribution_status IN ('heuristic_match', 'linked_settlement', 'exact_match', 'attributed') ) ) `; @@ -3992,6 +5097,21 @@ function timestampValue(value) { return Number.isFinite(parsed) ? parsed : -Infinity; } +function floorTimestamp(value, granularityMs) { + const parsed = Date.parse(value || ''); + const granularity = Number(granularityMs); + if (!Number.isFinite(parsed) || !Number.isInteger(granularity) || granularity <= 0) return null; + return new Date(Math.floor(parsed / granularity) * granularity).toISOString(); +} + +function laterTimestamp(left, right) { + const leftMs = timestampValue(left); + const rightMs = timestampValue(right); + if (!Number.isFinite(leftMs)) return toIsoTimestamp(right); + if (!Number.isFinite(rightMs)) return toIsoTimestamp(left); + return new Date(Math.max(leftMs, rightMs)).toISOString(); +} + function buildDepositStatusRowKey(payload) { const details = payload?.details || {}; return [ diff --git a/src/operator-dashboard/static/lib/format.js b/src/operator-dashboard/static/lib/format.js index 5d99dd5..5615e73 100644 --- a/src/operator-dashboard/static/lib/format.js +++ b/src/operator-dashboard/static/lib/format.js @@ -44,6 +44,20 @@ export function formatAgeFromTimestamp(value, now = Date.now()) { return formatAge(now - timestamp); } +export function formatBytes(value) { + const numeric = Number(value); + if (!Number.isFinite(numeric)) return 'Unavailable'; + const units = ['B', 'KiB', 'MiB', 'GiB', 'TiB']; + let amount = Math.max(0, numeric); + let unitIndex = 0; + while (amount >= 1024 && unitIndex < units.length - 1) { + amount /= 1024; + unitIndex += 1; + } + const digits = amount >= 100 || unitIndex === 0 ? 0 : amount >= 10 ? 1 : 2; + return `${amount.toFixed(digits)} ${units[unitIndex]}`; +} + export function formatEur(value) { if (value == null || value === '') return 'Unavailable'; const numeric = Number(value); diff --git a/src/operator-dashboard/static/pages/StrategyPage.jsx b/src/operator-dashboard/static/pages/StrategyPage.jsx index 9b25867..c2f2e8f 100644 --- a/src/operator-dashboard/static/pages/StrategyPage.jsx +++ b/src/operator-dashboard/static/pages/StrategyPage.jsx @@ -11,6 +11,7 @@ const TRADING_PAIR_MODES = new Set(['maker', 'taker', 'both']); const COMPETITIVENESS_GROUP_ROW_COUNT = 12; const COMPETITIVENESS_DETAIL_ROW_COUNT = 8; const COMPETITIVENESS_LATENCY_ROW_COUNT = 6; +const COMPETITIVENESS_ROLLUP_ROW_COUNT = 8; const QUOTE_LIFECYCLE_ROW_COUNT = 20; async function copyIdentifier(value) { @@ -316,11 +317,13 @@ function MakerCompetitivenessSection({ summary, pairConfig }) { const latestErrors = displaySummary?.latest_errors || []; const policySkips = displaySummary?.policy_skips || []; const latencyStages = displaySummary?.latency_stages || []; + const persistedRollups = displaySummary?.persisted_rollups || []; const groupRows = fixedRows(groups, COMPETITIVENESS_GROUP_ROW_COUNT); const ageBucketRows = fixedRows(ageBuckets, COMPETITIVENESS_GROUP_ROW_COUNT); const latestErrorRows = fixedRows(latestErrors, COMPETITIVENESS_DETAIL_ROW_COUNT); const policySkipRows = fixedRows(policySkips, COMPETITIVENESS_DETAIL_ROW_COUNT); const latencyStageRows = fixedRows(latencyStages, COMPETITIVENESS_LATENCY_ROW_COUNT); + const rollupRows = fixedRows(persistedRollups, COMPETITIVENESS_ROLLUP_ROW_COUNT); useEffect(() => { latestSummaryRef.current = summary || {}; @@ -362,7 +365,7 @@ function MakerCompetitivenessSection({ summary, pairConfig }) {
- Live rows update as they arrive. Fixed row slots and clamped cells keep the table height stable. + Live rows update as they arrive. Older pruned detail is represented by aggregate rollups, not per-quote evidence.
@@ -498,6 +501,61 @@ function MakerCompetitivenessSection({ summary, pairConfig }) {
+ + + + + + + + + + + + + + + + {rollupRows.map((rollup, index) => { + if (!rollup) { + return ( + + + + ); + } + return ( + + + + + + + + + + + ); + })} + +
Rollup windowPairRequestAge / notionalResultOutcomeCountsSource class
{index === 0 && !persistedRollups.length ? 'No persisted aggregate rollups are available yet.' : ''}
+
{formatTimestamp(rollup.window_end)}
+
{formatTimestamp(rollup.window_start)}
+ 		+
{pairDisplayLabel(rollup.pair, pairConfig)}
+
{truncateMiddle(rollup.pair || '', 42)}
+ 		{plainCodeLabel(rollup.request_kind)} +
{rollup.quote_age_bucket}
+
{rollup.notional_bucket}
+ 		+
{plainCodeLabel(rollup.result_code)}
+ {rollup.failure_category ?
{plainCodeLabel(rollup.failure_category)}
: null} + 		{plainCodeLabel(rollup.outcome_status)} +
{Number(rollup.quote_count || 0).toLocaleString()} quotes
+
{`${Number(rollup.completed_count || 0).toLocaleString()} completed / ${Number(rollup.relay_failed_count || 0).toLocaleString()} failed`}
+ 		{plainCodeLabel(rollup.source_data_class)}
+ +
diff --git a/src/operator-dashboard/static/pages/SystemPage.jsx b/src/operator-dashboard/static/pages/SystemPage.jsx index 2c71a9d..ab77ad2 100644 --- a/src/operator-dashboard/static/pages/SystemPage.jsx +++ b/src/operator-dashboard/static/pages/SystemPage.jsx @@ -1,9 +1,18 @@ import MetricCard from '../components/MetricCard.jsx'; import ServiceCard from '../components/ServiceCard.jsx'; import TableFrame from '../components/TableFrame.jsx'; -import { formatBoolean, formatTimestamp } from '../lib/format.js'; +import { formatAge, formatBoolean, formatBytes, formatTimestamp } from '../lib/format.js'; export default function SystemPage({ system, onControl }) { + const retention = system.persistence.retention_summary || {}; + const policy = retention.policy || {}; + const latestRun = retention.latest_run || {}; + const rollups = retention.rollups || {}; + const storage = retention.storage || {}; + const prunedByTable = new Map( + (latestRun.pruned_counts?.tables || []).map((entry) => [entry.table, entry]), + ); + return ( <>
@@ -119,6 +128,104 @@ export default function SystemPage({ system, onControl }) { value={formatTimestamp(system.persistence.last_quote_outcomes_at)} /> + +
+ + + + + + +
+ + +
+ + + + + + + + + + {storage.tables?.length ? ( + storage.tables.map((entry) => { + const pruned = prunedByTable.get(entry.table); + return ( + + + + + + + ); + }) + ) : ( + + + + )} + +
TableApprox rowsTotal sizeLatest pruned
{entry.table}{Number(entry.approximate_rows || 0).toLocaleString()}{formatBytes(entry.total_bytes)}{pruned ? Number(pruned.deletedCount || 0).toLocaleString() : '0'}
No quote lifecycle storage metrics are available yet.
+ + + + + + + + + + + + + + + + {system.persistence.quote_lifecycle_rollups?.length ? ( + system.persistence.quote_lifecycle_rollups.slice(0, 10).map((rollup) => ( + + + + + + + + + )) + ) : ( + + + + )} + +
Window endPairResultOutcomeQuotesCompleted
{formatTimestamp(rollup.window_end)}{rollup.pair || 'unknown'}{rollup.result_code || 'no_result'}{rollup.outcome_status || 'unknown'}{Number(rollup.quote_count || 0).toLocaleString()}{Number(rollup.completed_count || 0).toLocaleString()}
No aggregate quote lifecycle rollups are available yet.
+ diff --git a/src/operator-dashboard/static/styles.css b/src/operator-dashboard/static/styles.css index 4fe2341..1a4f942 100644 --- a/src/operator-dashboard/static/styles.css +++ b/src/operator-dashboard/static/styles.css @@ -327,6 +327,10 @@ select { min-width: 720px; } +.storage-retention-table td { + white-space: nowrap; +} + .table-wrap { overflow-x: auto; border: 1px solid var(--line); diff --git a/test/history-writer-static.test.mjs b/test/history-writer-static.test.mjs index f25fc72..43a86ad 100644 --- a/test/history-writer-static.test.mjs +++ b/test/history-writer-static.test.mjs @@ -20,20 +20,16 @@ test('history writer replays durable topics but joins the raw quote firehose liv assert.match(source, /runHistoryConsumer\(rawQuoteConsumer\)/); assert.match(source, /eachBatch/); assert.match(source, /insertHistoryEvents/); - assert.match(source, /rawQuoteHistoryPruneIntervalMs\s*=\s*60 \* 1000/); - assert.match(source, /rawQuoteHistoryRetainRecentMs\s*=\s*30 \* 60 \* 1000/); - assert.match(source, /rawQuoteHistoryPruneBatchSize\s*=\s*500_000/); - assert.match(source, /rawQuoteHistoryPruneMaxBatches\s*=\s*20/); - assert.match(source, /quoteLifecycleHistoryPruneIntervalMs\s*=\s*60 \* 1000/); - assert.match(source, /quoteLifecycleHistoryRetainRecentMs\s*=\s*30 \* 60 \* 1000/); - assert.match(source, /quoteLifecycleHistoryPruneBatchSize\s*=\s*100_000/); - assert.match(source, /quoteLifecycleHistoryPruneMaxBatches\s*=\s*10/); - assert.match(source, /pruneRawNearIntentsQuoteHistory/); - assert.match(source, /pruneNonSuccessfulQuoteLifecycleHistory/); - assert.match(source, /maxBatches:\s*rawQuoteHistoryPruneMaxBatches/); - assert.match(source, /maxBatches:\s*quoteLifecycleHistoryPruneMaxBatches/); - assert.match(source, /batch\.topic === config\.kafkaTopicRawNearIntentsQuote/); - assert.match(source, /maybePruneQuoteLifecycleHistory/); + assert.match(source, /quoteLifecycleRetentionMaintenanceIntervalMs\s*=\s*60 \* 1000/); + assert.match(source, /runQuoteLifecycleRetentionMaintenance/); + assert.match(source, /loadQuoteLifecycleRetentionSummary/); + assert.match(source, /maybeRunQuoteLifecycleRetentionMaintenance/); + assert.match(source, /retention_mode/); + assert.match(source, /storage_pressure/); + assert.doesNotMatch(source, /RetainRecentMs/); + assert.doesNotMatch(source, /PruneBatchSize/); + assert.doesNotMatch(source, /pruneRawNearIntentsQuoteHistory/); + assert.doesNotMatch(source, /pruneNonSuccessfulQuoteLifecycleHistory/); }); test('history writer passes tracked assets into portfolio valuation', () => { diff --git a/test/inventory-and-history.test.mjs b/test/inventory-and-history.test.mjs index ca4e5ee..78599aa 100644 --- a/test/inventory-and-history.test.mjs +++ b/test/inventory-and-history.test.mjs @@ -194,7 +194,7 @@ test('quote lifecycle retention preserves completed trades and drains old non-su for (const query of queries) { assert.match(query.sql, /DELETE FROM/); assert.match(query.sql, /outcome_status = 'completed'/); - assert.match(query.sql, /attribution_status IN \('heuristic_match', 'exact_match', 'attributed'\)/); + assert.match(query.sql, /attribution_status IN \('heuristic_match', 'linked_settlement', 'exact_match', 'attributed'\)/); assert.deepEqual(query.params, ['2026-06-05T11:30:00.000Z', 100_000]); } assert.match(queries.at(-1).sql, /DELETE FROM quote_outcome_attributions/); diff --git a/test/operator-dashboard-ui-static.test.mjs b/test/operator-dashboard-ui-static.test.mjs index c9489cc..3914bdd 100644 --- a/test/operator-dashboard-ui-static.test.mjs +++ b/test/operator-dashboard-ui-static.test.mjs @@ -146,6 +146,8 @@ test('strategy page exposes maker timing waterfall and competitiveness summaries assert.match(strategySource, /quote_to_relay_result_ms/); assert.match(strategySource, /quote_not_found_or_finished/); assert.match(strategySource, /maker_competitiveness/); + assert.match(strategySource, /persisted_rollups/); + assert.match(strategySource, /aggregate rollups/); assert.match(strategySource, /pairDisplayLabel/); assert.match(strategySource, /group\.edge_bps/); assert.match(strategySource, /Pause updates/); @@ -164,6 +166,16 @@ test('strategy page exposes maker timing waterfall and competitiveness summaries assert.match(stylesSource, /\.timing-waterfall-table/); }); +test('system page exposes quote lifecycle retention and storage rollup truth', () => { + assert.match(systemSource, /Retention mode/); + assert.match(systemSource, /Rollup freshness/); + assert.match(systemSource, /Preserved successes/); + assert.match(systemSource, /storage-retention-table/); + assert.match(systemSource, /quote_lifecycle_rollups/); + assert.match(systemSource, /formatBytes/); + assert.match(stylesSource, /\.storage-retention-table td/); +}); + test('strategy page distinguishes configured bps from gross quote edge percent', () => { assert.match(strategySource, /formatConfiguredEdgeBps/); assert.match(strategySource, /Configured edge/); diff --git a/test/operator-dashboard.test.mjs b/test/operator-dashboard.test.mjs index 83cb2da..2e6c4f0 100644 --- a/test/operator-dashboard.test.mjs +++ b/test/operator-dashboard.test.mjs @@ -232,6 +232,79 @@ test('control routing only resolves the allowlisted safe dashboard actions', () assert.equal(risky, null); }); +test('dashboard bootstrap exposes quote lifecycle retention and rollup storage truth', () => { + const payload = buildDashboardBootstrap({ + config: buildDualBtcConfig(), + 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: [], + quoteLifecycleRetention: { + retention_mode: 'normal', + detail_retention_ms: 30 * 60 * 1000, + raw_retention_ms: 30 * 60 * 1000, + rollup_required_until: '2026-06-06T10:30:00.000Z', + rollup_fresh: true, + policy: { + policy_key: 'quote_lifecycle_default', + raw_retention_ms: 30 * 60 * 1000, + }, + latest_run: { + completed_at: '2026-06-06T11:00:00.000Z', + pruned_counts: { + deletedCount: 7, + tables: [{ table: 'trade_decisions', deletedCount: 7 }], + }, + }, + storage: { + database: { total_bytes: 1000 }, + tables: [{ table: 'trade_decisions', total_bytes: 500, approximate_rows: 20 }], + }, + rollups: { + latest_rollup_at: '2026-06-06T10:59:00.000Z', + rollup_row_count: 3, + }, + preserved_successful_quote_count: 2, + }, + quoteLifecycleRollups: [{ + rollup_id: 'rollup-1', + window_end: '2026-06-06T10:30:00.000Z', + pair: 'a->b', + result_code: 'submission_failed', + outcome_status: 'relay_failed', + quote_count: 4, + completed_count: 0, + }], + recentIntentRequests: [], + recentAlertTransitions: [], + recentEnvironmentStatuses: [], + serviceSnapshots: [{ + service: 'history-writer', + label: 'History Writer', + reachable: true, + state: { + database_connectivity: true, + retention_mode: 'normal', + }, + health: { ok: true }, + }], + }); + + assert.equal(payload.system.persistence.retention_summary.retention_mode, 'normal'); + assert.equal(payload.system.persistence.preserved_successful_quote_count, 2); + assert.equal(payload.system.persistence.quote_lifecycle_rollups[0].rollup_id, 'rollup-1'); + assert.equal(payload.strategy.strategy_state.maker_competitiveness.persisted_rollup_count, 1); +}); + test('basic auth resolves operator identity and reuses a session cookie', () => { const authorizationHeader = `Basic ${Buffer.from('admin:secret-password').toString('base64')}`; const first = resolveDashboardRequestAuth({ diff --git a/test/ops-sentinel-static.test.mjs b/test/ops-sentinel-static.test.mjs index 2d43810..ff3b7fe 100644 --- a/test/ops-sentinel-static.test.mjs +++ b/test/ops-sentinel-static.test.mjs @@ -34,3 +34,11 @@ test('ops sentinel turns dashboard maker competitiveness truth into runtime aler assert.match(source, /buildMakerCompetitivenessRuntimeAlerts/); assert.match(source, /latest_maker_competitiveness/); }); + +test('ops sentinel turns quote lifecycle retention health into storage alerts', () => { + assert.match(source, /quote_lifecycle_retention_policy_invalid/); + assert.match(source, /quote_lifecycle_rollup_stale/); + assert.match(source, /quote_lifecycle_storage_pressure/); + assert.match(source, /writerState\.retention_mode/); + assert.match(source, /writerState\.retention_summary/); +}); diff --git a/test/quote-lifecycle-retention.test.mjs b/test/quote-lifecycle-retention.test.mjs new file mode 100644 index 0000000..180891a --- /dev/null +++ b/test/quote-lifecycle-retention.test.mjs @@ -0,0 +1,299 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { extendMakerTiming } from '../src/core/maker-timing.mjs'; +import { + DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + QUOTE_LIFECYCLE_DATA_CLASSES, + buildQuoteLifecycleRollups, + classifyQuoteLifecycleDataClass, + decideQuoteLifecycleRetentionMode, + normalizeQuoteLifecycleRetentionPolicy, +} from '../src/core/quote-lifecycle-retention.mjs'; +import { + countQuoteLifecyclePruneCandidates, + runQuoteLifecycleRetentionMaintenance, +} from '../src/lib/postgres.mjs'; + +test('quote lifecycle classifier separates success, in-flight, non-success, raw firehose, and rollups', () => { + assert.equal( + classifyQuoteLifecycleDataClass({ + table: 'quote_outcome_attributions', + row: { outcome_status: 'completed', attribution_status: 'heuristic_match' }, + }), + QUOTE_LIFECYCLE_DATA_CLASSES.successfulTradeEvidence, + ); + assert.equal( + classifyQuoteLifecycleDataClass({ + table: 'quote_outcome_attributions', + row: { outcome_status: 'submitted', attribution_status: 'linked_settlement' }, + }), + QUOTE_LIFECYCLE_DATA_CLASSES.successfulTradeEvidence, + ); + assert.equal( + classifyQuoteLifecycleDataClass({ + table: 'raw_near_intents_quotes', + row: { ingested_at: '2026-06-06T10:00:00.000Z', quote_id: null }, + now: '2026-06-06T11:00:00.000Z', + }), + QUOTE_LIFECYCLE_DATA_CLASSES.rawDebugFirehose, + ); + assert.equal( + classifyQuoteLifecycleDataClass({ + table: 'trade_decisions', + row: { ingested_at: '2026-06-06T10:55:00.000Z', quote_id: 'quote-recent' }, + now: '2026-06-06T11:00:00.000Z', + detailRetentionMs: 10 * 60 * 1000, + }), + QUOTE_LIFECYCLE_DATA_CLASSES.inFlightDetail, + ); + assert.equal( + classifyQuoteLifecycleDataClass({ + table: 'trade_execution_results', + row: { ingested_at: '2026-06-06T09:00:00.000Z', quote_id: 'quote-old' }, + now: '2026-06-06T11:00:00.000Z', + detailRetentionMs: 10 * 60 * 1000, + }), + QUOTE_LIFECYCLE_DATA_CLASSES.nonSuccessDetail, + ); + assert.equal( + classifyQuoteLifecycleDataClass({ + table: 'quote_lifecycle_rollups', + row: { data_class: 'aggregate_rollup' }, + }), + QUOTE_LIFECYCLE_DATA_CLASSES.aggregateRollup, + ); +}); + +test('retention policy validation fails closed on missing or unsafe policy fields', () => { + assert.equal(normalizeQuoteLifecycleRetentionPolicy(null).ok, false); + assert.match(normalizeQuoteLifecycleRetentionPolicy(null).block_reason, /missing/); + + const unsafe = normalizeQuoteLifecycleRetentionPolicy({ + ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + preserve_successful_evidence: false, + }); + assert.equal(unsafe.ok, false); + assert.match(unsafe.block_reason, /preserve_successful_evidence/); +}); + +test('retention mode enters pressure only from DB-backed policy thresholds', () => { + const normal = decideQuoteLifecycleRetentionMode({ + policy: DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + storageMetrics: { + database: { total_bytes: 10 }, + tables: [{ table: 'trade_decisions', total_bytes: 100 }], + }, + }); + assert.equal(normal.mode, 'normal'); + assert.equal(normal.detail_retention_ms, DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.normal_detail_retention_ms); + + const pressure = decideQuoteLifecycleRetentionMode({ + policy: DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + storageMetrics: { + database: { total_bytes: 10 }, + tables: [{ + table: 'trade_decisions', + total_bytes: DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.pressure_table_bytes, + }], + }, + }); + assert.equal(pressure.mode, 'pressure'); + assert.equal(pressure.detail_retention_ms, DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.pressure_detail_retention_ms); +}); + +test('quote lifecycle rollups group by pair, direction, request, edge, result, failure, age, notional, and outcome', () => { + const nbtc = 'nep141:nbtc.bridge.near'; + const eure = 'nep141:eure.omft.near'; + const usdc = 'nep141:usdc.omft.near'; + const rows = [ + { + quote_id: 'quote-completed', + activity_at: '2026-06-06T10:00:00.010Z', + quote: { + quote_id: 'quote-completed', + pair: `${nbtc}->${eure}`, + request_kind: 'exact_in', + maker_timing: { quote_received_at: '2026-06-06T10:00:00.000Z' }, + }, + decision: { + decision_id: 'decision-completed', + quote_id: 'quote-completed', + pair: `${nbtc}->${eure}`, + direction: 'base_to_quote', + request_kind: 'exact_in', + edge_bps: '49', + notional: '5', + notional_symbol: 'EURe', + }, + command: { + command_id: 'command-completed', + quote_id: 'quote-completed', + }, + execution: { + quote_id: 'quote-completed', + status: 'submitted', + result_code: 'quote_response_ok', + maker_timing: extendMakerTiming({ quote_received_at: '2026-06-06T10:00:00.000Z' }, { + relay_result_at: '2026-06-06T10:00:00.090Z', + }), + }, + outcome: { + quote_id: 'quote-completed', + outcome_status: 'completed', + attribution_status: 'heuristic_match', + }, + }, + { + quote_id: 'quote-failed', + activity_at: '2026-06-06T10:00:00.020Z', + decision: { + decision_id: 'decision-failed', + quote_id: 'quote-failed', + pair: `${nbtc}->${usdc}`, + direction: 'base_to_quote', + request_kind: 'exact_in', + edge_bps: '20', + notional: '8', + notional_symbol: 'USDC', + }, + execution: { + quote_id: 'quote-failed', + status: 'failed', + result_code: 'submission_failed', + failure_category: 'quote_not_found_or_finished', + maker_timing: extendMakerTiming({ quote_received_at: '2026-06-06T10:00:00.000Z' }, { + relay_result_at: '2026-06-06T10:00:00.400Z', + }), + }, + }, + ]; + + const first = buildQuoteLifecycleRollups({ + rows, + now: '2026-06-06T11:00:00.000Z', + computedAt: '2026-06-06T11:00:00.000Z', + }); + const second = buildQuoteLifecycleRollups({ + rows, + now: '2026-06-06T11:00:00.000Z', + computedAt: '2026-06-06T11:00:00.000Z', + }); + + assert.deepEqual(first, second); + assert.ok(first.some((rollup) => ( + rollup.pair === `${nbtc}->${eure}` + && rollup.request_kind === 'exact_in' + && rollup.edge_bps === '49' + && rollup.notional_bucket === '5-25 EURe' + && rollup.outcome_status === 'completed' + && rollup.completed_count === 1 + && rollup.relay_accepted_count === 1 + ))); + assert.ok(first.some((rollup) => ( + rollup.pair === `${nbtc}->${usdc}` + && rollup.result_code === 'submission_failed' + && rollup.failure_category === 'quote_not_found_or_finished' + && rollup.quote_age_bucket === '250-500ms' + && rollup.notional_bucket === '5-25 USDC' + && rollup.relay_failed_count === 1 + ))); +}); + +test('policy-driven prune candidate counting fails closed when policy is invalid', async () => { + const pool = { + async query() { + throw new Error('query should not run for invalid policy'); + }, + }; + const result = await countQuoteLifecyclePruneCandidates(pool, { + policy: { + ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + enabled: false, + }, + }); + + assert.equal(result.ok, false); + assert.equal(result.total, 0); + assert.match(result.block_reason, /enabled/); +}); + +test('retention maintenance writes rollups before deleting detail and preserves successful predicates', async () => { + const queries = []; + let watermarkReads = 0; + const policyRow = { + ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY, + updated_at: '2026-06-06T09:00:00.000Z', + }; + const pool = { + async query(sql, params = []) { + queries.push({ sql, params }); + if (sql.includes('FROM retention_policies')) return { rows: [policyRow], rowCount: 1 }; + if (sql.includes('pg_database_size')) return { rows: [{ total_bytes: 10 }], rowCount: 1 }; + if (sql.includes('pg_stat_user_tables')) return { rows: [], rowCount: 0 }; + if (sql.includes('COUNT(*)::INT AS count') && sql.includes('quote_outcome_attributions')) { + return { rows: [{ count: 1 }], rowCount: 1 }; + } + if (sql.includes('FROM quote_lifecycle_rollup_watermarks')) { + watermarkReads += 1; + return watermarkReads === 1 + ? { rows: [], rowCount: 0 } + : { + rows: [{ + rollup_key: 'quote_lifecycle', + covered_until: '2026-06-06T10:30:00.000Z', + computed_at: '2026-06-06T11:00:00.000Z', + status: 'success', + row_count: 1, + rollup_count: 1, + }], + rowCount: 1, + }; + } + if (sql.includes('MIN(activity_at)')) { + return { rows: [{ earliest_at: '2026-06-06T10:00:00.000Z' }], rowCount: 1 }; + } + if (sql.includes('WITH subject_events')) { + return { rows: [], rowCount: 0 }; + } + if (sql.includes('quote_id IS NULL')) { + return { + rows: [{ + event_id: 'raw-old', + quote_id: null, + activity_at: '2026-06-06T10:00:00.000Z', + pair: 'a->b', + quote_payload: { + pair: 'a->b', + request_kind: 'exact_in', + maker_timing: { quote_received_at: '2026-06-06T10:00:00.000Z' }, + }, + quote_observed_at: '2026-06-06T10:00:00.000Z', + quote_ingested_at: '2026-06-06T10:00:00.001Z', + }], + rowCount: 1, + }; + } + if (sql.includes('INSERT INTO quote_lifecycle_rollups')) return { rows: [], rowCount: 1 }; + if (sql.includes('INSERT INTO quote_lifecycle_rollup_watermarks')) return { rows: [], rowCount: 1 }; + if (sql.includes('DELETE FROM')) return { rows: [], rowCount: 2 }; + if (sql.includes('INSERT INTO quote_lifecycle_retention_runs')) return { rows: [], rowCount: 1 }; + return { rows: [], rowCount: 0 }; + }, + }; + + const result = await runQuoteLifecycleRetentionMaintenance(pool, { + now: '2026-06-06T11:00:00.000Z', + }); + const rollupIndex = queries.findIndex((query) => query.sql.includes('INSERT INTO quote_lifecycle_rollups')); + const firstDeleteIndex = queries.findIndex((query) => query.sql.includes('DELETE FROM')); + + assert.equal(result.status, 'success'); + assert.ok(rollupIndex >= 0); + assert.ok(firstDeleteIndex > rollupIndex); + assert.ok(queries.some((query) => ( + query.sql.includes('DELETE FROM quote_outcome_attributions') + && query.sql.includes("target.outcome_status = 'completed'") + && query.sql.includes('linked_settlement') + ))); +}); diff --git a/test/runtime-health.test.mjs b/test/runtime-health.test.mjs index 2ec1497..efe731e 100644 --- a/test/runtime-health.test.mjs +++ b/test/runtime-health.test.mjs @@ -123,3 +123,38 @@ test('maker competitiveness alerts are pair-scoped for high quote-finished relay assert.equal(alerts[0].details.quote_not_found_or_finished_count, 4); assert.equal(alerts[0].details.quote_not_found_or_finished_rate, 0.8); }); + +test('history writer retention pressure and stale rollups surface as runtime warnings', () => { + const pressure = deriveServiceHealth({ + service: 'history-writer', + snapshot: { + reachable: true, + state: { + database_connectivity: true, + last_write_at: '2026-06-06T10:59:59.000Z', + retention_mode: 'pressure', + }, + health: { ok: true }, + }, + now: '2026-06-06T11:00:00.000Z', + }); + assert.equal(pressure.status, 'warning'); + assert.equal(pressure.label, 'storage pressure'); + + const stale = deriveServiceHealth({ + service: 'history-writer', + snapshot: { + reachable: true, + state: { + database_connectivity: true, + last_write_at: '2026-06-06T10:59:59.000Z', + retention_mode: 'blocked_rollup_stale', + }, + health: { ok: true }, + }, + now: '2026-06-06T11:00:00.000Z', + }); + assert.equal(stale.status, 'warning'); + assert.equal(stale.label, 'rollup stale'); + assert.ok(stale.reasons.some((reason) => /fail-closed/.test(reason))); +}); diff --git a/test/strategy-threshold-config.test.mjs b/test/strategy-threshold-config.test.mjs index a716cca..49081a5 100644 --- a/test/strategy-threshold-config.test.mjs +++ b/test/strategy-threshold-config.test.mjs @@ -21,6 +21,9 @@ test('kubernetes production config does not carry pair, asset, or edge env vars' assert.doesNotMatch(manifest, /STRATEGY_MAX_NOTIONAL_EURE/); assert.doesNotMatch(manifest, /INTENT_REQUEST_DEFAULT_AMOUNT_EURE/); assert.doesNotMatch(manifest, /INTENT_REQUEST_MAX_AMOUNT_EURE/); + assert.doesNotMatch(manifest, /RETENTION_/); + assert.doesNotMatch(manifest, /QUOTE_LIFECYCLE_RETENTION/); + assert.doesNotMatch(manifest, /RAW_QUOTE_RETENTION/); }); test('live market-maker watch script reads DB assets instead of removed trading env vars', () => {