Fix quote statistics refresh collapse

Proof: reproduced live quote lifecycle statistics refresh stalling while persisted subject and window counts remained nonzero; fixed refresh to incrementally scan recent durable source evidence from the retained subject table, canonicalized statistic window ids, de-duped duplicate persisted grain/window rows on load, and validated with targeted stats/dashboard tests, full npm test, and dashboard bundle build.

Assumptions: quote lifecycle statistics are durable retained unique quote subjects, and a 5 minute source overlap is enough to catch ordinary insert/refresh races without re-reading the whole retained history on every dashboard refresh.

Still fake: this does not create venue-native fill truth where no durable settlement evidence exists, does not repair WebSocket client backpressure from high live quote volume, and does not reconstruct pruned raw quote detail.

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}`;
 }
 

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

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/);
 });

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 = [
     {