diff --git a/src/core/quote-lifecycle-statistics.mjs b/src/core/quote-lifecycle-statistics.mjs index a48d893..403dbd1 100644 --- a/src/core/quote-lifecycle-statistics.mjs +++ b/src/core/quote-lifecycle-statistics.mjs @@ -275,7 +275,7 @@ export function normalizeQuoteLifecycleStatisticsRow(row = {}) { export function quoteLifecycleStatisticId(row = {}) { const grain = row.grain || 'all_time'; - const windowStart = row.window_start || 'all'; + const windowStart = toIsoTimestamp(row.window_start) || 'all'; return `quote-lifecycle-stat:${grain}:${windowStart}`; } diff --git a/src/lib/postgres.mjs b/src/lib/postgres.mjs index 72c8f3b..c78a078 100644 --- a/src/lib/postgres.mjs +++ b/src/lib/postgres.mjs @@ -62,6 +62,7 @@ const QUOTE_LIFECYCLE_RETENTION_RUNS_TABLE = 'quote_lifecycle_retention_runs'; const QUOTE_LIFECYCLE_STATISTICS_TABLE = 'quote_lifecycle_statistics'; const QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE = 'quote_lifecycle_stat_subjects'; const QUOTE_LIFECYCLE_ROLLUP_WINDOWS_PER_PASS = 48; +const QUOTE_LIFECYCLE_STATISTICS_SOURCE_OVERLAP_MS = 5 * 60 * 1000; const SUPPORTED_ASSET_IMPORT_RUNS_TABLE = 'supported_asset_import_runs'; const TRADING_ASSETS_TABLE = 'trading_assets'; const TRADING_PAIRS_TABLE = 'trading_pairs'; @@ -439,6 +440,10 @@ export async function ensureQuoteLifecycleRetentionSchema(pool) { CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}_bucket_idx ON ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} (bucket, latest_stage_at DESC NULLS LAST) `); + await pool.query(` + CREATE INDEX IF NOT EXISTS ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}_updated_idx + ON ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} (updated_at DESC) + `); await pool.query(` CREATE TABLE IF NOT EXISTS ${QUOTE_LIFECYCLE_ROLLUP_WATERMARKS_TABLE} ( @@ -3204,8 +3209,10 @@ export async function refreshQuoteLifecycleStatistics(pool, { now = new Date().toISOString(), grains = QUOTE_LIFECYCLE_STATISTIC_GRAINS, } = {}) { + const sourceSince = await loadQuoteLifecycleStatisticSourceSince(pool); const subjectUpsertResult = await upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { now, + sourceSince, }); const statistics = await buildQuoteLifecycleStatisticsFromSubjects(pool, { grains, @@ -3219,13 +3226,36 @@ export async function refreshQuoteLifecycleStatistics(pool, { evidence_counts: { statistic_subjects: subjectUpsertResult.statistic_subjects, statistic_subject_upserts: subjectUpsertResult.upserted_count, + source_since: subjectUpsertResult.source_since, }, statistics, }; } +async function loadQuoteLifecycleStatisticSourceSince(pool) { + const result = await pool.query(` + SELECT + EXISTS ( + SELECT 1 + FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} + LIMIT 1 + ) AS has_statistic_subjects, + MAX(updated_at) AS latest_subject_updated_at + FROM ${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE} + `); + const row = result.rows[0] || {}; + const hasSubjects = row.has_statistic_subjects === true || row.has_statistic_subjects === 't'; + const latestUpdatedAt = toIsoTimestamp(row.latest_subject_updated_at); + if (!hasSubjects || !latestUpdatedAt) return null; + + const cutoffMs = Date.parse(latestUpdatedAt) - QUOTE_LIFECYCLE_STATISTICS_SOURCE_OVERLAP_MS; + if (!Number.isFinite(cutoffMs)) return null; + return new Date(cutoffMs).toISOString(); +} + async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { now = new Date().toISOString(), + sourceSince = null, } = {}) { const result = await pool.query( ` @@ -3251,6 +3281,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { ) AS payload FROM swap_demand_events WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + AND ($2::timestamptz IS NULL OR COALESCE(ingested_at, observed_at) >= $2::timestamptz) UNION ALL @@ -3290,6 +3321,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { ) AS payload FROM trade_decisions WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + AND ($2::timestamptz IS NULL OR COALESCE(ingested_at, observed_at) >= $2::timestamptz) UNION ALL @@ -3314,6 +3346,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { ) AS payload FROM execute_trade_commands WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + AND ($2::timestamptz IS NULL OR COALESCE(ingested_at, observed_at) >= $2::timestamptz) UNION ALL @@ -3366,6 +3399,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { ) AS payload FROM trade_execution_results WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + AND ($2::timestamptz IS NULL OR COALESCE(ingested_at, observed_at) >= $2::timestamptz) UNION ALL @@ -3415,6 +3449,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { ) AS payload FROM ${QUOTE_OUTCOMES_TABLE} WHERE NULLIF(COALESCE(quote_id, payload->>'quote_id'), '') IS NOT NULL + AND ($2::timestamptz IS NULL OR COALESCE(computed_at, outcome_observed_at, submitted_at, command_at) >= $2::timestamptz) ), source_ranked AS ( SELECT *, @@ -3515,7 +3550,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { END, updated_at = GREATEST(${QUOTE_LIFECYCLE_STATISTIC_SUBJECTS_TABLE}.updated_at, EXCLUDED.updated_at) `, - [now], + [now, sourceSince], ); const countResult = await pool.query(` @@ -3525,6 +3560,7 @@ async function upsertQuoteLifecycleStatisticSubjectsFromHistory(pool, { return { upserted_count: Number(result.rowCount || 0), statistic_subjects: Number(countResult.rows[0]?.subject_count || 0), + source_since: sourceSince, }; } @@ -3692,15 +3728,24 @@ export async function loadQuoteLifecycleStatistics(pool, { const normalizedLimit = Math.max(1, Math.min(1000, Number(limit) || 96)); const result = await pool.query( ` - WITH ranked AS ( + WITH deduped AS ( + SELECT + *, + ROW_NUMBER() OVER ( + PARTITION BY grain, window_start, window_end + ORDER BY computed_at DESC, stat_id DESC + ) AS duplicate_rank + FROM ${QUOTE_LIFECYCLE_STATISTICS_TABLE} + WHERE grain = ANY($1::text[]) + ), ranked AS ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY grain - ORDER BY window_start DESC NULLS LAST + ORDER BY window_start DESC NULLS LAST, computed_at DESC ) AS grain_rank - FROM ${QUOTE_LIFECYCLE_STATISTICS_TABLE} - WHERE grain = ANY($1::text[]) + FROM deduped + WHERE duplicate_rank = 1 ) SELECT * FROM ranked diff --git a/test/postgres-quote-lifecycle-statistics.test.mjs b/test/postgres-quote-lifecycle-statistics.test.mjs index b63a403..ff0e799 100644 --- a/test/postgres-quote-lifecycle-statistics.test.mjs +++ b/test/postgres-quote-lifecycle-statistics.test.mjs @@ -17,6 +17,14 @@ test('postgres quote lifecycle statistics refresh idempotently upserts state buc const pool = { async query(sql, params = []) { issuedSql.push(sql); + if (sql.includes('MAX(updated_at) AS latest_subject_updated_at')) { + return { + rows: [{ + has_statistic_subjects: subjects.size > 0, + latest_subject_updated_at: subjects.size ? '2026-06-12T17:32:00.000Z' : null, + }], + }; + } if (sql.includes('INSERT INTO quote_lifecycle_stat_subjects')) { subjects.set('quote-stat', { quote_id: 'quote-stat', @@ -28,6 +36,9 @@ test('postgres quote lifecycle statistics refresh idempotently upserts state buc lifecycle_state: includeFailure ? 'failed' : 'command_emitted', reason_code: includeFailure ? 'quote_not_found_or_finished' : 'awaiting_executor', }); + if (subjects.size > 0 && includeFailure) { + assert.equal(params[1], '2026-06-12T17:27:00.000Z'); + } return { rows: [], rowCount: 1 }; } if (sql.includes('SELECT COUNT(*)::INT AS subject_count')) { @@ -78,6 +89,10 @@ test('postgres quote lifecycle statistics refresh idempotently upserts state buc issuedSql.some((sql) => sql.includes('ORDER BY COALESCE') && sql.includes('FROM swap_demand_events')), false, ); + assert.equal( + issuedSql.some((sql) => sql.includes('COALESCE(ingested_at, observed_at) >= $2::timestamptz')), + true, + ); }); test('postgres quote lifecycle statistics retain per-quote buckets after detail pruning', async () => { @@ -86,6 +101,14 @@ test('postgres quote lifecycle statistics retain per-quote buckets after detail let sourceAvailable = true; const pool = { async query(sql, params = []) { + if (sql.includes('MAX(updated_at) AS latest_subject_updated_at')) { + return { + rows: [{ + has_statistic_subjects: subjects.size > 0, + latest_subject_updated_at: subjects.size ? '2026-06-12T14:12:00.000Z' : null, + }], + }; + } if (sql.includes('INSERT INTO quote_lifecycle_stat_subjects')) { if (sourceAvailable) { subjects.set('quote-completed', { @@ -135,8 +158,10 @@ test('postgres quote lifecycle statistics retain per-quote buckets after detail }); test('postgres quote lifecycle statistics loader returns all-time and clamped grain rows', async () => { + let issuedSql = ''; const pool = { async query(sql, params = []) { + issuedSql = sql; assert.match(sql, /FROM quote_lifecycle_statistics/); assert.deepEqual(params, [['all_time', 'five_minute'], 2]); return { @@ -164,4 +189,6 @@ test('postgres quote lifecycle statistics loader returns all-time and clamped gr assert.equal(rows[0].quote_count, 3); assert.equal(rows[0].bucket_counts.success, 1); assert.equal(rows[0].bucket_counts.submitted_no_reply, 0); + assert.match(issuedSql, /PARTITION BY grain, window_start, window_end/); + assert.match(issuedSql, /duplicate_rank = 1/); }); diff --git a/test/quote-lifecycle-statistics.test.mjs b/test/quote-lifecycle-statistics.test.mjs index 73305da..bd131c1 100644 --- a/test/quote-lifecycle-statistics.test.mjs +++ b/test/quote-lifecycle-statistics.test.mjs @@ -4,6 +4,7 @@ import assert from 'node:assert/strict'; import { buildQuoteLifecycleStatistics, classifyQuoteLifecycleStatisticsBucket, + quoteLifecycleStatisticId, quoteLifecycleStatisticWindow, } from '../src/core/quote-lifecycle-statistics.mjs'; @@ -66,6 +67,16 @@ test('quote lifecycle statistics use UTC grain boundaries with Monday UTC weeks' }); }); +test('quote lifecycle statistic ids canonicalize Date window values', () => { + assert.equal( + quoteLifecycleStatisticId({ + grain: 'five_minute', + window_start: new Date('2026-06-14T18:40:00.000Z'), + }), + 'quote-lifecycle-stat:five_minute:2026-06-14T18:40:00.000Z', + ); +}); + test('quote lifecycle statistics dedupe quotes and refresh later results in the original quote window', () => { const rows = [ {