From 41bda076a7052140b72522ee83f74afed25138bf Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 13:27:01 +0100 Subject: [PATCH 01/10] feat(webapp,run-engine,core,clickhouse): surface total concurrency in metrics and dashboard Queues with a totalConcurrencyLimit now report how they use it. The gauge pipeline emits total running and the stored cap, ClickHouse aggregates them into the queue metrics tiers, the queues list gets a Total column, the queue detail page charts total running against the cap, and the per-key table shows each key's effective limit including per-key overrides. Queue retrieve and list API responses include the same totals. --- .changeset/queue-total-concurrency-stats.md | 5 + .../v3/QueueListPresenter.server.ts | 25 +++- .../v3/QueueRetrievePresenter.server.ts | 20 ++++ .../route.tsx | 32 ++++- .../route.tsx | 69 ++++++++++- ...ueueParam.concurrency.combined.override.ts | 3 + ....$queueParam.concurrency.combined.reset.ts | 3 + ...es.$queueParam.concurrency.key.override.ts | 3 + ...ueues.$queueParam.concurrency.key.reset.ts | 3 + ...queues.$queueParam.concurrency.override.ts | 3 + ...v1.queues.$queueParam.concurrency.reset.ts | 3 + .../resources.queues.concurrency-keys.ts | 10 +- apps/webapp/app/v3/querySchemas.ts | 24 ++++ apps/webapp/app/v3/queueMetricsMapping.ts | 2 + ...42_add_queue_metrics_total_concurrency.sql | 109 ++++++++++++++++++ .../clickhouse/src/queueMetrics.ts | 2 + internal-packages/metrics-pipeline/src/lua.ts | 18 ++- .../run-engine/src/engine/index.ts | 14 +++ .../run-engine/src/run-queue/index.ts | 49 +++++++- packages/core/src/v3/schemas/queues.ts | 15 +++ 20 files changed, 400 insertions(+), 12 deletions(-) create mode 100644 .changeset/queue-total-concurrency-stats.md create mode 100644 internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql diff --git a/.changeset/queue-total-concurrency-stats.md b/.changeset/queue-total-concurrency-stats.md new file mode 100644 index 00000000000..a70da24d1fb --- /dev/null +++ b/.changeset/queue-total-concurrency-stats.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Queue retrieve and list API responses now report total concurrency usage. When a queue has a `totalConcurrencyLimit`, `concurrency.total` includes the effective cap, the declared base, any active override, and how many runs are in flight across all concurrency keys. diff --git a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts index 0dc3daa9856..d78e98ade4c 100644 --- a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts @@ -9,7 +9,10 @@ import { engine } from "~/v3/runEngine.server"; import { BasePresenter } from "./basePresenter.server"; import { toQueueItem } from "./QueueRetrievePresenter.server"; -type QueueListEngine = Pick; +type QueueListEngine = Pick< + RunEngine, + "lengthOfQueues" | "currentConcurrencyOfQueues" | "totalConcurrencyOfQueues" +>; export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25; const MAX_ITEMS_PER_PAGE = 100; @@ -34,6 +37,9 @@ const queueListSelect = { concurrencyLimitOverriddenAt: true, concurrencyLimitOverriddenBy: true, concurrencyLimitOverridePercent: true, + totalConcurrencyLimit: true, + totalConcurrencyLimitBase: true, + totalConcurrencyLimitOverriddenAt: true, type: true, paused: true, } satisfies Prisma.TaskQueueSelect; @@ -333,11 +339,15 @@ export class QueueListPresenter extends BasePresenter { concurrencyLimitOverriddenAt: Date | null; concurrencyLimitOverriddenBy: string | null; concurrencyLimitOverridePercent: Prisma.Decimal | null; + totalConcurrencyLimit: number | null; + totalConcurrencyLimitBase: number | null; + totalConcurrencyLimitOverriddenAt: Date | null; type: TaskQueueType; paused: boolean; }[] ): Promise { - const [queuedByQueue, runningByQueue] = await Promise.all([ + const queuesWithTotalCap = queues.filter((q) => q.totalConcurrencyLimit !== null); + const [queuedByQueue, runningByQueue, totalRunningByQueue] = await Promise.all([ this.engineClient.lengthOfQueues( environment, queues.map((q) => q.name) @@ -346,6 +356,12 @@ export class QueueListPresenter extends BasePresenter { environment, queues.map((q) => q.name) ), + queuesWithTotalCap.length > 0 + ? this.engineClient.totalConcurrencyOfQueues( + environment, + queuesWithTotalCap.map((q) => q.name) + ) + : Promise.resolve({} as Record), ]); // Manually "join" the overridden users because there is no way to implement the relationship @@ -373,6 +389,11 @@ export class QueueListPresenter extends BasePresenter { ? (overriddenByMap.get(queue.concurrencyLimitOverriddenBy) ?? null) : null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, + totalRunning: + queue.totalConcurrencyLimit !== null ? (totalRunningByQueue[queue.name] ?? 0) : null, }), // Prisma returns Decimal; the client only needs a plain number (null for absolute overrides). concurrencyLimitOverridePercent: diff --git a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts index f6918394e5c..e4777ceb13e 100644 --- a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts @@ -90,6 +90,7 @@ export class QueueRetrievePresenter extends BasePresenter { const results = await Promise.all([ engine.lengthOfQueues(environment, [queue.name]), engine.currentConcurrencyOfQueues(environment, [queue.name]), + engine.totalConcurrencyOfQueues(environment, [queue.name]), ]); // Transform queues to include running and queued counts @@ -107,6 +108,11 @@ export class QueueRetrievePresenter extends BasePresenter { concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null, concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy ?? null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit ?? null, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase ?? null, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt ?? null, + totalRunning: + queue.totalConcurrencyLimit != null ? (results[2]?.[queue.name] ?? 0) : null, }), // The percent source-of-truth for percent-based overrides isn't part of the shared // `QueueItem` schema (that's a public contract), so we surface it as an extra field on @@ -148,6 +154,10 @@ export function toQueueItem(data: { concurrencyLimitOverriddenAt: Date | null; concurrencyLimitOverriddenBy: User | null; paused: boolean; + totalConcurrencyLimit?: number | null; + totalConcurrencyLimitBase?: number | null; + totalConcurrencyLimitOverriddenAt?: Date | null; + totalRunning?: number | null; }): QueueItem & { releaseConcurrencyOnWaitpoint: boolean } { return { id: data.friendlyId, @@ -164,6 +174,16 @@ export function toQueueItem(data: { override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null, overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy), overriddenAt: data.concurrencyLimitOverriddenAt, + total: + data.totalConcurrencyLimit !== undefined + ? { + current: data.totalConcurrencyLimit, + base: data.totalConcurrencyLimitBase ?? null, + override: data.totalConcurrencyLimitOverriddenAt ? data.totalConcurrencyLimit : null, + overriddenAt: data.totalConcurrencyLimitOverriddenAt ?? null, + running: data.totalRunning ?? null, + } + : undefined, }, // TODO: This needs to be removed but keeping this here for now to avoid breaking existing clients releaseConcurrencyOnWaitpoint: true, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index 4b5673c6f0e..6e4fd9f9347 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -698,6 +698,13 @@ function QueuesWithMetricsView() { Queued Running Limit + + Total + + = + Math.min( + queue.concurrency.total.current, + environment.concurrencyLimit + ) && + "text-warning" + )} + > + {queue.concurrency?.total?.current != null + ? `${queue.concurrency.total.running ?? 0}/${Math.min( + queue.concurrency.total.current, + environment.concurrencyLimit + )}` + : "–"} + - +
{hasFilters ? "No queues found matching your filters" : "No queues found"} diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx index 8f8a91dca4f..22279ffcaba 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx @@ -392,7 +392,12 @@ export default function Page() { ) ) : ( - + )} @@ -402,7 +407,13 @@ export default function Page() { {view === "keys" && hasKeys ? ( <> - + {selectedKey ? ( @@ -436,10 +447,12 @@ function OverviewCharts({ ids, timeRange, queueName, + hasTotalLimit, }: { ids: Ids; timeRange: TimeRangeParams; queueName: string; + hasTotalLimit: boolean; }) { const zoomToTimeFilter = useZoomToTimeFilter(); return ( @@ -479,6 +492,37 @@ function OverviewCharts({ // leading zeros so the reference line doesn't start with a false 0→limit step. carryBackfill={["limit"]} /> + {hasTotalLimit ? ( + + Runs in flight across ALL concurrency keys ( + ) versus the queue's total limit ( + + ). + + } + showLegend + className="aspect-[2/1]" + query={`SELECT timeBucket() AS t, max(max_total_running) AS running, least(max(max_total_limit), max(max_env_limit)) AS cap\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} + fillGaps + minBucketSeconds={SYNCED_CHART_MIN_BUCKET_SECONDS} + ids={ids} + timeRange={timeRange} + queueName={queueName} + series={[ + { key: "cap", label: "Total limit", color: COLORS.limit }, + { key: "running", label: "Running", color: COLORS.running }, + ]} + thresholdStroke={{ + series: "running", + valueFromSeries: "cap", + aboveColor: "var(--color-warning)", + }} + carryBackfill={["cap"]} + /> + ) : null} Key Queued now Running now + + Limit + Oldest wait Started Peak backlog @@ -976,11 +1031,11 @@ function KeyStatsTable({ {showLoading ? ( - + Loading… ) : rows.length === 0 ? ( - + {search ? `No keys match “${search}”` : "No concurrency keys"} ) : ( @@ -994,6 +1049,12 @@ function KeyStatsTable({ {row.key} {row.queued.toLocaleString()} {row.running.toLocaleString()} + + {Math.min(row.limitOverride ?? defaultKeyLimit, envLimit).toLocaleString()} + {row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)} diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts index c643b77965a..77688a9fcc5 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts @@ -46,6 +46,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts index b2841f1efe6..0e588716658 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts @@ -45,6 +45,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts index 5a37b4526ec..3758a882c93 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts @@ -52,6 +52,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts index 51d14642e2c..e30bf3f6360 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts @@ -46,6 +46,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts index 90f5772c5d3..42bb2008682 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts @@ -61,6 +61,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts index 503d875e471..3f36e629f09 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts @@ -43,6 +43,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts index 67c2b9f500a..662013694ef 100644 --- a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts +++ b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts @@ -43,6 +43,8 @@ export type ConcurrencyKeyRow = { peakBacklog: number; peakRunning: number; meanWaitMs: number; + /** Per-key concurrency limit override, when one is set for this key (null = inherits the queue limit). */ + limitOverride: number | null; }; export type ConcurrencyKeysResponse = @@ -151,8 +153,11 @@ export const action = async ({ request }: ActionFunctionArgs) => { const total = rankingRows?.[0]?.ranked_total ?? 0; const keys = (rankingRows ?? []).map((r) => r.concurrency_key); - // Enrich just this page's keys with live "now" counts from Redis. - const live = await engine.concurrencyKeyLiveStats(environment, queueName, keys); + // Enrich just this page's keys with live "now" counts and any per-key limit overrides from Redis. + const [live, keyLimitOverrides] = await Promise.all([ + engine.concurrencyKeyLiveStats(environment, queueName, keys), + engine.runQueue.getQueueConcurrencyKeyLimits(environment, queueName), + ]); const loadedAt = Date.now(); const rows: ConcurrencyKeyRow[] = (rankingRows ?? []).map((r) => { @@ -168,6 +173,7 @@ export const action = async ({ request }: ActionFunctionArgs) => { peakBacklog: r.peak_backlog, peakRunning: r.peak_running, meanWaitMs: r.mean_wait_ms, + limitOverride: keyLimitOverrides[r.concurrency_key] ?? null, }; }); diff --git a/apps/webapp/app/v3/querySchemas.ts b/apps/webapp/app/v3/querySchemas.ts index 690bbaf5396..8267a0f8020 100644 --- a/apps/webapp/app/v3/querySchemas.ts +++ b/apps/webapp/app/v3/querySchemas.ts @@ -770,6 +770,22 @@ const queueMetricsSchema: TableSchema = { fillMode: "carry", }), }, + max_total_running: { + name: "max_total_running", + ...column("UInt32", { + description: + "Peak in-flight runs across ALL concurrency keys of the queue in the bucket (only emitted for keyed queues). Aggregate with max().", + fillMode: "carry", + }), + }, + max_total_limit: { + name: "max_total_limit", + ...column("UInt32", { + description: + "The queue's total concurrency limit across all keys, as stored (0 = no cap; clamp against max_env_limit). Aggregate with max().", + fillMode: "carry", + }), + }, max_ck_backlogged: { name: "max_ck_backlogged", ...column("UInt32", { @@ -1406,6 +1422,14 @@ const queueMetricsByKeySchema: TableSchema = { fillMode: "carry", }), }, + max_limit: { + name: "max_limit", + ...column("UInt32", { + description: + "The effective concurrency limit for this key (the queue limit, or its per-key override). Aggregate with max().", + fillMode: "carry", + }), + }, wait_ms_sum: { name: "wait_ms_sum", ...column("UInt64", { diff --git a/apps/webapp/app/v3/queueMetricsMapping.ts b/apps/webapp/app/v3/queueMetricsMapping.ts index 9433b361a88..d341f4a63cd 100644 --- a/apps/webapp/app/v3/queueMetricsMapping.ts +++ b/apps/webapp/app/v3/queueMetricsMapping.ts @@ -131,6 +131,8 @@ export function mapEntryToRows( throttled: num(f.thr), ck_backlogged: num(f.ckq), ck_max_wait_ms: num(f.ckw), + total_running: num(f.tcc), + total_limit: num(f.tlim), }, ]; } diff --git a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql b/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql new file mode 100644 index 00000000000..7711effa6e3 --- /dev/null +++ b/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql @@ -0,0 +1,109 @@ +-- +goose Up + +-- Total-concurrency gauges: total_running is the in-flight count across ALL +-- concurrency-key variants of a queue (the groupConcurrency set), total_limit the +-- RAW stored total cap (0 = none; readers clamp against max_env_limit). Emitted on +-- base-queue gauge rows only. Per-key gauge rows now carry the EFFECTIVE per-key +-- limit in queue_limit (override-aware), surfaced in the ck tier as max_limit. + +ALTER TABLE trigger_dev.queue_metrics_raw_v1 + ADD COLUMN IF NOT EXISTS total_running UInt32 DEFAULT 0, + ADD COLUMN IF NOT EXISTS total_limit UInt32 DEFAULT 0; + +ALTER TABLE trigger_dev.queue_metrics_v1 + ADD COLUMN IF NOT EXISTS max_total_running SimpleAggregateFunction(max, UInt32), + ADD COLUMN IF NOT EXISTS max_total_limit SimpleAggregateFunction(max, UInt32); + +ALTER TABLE trigger_dev.queue_metrics_5m_v1 + ADD COLUMN IF NOT EXISTS max_total_running SimpleAggregateFunction(max, UInt32), + ADD COLUMN IF NOT EXISTS max_total_limit SimpleAggregateFunction(max, UInt32); + +ALTER TABLE trigger_dev.queue_metrics_ck_v1 + ADD COLUMN IF NOT EXISTS max_limit SimpleAggregateFunction(max, UInt32); + +-- Materialized views cannot be altered: recreate them with the new columns. The 5m +-- MV MUST keep reading raw, never cascade off queue_metrics_v1 (out-of-time-order +-- deltaSumTimestamp merges double-count bridging spans). + +DROP VIEW IF EXISTS trigger_dev.queue_metrics_mv_v1; +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_mv_v1 +TO trigger_dev.queue_metrics_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + max(total_running) AS max_total_running, + max(total_limit) AS max_total_limit, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +DROP VIEW IF EXISTS trigger_dev.queue_metrics_5m_mv_v1; +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_5m_mv_v1 +TO trigger_dev.queue_metrics_5m_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 5 MINUTE) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + max(total_running) AS max_total_running, + max(total_limit) AS max_total_limit, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +DROP VIEW IF EXISTS trigger_dev.queue_metrics_ck_mv_v1; +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_ck_mv_v1 +TO trigger_dev.queue_metrics_ck_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, concurrency_key, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack') AS ack_delta, + maxIf(queued, op = 'gauge') AS max_queued, + maxIf(running, op = 'gauge') AS max_running, + maxIf(queue_limit, op = 'gauge') AS max_limit, + sumIf(wait_ms, op = 'started') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0) AS wait_ms_count +FROM trigger_dev.queue_metrics_raw_v1 +WHERE concurrency_key != '' +GROUP BY organization_id, project_id, environment_id, queue_name, concurrency_key, bucket_start; + +-- +goose Down +DROP VIEW IF EXISTS trigger_dev.queue_metrics_ck_mv_v1; +DROP VIEW IF EXISTS trigger_dev.queue_metrics_5m_mv_v1; +DROP VIEW IF EXISTS trigger_dev.queue_metrics_mv_v1; +ALTER TABLE trigger_dev.queue_metrics_ck_v1 DROP COLUMN IF EXISTS max_limit; +ALTER TABLE trigger_dev.queue_metrics_5m_v1 DROP COLUMN IF EXISTS max_total_running, DROP COLUMN IF EXISTS max_total_limit; +ALTER TABLE trigger_dev.queue_metrics_v1 DROP COLUMN IF EXISTS max_total_running, DROP COLUMN IF EXISTS max_total_limit; +ALTER TABLE trigger_dev.queue_metrics_raw_v1 DROP COLUMN IF EXISTS total_running, DROP COLUMN IF EXISTS total_limit; diff --git a/internal-packages/clickhouse/src/queueMetrics.ts b/internal-packages/clickhouse/src/queueMetrics.ts index 39576b4a0a3..aa3cf5296d2 100644 --- a/internal-packages/clickhouse/src/queueMetrics.ts +++ b/internal-packages/clickhouse/src/queueMetrics.ts @@ -21,6 +21,8 @@ export const QueueMetricsRawV1Input = z.object({ throttled: z.number().optional(), ck_backlogged: z.number().optional(), ck_max_wait_ms: z.number().optional(), + total_running: z.number().optional(), + total_limit: z.number().optional(), wait_ms: z.number().optional(), cumulative: z.number().optional(), }); diff --git a/internal-packages/metrics-pipeline/src/lua.ts b/internal-packages/metrics-pipeline/src/lua.ts index 64f3b896c0d..701f608308a 100644 --- a/internal-packages/metrics-pipeline/src/lua.ts +++ b/internal-packages/metrics-pipeline/src/lua.ts @@ -17,6 +17,10 @@ export type GaugeComputeLuaParams = { // CK-health extras (both or neither): appended as an optional gauge tail, gauge[8]/gauge[9]. ckBacklogged?: string; ckMaxWaitMs?: string; + // Total-concurrency extras (both or neither, and only with the CK extras): appended as + // gauge[10]/gauge[11]. totalLimit is the RAW stored limit (0 = none); readers clamp. + totalRunning?: string; + totalLimit?: string; }; // Computes an op=gauge snapshot into the enclosing script's `__qm_g` local (a flat @@ -26,11 +30,21 @@ export type GaugeComputeLuaParams = { export function createMetricsGaugeComputeLua(params: GaugeComputeLuaParams): string { const throttled = params.throttledExpr ?? "__cc >= __lim and __ql > 0"; const hasCk = params.ckBacklogged != null && params.ckMaxWaitMs != null; - const gauge = hasCk + const hasTotal = params.totalRunning != null && params.totalLimit != null; + if (hasTotal && !hasCk) { + throw new Error("gauge totalRunning/totalLimit extras require the CK extras"); + } + const gauge = hasTotal ? ` local __ckq = tonumber(${params.ckBacklogged}) or 0 local __ckw = tonumber(${params.ckMaxWaitMs}) or 0 + local __tcc = tonumber(${params.totalRunning}) or 0 + local __tlim = tonumber(${params.totalLimit}) or 0 + __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr, __ckq, __ckw, __tcc, __tlim}` + : hasCk + ? ` local __ckq = tonumber(${params.ckBacklogged}) or 0 + local __ckw = tonumber(${params.ckMaxWaitMs}) or 0 __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr, __ckq, __ckw}` - : ` __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr}`; + : ` __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr}`; return ` if ${params.enabledArg} then diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 9d7bb8ff947..504673bf052 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -1740,6 +1740,20 @@ export class RunEngine { return this.runQueue.currentConcurrencyOfQueues(environment, queues); } + async totalConcurrencyOfQueues( + environment: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + return this.runQueue.totalConcurrencyOfQueues(environment, queues); + } + + async totalConcurrencyLimitsOfQueues( + environment: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + return this.runQueue.totalConcurrencyLimitsOfQueues(environment, queues); + } + async concurrencyKeyBreakdown( environment: MinimalAuthenticatedEnvironment, queue: string, diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 7f6493dfc2b..6071106db26 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -213,7 +213,8 @@ const QUEUE_METRICS_CK_ENQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ enabledArg: "ARGV[#ARGV] == '1'", queued: "redis.call('ZCARD', queueKey)", running: "redis.call('SCARD', queueCurrentConcurrencyKey)", - queueLimit: "redis.call('GET', queueConcurrencyLimitKey) or '1000000'", + queueLimit: + "redis.call('HGET', ckLimitsKey, queueName) or redis.call('GET', queueConcurrencyLimitKey) or '1000000'", envQueued: "redis.call('ZCARD', envQueueKey)", envRunning: "redis.call('SCARD', envCurrentConcurrencyKey)", envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", @@ -244,6 +245,8 @@ const QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", throttledExpr: "false", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + totalRunning: "redis.call('SCARD', groupConcurrencyKey)", + totalLimit: "redis.call('GET', totalConcurrencyLimitKey) or '0'", }); /** Injected queue-metrics stream emitter; all calls are no-ops when metrics are disabled. */ @@ -702,6 +705,46 @@ export class RunQueue { return limits; } + /** Batch variant of totalConcurrencyOfQueue: one pipeline of group SCARDs. */ + public async totalConcurrencyOfQueues( + env: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + const pipeline = this.redis.pipeline(); + queues.forEach((queue) => { + pipeline.scard(this.keys.queueGroupConcurrencyKey(env, queue)); + }); + + const results = await pipeline.exec(); + + return queues.reduce( + (acc, queue, index) => { + const value = results?.[index]?.[1]; + acc[queue] = typeof value === "number" ? value : 0; + return acc; + }, + {} as Record + ); + } + + /** Batch read of the RAW stored total concurrency limits (undefined = no cap). */ + public async totalConcurrencyLimitsOfQueues( + env: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + const keys = queues.map((queue) => this.keys.queueTotalConcurrencyLimitKey(env, queue)); + const values = keys.length > 0 ? await this.redis.mget(...keys) : []; + + return queues.reduce( + (acc, queue, index) => { + const value = values[index]; + acc[queue] = value != null ? Number(value) : undefined; + return acc; + }, + {} as Record + ); + } + public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) { await this.#callUpdateEnvironmentConcurrencyLimits({ envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env), @@ -2333,6 +2376,10 @@ export class RunQueue { fields.ckq = ckq; fields.ckw = ckw; } + if (gauge.length >= 11) { + fields.tcc = gauge[9]; + fields.tlim = gauge[10]; + } this.options.queueMetrics?.emitGauge(queue, fields); } diff --git a/packages/core/src/v3/schemas/queues.ts b/packages/core/src/v3/schemas/queues.ts index 34a47b34e3e..f3039f8075a 100644 --- a/packages/core/src/v3/schemas/queues.ts +++ b/packages/core/src/v3/schemas/queues.ts @@ -45,6 +45,21 @@ export const QueueItem = z.object({ overriddenAt: z.coerce.date().nullable(), /** Who overrode the concurrency limit (will be null if overridden via the API) */ overriddenBy: z.string().nullable(), + /** The total concurrency cap across all concurrencyKey values of the queue */ + total: z + .object({ + /** The effective/current total concurrency limit (null = no cap) */ + current: z.number().nullable(), + /** The declared total limit an override reverts to on reset */ + base: z.number().nullable(), + /** The overridden total limit, when an override is active */ + override: z.number().nullable(), + /** When the total override was applied */ + overriddenAt: z.coerce.date().nullable(), + /** Runs currently in flight across all concurrencyKey values */ + running: z.number().nullable(), + }) + .optional(), }) .optional(), }); From 70b70d06083891ea0d2dea205313b499edae1a63 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 13:38:22 +0100 Subject: [PATCH 02/10] fix(run-engine,clickhouse,webapp): total gauges on enqueue paths; restore views on rollback The CK enqueue gauges (fast path and queued path) now sample total running and the stored cap, so metric buckets fed only by enqueues no longer record zero totals. The migration's down section recreates the pre-existing materialized view definitions so ingestion keeps flowing after a rollback. The per-key table reads only the page's overrides with one HMGET instead of loading the queue's whole override hash. --- .../resources.queues.concurrency-keys.ts | 2 +- ...42_add_queue_metrics_total_concurrency.sql | 68 +++++++++++++++++++ .../run-engine/src/run-queue/index.ts | 36 +++++++++- 3 files changed, 103 insertions(+), 3 deletions(-) diff --git a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts index 662013694ef..8c590554e51 100644 --- a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts +++ b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts @@ -156,7 +156,7 @@ export const action = async ({ request }: ActionFunctionArgs) => { // Enrich just this page's keys with live "now" counts and any per-key limit overrides from Redis. const [live, keyLimitOverrides] = await Promise.all([ engine.concurrencyKeyLiveStats(environment, queueName, keys), - engine.runQueue.getQueueConcurrencyKeyLimits(environment, queueName), + engine.runQueue.getQueueConcurrencyKeyLimitsForKeys(environment, queueName, keys), ]); const loadedAt = Date.now(); diff --git a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql b/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql index 7711effa6e3..8bb19faeef1 100644 --- a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql +++ b/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql @@ -107,3 +107,71 @@ ALTER TABLE trigger_dev.queue_metrics_ck_v1 DROP COLUMN IF EXISTS max_limit; ALTER TABLE trigger_dev.queue_metrics_5m_v1 DROP COLUMN IF EXISTS max_total_running, DROP COLUMN IF EXISTS max_total_limit; ALTER TABLE trigger_dev.queue_metrics_v1 DROP COLUMN IF EXISTS max_total_running, DROP COLUMN IF EXISTS max_total_limit; ALTER TABLE trigger_dev.queue_metrics_raw_v1 DROP COLUMN IF EXISTS total_running, DROP COLUMN IF EXISTS total_limit; + +-- Recreate the pre-042 materialized views (the definitions from 036) so ingestion keeps +-- feeding every aggregate table after a rollback. +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_mv_v1 +TO trigger_dev.queue_metrics_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_5m_mv_v1 +TO trigger_dev.queue_metrics_5m_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 5 MINUTE) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_ck_mv_v1 +TO trigger_dev.queue_metrics_ck_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, concurrency_key, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack') AS ack_delta, + maxIf(queued, op = 'gauge') AS max_queued, + maxIf(running, op = 'gauge') AS max_running, + sumIf(wait_ms, op = 'started') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0) AS wait_ms_count +FROM trigger_dev.queue_metrics_raw_v1 +WHERE concurrency_key != '' +GROUP BY organization_id, project_id, environment_id, queue_name, concurrency_key, bucket_start; diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 6071106db26..f873e2f5f68 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -208,6 +208,14 @@ const QUEUE_METRICS_CK_GAUGE_EXTRAS = { ckMaxWaitMs: "__ckwait", }; +// Total-concurrency tail (gauge[10]/gauge[11]): live group cardinality + raw stored cap. +// Requires groupConcurrencyKey/totalConcurrencyLimitKey locals; the CK scripts that actually +// run (the Tracked variants and the CK dequeue) all declare them for the total-cap gate. +const QUEUE_METRICS_TOTAL_GAUGE_EXTRAS = { + totalRunning: "redis.call('SCARD', groupConcurrencyKey)", + totalLimit: "redis.call('GET', totalConcurrencyLimitKey) or '0'", +}; + // CK enqueue variants of the two gauges above, extended with the CK-health tail. const QUEUE_METRICS_CK_ENQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ enabledArg: "ARGV[#ARGV] == '1'", @@ -219,6 +227,7 @@ const QUEUE_METRICS_CK_ENQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ envRunning: "redis.call('SCARD', envCurrentConcurrencyKey)", envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); const QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA = createMetricsGaugeComputeLua({ @@ -230,6 +239,7 @@ const QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA = createMetricsGaugeComputeLua envRunning: "envCurrent", envLimit: "envLimit", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); // CK dequeue: depth/running from the per-base-queue aggregate counters the run-queue already @@ -245,8 +255,7 @@ const QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", throttledExpr: "false", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, - totalRunning: "redis.call('SCARD', groupConcurrencyKey)", - totalLimit: "redis.call('GET', totalConcurrencyLimitKey) or '0'", + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); /** Injected queue-metrics stream emitter; all calls are no-ops when metrics are disabled. */ @@ -705,6 +714,29 @@ export class RunQueue { return limits; } + /** Per-key limit overrides for just the given keys: one HMGET, O(keys) not O(overrides). */ + public async getQueueConcurrencyKeyLimitsForKeys( + env: MinimalAuthenticatedEnvironment, + queue: string, + concurrencyKeys: string[] + ): Promise> { + if (concurrencyKeys.length === 0) { + return {}; + } + + const fields = concurrencyKeys.map((key) => this.keys.queueKey(env, queue, key)); + const values = await this.redis.hmget(this.keys.queueCkLimitsKey(env, queue), ...fields); + + const limits: Record = {}; + concurrencyKeys.forEach((key, index) => { + const value = values[index]; + if (value != null) { + limits[key] = Number(value); + } + }); + return limits; + } + /** Batch variant of totalConcurrencyOfQueue: one pipeline of group SCARDs. */ public async totalConcurrencyOfQueues( env: MinimalAuthenticatedEnvironment, From a0081088fa07506b84402489469589d317b9312e Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 16:48:06 +0100 Subject: [PATCH 03/10] fix(clickhouse): keep migration comments semicolon-free The test harness splits a migration's up section on semicolons, so a semicolon inside a comment yields a comment-only statement that ClickHouse rejects as an empty query and every container-backed suite fails at setup. --- .../schema/042_add_queue_metrics_total_concurrency.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql b/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql index 8bb19faeef1..f45e6421ae3 100644 --- a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql +++ b/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql @@ -2,7 +2,7 @@ -- Total-concurrency gauges: total_running is the in-flight count across ALL -- concurrency-key variants of a queue (the groupConcurrency set), total_limit the --- RAW stored total cap (0 = none; readers clamp against max_env_limit). Emitted on +-- RAW stored total cap (0 = none, readers clamp against max_env_limit). Emitted on -- base-queue gauge rows only. Per-key gauge rows now carry the EFFECTIVE per-key -- limit in queue_limit (override-aware), surfaced in the ck tier as max_limit. From 03146bf81621ee9c2da11db27c71e196d2343e2f Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 17:01:59 +0100 Subject: [PATCH 04/10] fix(webapp): skip the total concurrency read when the queue has no cap Queue retrieve only asks the engine for total running when a total limit is set, matching the list presenter and avoiding a pointless read for the common uncapped case. --- .../webapp/app/presenters/v3/QueueRetrievePresenter.server.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts index e4777ceb13e..26811a8800e 100644 --- a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts @@ -90,7 +90,9 @@ export class QueueRetrievePresenter extends BasePresenter { const results = await Promise.all([ engine.lengthOfQueues(environment, [queue.name]), engine.currentConcurrencyOfQueues(environment, [queue.name]), - engine.totalConcurrencyOfQueues(environment, [queue.name]), + queue.totalConcurrencyLimit != null + ? engine.totalConcurrencyOfQueues(environment, [queue.name]) + : undefined, ]); // Transform queues to include running and queued counts From a6eb24e8122bbeb461f5d33e4292defeaac5859b Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 18:21:57 +0100 Subject: [PATCH 05/10] feat(webapp): show the Total column in the non-metrics queues table too The total concurrency numbers come from live Redis, not the metrics pipeline, so the column belongs in both tables rather than only behind the queue metrics UI gate. --- .../route.tsx | 29 ++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index 6e4fd9f9347..3a5d0ea9954 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -1806,6 +1806,12 @@ function ClassicQueuesView() { Queued Running Limit + + Total + {limit} + = + Math.min( + queue.concurrency.total.current, + environment.concurrencyLimit + ) && + "text-warning" + )} + > + {queue.concurrency?.total?.current != null + ? `${queue.concurrency.total.running ?? 0}/${Math.min( + queue.concurrency.total.current, + environment.concurrencyLimit + )}` + : "–"} + - +
{hasFilters ? "No queues found matching your filters" : "No queues found"} From 638e2453acab5285df38c31e8258b9eec9b2a672 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 18:33:02 +0100 Subject: [PATCH 06/10] feat(webapp): fold the total cap into the Limit column A separate Total column implied every queue should have one, and its dash read as a missing limit on queues that never use concurrency keys. Only queues that declare a totalConcurrencyLimit now change: their Limit cell reads as per-key plus total (e.g. 1 /key, 3 total) and Running turns warning-colored when the total cap is saturated. Plain queues are unchanged. --- .../route.tsx | 88 ++++++++----------- 1 file changed, 38 insertions(+), 50 deletions(-) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index 3a5d0ea9954..b62f89d4985 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -697,13 +697,12 @@ function QueuesWithMetricsView() { Name Queued Running - Limit - Total + Limit 0 && "text-text-bright" + queue.concurrency?.total?.current != null && + queue.running >= + Math.min( + queue.concurrency.total.current, + environment.concurrencyLimit + ) + ? "text-warning" + : queue.running > 0 && "text-text-bright" )} > {queue.running} @@ -872,29 +878,16 @@ function QueuesWithMetricsView() { ) : ( limit )} - - = - Math.min( - queue.concurrency.total.current, - environment.concurrencyLimit - ) && - "text-warning" - )} - > - {queue.concurrency?.total?.current != null - ? `${queue.concurrency.total.running ?? 0}/${Math.min( + {queue.concurrency?.total?.current != null ? ( + + /key ·{" "} + {Math.min( queue.concurrency.total.current, environment.concurrencyLimit - )}` - : "–"} + )}{" "} + total + + ) : null} - +
{hasFilters ? "No queues found matching your filters" : "No queues found"} @@ -1805,12 +1798,11 @@ function ClassicQueuesView() { Name Queued Running - Limit - Total + Limit 0 && "text-text-bright", + queue.concurrency?.total?.current != null && + queue.running >= + Math.min( + queue.concurrency.total.current, + environment.concurrencyLimit + ) + ? "text-warning" + : queue.running > 0 && "text-text-bright", isAtConcurrencyLimit && "text-warning" )} > @@ -1933,27 +1932,16 @@ function ClassicQueuesView() { )} > {limit} - - = - Math.min( - queue.concurrency.total.current, - environment.concurrencyLimit - ) && - "text-warning" - )} - > - {queue.concurrency?.total?.current != null - ? `${queue.concurrency.total.running ?? 0}/${Math.min( + {queue.concurrency?.total?.current != null ? ( + + /key ·{" "} + {Math.min( queue.concurrency.total.current, environment.concurrencyLimit - )}` - : "–"} + )}{" "} + total + + ) : null} - +
{hasFilters ? "No queues found matching your filters" : "No queues found"} From 7dc9c8ad253b7c1df23d006a6018dfc9f5d69933 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 18:39:14 +0100 Subject: [PATCH 07/10] fix(webapp): saturate the total-cap warning on keyed runs only The total cap gates keyed admissions, so the Running cell now warns off the group count rather than the aggregate that also includes unkeyed runs. --- .../route.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index b62f89d4985..c21a67cd633 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -847,7 +847,7 @@ function QueuesWithMetricsView() { "w-[1%]", queue.paused ? "opacity-50" : undefined, queue.concurrency?.total?.current != null && - queue.running >= + (queue.concurrency.total.running ?? 0) >= Math.min( queue.concurrency.total.current, environment.concurrencyLimit @@ -1911,7 +1911,7 @@ function ClassicQueuesView() { "w-[1%] pl-16 tabular-nums", queue.paused ? "opacity-50" : undefined, queue.concurrency?.total?.current != null && - queue.running >= + (queue.concurrency.total.running ?? 0) >= Math.min( queue.concurrency.total.current, environment.concurrencyLimit From ecfbdbb3f00edf5fb5072e070f80349c8b320357 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 19:11:06 +0100 Subject: [PATCH 08/10] refactor(webapp,core,clickhouse): combined concurrency in responses, dashboard and metrics Queue API responses expose concurrency.combined, the dashboard says combined, and the new metrics columns are named combined_running and combined_limit. --- .../v3/QueueRetrievePresenter.server.ts | 2 +- .../route.tsx | 28 ++++++++--------- .../route.tsx | 10 +++---- apps/webapp/app/v3/querySchemas.ts | 10 +++---- apps/webapp/app/v3/queueMetricsMapping.ts | 4 +-- ...dd_queue_metrics_combined_concurrency.sql} | 30 +++++++++---------- .../clickhouse/src/queueMetrics.ts | 4 +-- packages/core/src/v3/schemas/queues.ts | 12 ++++---- 8 files changed, 50 insertions(+), 50 deletions(-) rename internal-packages/clickhouse/schema/{042_add_queue_metrics_total_concurrency.sql => 042_add_queue_metrics_combined_concurrency.sql} (91%) diff --git a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts index 26811a8800e..1ce08e4c628 100644 --- a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts @@ -176,7 +176,7 @@ export function toQueueItem(data: { override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null, overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy), overriddenAt: data.concurrencyLimitOverriddenAt, - total: + combined: data.totalConcurrencyLimit !== undefined ? { current: data.totalConcurrencyLimit, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index c21a67cd633..5f4337ac161 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -700,7 +700,7 @@ function QueuesWithMetricsView() { Limit @@ -846,10 +846,10 @@ function QueuesWithMetricsView() { className={cn( "w-[1%]", queue.paused ? "opacity-50" : undefined, - queue.concurrency?.total?.current != null && - (queue.concurrency.total.running ?? 0) >= + queue.concurrency?.combined?.current != null && + (queue.concurrency.combined.running ?? 0) >= Math.min( - queue.concurrency.total.current, + queue.concurrency.combined.current, environment.concurrencyLimit ) ? "text-warning" @@ -878,14 +878,14 @@ function QueuesWithMetricsView() { ) : ( limit )} - {queue.concurrency?.total?.current != null ? ( + {queue.concurrency?.combined?.current != null ? ( /key ·{" "} {Math.min( - queue.concurrency.total.current, + queue.concurrency.combined.current, environment.concurrencyLimit )}{" "} - total + combined ) : null} @@ -1800,7 +1800,7 @@ function ClassicQueuesView() { Running Limit @@ -1910,10 +1910,10 @@ function ClassicQueuesView() { className={cn( "w-[1%] pl-16 tabular-nums", queue.paused ? "opacity-50" : undefined, - queue.concurrency?.total?.current != null && - (queue.concurrency.total.running ?? 0) >= + queue.concurrency?.combined?.current != null && + (queue.concurrency.combined.running ?? 0) >= Math.min( - queue.concurrency.total.current, + queue.concurrency.combined.current, environment.concurrencyLimit ) ? "text-warning" @@ -1932,14 +1932,14 @@ function ClassicQueuesView() { )} > {limit} - {queue.concurrency?.total?.current != null ? ( + {queue.concurrency?.combined?.current != null ? ( /key ·{" "} {Math.min( - queue.concurrency.total.current, + queue.concurrency.combined.current, environment.concurrencyLimit )}{" "} - total + combined ) : null} diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx index 22279ffcaba..7b3bddec64a 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues_.$queueParam/route.tsx @@ -396,7 +396,7 @@ export default function Page() { ids={ids} timeRange={timeRange} queueName={fullName} - hasTotalLimit={queue.concurrency?.total?.current != null} + hasTotalLimit={queue.concurrency?.combined?.current != null} /> )} @@ -494,25 +494,25 @@ function OverviewCharts({ /> {hasTotalLimit ? ( Runs in flight across ALL concurrency keys ( - ) versus the queue's total limit ( + ) versus the queue's combined limit ( ). } showLegend className="aspect-[2/1]" - query={`SELECT timeBucket() AS t, max(max_total_running) AS running, least(max(max_total_limit), max(max_env_limit)) AS cap\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} + query={`SELECT timeBucket() AS t, max(max_combined_running) AS running, least(max(max_combined_limit), max(max_env_limit)) AS cap\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} fillGaps minBucketSeconds={SYNCED_CHART_MIN_BUCKET_SECONDS} ids={ids} timeRange={timeRange} queueName={queueName} series={[ - { key: "cap", label: "Total limit", color: COLORS.limit }, + { key: "cap", label: "Combined limit", color: COLORS.limit }, { key: "running", label: "Running", color: COLORS.running }, ]} thresholdStroke={{ diff --git a/apps/webapp/app/v3/querySchemas.ts b/apps/webapp/app/v3/querySchemas.ts index 8267a0f8020..05cd7f0b394 100644 --- a/apps/webapp/app/v3/querySchemas.ts +++ b/apps/webapp/app/v3/querySchemas.ts @@ -770,19 +770,19 @@ const queueMetricsSchema: TableSchema = { fillMode: "carry", }), }, - max_total_running: { - name: "max_total_running", + max_combined_running: { + name: "max_combined_running", ...column("UInt32", { description: "Peak in-flight runs across ALL concurrency keys of the queue in the bucket (only emitted for keyed queues). Aggregate with max().", fillMode: "carry", }), }, - max_total_limit: { - name: "max_total_limit", + max_combined_limit: { + name: "max_combined_limit", ...column("UInt32", { description: - "The queue's total concurrency limit across all keys, as stored (0 = no cap; clamp against max_env_limit). Aggregate with max().", + "The queue's combined concurrency limit across all keys, as stored (0 = no cap; clamp against max_env_limit). Aggregate with max().", fillMode: "carry", }), }, diff --git a/apps/webapp/app/v3/queueMetricsMapping.ts b/apps/webapp/app/v3/queueMetricsMapping.ts index d341f4a63cd..f093dc3f027 100644 --- a/apps/webapp/app/v3/queueMetricsMapping.ts +++ b/apps/webapp/app/v3/queueMetricsMapping.ts @@ -131,8 +131,8 @@ export function mapEntryToRows( throttled: num(f.thr), ck_backlogged: num(f.ckq), ck_max_wait_ms: num(f.ckw), - total_running: num(f.tcc), - total_limit: num(f.tlim), + combined_running: num(f.tcc), + combined_limit: num(f.tlim), }, ]; } diff --git a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql b/internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql similarity index 91% rename from internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql rename to internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql index f45e6421ae3..03cb133799a 100644 --- a/internal-packages/clickhouse/schema/042_add_queue_metrics_total_concurrency.sql +++ b/internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql @@ -1,22 +1,22 @@ -- +goose Up --- Total-concurrency gauges: total_running is the in-flight count across ALL --- concurrency-key variants of a queue (the groupConcurrency set), total_limit the +-- Total-concurrency gauges: combined_running is the in-flight count across ALL +-- concurrency-key variants of a queue (the groupConcurrency set), combined_limit the -- RAW stored total cap (0 = none, readers clamp against max_env_limit). Emitted on -- base-queue gauge rows only. Per-key gauge rows now carry the EFFECTIVE per-key -- limit in queue_limit (override-aware), surfaced in the ck tier as max_limit. ALTER TABLE trigger_dev.queue_metrics_raw_v1 - ADD COLUMN IF NOT EXISTS total_running UInt32 DEFAULT 0, - ADD COLUMN IF NOT EXISTS total_limit UInt32 DEFAULT 0; + ADD COLUMN IF NOT EXISTS combined_running UInt32 DEFAULT 0, + ADD COLUMN IF NOT EXISTS combined_limit UInt32 DEFAULT 0; ALTER TABLE trigger_dev.queue_metrics_v1 - ADD COLUMN IF NOT EXISTS max_total_running SimpleAggregateFunction(max, UInt32), - ADD COLUMN IF NOT EXISTS max_total_limit SimpleAggregateFunction(max, UInt32); + ADD COLUMN IF NOT EXISTS max_combined_running SimpleAggregateFunction(max, UInt32), + ADD COLUMN IF NOT EXISTS max_combined_limit SimpleAggregateFunction(max, UInt32); ALTER TABLE trigger_dev.queue_metrics_5m_v1 - ADD COLUMN IF NOT EXISTS max_total_running SimpleAggregateFunction(max, UInt32), - ADD COLUMN IF NOT EXISTS max_total_limit SimpleAggregateFunction(max, UInt32); + ADD COLUMN IF NOT EXISTS max_combined_running SimpleAggregateFunction(max, UInt32), + ADD COLUMN IF NOT EXISTS max_combined_limit SimpleAggregateFunction(max, UInt32); ALTER TABLE trigger_dev.queue_metrics_ck_v1 ADD COLUMN IF NOT EXISTS max_limit SimpleAggregateFunction(max, UInt32); @@ -45,8 +45,8 @@ SELECT max(env_limit) AS max_env_limit, max(ck_backlogged) AS max_ck_backlogged, max(ck_max_wait_ms) AS max_ck_wait_ms, - max(total_running) AS max_total_running, - max(total_limit) AS max_total_limit, + max(combined_running) AS max_combined_running, + max(combined_limit) AS max_combined_limit, sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles @@ -73,8 +73,8 @@ SELECT max(env_limit) AS max_env_limit, max(ck_backlogged) AS max_ck_backlogged, max(ck_max_wait_ms) AS max_ck_wait_ms, - max(total_running) AS max_total_running, - max(total_limit) AS max_total_limit, + max(combined_running) AS max_combined_running, + max(combined_limit) AS max_combined_limit, sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles @@ -104,9 +104,9 @@ DROP VIEW IF EXISTS trigger_dev.queue_metrics_ck_mv_v1; DROP VIEW IF EXISTS trigger_dev.queue_metrics_5m_mv_v1; DROP VIEW IF EXISTS trigger_dev.queue_metrics_mv_v1; ALTER TABLE trigger_dev.queue_metrics_ck_v1 DROP COLUMN IF EXISTS max_limit; -ALTER TABLE trigger_dev.queue_metrics_5m_v1 DROP COLUMN IF EXISTS max_total_running, DROP COLUMN IF EXISTS max_total_limit; -ALTER TABLE trigger_dev.queue_metrics_v1 DROP COLUMN IF EXISTS max_total_running, DROP COLUMN IF EXISTS max_total_limit; -ALTER TABLE trigger_dev.queue_metrics_raw_v1 DROP COLUMN IF EXISTS total_running, DROP COLUMN IF EXISTS total_limit; +ALTER TABLE trigger_dev.queue_metrics_5m_v1 DROP COLUMN IF EXISTS max_combined_running, DROP COLUMN IF EXISTS max_combined_limit; +ALTER TABLE trigger_dev.queue_metrics_v1 DROP COLUMN IF EXISTS max_combined_running, DROP COLUMN IF EXISTS max_combined_limit; +ALTER TABLE trigger_dev.queue_metrics_raw_v1 DROP COLUMN IF EXISTS combined_running, DROP COLUMN IF EXISTS combined_limit; -- Recreate the pre-042 materialized views (the definitions from 036) so ingestion keeps -- feeding every aggregate table after a rollback. diff --git a/internal-packages/clickhouse/src/queueMetrics.ts b/internal-packages/clickhouse/src/queueMetrics.ts index aa3cf5296d2..f3a6be695e4 100644 --- a/internal-packages/clickhouse/src/queueMetrics.ts +++ b/internal-packages/clickhouse/src/queueMetrics.ts @@ -21,8 +21,8 @@ export const QueueMetricsRawV1Input = z.object({ throttled: z.number().optional(), ck_backlogged: z.number().optional(), ck_max_wait_ms: z.number().optional(), - total_running: z.number().optional(), - total_limit: z.number().optional(), + combined_running: z.number().optional(), + combined_limit: z.number().optional(), wait_ms: z.number().optional(), cumulative: z.number().optional(), }); diff --git a/packages/core/src/v3/schemas/queues.ts b/packages/core/src/v3/schemas/queues.ts index f3039f8075a..9ca282fb33d 100644 --- a/packages/core/src/v3/schemas/queues.ts +++ b/packages/core/src/v3/schemas/queues.ts @@ -45,16 +45,16 @@ export const QueueItem = z.object({ overriddenAt: z.coerce.date().nullable(), /** Who overrode the concurrency limit (will be null if overridden via the API) */ overriddenBy: z.string().nullable(), - /** The total concurrency cap across all concurrencyKey values of the queue */ - total: z + /** The combined concurrency cap across all concurrencyKey values of the queue */ + combined: z .object({ - /** The effective/current total concurrency limit (null = no cap) */ + /** The effective/current combined concurrency limit (null = no cap) */ current: z.number().nullable(), - /** The declared total limit an override reverts to on reset */ + /** The declared combined limit an override reverts to on reset */ base: z.number().nullable(), - /** The overridden total limit, when an override is active */ + /** The overridden combined limit, when an override is active */ override: z.number().nullable(), - /** When the total override was applied */ + /** When the combined override was applied */ overriddenAt: z.coerce.date().nullable(), /** Runs currently in flight across all concurrencyKey values */ running: z.number().nullable(), From 3fb8f96601ea8c130b9797803551ba87214b6f28 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Sat, 29 Aug 2026 19:35:52 +0100 Subject: [PATCH 09/10] feat(webapp): bracketed combined limit in the Limit column Queues that set a combinedConcurrencyLimit show it bracketed next to the per-key limit with a fine dashed underline and an explanatory tooltip; the Limit header tooltip is width-capped. Queues without one are unchanged. --- .../route.tsx | 74 ++++++++++++++----- 1 file changed, 56 insertions(+), 18 deletions(-) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index 5f4337ac161..4a13711ab54 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -700,7 +700,8 @@ function QueuesWithMetricsView() { Limit @@ -879,14 +880,32 @@ function QueuesWithMetricsView() { limit )} {queue.concurrency?.combined?.current != null ? ( - - /key ·{" "} - {Math.min( - queue.concurrency.combined.current, - environment.concurrencyLimit - )}{" "} - combined - + + ( + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )} + ) + + } + content={ + <> + Combined limit: at most{" "} + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )}{" "} + runs across all concurrency keys of this queue. The main limit + applies to each key separately. + + } + className="max-w-[260px]" + /> ) : null} Running Limit @@ -1933,14 +1953,32 @@ function ClassicQueuesView() { > {limit} {queue.concurrency?.combined?.current != null ? ( - - /key ·{" "} - {Math.min( - queue.concurrency.combined.current, - environment.concurrencyLimit - )}{" "} - combined - + + ( + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )} + ) + + } + content={ + <> + Combined limit: at most{" "} + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )}{" "} + runs across all concurrency keys of this queue. The main limit + applies to each key separately. + + } + className="max-w-[260px]" + /> ) : null} Date: Sat, 29 Aug 2026 19:40:12 +0100 Subject: [PATCH 10/10] fix(webapp): combined-limit tooltip renders beside the cell link The tooltip trigger is a button, so nesting it in the Limit cell's link made clicking it navigate; it now renders as the cell's trailing adornment. --- .../route.tsx | 61 ++++++++++--------- 1 file changed, 33 insertions(+), 28 deletions(-) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index 4a13711ab54..7b10b8d6d14 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -868,6 +868,39 @@ function QueuesWithMetricsView() { queue.paused ? "opacity-50" : undefined, queue.concurrency?.overriddenAt && "font-medium text-text-bright" )} + // The combined-limit hint is a tooltip button, so it renders beside the + // link (trailing) rather than nested inside the ; the number stays the + // link. + trailingContent={ + queue.concurrency?.combined?.current != null ? ( + + ( + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )} + ) + + } + content={ + <> + Combined limit: at most{" "} + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )}{" "} + runs across all concurrency keys of this queue. The main limit + applies to each key separately. + + } + className="max-w-[260px]" + /> + ) : undefined + } > {queue.concurrencyLimitOverridePercent !== null ? ( <> @@ -879,34 +912,6 @@ function QueuesWithMetricsView() { ) : ( limit )} - {queue.concurrency?.combined?.current != null ? ( - - ( - {Math.min( - queue.concurrency.combined.current, - environment.concurrencyLimit - )} - ) - - } - content={ - <> - Combined limit: at most{" "} - {Math.min( - queue.concurrency.combined.current, - environment.concurrencyLimit - )}{" "} - runs across all concurrency keys of this queue. The main limit - applies to each key separately. - - } - className="max-w-[260px]" - /> - ) : null}