fix(webapp): timestamp live metric responses (#4719)

## Summary

Records when live metric responses arrive and uses that timestamp to
evaluate gauge freshness and waiting duration. Cached or failed
responses remain untrusted until revalidated, while rendered values stay
stable between polling updates.

Author: carderne

apps/webapp/app/hooks/useMetricResourceQuery.ts

@@ -13,6 +13,30 @@ export type MetricResourceTimeRange = {
   to: string | null;
 };
 
+export function useIsMetricResponseFresh(
+  responseReceivedAt: number | null,
+  dataTimestamp: number,
+  maxAgeMs: number
+) {
+  const expiresAt =
+    responseReceivedAt !== null && Number.isFinite(dataTimestamp) ? dataTimestamp + maxAgeMs : null;
+  const [expiredAt, setExpiredAt] = useState<number | null>(null);
+
+  useEffect(() => {
+    if (expiresAt === null) return;
+
+    const timeout = setTimeout(() => setExpiredAt(expiresAt), Math.max(0, expiresAt - Date.now()));
+    return () => clearTimeout(timeout);
+  }, [expiresAt]);
+
+  return (
+    expiresAt !== null &&
+    responseReceivedAt !== null &&
+    responseReceivedAt < expiresAt &&
+    expiredAt !== expiresAt
+  );
+}
+
 export type MetricResourceQueryOptions = {
   organizationId: string;
   projectId: string;
@@ -102,6 +126,8 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO
   );
   const [isLoading, setIsLoading] = useState(true);
   const [failed, setFailed] = useState(false);
+  const [responseReceivedAt, setResponseReceivedAt] = useState<number | null>(null);
+  const [lastSuccessfulResponseAt, setLastSuccessfulResponseAt] = useState<number | null>(null);
   const abortRef = useRef<AbortController | null>(null);
   const loadedKeyRef = useRef<string | null>(null);
 
@@ -111,6 +137,8 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO
       loadedKeyRef.current = cacheKey;
       setRows(null);
       setFailed(false);
+      setResponseReceivedAt(null);
+      setLastSuccessfulResponseAt(null);
       setIsLoading(false);
       return;
     }
@@ -125,6 +153,8 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO
       loadedKeyRef.current = cacheKey;
       setRows(responseCache.get(cacheKey) ?? null);
       setFailed(false);
+      setResponseReceivedAt(null);
+      setLastSuccessfulResponseAt(null);
     }
     setIsLoading(true);
     fetch("/resources/metric", {
@@ -150,17 +180,22 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO
         if (controller.signal.aborted) return;
         if (data.success) {
           cacheSet(cacheKey, data.data.rows);
+          const receivedAt = Date.now();
           setRows(data.data.rows);
           setFailed(false);
+          setResponseReceivedAt(receivedAt);
+          setLastSuccessfulResponseAt(receivedAt);
         } else {
           setFailed(true);
+          setResponseReceivedAt(null);
         }
         setIsLoading(false);
       })
       .catch((error) => {
         if (error instanceof DOMException && error.name === "AbortError") return;
         if (!controller.signal.aborted) {
           setFailed(true);
+          setResponseReceivedAt(null);
           setIsLoading(false);
         }
       });
@@ -191,5 +226,12 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO
     callback: load,
   });
 
-  return { rows: rows ?? [], isLoading, showLoading: isLoading && !rows, failed };
+  return {
+    rows: rows ?? [],
+    isLoading,
+    showLoading: isLoading && !rows,
+    failed,
+    responseReceivedAt,
+    lastSuccessfulResponseAt,
+  };
 }

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx

@@ -74,6 +74,7 @@ import { ChartCard } from "~/components/primitives/charts/ChartCard";
 import { ChartSyncProvider } from "~/components/primitives/charts/ChartSyncContext";
 import { useZoomToTimeFilter } from "~/hooks/useZoomToTimeFilter";
 import {
+  useIsMetricResponseFresh,
   useMetricResourceQuery,
   type MetricResourceTimeRange,
 } from "~/hooks/useMetricResourceQuery";
