fix(webapp): timestamp live metric responses

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;
@@ -1148,6 +1157,7 @@ function RunBody({
                   waiting={queueMetrics.waiting}
                   status={run.status}
                   createdAt={run.createdAt}
+                  loadedAt={loadedAt}
                   runFriendlyId={run.friendlyId}
                 />
               ) : null}
@@ -1283,6 +1293,7 @@ function WaitingInQueueBlock({
   waiting,
   status,
   createdAt,
+  loadedAt,
   runFriendlyId,
 }: {
   queueName: string;
@@ -1291,11 +1302,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,
@@ -1307,10 +1323,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;
@@ -1325,7 +1344,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.