feat(webapp,run-engine,core,clickhouse): concurrency limits management API, metrics and the Concurrency page

.changeset/task-concurrency.md

@@ -21,3 +21,5 @@ export const generateSummary = task({
 `perKey` caps each `concurrencyKey` pool and `total` caps across everything, keys or not. The queue-level `concurrencyLimit` option keeps working unchanged and is deprecated in favor of `concurrency`. Enforcement happens server-side; servers without support accept the option but do not enforce it yet.
 
 Manage limits at runtime with the new `concurrencyLimits` namespace: `list()` and `retrieve(name)` report each limit's bounds plus its live `running` and `queued` counts, `override(name, { perKey, total })` changes only the given bounds (overriding `total` to `0` pauses the limit), and `reset(name)` restores the declared values.
+
+Queue reads (`queues.list()` and `queues.retrieve()`) now report a `version` that discriminates the shape: `V1` queues keep today's fields (their own `concurrencyLimit` and its override state), while `V2` queues (tasks declared with `concurrency`) carry no queue-level concurrency, since their limits are read and overridden through `concurrencyLimits` (a task's inline limit under its derived `task/<task-id>` name). Existing reads keep compiling: a `V2` queue reports `concurrencyLimit` as null and `concurrency` as undefined.

apps/webapp/app/components/billing/OrgBanner.tsx

@@ -14,7 +14,7 @@ import {
 import { useOptionalProject, useProject } from "~/hooks/useProject";
 import { useShowSelfServe } from "~/hooks/useShowSelfServe";
 import { useCurrentPlan } from "~/routes/_app.orgs.$organizationSlug/route";
-import { v3BillingLimitsPath, v3BillingPath, v3QueuesPath } from "~/utils/pathBuilder";
+import { v3BillingLimitsPath, v3BillingPath, concurrencyPath } from "~/utils/pathBuilder";
 import { ENVIRONMENT_PAUSE_SOURCE_BILLING_LIMIT } from "~/utils/environmentPauseSource";
 
 function getUpgradeResetDate(): Date {
@@ -54,7 +54,7 @@ export function OrgBanner() {
     showSelfServe,
   });
 
-  const hideQueuesButton = location.pathname.endsWith("/queues");
+  const hideConcurrencyButton = location.pathname.endsWith("/concurrency");
   const hideBillingLimitBanner = location.pathname.endsWith("/settings/billing-limits");
 
   switch (bannerKind) {
@@ -70,7 +70,7 @@ export function OrgBanner() {
       return isArchived ? (
         <ArchivedEnvironmentBanner />
       ) : (
-        <PausedEnvironmentBanner hideButton={hideQueuesButton} />
+        <PausedEnvironmentBanner hideButton={hideConcurrencyButton} />
       );
     default:
       return null;
@@ -209,7 +209,7 @@ function PausedEnvironmentBanner({ hideButton }: { hideButton: boolean }) {
         hideButton ? undefined : (
           <LinkButton
             variant="tertiary/small"
-            to={v3QueuesPath(organization, project, environment)}
+            to={concurrencyPath(organization, project, environment)}
           >
             Manage
           </LinkButton>

apps/webapp/app/components/dashboard-agent/DashboardAgentPanel.tsx

@@ -50,7 +50,7 @@ import {
 } from "./unread-counts";
 import { AgentPanelColumn } from "./panel-layout";
 import { markerAfterActiveChat, markerAfterActivity } from "./thinking-marker";
-import { concurrencyPath } from "~/utils/pathBuilder";
+import { concurrencyLimitsPath } from "~/utils/pathBuilder";
 
 function serializePageContext(pageContext: AgentPageContext): string | undefined {
   try {
@@ -131,7 +131,7 @@ export function DashboardAgentPanel({
   const currentPage = agentPageLabel(pageContext, location.pathname);
 
   const pagePaths = useMemo<Record<string, string>>(
-    () => ({ raise_env_limit: concurrencyPath(organization, project, environment) }),
+    () => ({ raise_env_limit: concurrencyLimitsPath(organization, project, environment) }),
     [organization, project, environment]
   );
 

apps/webapp/app/components/dashboard-agent/page-label.ts

@@ -24,7 +24,7 @@ const KIND_LABELS: Record<Exclude<AgentPage["kind"], "other">, string> = {
   alerts: "Alerts",
   apikeys: "API keys",
   envvars: "Environment variables",
-  concurrency: "Concurrency",
+  concurrency: "Concurrency limits",
   regions: "Regions",
   settings: "Settings",
   waitpoints: "Waitpoints",
@@ -50,6 +50,7 @@ const SECTION_LABELS: Record<string, string> = {
   "bulk-actions": "Bulk actions",
   branches: "Branches",
   concurrency: "Concurrency",
+  "concurrency-limits": "Concurrency limits",
   dashboards: "Dashboards",
   deployments: "Deployments",
   "dev-branches": "Branches",

apps/webapp/app/components/navigation/favoritePages.tsx

@@ -223,7 +223,8 @@ const ENV_PAGE_META: Record<string, PageMeta> = {
   logs: { icon: "logs", name: "Logs" },
   errors: { icon: "errors", name: "Errors", singular: "Error" },
   query: { icon: "query", name: "Query" },
-  queues: { icon: "queues", name: "Queues", singular: "Queue" },
+  queues: { icon: "queues", name: "Concurrency" },
+  concurrency: { icon: "queues", name: "Concurrency" },
   dashboards: { icon: "dashboards", name: "Dashboards", singular: "Dashboard" },
   deployments: { icon: "deployments", name: "Deploys", singular: "Deploy" },
   "environment-variables": { icon: "environment-variables", name: "Environment variables" },
@@ -234,7 +235,7 @@ const ENV_PAGE_META: Record<string, PageMeta> = {
   "bulk-actions": { icon: "bulk-actions", name: "Bulk actions", singular: "Bulk action" },
   apikeys: { icon: "apikeys", name: "API keys" },
   alerts: { icon: "alerts", name: "Alerts", singular: "Alert" },
-  concurrency: { icon: "concurrency", name: "Concurrency" },
+  "concurrency-limits": { icon: "concurrency", name: "Concurrency limits" },
   limits: { icon: "limits", name: "Limits" },
   schedules: { icon: "schedules", name: "Schedules", singular: "Schedule" },
   test: { icon: "test", name: "Test", singular: "Test" },

apps/webapp/app/components/navigation/sideMenuSections.tsx

@@ -23,7 +23,7 @@ import {
   type OrgForPath,
   type ProjectForPath,
   branchesPath,
-  concurrencyPath,
+  concurrencyLimitsPath,
   limitsPath,
   queryPath,
   regionsPath,
@@ -39,7 +39,7 @@ import {
   v3ProjectAlertsPath,
   v3ProjectSettingsIntegrationsPath,
   v3PromptsPath,
-  v3QueuesPath,
+  concurrencyPath,
   v3WaitpointTokensPath,
 } from "~/utils/pathBuilder";
 import { AlphaBadge, NewBadge } from "../FeatureBadges";
@@ -160,10 +160,10 @@ export function buildSideMenuSections({
               } satisfies SideMenuItemConfig,
               {
                 id: "queues",
-                name: "Queues",
+                name: "Concurrency",
                 icon: QueuesIcon,
                 activeIconColor: "text-queues",
-                to: v3QueuesPath(organization, project, environment),
+                to: concurrencyPath(organization, project, environment),
                 dataAction: "queues",
               } satisfies SideMenuItemConfig,
               {
@@ -269,10 +269,10 @@ export function buildSideMenuSections({
         ? [
             {
               id: "concurrency",
-              name: "Concurrency",
+              name: "Concurrency limits",
               icon: ConcurrencyIcon,
               activeIconColor: "text-text-bright",
-              to: concurrencyPath(organization, project, environment),
+              to: concurrencyLimitsPath(organization, project, environment),
               dataAction: "concurrency",
             } satisfies SideMenuItemConfig,
           ]

apps/webapp/app/components/queues/QueueControls.tsx

@@ -1,9 +1,9 @@
 import { AdjustmentsHorizontalIcon, PauseIcon, PlayIcon } from "@heroicons/react/20/solid";
 import { DialogClose } from "@radix-ui/react-dialog";
 import { Form, useNavigation } from "@remix-run/react";
-import type { QueueItem } from "@trigger.dev/core/v3/schemas";
 import { useEffect, useState } from "react";
 import { cn } from "~/utils/cn";
+import type { QueueLimits } from "~/components/queues/queue-limits";
 import { Button, type ButtonVariant } from "~/components/primitives/Buttons";
 import { Dialog, DialogContent, DialogHeader, DialogTrigger } from "~/components/primitives/Dialog";
 import { FormButtons } from "~/components/primitives/FormButtons";
@@ -159,7 +159,12 @@ export function QueueOverrideConcurrencyButton({
   environmentConcurrencyLimit,
   trigger,
 }: {
-  queue: QueueItem & { concurrencyLimitOverridePercent: number | null };
+  queue: {
+    id: string;
+    name: string;
+    limits: QueueLimits;
+    concurrencyLimitOverridePercent: number | null;
+  };
   environmentConcurrencyLimit: number;
   /** How to render the dialog trigger. "menu-item" (default) is a PopoverMenuItem for row menus;
    * "button" is a standalone labeled button; "icon" is an icon-only button with the label in a
@@ -172,14 +177,14 @@ export function QueueOverrideConcurrencyButton({
     queue.concurrencyLimitOverridePercent !== null ? "percent" : "absolute"
   );
   const [concurrencyLimit, setConcurrencyLimit] = useState<string>(
-    queue.concurrencyLimit?.toString() ?? environmentConcurrencyLimit.toString()
+    queue.limits.perKey.current?.toString() ?? environmentConcurrencyLimit.toString()
   );
   const [percent, setPercent] = useState<string>(
     queue.concurrencyLimitOverridePercent?.toString() ?? "100"
   );
 
-  const isOverridden = !!queue.concurrency?.overriddenAt;
-  const currentLimit = queue.concurrencyLimit ?? environmentConcurrencyLimit;
+  const isOverridden = !!queue.limits.perKey.overriddenAt;
+  const currentLimit = queue.limits.perKey.current ?? environmentConcurrencyLimit;
 
   useEffect(() => {
     if (navigation.state === "loading" || navigation.state === "idle") {
@@ -277,18 +282,18 @@ export function QueueOverrideConcurrencyButton({
           {isOverridden ? (
             <Paragraph variant="small">
               This queue's concurrency limit is currently overridden to {currentLimit}.
-              {typeof queue.concurrency?.base === "number" &&
-                ` The original limit set in code was ${queue.concurrency.base}.`}{" "}
+              {typeof queue.limits.perKey.base === "number" &&
+                ` The original limit set in code was ${queue.limits.perKey.base}.`}{" "}
               You can update the override or remove it to restore the{" "}
-              {typeof queue.concurrency?.base === "number"
+              {typeof queue.limits.perKey.base === "number"
                 ? "limit set in code"
                 : "environment concurrency limit"}
               .
             </Paragraph>
           ) : (
             <Paragraph variant="small">
               Override this queue's concurrency limit. The current limit is {currentLimit}, which is
-              set {queue.concurrencyLimit !== null ? "in code" : "by the environment"}.
+              set {queue.limits.perKey.current !== null ? "in code" : "by the environment"}.
             </Paragraph>
           )}
           <Form method="post" onSubmit={() => setIsOpen(false)} className="space-y-3">

apps/webapp/app/components/queues/QueueMetricCards.tsx

@@ -102,6 +102,12 @@ type QueueMetricChartProps = {
    * are config values that existed all along, so carry the first value backward instead.
    */
   carryBackfill?: string[];
+  /**
+   * Column that marks a bucket as genuinely sampled. When set, carryBackfill only
+   * overwrites buckets where this column is absent or zero, so history from before
+   * a config value existed keeps its truthful gap instead of inheriting the value.
+   */
+  carryBackfillGuard?: string;
   /** Show the series legend below the chart (use for multi-series charts). */
   showLegend?: boolean;
   /**
@@ -141,6 +147,7 @@ export function QueueMetricChart({
   defaultPeriod,
   warningOverlay,
   carryBackfill,
+  carryBackfillGuard,
   thresholdStroke,
   onHasDataChange,
   minBucketSeconds,
@@ -163,6 +170,7 @@ export function QueueMetricChart({
         };
         const hasSamples = sampleCountColumn ? toNumber(r[sampleCountColumn]) > 0 : true;
         for (const s of series) point[s.key] = hasSamples ? toNumber(r[s.key]) : null;
+        if (carryBackfillGuard) point[carryBackfillGuard] = toNumber(r[carryBackfillGuard]);
         return point;
       })
       .filter((p) => Number.isFinite(p.bucket));
@@ -174,12 +182,15 @@ export function QueueMetricChart({
         const first = points.findIndex((p) => toNumber(p[key]) > 0);
         if (first > 0) {
           const value = points[first]![key]!;
-          for (let i = 0; i < first; i++) points[i]![key] = value;
+          for (let i = 0; i < first; i++) {
+            if (carryBackfillGuard && toNumber(points[i]![carryBackfillGuard]) > 0) continue;
+            points[i]![key] = value;
+          }
         }
       }
     }
     return points;
-  }, [rows, series, carryBackfill, sampleCountColumn]);
+  }, [rows, series, carryBackfill, carryBackfillGuard, sampleCountColumn]);
 
   const chartConfig = useMemo(() => {
     const cfg: ChartConfig = {};
@@ -340,7 +351,7 @@ export function QueueSidebarStats({
   };
 
   const { rows, showLoading } = useQueueMetric(
-    `SELECT max(max_queued) AS peak_queued,\n  round(quantilesMerge(0.5, 0.9, 0.95, 0.99)(wait_quantiles)[3]) AS worst_p95\nFROM queue_metrics`,
+    `SELECT max(max_queued) AS peak_queued,\n  round(quantilesMerge(0.5, 0.9, 0.95, 0.99)(wait_quantiles)[3]) AS worst_p95\nFROM concurrency_metrics`,
     { ids, timeRange, queueName, defaultPeriod }
   );
   const row = rows[0];

apps/webapp/app/components/queues/queue-limits.ts

@@ -0,0 +1,32 @@
+/**
+ * Dashboard-only view of a queue row's configured bounds. The public QueueItem
+ * shape is version-discriminated (V2 queues carry no queue-level concurrency),
+ * but the dashboard shows configured limits for every row, so presenters attach
+ * this alongside the public fields.
+ */
+type QueueLimitBound = {
+  /** The enforced value right now (declared, or the override when one is active) */
+  current: number | null;
+  /** The declared value an override reverts to */
+  base: number | null;
+  /** The overridden value, when an override is active */
+  override: number | null;
+  overriddenAt: Date | null;
+  /** Display name of who applied the override (null when via the API) */
+  overriddenBy: string | null;
+};
+
+type QueueTotalBound = {
+  current: number;
+  base: number | null;
+  override: number | null;
+  overriddenAt: Date | null;
+  /** Runs in flight across every pool of the row (keyed and keyless) */
+  running: number | null;
+};
+
+export type QueueLimits = {
+  perKey: QueueLimitBound;
+  /** Null when the row declares no total bound */
+  total: QueueTotalBound | null;
+};

apps/webapp/app/components/runs/v3/QueueName.tsx

@@ -1,19 +1,49 @@
 import { TasksIcon } from "~/assets/icons/TasksIcon";
+import { ConcurrencyIcon } from "~/assets/icons/ConcurrencyIcon";
 import { SimpleTooltip } from "~/components/primitives/Tooltip";
 import { cn } from "~/utils/cn";
 import { RectangleStackIcon } from "@heroicons/react/20/solid";
 
+const LIMIT_PREFIX = "limit/";
+const TASK_PREFIX = "task/";
+
 export function QueueName({
   name,
   type,
+  kind,
   paused,
   className,
 }: {
   name: string;
   type: "task" | "custom";
+  /** "limit" rows are named concurrency limits rather than queues. */
+  kind?: "queue" | "limit";
   paused?: boolean;
   className?: string;
 }) {
+  if (kind === "limit") {
+    const displayName = name.startsWith(LIMIT_PREFIX) ? name.slice(LIMIT_PREFIX.length) : name;
+    return (
+      <span className={cn("flex items-center gap-1", className)}>
+        <SimpleTooltip
+          button={
+            <ConcurrencyIcon
+              className={cn("size-[1.125rem] text-amber-500", paused && "opacity-50")}
+            />
+          }
+          content={
+            displayName.startsWith(TASK_PREFIX)
+              ? `This is the inline concurrency limit of your "${displayName.slice(
+                  TASK_PREFIX.length
+                )}" task`
+              : "This is a named concurrency limit declared in your code."
+          }
+        />
+        <span className={paused ? "opacity-50" : undefined}>{displayName}</span>
+      </span>
+    );
+  }
+
   return (
     <span className={cn("flex items-center gap-1", className)}>
       {type === "task" ? (

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

@@ -634,7 +634,7 @@ const queuesDashboard: BuiltInDashboard = {
       "t-pressure": { title: "Queue pressure", query: "", display: { type: "title" } },
       pressure: {
         title: "Queue pressure",
-        query: `SELECT queue,\n  argMax(max_running, bucket_start) AS running,\n  argMax(max_queued, bucket_start) AS queued,\n  argMax(max_limit, bucket_start) AS limit,\n  running + queued AS demand,\n  max(max_queued) AS peak_queued,\n  sum(throttled_count) AS throttled,\n  multiIf(running >= limit AND queued > 0, 'queue-limited', queued > 0, 'backlogged', 'healthy') AS status\nFROM queue_metrics\nGROUP BY queue\nORDER BY peak_queued DESC`,
+        query: `SELECT queue,\n  argMax(max_running, bucket_start) AS running,\n  argMax(max_queued, bucket_start) AS queued,\n  argMax(max_limit, bucket_start) AS limit,\n  running + queued AS demand,\n  max(max_queued) AS peak_queued,\n  sum(throttled_count) AS throttled,\n  multiIf(running >= limit AND queued > 0, 'queue-limited', queued > 0, 'backlogged', 'healthy') AS status\nFROM concurrency_metrics\nGROUP BY queue\nORDER BY peak_queued DESC`,
         display: {
           type: "table",
           prettyFormatting: true,
@@ -644,7 +644,7 @@ const queuesDashboard: BuiltInDashboard = {
       "t-trends": { title: "Per-queue trends", query: "", display: { type: "title" } },
       "running-q": {
         title: "Running by queue",
-        query: `SELECT timeBucket() AS t, queue, max(max_running) AS running\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`,
+        query: `SELECT timeBucket() AS t, queue, max(max_running) AS running\nFROM concurrency_metrics\nGROUP BY t, queue\nORDER BY t`,
         // Grouped gauge: carry each queue's running across idle buckets (per-group LOCF).
         fillGaps: true,
         display: {
@@ -661,7 +661,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "queued-q": {
         title: "Queue depth (backlog) by queue",
-        query: `SELECT timeBucket() AS t, queue, max(max_queued) AS queued\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`,
+        query: `SELECT timeBucket() AS t, queue, max(max_queued) AS queued\nFROM concurrency_metrics\nGROUP BY t, queue\nORDER BY t`,
         // Grouped gauge: carry each queue's backlog across idle buckets (per-group LOCF).
         fillGaps: true,
         display: {
@@ -678,7 +678,7 @@ const queuesDashboard: BuiltInDashboard = {
       },
       "throttled-q": {
         title: "Throttled buckets by queue",
-        query: `SELECT timeBucket() AS t, queue, sum(throttled_count) AS throttled\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`,
+        query: `SELECT timeBucket() AS t, queue, sum(throttled_count) AS throttled\nFROM concurrency_metrics\nGROUP BY t, queue\nORDER BY t`,
         // Grouped counter: per-group zero-fill so idle buckets read 0, not a gap.
         fillGaps: true,
         display: {
@@ -697,7 +697,7 @@ const queuesDashboard: BuiltInDashboard = {
         title: "Enqueued vs started",
         // 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`,
+        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 concurrency_metrics\n  GROUP BY t, queue\n)\nGROUP BY t\nORDER BY t`,
         display: {
           type: "chart",
           chartType: "line",