@@ -420,27 +421,30 @@ function QueuesWithMetricsView() {
   // Empty rows (quiet env, or the very first fetch still in flight) fall back to the loader values,
   // so we never flash a stale 0. Fixed 15m window, env-wide (no queue filter), CH-only recurring
   // load; pauses while the tab is hidden (handled inside the hook).
-  const { rows: liveBlockRows } = useMetricResourceQuery(QUEUE_LIVE_BLOCKS_QUERY, {
-    organizationId: organization.id,
-    projectId: project.id,
-    environmentId: env.id,
-    timeRange: { period: QUEUE_LIVE_BLOCKS_PERIOD, from: null, to: null },
-    defaultPeriod: QUEUE_LIVE_BLOCKS_PERIOD,
-    fillGaps: false,
-    refreshIntervalMs: 15_000,
-  });
+  const { rows: liveBlockRows, responseReceivedAt } = useMetricResourceQuery(
+    QUEUE_LIVE_BLOCKS_QUERY,
+    {
+      organizationId: organization.id,
+      projectId: project.id,
+      environmentId: env.id,
+      timeRange: { period: QUEUE_LIVE_BLOCKS_PERIOD, from: null, to: null },
+      defaultPeriod: QUEUE_LIVE_BLOCKS_PERIOD,
+      fillGaps: false,
+      refreshIntervalMs: 15_000,
+    }
+  );
   const lastLiveBlockRow =
     liveBlockRows.length > 0 ? liveBlockRows[liveBlockRows.length - 1] : null;
   // Only trust the gauge while its newest bucket is fresh. A row painted from the hook's cache on
   // client-side nav-back (responseCache), or a quiet env whose latest bucket is minutes old, must
   // not override the loader's Redis-exact live values with a stale count.
   const lastLiveBucketMs = lastLiveBlockRow ? tileTimeToMs(lastLiveBlockRow.t) : NaN;
-  const freshLiveBlockRow =
-    lastLiveBlockRow &&
-    Number.isFinite(lastLiveBucketMs) &&
-    Date.now() - lastLiveBucketMs < LIVE_GAUGE_FRESH_MS
-      ? lastLiveBlockRow
-      : null;
+  const liveBlockIsFresh = useIsMetricResponseFresh(
+    responseReceivedAt,
+    lastLiveBucketMs,
+    LIVE_GAUGE_FRESH_MS
+  );
+  const freshLiveBlockRow = lastLiveBlockRow && liveBlockIsFresh ? lastLiveBlockRow : null;
   const envQueuedLive = freshLiveBlockRow
     ? tileNumber(freshLiveBlockRow.env_queued)
     : environment.queued;

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx

@@ -33,6 +33,7 @@ import {
   toNumber,
   useQueueMetric,
 } from "~/components/queues/QueueMetricCards";
+import { useIsMetricResponseFresh } from "~/hooks/useMetricResourceQuery";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { QueueRetrievePresenter } from "~/presenters/v3/QueueRetrievePresenter.server";
@@ -1121,7 +1122,7 @@ function QueueStats({
   // Latest gauges from ClickHouse, polled every 15s so the live blocks keep ticking after first
   // paint. Read the newest bucket (largest t); until the first poll lands liveRows is empty and the
   // *Live values stay null, so the blocks show the loader values instead of flashing 0.
-  const { rows: liveRows } = useQueueMetric(
+  const { rows: liveRows, responseReceivedAt } = useQueueMetric(
     `SELECT timeBucket() AS t, max(max_running) AS running, max(max_queued) AS queued, max(max_limit) AS q_limit, max(max_ck_wait_ms) AS ck_wait FROM queue_metrics GROUP BY t ORDER BY t`,
     {
       ids,
@@ -1137,8 +1138,11 @@ function QueueStats({
   // Redis/PG value instead of lingering on a stale count.
   const latest = liveRows.length > 0 ? liveRows[liveRows.length - 1] : undefined;
   const latestBucketMs = latest ? clickhouseTimeToMs(latest.t) : NaN;
-  const liveFresh =
-    Number.isFinite(latestBucketMs) && Date.now() - latestBucketMs < LIVE_GAUGE_FRESH_MS;
+  const liveFresh = useIsMetricResponseFresh(
+    responseReceivedAt,
+    latestBucketMs,
+    LIVE_GAUGE_FRESH_MS
+  );
   const fresh = latest && liveFresh ? latest : undefined;
   const runningLive = fresh ? toNumber(fresh.running) : null;
   const queuedLive = fresh ? toNumber(fresh.queued) : null;

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -91,6 +91,7 @@ import { useEnvironment } from "~/hooks/useEnvironment";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { useSearchParams } from "~/hooks/useSearchParam";
+import { useIsMetricResponseFresh } from "~/hooks/useMetricResourceQuery";
 import { useHasAdminAccess } from "~/hooks/useUser";
 import { redirectWithErrorMessage } from "~/models/message.server";
 import {
@@ -176,7 +177,12 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
         envParam,
         run: result.run,
       });
-      return typedjson({ type: "run" as const, run: result.run, queueMetrics });
+      return typedjson({
+        type: "run" as const,
+        run: result.run,
+        queueMetrics,
+        loadedAt: Date.now(),
+      });
     }
     return typedjson({ type: "span" as const, span: result.span });
   } catch (error) {
@@ -270,6 +276,7 @@ export function SpanView({
         <RunBody
           run={fetcher.data.run}
           queueMetrics={fetcher.data.queueMetrics}
+          loadedAt={fetcher.data.loadedAt}
           runParam={runParam}
           spanId={spanId}
           closePanel={closePanel}
@@ -395,12 +402,14 @@ function applySpanOverrides(span: Span, spanOverrides?: SpanOverride): Span {
 function RunBody({
   run,
   queueMetrics,
+  loadedAt,
   runParam,
   spanId,
   closePanel,
 }: {
   run: SpanRun;
   queueMetrics: RunQueueMetrics | null;
+  loadedAt: number;
   runParam: string;
   spanId: string;
   closePanel?: () => void;
@@ -1154,6 +1163,7 @@ function RunBody({
                   waiting={queueMetrics.waiting}
                   status={run.status}
                   createdAt={run.createdAt}
+                  loadedAt={loadedAt}
                   runFriendlyId={run.friendlyId}
                 />
               ) : null}
@@ -1310,6 +1320,7 @@ function WaitingInQueueBlock({
   waiting,
   status,
   createdAt,
+  loadedAt,
   runFriendlyId,
 }: {
   queueName: string;
@@ -1318,11 +1329,16 @@ function WaitingInQueueBlock({
   waiting: RunQueueWaiting;
   status: SpanRun["status"];
   createdAt: Date;
+  loadedAt: number;
   runFriendlyId: string;
 }) {
   // Latest gauges from ClickHouse (as on the queue page), polled so the blocks keep ticking. Trust
   // the newest bucket only while fresh; otherwise fall back to the loader's live values.
-  const { rows: liveRows } = useQueueMetric(
+  const {
+    rows: liveRows,
+    responseReceivedAt,
+    lastSuccessfulResponseAt,
+  } = useQueueMetric(
     `SELECT timeBucket() AS t, max(max_running) AS running, max(max_queued) AS queued, max(max_limit) AS q_limit\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
     {
       ids: waiting.ids,
@@ -1334,10 +1350,13 @@ function WaitingInQueueBlock({
   );
   const latest = liveRows.length > 0 ? liveRows[liveRows.length - 1] : undefined;
   const latestBucketMs = latest ? clickhouseTimeToMs(latest.t) : NaN;
-  const fresh =
-    latest && Number.isFinite(latestBucketMs) && Date.now() - latestBucketMs < LIVE_GAUGE_FRESH_MS
-      ? latest
-      : undefined;
+  const now = Math.max(loadedAt, lastSuccessfulResponseAt ?? loadedAt);
+  const liveFresh = useIsMetricResponseFresh(
+    responseReceivedAt,
+    latestBucketMs,
+    LIVE_GAUGE_FRESH_MS
+  );
+  const fresh = latest && liveFresh ? latest : undefined;
 
   const key = waiting.concurrencyKey;
   const running = fresh ? toNumber(fresh.running) : waiting.running;
@@ -1352,7 +1371,7 @@ function WaitingInQueueBlock({
   const showAtLimit = status === "PENDING" && atLimit && !paused;
   const pct =
     limit && limit > 0 ? Math.min(100, Math.round((runningAgainstLimit / limit) * 100)) : null;
-  const waitedMs = Math.max(0, Date.now() - new Date(createdAt).getTime());
+  const waitedMs = Math.max(0, now - new Date(createdAt).getTime());
 
   // Why the run is held, surfaced as a warning icon on the Status tile (queue-page style) rather
   // than a separate sentence.