diff --git a/src/apps/history-writer.mjs b/src/apps/history-writer.mjs index 777b04f..bc0c889 100644 --- a/src/apps/history-writer.mjs +++ b/src/apps/history-writer.mjs @@ -26,6 +26,7 @@ import { insertHistoryEvents, loadLatestPortfolioMetric, loadPortfolioMetricInputs, + pruneNonSuccessfulQuoteLifecycleHistory, pruneRawNearIntentsQuoteHistory, refreshIntentRequestOutcomes, refreshQuoteOutcomes, @@ -125,6 +126,10 @@ 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 liveEvidenceTopics = [ config.kafkaTopicNormSwapDemand, config.kafkaTopicDecisionTradeDecision, @@ -197,6 +202,9 @@ const state = { last_raw_quote_prune_at: null, raw_quote_prune_deleted_count: 0, raw_quote_prune_error: null, + last_quote_lifecycle_prune_at: null, + quote_lifecycle_prune_deleted_count: 0, + quote_lifecycle_prune_error: null, derived_refresh_skipped_count: 0, last_derived_refresh_skipped_at: null, last_derived_refresh_skipped_topic: null, @@ -287,6 +295,7 @@ async function runHistoryConsumer(historyConsumer) { if (batch.topic === config.kafkaTopicRawNearIntentsQuote) { await maybePruneRawQuoteHistory(); } + await maybePruneQuoteLifecycleHistory(); if (state.draining) { setTimeout(() => shutdown(), 0); @@ -335,6 +344,46 @@ async function maybePruneRawQuoteHistory({ force = false } = {}) { } } +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; + } + + 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; + state.quote_lifecycle_prune_error = null; + if (result.deletedCount > 0) { + logger.info('non_successful_quote_lifecycle_history_pruned', { + details: result, + }); + } + 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; + } +} + async function handleWrittenHistoryEvent({ topic, partition, diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index f5a1693..ed3f2dc 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -2190,6 +2190,92 @@ export async function pruneRawNearIntentsQuoteHistory(pool, { }; } +export async function pruneNonSuccessfulQuoteLifecycleHistory(pool, { + now = new Date().toISOString(), + retainRecentMs = 30 * 60 * 1000, + batchSize = 100_000, + maxBatches = 1, +} = {}) { + const nowMs = Date.parse(now); + if (!Number.isFinite(nowMs)) throw new Error('now must be a valid timestamp'); + if (!Number.isInteger(retainRecentMs) || retainRecentMs <= 0) { + throw new Error('retain_recent_ms must be a positive integer'); + } + const boundedBatchSize = Math.max(1, Math.min(500_000, Math.floor(Number(batchSize) || 0))); + const boundedMaxBatches = Math.max(1, Math.min(100, Math.floor(Number(maxBatches) || 0))); + const cutoff = new Date(nowMs - retainRecentMs).toISOString(); + const tables = [ + { 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 }, + ]; + + const perTable = []; + let deletedCount = 0; + for (const spec of tables) { + let tableDeletedCount = 0; + let batches = 0; + for (let batch = 0; batch < boundedMaxBatches; batch += 1) { + const successPredicate = spec.selfOutcome + ? ` + ( + target.outcome_status = 'completed' + OR target.attribution_status IN ('heuristic_match', 'exact_match', 'attributed') + ) + ` + : ` + EXISTS ( + SELECT 1 + FROM quote_outcome_attributions outcome + WHERE outcome.quote_id = target.quote_id + AND ( + outcome.outcome_status = 'completed' + OR outcome.attribution_status IN ('heuristic_match', 'exact_match', 'attributed') + ) + ) + `; + 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 ${successPredicate} + 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 batchDeleted = Number(result.rowCount || 0); + tableDeletedCount += batchDeleted; + deletedCount += batchDeleted; + batches += 1; + if (batchDeleted < boundedBatchSize) break; + } + perTable.push({ + table: spec.table, + deletedCount: tableDeletedCount, + batches, + }); + } + + return { + deletedCount, + cutoff, + retainRecentMs, + batchSize: boundedBatchSize, + maxBatches: boundedMaxBatches, + tables: perTable, + }; +} + export async function insertEnvironmentStatusChange(pool, { topic, event, record }) { const fingerprint = event.payload?.status_fingerprint || null; const environmentKey = event.payload?.environment_key || record.decision_key || null; diff --git a/test/history-writer-static.test.mjs b/test/history-writer-static.test.mjs index 6c02160..f25fc72 100644 --- a/test/history-writer-static.test.mjs +++ b/test/history-writer-static.test.mjs @@ -24,9 +24,16 @@ test('history writer replays durable topics but joins the raw quote firehose liv 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/); }); 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 f612b67..ca4e5ee 100644 --- a/test/inventory-and-history.test.mjs +++ b/test/inventory-and-history.test.mjs @@ -3,7 +3,11 @@ import assert from 'node:assert/strict'; import { buildInventorySnapshot } from '../src/core/inventory.mjs'; import { routeHistoryRecord } from '../src/core/history-records.mjs'; -import { insertHistoryEvents, pruneRawNearIntentsQuoteHistory } from '../src/lib/postgres.mjs'; +import { + insertHistoryEvents, + pruneNonSuccessfulQuoteLifecycleHistory, + pruneRawNearIntentsQuoteHistory, +} from '../src/lib/postgres.mjs'; test('inventory snapshot keeps pending funding out of spendable balances', () => { const snapshot = buildInventorySnapshot({ @@ -157,6 +161,45 @@ test('raw quote retention drains multiple bounded batches when firehose backlog } }); +test('quote lifecycle retention preserves completed trades and drains old non-success rows', async () => { + const rowCounts = [100_000, 12, 7, 4, 3, 2, 1]; + const queries = []; + const pool = { + async query(sql, params) { + queries.push({ sql, params }); + return { rows: [], rowCount: rowCounts.shift() ?? 0 }; + }, + }; + + const result = await pruneNonSuccessfulQuoteLifecycleHistory(pool, { + now: '2026-06-05T12:00:00.000Z', + retainRecentMs: 30 * 60 * 1000, + batchSize: 100_000, + maxBatches: 10, + }); + + assert.equal(result.deletedCount, 100_029); + assert.equal(result.cutoff, '2026-06-05T11:30:00.000Z'); + assert.equal(result.batchSize, 100_000); + assert.equal(result.maxBatches, 10); + assert.deepEqual(result.tables.map((entry) => entry.table), [ + 'raw_near_intents_quotes', + 'swap_demand_events', + 'trade_decisions', + 'execute_trade_commands', + 'trade_execution_results', + 'quote_outcome_attributions', + ]); + assert.equal(queries.length, 7); + 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.deepEqual(query.params, ['2026-06-05T11:30:00.000Z', 100_000]); + } + assert.match(queries.at(-1).sql, /DELETE FROM quote_outcome_attributions/); +}); + function historyEvent(eventId, payload) { return { event_id: eventId,