Add quote lifecycle retention rollups
All checks were successful
deploy / deploy (push) Successful in 57s
All checks were successful
deploy / deploy (push) Successful in 57s
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.
This commit is contained in:
philipp
parent
52ab682fab
commit
2d329c9b2f
20 changed files with 2527 additions and 83 deletions
86
scripts/ops/quote_lifecycle_retention.mjs
Executable file
86
scripts/ops/quote_lifecycle_retention.mjs
Executable file
|
|
@ -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(() => {});
|
||||
}
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
492
src/core/quote-lifecycle-retention.mjs
Normal file
492
src/core/quote-lifecycle-retention.mjs
Normal file
|
|
@ -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();
|
||||
}
|
||||
|
|
@ -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') {
|
||||
|
|
|
|||
1124
src/lib/postgres.mjs
1124
src/lib/postgres.mjs
File diff suppressed because it is too large
Load diff
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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 }) {
|
|||
</div>
|
||||
</div>
|
||||
<div className="status-subtle competitiveness-snapshot-note">
|
||||
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.
|
||||
</div>
|
||||
<div className="metric-stack">
|
||||
<MetricCard label="Accepted" meta={formatRate(total.accepted_rate)} value={String(total.accepted_count || 0)} />
|
||||
|
|
@ -498,6 +501,61 @@ function MakerCompetitivenessSection({ summary, pairConfig }) {
|
|||
</TableFrame>
|
||||
</div>
|
||||
|
||||
<TableFrame className="competitiveness-frame">
|
||||
<table className="competitiveness-table competitiveness-rollup-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Rollup window</th>
|
||||
<th>Pair</th>
|
||||
<th>Request</th>
|
||||
<th>Age / notional</th>
|
||||
<th>Result</th>
|
||||
<th>Outcome</th>
|
||||
<th>Counts</th>
|
||||
<th>Source class</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{rollupRows.map((rollup, index) => {
|
||||
if (!rollup) {
|
||||
return (
|
||||
<tr className="competitiveness-placeholder-row" key={`rollup-placeholder:${index}`}>
|
||||
<td colSpan={8}>{index === 0 && !persistedRollups.length ? 'No persisted aggregate rollups are available yet.' : ''}</td>
|
||||
</tr>
|
||||
);
|
||||
}
|
||||
return (
|
||||
<tr key={rollup.rollup_id}>
|
||||
<td>
|
||||
<div>{formatTimestamp(rollup.window_end)}</div>
|
||||
<div className="status-subtle">{formatTimestamp(rollup.window_start)}</div>
|
||||
</td>
|
||||
<td>
|
||||
<div>{pairDisplayLabel(rollup.pair, pairConfig)}</div>
|
||||
<div className="status-subtle mono">{truncateMiddle(rollup.pair || '', 42)}</div>
|
||||
</td>
|
||||
<td>{plainCodeLabel(rollup.request_kind)}</td>
|
||||
<td>
|
||||
<div>{rollup.quote_age_bucket}</div>
|
||||
<div className="status-subtle">{rollup.notional_bucket}</div>
|
||||
</td>
|
||||
<td>
|
||||
<div>{plainCodeLabel(rollup.result_code)}</div>
|
||||
{rollup.failure_category ? <div className="status-subtle">{plainCodeLabel(rollup.failure_category)}</div> : null}
|
||||
</td>
|
||||
<td>{plainCodeLabel(rollup.outcome_status)}</td>
|
||||
<td>
|
||||
<div>{Number(rollup.quote_count || 0).toLocaleString()} quotes</div>
|
||||
<div className="status-subtle">{`${Number(rollup.completed_count || 0).toLocaleString()} completed / ${Number(rollup.relay_failed_count || 0).toLocaleString()} failed`}</div>
|
||||
</td>
|
||||
<td>{plainCodeLabel(rollup.source_data_class)}</td>
|
||||
</tr>
|
||||
);
|
||||
})}
|
||||
</tbody>
|
||||
</table>
|
||||
</TableFrame>
|
||||
|
||||
<div className="stack-grid competitiveness-table-stack">
|
||||
<TableFrame className="competitiveness-frame">
|
||||
<table className="competitiveness-table competitiveness-detail-table">
|
||||
|
|
|
|||
|
|
@ -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 (
|
||||
<>
|
||||
<section className="panel">
|
||||
|
|
@ -119,6 +128,104 @@ export default function SystemPage({ system, onControl }) {
|
|||
value={formatTimestamp(system.persistence.last_quote_outcomes_at)}
|
||||
/>
|
||||
</div>
|
||||
|
||||
<div className="metric-grid" style={{ marginTop: 14 }}>
|
||||
<MetricCard
|
||||
label="Retention mode"
|
||||
meta={retention.mode_reason || `Policy ${policy.policy_key || 'unavailable'}`}
|
||||
value={retention.retention_mode || system.persistence.retention_mode || 'Unavailable'}
|
||||
/>
|
||||
<MetricCard
|
||||
label="Rollup freshness"
|
||||
meta={`Required through ${formatTimestamp(retention.rollup_required_until)}`}
|
||||
value={retention.rollup_fresh ? 'Fresh' : 'Stale'}
|
||||
/>
|
||||
<MetricCard
|
||||
label="Latest rollup"
|
||||
meta={`${rollups.rollup_row_count || 0} aggregate rows`}
|
||||
value={formatTimestamp(rollups.latest_rollup_at || system.persistence.last_quote_lifecycle_rollup_at)}
|
||||
/>
|
||||
<MetricCard
|
||||
label="Latest prune"
|
||||
meta={`${latestRun.pruned_counts?.deletedCount || 0} rows in latest run`}
|
||||
value={formatTimestamp(latestRun.completed_at || system.persistence.last_quote_lifecycle_prune_at)}
|
||||
/>
|
||||
<MetricCard
|
||||
label="Preserved successes"
|
||||
meta="Completed or successfully attributed quote evidence"
|
||||
value={String(retention.preserved_successful_quote_count ?? system.persistence.preserved_successful_quote_count ?? 0)}
|
||||
/>
|
||||
<MetricCard
|
||||
label="Detail retention"
|
||||
meta={`Raw ${formatAge(policy.raw_retention_ms)}`}
|
||||
value={formatAge(retention.detail_retention_ms)}
|
||||
/>
|
||||
</div>
|
||||
|
||||
<TableFrame style={{ marginTop: 14 }}>
|
||||
<table className="storage-retention-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Table</th>
|
||||
<th>Approx rows</th>
|
||||
<th>Total size</th>
|
||||
<th>Latest pruned</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{storage.tables?.length ? (
|
||||
storage.tables.map((entry) => {
|
||||
const pruned = prunedByTable.get(entry.table);
|
||||
return (
|
||||
<tr key={entry.table}>
|
||||
<td className="mono">{entry.table}</td>
|
||||
<td>{Number(entry.approximate_rows || 0).toLocaleString()}</td>
|
||||
<td>{formatBytes(entry.total_bytes)}</td>
|
||||
<td>{pruned ? Number(pruned.deletedCount || 0).toLocaleString() : '0'}</td>
|
||||
</tr>
|
||||
);
|
||||
})
|
||||
) : (
|
||||
<tr>
|
||||
<td colSpan="4">No quote lifecycle storage metrics are available yet.</td>
|
||||
</tr>
|
||||
)}
|
||||
</tbody>
|
||||
</table>
|
||||
</TableFrame>
|
||||
|
||||
<TableFrame style={{ marginTop: 14 }}>
|
||||
<table className="storage-retention-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Window end</th>
|
||||
<th>Pair</th>
|
||||
<th>Result</th>
|
||||
<th>Outcome</th>
|
||||
<th>Quotes</th>
|
||||
<th>Completed</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{system.persistence.quote_lifecycle_rollups?.length ? (
|
||||
system.persistence.quote_lifecycle_rollups.slice(0, 10).map((rollup) => (
|
||||
<tr key={rollup.rollup_id}>
|
||||
<td>{formatTimestamp(rollup.window_end)}</td>
|
||||
<td className="mono">{rollup.pair || 'unknown'}</td>
|
||||
<td>{rollup.result_code || 'no_result'}</td>
|
||||
<td>{rollup.outcome_status || 'unknown'}</td>
|
||||
<td>{Number(rollup.quote_count || 0).toLocaleString()}</td>
|
||||
<td>{Number(rollup.completed_count || 0).toLocaleString()}</td>
|
||||
</tr>
|
||||
))
|
||||
) : (
|
||||
<tr>
|
||||
<td colSpan="6">No aggregate quote lifecycle rollups are available yet.</td>
|
||||
</tr>
|
||||
)}
|
||||
</tbody>
|
||||
</table>
|
||||
</TableFrame>
|
||||
<TableFrame style={{ marginTop: 14 }}>
|
||||
<table>
|
||||
<thead>
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -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/);
|
||||
|
|
|
|||
|
|
@ -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/);
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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/);
|
||||
});
|
||||
|
|
|
|||
299
test/quote-lifecycle-retention.test.mjs
Normal file
299
test/quote-lifecycle-retention.test.mjs
Normal file
|
|
@ -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')
|
||||
)));
|
||||
});
|
||||
|
|
@ -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)));
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue