feat(webapp): serve queue metrics reads from rollups and fix env totals

The built-in queues dashboard's enqueued vs started chart merged counter
states across queues, which mixes unrelated cumulative counters and
returns wrong totals; it now merges per queue and sums outside. Env
header tiles and saturation charts read the environment rollup, so their
cost no longer scales with queue count, and coarse-bucket ranges are
served from the 5m rollup automatically. Queue list ranking runs as one
query, time bounds are aligned to the bucket grid, and repeated
auto-refresh reads share ClickHouse query-cache entries.

Author: ericallam

apps/webapp/app/presenters/v3/BuiltInDashboards.server.ts

@@ -575,22 +575,22 @@ const queuesDashboard: BuiltInDashboard = {
     widgets: {
       "env-used": {
         title: "Concurrency in use",
-        query: `SELECT argMax(max_env_running, bucket_start) AS in_use\nFROM queue_metrics`,
+        query: `SELECT argMax(max_env_running, bucket_start) AS in_use\nFROM env_metrics`,
         display: { type: "bignumber", column: "in_use", aggregation: "max", abbreviate: false },
       },
       "env-limit": {
         title: "Environment limit",
-        query: `SELECT argMax(max_env_limit, bucket_start) AS env_limit\nFROM queue_metrics`,
+        query: `SELECT argMax(max_env_limit, bucket_start) AS env_limit\nFROM env_metrics`,
         display: { type: "bignumber", column: "env_limit", aggregation: "max", abbreviate: false },
       },
       "env-avail": {
         title: "Available slots",
-        query: `SELECT argMax(max_env_limit, bucket_start) - argMax(max_env_running, bucket_start) AS available\nFROM queue_metrics`,
+        query: `SELECT argMax(max_env_limit, bucket_start) - argMax(max_env_running, bucket_start) AS available\nFROM env_metrics`,
         display: { type: "bignumber", column: "available", aggregation: "max", abbreviate: false },
       },
       "env-sat": {
         title: "Env saturation",
-        query: `SELECT round(argMax(max_env_running, bucket_start) * 100.0 / nullIf(argMax(max_env_limit, bucket_start), 0), 1) AS saturation\nFROM queue_metrics`,
+        query: `SELECT round(argMax(max_env_running, bucket_start) * 100.0 / nullIf(argMax(max_env_limit, bucket_start), 0), 1) AS saturation\nFROM env_metrics`,
         display: {
           type: "bignumber",
           column: "saturation",
@@ -601,7 +601,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "sat-time": {
         title: "Environment saturation over time",
-        query: `SELECT timeBucket() AS t,\n  round(max(max_env_running) * 100.0 / nullIf(max(max_env_limit), 0), 1) AS saturation\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+        query: `SELECT timeBucket() AS t,\n  round(max(max_env_running) * 100.0 / nullIf(max(max_env_limit), 0), 1) AS saturation\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
         display: {
           type: "chart",
           chartType: "line",
@@ -616,7 +616,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "used-limit": {
         title: "Concurrency used vs limit",
-        query: `SELECT timeBucket() AS t,\n  max(max_env_running) AS used,\n  max(max_env_limit) AS limit\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+        query: `SELECT timeBucket() AS t,\n  max(max_env_running) AS used,\n  max(max_env_limit) AS limit\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
         // Single-series gauge: carry the last known used/limit across idle buckets instead of dropping to 0.
         fillGaps: true,
         display: {
@@ -695,9 +695,9 @@ const queuesDashboard: BuiltInDashboard = {
       },
       throughput: {
         title: "Enqueued vs started",
-        query: `SELECT timeBucket() AS t,\n  deltaSumTimestampMerge(enqueue_delta) AS enqueued,\n  deltaSumTimestampMerge(started_delta) AS started\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
-        // Single-series counters: zero-fill idle buckets so the line returns to 0 rather than interpolating across gaps.
-        fillGaps: true,
+        // Counter states merge per queue, then sum outside: a single merge across queues
+        // mixes unrelated odometers and returns wrong totals.
+        query: `SELECT t, sum(enq) AS enqueued, sum(st) AS started\nFROM (\n  SELECT timeBucket() AS t, queue,\n    deltaSumTimestampMerge(enqueue_delta) AS enq,\n    deltaSumTimestampMerge(started_delta) AS st\n  FROM queue_metrics\n  GROUP BY t, queue\n)\nGROUP BY t\nORDER BY t`,
         display: {
           type: "chart",
           chartType: "line",
@@ -712,7 +712,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "wait-pct": {
         title: "Scheduling delay p50/p95/p99 (ms)",
-        query: `SELECT timeBucket() AS t,\n  round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[1]) AS p50,\n  round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95,\n  round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[3]) AS p99\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+        query: `SELECT timeBucket() AS t,\n  round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[1]) AS p50,\n  round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95,\n  round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[3]) AS p99\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
         display: {
           type: "chart",
           chartType: "line",

apps/webapp/app/presenters/v3/QueueListPresenter.server.ts

@@ -217,40 +217,49 @@ export class QueueListPresenter extends BasePresenter {
       "query"
     );
 
+    // The window start is aligned to the minute so repeated page loads produce identical
+    // query text and can share ClickHouse query-cache entries.
+    const windowStartMs =
+      Math.floor((Date.now() - QUEUE_RANKING_WINDOW_MINUTES * 60 * 1000) / 60_000) * 60_000;
     const rankingArgs = {
       organizationId: environment.organizationId,
       projectId: environment.projectId,
       environmentId: environment.id,
-      startTime: formatClickhouseDateTime(
-        new Date(Date.now() - QUEUE_RANKING_WINDOW_MINUTES * 60 * 1000)
-      ),
+      startTime: formatClickhouseDateTime(new Date(windowStartMs)),
       nameContains: query?.trim() ?? "",
     };
 
-    const [countError, countRows] = await clickhouse.queueMetrics.rankingCount(rankingArgs);
-    if (countError) {
-      throw countError;
+    const offset = (page - 1) * this.perPage;
+
+    // One scan returns the page and the total ranked count (window function).
+    const [pageError, pageRows] = await clickhouse.queueMetrics.ranking({
+      ...rankingArgs,
+      byQueuedOnly: sort === "queued" ? 1 : 0,
+      limit: this.perPage,
+      offset,
+    });
+    if (pageError) {
+      throw pageError;
+    }
+
+    let ranked = pageRows?.[0]?.ranked_total ?? 0;
+    if (ranked === 0 && offset > 0) {
+      // Empty page past the ranked head: fetch the count alone for the tail slot math.
+      const [countError, countRows] = await clickhouse.queueMetrics.rankingCount(rankingArgs);
+      if (countError) {
+        throw countError;
+      }
+      ranked = countRows?.[0]?.ranked ?? 0;
     }
-    const ranked = countRows?.[0]?.ranked ?? 0;
     if (ranked > MAX_RANKED_QUEUES) {
       return null;
     }
 
     const where = buildQueueListWhere(environment.id, query, type);
     const totalQueues = await this._replica.taskQueue.count({ where });
-    const offset = (page - 1) * this.perPage;
 
     let rankedPageQueues: QueueListRow[] = [];
-    if (offset < ranked) {
-      const [pageError, pageRows] = await clickhouse.queueMetrics.rankingPage({
-        ...rankingArgs,
-        byQueuedOnly: sort === "queued" ? 1 : 0,
-        limit: this.perPage,
-        offset,
-      });
-      if (pageError) {
-        throw pageError;
-      }
+    if ((pageRows?.length ?? 0) > 0) {
       const rankedNames = (pageRows ?? []).map((row) => row.queue_name);
       rankedPageQueues = await this.findQueuesByNames(where, rankedNames);
     }
@@ -263,11 +272,9 @@ export class QueueListPresenter extends BasePresenter {
     if (tailNeeded > 0) {
       let excludedNames: string[] = [];
       if (ranked > 0) {
-        const [allError, allRows] = await clickhouse.queueMetrics.rankingPage({
+        const [allError, allRows] = await clickhouse.queueMetrics.rankingNames({
           ...rankingArgs,
-          byQueuedOnly: 0,
           limit: MAX_RANKED_QUEUES,
-          offset: 0,
         });
         if (allError) {
           throw allError;

apps/webapp/app/presenters/v3/QueueMetricsPresenter.server.ts

@@ -67,13 +67,16 @@ export class QueueMetricsPresenter {
         "query"
       );
 
+      // End bound snaps up to the bucket grid so repeated loads within a bucket produce
+      // identical params and share ClickHouse query-cache entries.
+      const endMs = Math.ceil(to.getTime() / bucketIntervalMs) * bucketIntervalMs;
       const ids = {
         organizationId: environment.organizationId,
         projectId: environment.projectId,
         environmentId: environment.id,
         queueNames,
         startTime: formatClickhouseDateTime(new Date(bucketStartMs)),
-        endTime: formatClickhouseDateTime(to),
+        endTime: formatClickhouseDateTime(new Date(endMs)),
       };
 
       const [summaryResult, sparklineResult] = await Promise.all([

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

@@ -1327,7 +1327,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
     id: "saturation",
     label: "Env saturation",
     color: "#6366F1",
-    query: `SELECT timeBucket() AS t,\n  max(max_env_running) AS used,\n  max(max_env_limit) AS env_limit\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+    query: `SELECT timeBucket() AS t,\n  max(max_env_running) AS used,\n  max(max_env_limit) AS env_limit\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
     formatValue: (v) => `${v}%`,
     derive: (rows) => {
       const sparkline = rows.map((r) => {
@@ -1342,7 +1342,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
     id: "backlog",
     label: "Backlog",
     color: "#A78BFA",
-    query: `SELECT timeBucket() AS t,\n  max(max_env_queued) AS queued\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+    query: `SELECT timeBucket() AS t,\n  max(max_env_queued) AS queued\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
     derive: (rows) => {
       const sparkline = rows.map((r) => tileNumber(r.queued));
       const peak = sparkline.reduce((max, v) => Math.max(max, v), 0);
@@ -1353,7 +1353,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
     id: "p95",
     label: "Scheduling delay p95",
     color: "#F59E0B",
-    query: `SELECT timeBucket() AS t,\n  round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+    query: `SELECT timeBucket() AS t,\n  round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
     formatValue: formatWaitMs,
     derive: (rows) => {
       const sparkline = rows.map((r) => tileNumber(r.p95));
@@ -1370,7 +1370,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
     id: "throttled",
     label: "Throttled",
     color: "#F59E0B",
-    query: `SELECT timeBucket() AS t,\n  sum(throttled_count) AS throttled\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
+    query: `SELECT timeBucket() AS t,\n  sum(throttled_count) AS throttled\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
     derive: (rows) => {
       const sparkline = rows.map((r) => tileNumber(r.throttled));
       const total = sparkline.reduce((sum, v) => sum + v, 0);

apps/webapp/app/services/queryService.server.ts

@@ -7,7 +7,12 @@ import {
   type TSQLQueryResult,
 } from "@internal/clickhouse";
 import type { CustomerQuerySource } from "@trigger.dev/database";
-import type { TableSchema, WhereClauseCondition } from "@internal/tsql";
+import {
+  calculateTimeBucketInterval,
+  type TableSchema,
+  type TimeBucketInterval,
+  type WhereClauseCondition,
+} from "@internal/tsql";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
@@ -110,6 +115,41 @@ export type ExecuteQueryResult<T> =
     }
   | { success: false; error: Error };
 
+const INTERVAL_UNIT_SECONDS: Record<TimeBucketInterval["unit"], number> = {
+  SECOND: 1,
+  MINUTE: 60,
+  HOUR: 3_600,
+  DAY: 86_400,
+  WEEK: 604_800,
+  MONTH: 2_592_000,
+};
+
+function floorToSeconds(date: Date, alignSeconds: number): Date {
+  const ms = alignSeconds * 1000;
+  return new Date(Math.floor(date.getTime() / ms) * ms);
+}
+
+/**
+ * Swap a table for one of its rollups when the query's bucket interval is at least the
+ * rollup's granularity. The rollup has identical logical columns, so only the physical
+ * table (and therefore rows read) changes.
+ */
+function resolveRollup(schema: TableSchema, timeRange: { from: Date; to: Date }): TableSchema {
+  if (!schema.rollups || schema.rollups.length === 0) {
+    return schema;
+  }
+  const interval = calculateTimeBucketInterval(
+    timeRange.from,
+    timeRange.to,
+    schema.timeBucketThresholds
+  );
+  const intervalSeconds = interval.value * INTERVAL_UNIT_SECONDS[interval.unit];
+  const best = [...schema.rollups]
+    .sort((a, b) => b.minIntervalSeconds - a.minIntervalSeconds)
+    .find((r) => r.minIntervalSeconds <= intervalSeconds);
+  return best ? { ...schema, clickhouseName: best.clickhouseName } : schema;
+}
+
 export async function getDefaultPeriod(organizationId: string): Promise<string> {
   const idealDefaultPeriodDays = 7;
   const maxQueryPeriod = await getLimit(organizationId, "queryPeriodDays", 30);
@@ -183,6 +223,14 @@ export async function executeQuery<TOut extends z.ZodSchema>(
     defaultPeriod,
   });
 
+  // Align the time bounds so repeated auto-refresh queries produce identical query
+  // params and can share ClickHouse query-cache entries (params are part of the key).
+  const alignSeconds = matchedSchema?.queryCache?.alignSeconds;
+  if (alignSeconds) {
+    if (timeFilter.from) timeFilter.from = floorToSeconds(timeFilter.from, alignSeconds);
+    if (timeFilter.to) timeFilter.to = floorToSeconds(timeFilter.to, alignSeconds);
+  }
+
   // Calculate the effective "from" date the user is requesting (for period clipping check)
   // This is null only when the user specifies just a "to" date (rare case)
   let requestedFromDate: Date | null = null;
@@ -192,6 +240,9 @@ export async function executeQuery<TOut extends z.ZodSchema>(
     // Period specified (or default) - calculate from now
     const periodMs = parse(timeFilter.period ?? defaultPeriod) ?? 7 * 24 * 60 * 60 * 1000;
     requestedFromDate = new Date(Date.now() - periodMs);
+    if (alignSeconds) {
+      requestedFromDate = floorToSeconds(requestedFromDate, alignSeconds);
+    }
   }
 
   // Build the fallback WHERE condition based on what the user specified
@@ -207,7 +258,10 @@ export async function executeQuery<TOut extends z.ZodSchema>(
   }
 
   const maxQueryPeriod = await getLimit(organizationId, "queryPeriodDays", 30);
-  const maxQueryPeriodDate = new Date(Date.now() - maxQueryPeriod * 24 * 60 * 60 * 1000);
+  let maxQueryPeriodDate = new Date(Date.now() - maxQueryPeriod * 24 * 60 * 60 * 1000);
+  if (alignSeconds) {
+    maxQueryPeriodDate = floorToSeconds(maxQueryPeriodDate, alignSeconds);
+  }
 
   // Check if the requested time period exceeds the plan limit
   const periodClipped = requestedFromDate !== null && requestedFromDate < maxQueryPeriodDate;
@@ -255,6 +309,10 @@ export async function executeQuery<TOut extends z.ZodSchema>(
     to: to ?? undefined,
     defaultPeriod,
   });
+  if (alignSeconds) {
+    timeRange.from = floorToSeconds(timeRange.from, alignSeconds);
+    timeRange.to = floorToSeconds(timeRange.to, alignSeconds);
+  }
 
   try {
     // Build field mappings for project_ref → project_id and environment_id → slug translation
@@ -277,10 +335,19 @@ export async function executeQuery<TOut extends z.ZodSchema>(
       organizationId,
       "query"
     );
+    // Serve coarse-bucket queries from the table's rollup when one qualifies.
+    const effectiveSchemas = matchedSchema?.rollups
+      ? querySchemas.map((s) => (s === matchedSchema ? resolveRollup(s, timeRange) : s))
+      : querySchemas;
+
+    const queryCacheSettings: ClickHouseSettings = matchedSchema?.queryCache
+      ? { use_query_cache: 1, query_cache_ttl: matchedSchema.queryCache.ttlSeconds }
+      : {};
+
     const result = await executeTSQL(queryClickhouse.reader, {
       ...baseOptions,
       schema: z.record(z.any()),
-      tableSchema: querySchemas,
+      tableSchema: effectiveSchemas,
       transformValues: true,
       enforcedWhereClause,
       fieldMappings,
@@ -290,6 +357,7 @@ export async function executeQuery<TOut extends z.ZodSchema>(
       timeRange,
       clickhouseSettings: {
         ...getDefaultClickhouseSettings(),
+        ...queryCacheSettings,
         ...baseOptions.clickhouseSettings, // Allow caller overrides if needed
       },
       querySettings: {