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..1ce08e4c628 100644 --- a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts @@ -90,6 +90,9 @@ export class QueueRetrievePresenter extends BasePresenter { const results = await Promise.all([ engine.lengthOfQueues(environment, [queue.name]), engine.currentConcurrencyOfQueues(environment, [queue.name]), + queue.totalConcurrencyLimit != null + ? engine.totalConcurrencyOfQueues(environment, [queue.name]) + : undefined, ]); // Transform queues to include running and queued counts @@ -107,6 +110,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 +156,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 +176,16 @@ export function toQueueItem(data: { override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null, overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy), overriddenAt: data.concurrencyLimitOverriddenAt, + combined: + 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 05ac0e7b47d..66a188e810e 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 @@ -115,6 +115,7 @@ import { import { queueMetricsMaxPeriodDays } from "~/components/queues/queueMetricsPeriod.server"; import { isQueueAtCapacity } from "~/components/queues/queue-thresholds"; import { pageMeta } from "~/utils/pageTitle"; +import { InlineCode } from "~/components/code/InlineCode"; const SearchParamsSchema = z.object({ query: z.string().optional(), @@ -218,8 +219,13 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const timeRange = clipQueueMetricsWindow( timeFilterFromTo({ period: - resolveQueueMetricsPeriod({ period, from, to, defaultPeriod, maxPeriodDays }) ?? - undefined, + resolveQueueMetricsPeriod({ + period, + from, + to, + defaultPeriod, + maxPeriodDays, + }) ?? undefined, from: parseFiniteInt(from), to: parseFiniteInt(to), defaultPeriod, @@ -243,7 +249,9 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { }; } } catch (error) { - logger.warn("Queue list metrics unavailable, rendering without them", { error }); + logger.warn("Queue list metrics unavailable, rendering without them", { + error, + }); } } @@ -706,7 +714,14 @@ function QueuesWithMetricsView() { Name Queued Running - Limit + + Limit + 0 && "text-text-bright" + queue.concurrency?.combined?.current != null && + (queue.concurrency.combined.running ?? 0) >= + Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + ) + ? "text-warning" + : queue.running > 0 && "text-text-bright" )} > {queue.running} @@ -863,12 +885,46 @@ 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 ? ( <> {limit} - ({formatOverridePercent(queue.concurrencyLimitOverridePercent)}%) + ({formatOverridePercent(queue.concurrencyLimitOverridePercent)} + %) ) : ( @@ -1265,7 +1321,11 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ value: limit > 0 ? Math.round((tileNumber(r.running) / limit) * 100) : 0, }; }); - return { points, total: peakOf(points), formatTotal: (v) => `${v}% peak` }; + return { + points, + total: peakOf(points), + formatTotal: (v) => `${v}% peak`, + }; }, }, { @@ -1279,7 +1339,11 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ bucket: tileTimeToMs(r.t), value: tileNumber(r.queued), })); - return { points, total: peakOf(points), formatTotal: (v) => `${v.toLocaleString()} peak` }; + return { + points, + total: peakOf(points), + formatTotal: (v) => `${v.toLocaleString()} peak`, + }; }, }, { @@ -1784,7 +1848,13 @@ function ClassicQueuesView() { Name Queued Running - Limit + + Limit + 0 && "text-text-bright", + queue.concurrency?.combined?.current != null && + (queue.concurrency.combined.running ?? 0) >= + Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + ) + ? "text-warning" + : queue.running > 0 && "text-text-bright", isAtConcurrencyLimit && "text-warning" )} > @@ -1906,6 +1983,34 @@ function ClassicQueuesView() { )} > {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} ); } + +const limitTooltip = ( + <> + + How many runs can execute at once.{" "} + + + 1 (20) means 1 run per concurrency key, + but at most 20 runs across all keys. Set using{" "} + combinedConcurrencyLimit in your code. + + +); 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..aa7400452a7 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() { ) ) : ( - + )} @@ -436,10 +441,12 @@ function OverviewCharts({ ids, timeRange, queueName, + hasTotalLimit, }: { ids: Ids; timeRange: TimeRangeParams; queueName: string; + hasTotalLimit: boolean; }) { const zoomToTimeFilter = useZoomToTimeFilter(); return ( @@ -479,6 +486,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 combined limit ( + + ). + + } + showLegend + className="aspect-[2/1]" + 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: "Combined limit", color: COLORS.limit }, + { key: "running", label: "Running", color: COLORS.running }, + ]} + thresholdStroke={{ + series: "running", + valueFromSeries: "cap", + aboveColor: "var(--color-warning)", + }} + carryBackfill={["cap"]} + /> + ) : null} 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(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 +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_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. +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/clickhouse/src/queueMetrics.ts b/internal-packages/clickhouse/src/queueMetrics.ts index 39576b4a0a3..f3a6be695e4 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(), + combined_running: z.number().optional(), + combined_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..bdd4c30adb7 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, @@ -2979,6 +2993,7 @@ export class RunEngine { { select: { queue: true, + concurrencyKey: true, }, }, this.prisma @@ -3000,6 +3015,7 @@ export class RunEngine { runId, orgId: latestSnapshot.organizationId, queue: taskRun.queue, + concurrencyKey: taskRun.concurrencyKey ?? undefined, env: { id: latestSnapshot.environmentId, type: latestSnapshot.environmentType, diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 7f6493dfc2b..dec9949d6f2 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -108,18 +108,12 @@ local function __gateReconcile(setKey, msgKeyPrefix, reconcileKeyPrefix) end end -local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msgKeyPrefix, ckOverridesEnabled) +local function __gatesHaveCapacity(gatesKeyPrefix, msg, messageId, envLimit, msgKeyPrefix) if not msg.gates then return true end for _, gate in ipairs(msg.gates) do local base, variant, gateKey = __gateKeys(gatesKeyPrefix, msg, gate) local occupancy = tonumber(redis.call('SCARD', variant .. ':currentConcurrency') or '0') local perKeyLimit = math.min(tonumber(redis.call('GET', base .. ':concurrency') or '1000000'), envLimit) - if ckOverridesEnabled and gateKey and gateKey ~= '' then - local gateOverride = redis.call('HGET', base .. ':ckLimits', string.sub(variant, #gatesKeyPrefix + 1)) - if gateOverride then - perKeyLimit = math.min(tonumber(gateOverride), envLimit) - end - end if occupancy >= perKeyLimit and redis.call('SISMEMBER', variant .. ':currentConcurrency', messageId) == 0 then __gateReconcile(variant .. ':currentConcurrency', msgKeyPrefix, gatesKeyPrefix) return false @@ -208,6 +202,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'", @@ -218,6 +220,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({ @@ -229,6 +232,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 @@ -244,6 +248,7 @@ const QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", throttledExpr: "false", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); /** Injected queue-metrics stream emitter; all calls are no-ops when metrics are disabled. */ @@ -257,13 +262,6 @@ export interface RunQueueMetricsEmitter { emitGauge(shardKey: string, fields: Record): void; } -export class RunQueueConcurrencyKeyLimitExceededError extends Error { - constructor(message: string) { - super(message); - this.name = "RunQueueConcurrencyKeyLimitExceededError"; - } -} - export type RunQueueOptions = { name: string; tracer: Tracer; @@ -325,8 +323,6 @@ export type RunQueueOptions = { * the total cap covering releases from builds without the mirror. */ gatesEnabled?: boolean; - /** Cap on per-concurrency-key limit overrides stored per queue. Default 1000. */ - maxConcurrencyKeyOverridesPerQueue?: number; workerOptions?: { pollIntervalMs?: number; immediatePollIntervalMs?: number; @@ -440,7 +436,6 @@ export class RunQueue { private queueSelectionStrategy: RunQueueSelectionStrategy; private shardCount: number; private counterTtlSeconds: number; - private maxConcurrencyKeyOverridesPerQueue: number; private abortController: AbortController; private worker: Worker; private workerQueueResolver: WorkerQueueResolver; @@ -451,7 +446,6 @@ export class RunQueue { constructor(public readonly options: RunQueueOptions) { this.shardCount = options.shardCount ?? 2; this.counterTtlSeconds = options.counterTtlSeconds ?? 86400; - this.maxConcurrencyKeyOverridesPerQueue = options.maxConcurrencyKeyOverridesPerQueue ?? 1000; this.retryOptions = options.retryOptions ?? defaultRetrySettings; this.redis = createRedisClient(options.redis, { onError: (error) => { @@ -646,60 +640,44 @@ export class RunQueue { return this.redis.scard(this.keys.queueGroupConcurrencyKey(env, queue)); } - /** - * Sets a per-concurrency-key limit override for a queue. The stored value is the - * raw requested limit; admit paths clamp to the environment limit at read time. - * Throws RunQueueConcurrencyKeyLimitExceededError when a NEW key would push the - * queue past maxConcurrencyKeyOverridesPerQueue (updates to existing keys always - * succeed). - */ - public async updateQueueConcurrencyKeyLimit( + /** Batch variant of totalConcurrencyOfQueue: one pipeline of group SCARDs. */ + public async totalConcurrencyOfQueues( env: MinimalAuthenticatedEnvironment, - queue: string, - concurrencyKey: string, - limit: number - ) { - const result = await this.redis.setQueueConcurrencyKeyLimit( - this.keys.queueCkLimitsKey(env, queue), - this.keys.queueKey(env, queue, concurrencyKey), - String(limit), - String(this.maxConcurrencyKeyOverridesPerQueue) - ); + queues: string[] + ): Promise> { + const pipeline = this.redis.pipeline(); + queues.forEach((queue) => { + pipeline.scard(this.keys.queueGroupConcurrencyKey(env, queue)); + }); - if (result === 0) { - throw new RunQueueConcurrencyKeyLimitExceededError( - `Cannot add a concurrency key override to queue ${queue}: the queue already has ${this.maxConcurrencyKeyOverridesPerQueue} overrides` - ); - } - } + const results = await pipeline.exec(); - public async removeQueueConcurrencyKeyLimit( - env: MinimalAuthenticatedEnvironment, - queue: string, - concurrencyKey: string - ) { - return this.redis.hdel( - this.keys.queueCkLimitsKey(env, queue), - this.keys.queueKey(env, queue, concurrencyKey) + return queues.reduce( + (acc, queue, index) => { + const value = results?.[index]?.[1]; + acc[queue] = typeof value === "number" ? value : 0; + return acc; + }, + {} as Record ); } - /** Returns the raw per-concurrency-key limit overrides for a queue, keyed by concurrency key value. */ - public async getQueueConcurrencyKeyLimits( + /** Batch read of the RAW stored total concurrency limits (undefined = no cap). */ + public async totalConcurrencyLimitsOfQueues( env: MinimalAuthenticatedEnvironment, - queue: string - ): Promise> { - const raw = await this.redis.hgetall(this.keys.queueCkLimitsKey(env, queue)); + queues: string[] + ): Promise> { + const keys = queues.map((queue) => this.keys.queueTotalConcurrencyLimitKey(env, queue)); + const values = keys.length > 0 ? await this.redis.mget(...keys) : []; - const limits: Record = {}; - for (const [variantName, value] of Object.entries(raw)) { - const ckIndex = variantName.indexOf(":ck:"); - if (ckIndex === -1) { - continue; - } - limits[variantName.slice(ckIndex + 4)] = Number(value); - } - return limits; + 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) { @@ -1550,6 +1528,7 @@ export class RunQueue { runId: string; orgId: string; queue: string; + concurrencyKey?: string; env: RunQueueKeyProducerEnvironment; }) { return this.#callClearMessageFromConcurrencySets(params); @@ -2333,6 +2312,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); } @@ -2435,7 +2418,6 @@ export class RunQueue { const totalConcurrencyLimitKey = this.keys.queueTotalConcurrencyLimitKeyFromQueue( message.queue ); - const ckLimitsKey = this.keys.queueCkLimitsKeyFromQueue(message.queue); const totalConcurrencyEnabledArg = this.options.totalConcurrencyEnabled ? "1" : "0"; if (ttlInfo) { @@ -2459,7 +2441,6 @@ export class RunQueue { baseQueueKey, groupConcurrencyKey, totalConcurrencyLimitKey, - ckLimitsKey, // args queueName, messageId, @@ -2499,7 +2480,6 @@ export class RunQueue { baseQueueKey, groupConcurrencyKey, totalConcurrencyLimitKey, - ckLimitsKey, // args queueName, messageId, @@ -2790,7 +2770,6 @@ export class RunQueue { runningCounterKey, this.keys.queueGroupConcurrencyKeyFromQueue(ckWildcardQueue), this.keys.queueTotalConcurrencyLimitKeyFromQueue(ckWildcardQueue), - this.keys.queueCkLimitsKeyFromQueue(ckWildcardQueue), //args ckWildcardQueue, String(Date.now()), @@ -3081,18 +3060,30 @@ export class RunQueue { runId, orgId, queue, + concurrencyKey, env, }: { runId: string; orgId: string; queue: string; + concurrencyKey?: string; env: RunQueueKeyProducerEnvironment; }) { const messageId = runId; const messageKey = this.keys.messageKey(orgId, messageId); - const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKey(env, queue); + /** + * Callers pass the bare TaskRun queue name plus its concurrencyKey; the run's + * slots live on the ck variant, and the tracked clear additionally mirrors the + * group set and counters that only keyed queues maintain. + */ + const fullQueue = concurrencyKey ? this.keys.queueKey(env, queue, concurrencyKey) : queue; + const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKey( + env, + queue, + concurrencyKey + ); const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKey(env); - const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKey(env, queue); + const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKey(env, queue, concurrencyKey); const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKey(env); this.logger.debug("Calling clearMessageFromConcurrencySets", { @@ -3107,15 +3098,15 @@ export class RunQueue { service: this.name, }); - if (queue.includes(":ck:")) { + if (fullQueue.includes(":ck:")) { return this.redis.clearMessageFromConcurrencySetsTracked( queueCurrentConcurrencyKey, envCurrentConcurrencyKey, queueCurrentDequeuedKey, envCurrentDequeuedKey, - this.keys.queueRunningCounterKeyFromQueue(queue), - this.keys.ckIndexKeyFromQueue(queue), - this.keys.queueGroupConcurrencyKeyFromQueue(queue), + this.keys.queueRunningCounterKeyFromQueue(fullQueue), + this.keys.ckIndexKeyFromQueue(fullQueue), + this.keys.queueGroupConcurrencyKeyFromQueue(fullQueue), messageKey, messageId, this.options.redis.keyPrefix ?? "", @@ -3687,7 +3678,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) end end @@ -3802,7 +3793,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) end end @@ -4081,7 +4072,7 @@ return __qmret(0) // *Tracked variants of dequeueMessageFromKey and the ack/nack/dlq/release/clear // scripts. this.redis.defineCommand("enqueueMessageCkTracked", { - numberOfKeys: 18, + numberOfKeys: 17, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -4103,7 +4094,6 @@ local baseQueueKey = KEYS[15] -- Total-cap keys (KEYS 16-17) local groupConcurrencyKey = KEYS[16] local totalConcurrencyLimitKey = KEYS[17] -local ckLimitsKey = KEYS[18] local queueName = ARGV[1] local messageId = ARGV[2] @@ -4141,12 +4131,6 @@ if enableFastPath == '1' then tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envLimit ) - if totalConcurrencyEnabled then - local perKeyOverride = redis.call('HGET', ckLimitsKey, queueName) - if perKeyOverride then - queueLimit = math.min(tonumber(perKeyOverride), envLimit) - end - end if queueCurrent < queueLimit then -- Total-cap gate: a fast-path admit consumes a group slot, so it must @@ -4169,7 +4153,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) end end @@ -4258,7 +4242,7 @@ return __qmret(0) }); this.redis.defineCommand("enqueueMessageWithTtlCkTracked", { - numberOfKeys: 19, + numberOfKeys: 18, lua: ` local masterQueueKey = KEYS[1] local queueKey = KEYS[2] @@ -4281,7 +4265,6 @@ local baseQueueKey = KEYS[16] -- Total-cap keys (KEYS 17-18) local groupConcurrencyKey = KEYS[17] local totalConcurrencyLimitKey = KEYS[18] -local ckLimitsKey = KEYS[19] local queueName = ARGV[1] local messageId = ARGV[2] @@ -4321,12 +4304,6 @@ if enableFastPath == '1' then tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envLimit ) - if totalConcurrencyEnabled then - local perKeyOverride = redis.call('HGET', ckLimitsKey, queueName) - if perKeyOverride then - queueLimit = math.min(tonumber(perKeyOverride), envLimit) - end - end if queueCurrent < queueLimit then -- Total-cap gate: see enqueueMessageCkTracked. @@ -4347,7 +4324,7 @@ if enableFastPath == '1' then local okDecode, decoded = pcall(cjson.decode, messageData) if okDecode and type(decoded) == 'table' and decoded.gates then gateMsg = decoded - gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil, totalConcurrencyEnabled) + gatesAllowFastPath = __gatesHaveCapacity(keyPrefix, decoded, messageId, envLimit, nil) end end @@ -4750,7 +4727,7 @@ for i = 1, #messages, 2 do else local gatesAllow = true if gatesEnabled then - gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix, totalConcurrencyEnabled) + gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix) end if gatesAllow then @@ -4792,6 +4769,10 @@ else redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName) end +-- Re-sample the gauge so the emitted snapshot includes this batch's admissions; +-- the top-of-script sample only covers the early returns where nothing was admitted. +${QUEUE_METRICS_GAUGE_LUA} + -- Return results as a flat array: [messageId1, messageScore1, messagePayload1, messageId2, messageScore2, messagePayload2, ...] return __qmret(results) `, @@ -4952,7 +4933,7 @@ return results // (normal dequeue, TTL-expired, or stale-orphan path — all of which were // counted at enqueue time). this.redis.defineCommand("dequeueMessagesFromCkQueueTracked", { - numberOfKeys: 14, + numberOfKeys: 13, lua: ` local ckIndexKey = KEYS[1] local queueConcurrencyLimitKey = KEYS[2] @@ -4967,7 +4948,6 @@ local lengthCounterKey = KEYS[10] local runningCounterKey = KEYS[11] local groupConcurrencyKey = KEYS[12] local totalConcurrencyLimitKey = KEYS[13] -local ckLimitsKey = KEYS[14] local ckWildcardName = ARGV[1] local currentTime = tonumber(ARGV[2]) @@ -5060,12 +5040,6 @@ for _, ckQueueName in ipairs(ckQueues) do local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0') local perKeyLimit = queueConcurrencyLimit - if totalConcurrencyEnabled then - local perKeyOverride = redis.call('HGET', ckLimitsKey, ckQueueName) - if perKeyOverride then - perKeyLimit = math.min(tonumber(perKeyOverride), envConcurrencyLimit) - end - end if ckCurrentConcurrency >= perKeyLimit then -- Back a blocked variant off so it cannot pin the bounded candidate window @@ -5100,7 +5074,7 @@ for _, ckQueueName in ipairs(ckQueues) do else local gatesAllow = true if gatesEnabled then - gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix, totalConcurrencyEnabled) + gatesAllow = __gatesHaveCapacity(keyPrefix, messageData, messageId, envConcurrencyLimit, messageKeyPrefix) end if not gatesAllow then blockedByGates = true @@ -5175,6 +5149,10 @@ else redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName) end +-- Re-sample the gauge so the emitted snapshot includes this batch's admissions; +-- the top-of-script sample only covers the early returns where nothing was admitted. +${QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA} + return __qmret(results) `, }); @@ -5998,26 +5976,6 @@ __gatesRelease(keyPrefix, redis.call('GET', messageKey), messageId) `, }); - this.redis.defineCommand("setQueueConcurrencyKeyLimit", { - numberOfKeys: 1, - lua: ` -local ckLimitsKey = KEYS[1] - -local fieldName = ARGV[1] -local limit = ARGV[2] -local maxFields = tonumber(ARGV[3]) - -if redis.call('HEXISTS', ckLimitsKey, fieldName) == 0 then - if redis.call('HLEN', ckLimitsKey) >= maxFields then - return 0 - end -end - -redis.call('HSET', ckLimitsKey, fieldName, limit) -return 1 -`, - }); - this.redis.defineCommand("updateEnvironmentConcurrencyLimits", { numberOfKeys: 2, lua: ` @@ -6382,14 +6340,6 @@ declare module "@internal/redis" { callback?: Callback ): Result; - setQueueConcurrencyKeyLimit( - ckLimitsKey: string, - fieldName: string, - limit: string, - maxFields: string, - callback?: Callback - ): Result; - updateEnvironmentConcurrencyLimits( // keys envConcurrencyLimitKey: string, @@ -6576,7 +6526,6 @@ declare module "@internal/redis" { baseQueueKey: string, groupConcurrencyKey: string, totalConcurrencyLimitKey: string, - ckLimitsKey: string, queueName: string, messageId: string, messageData: string, @@ -6614,7 +6563,6 @@ declare module "@internal/redis" { baseQueueKey: string, groupConcurrencyKey: string, totalConcurrencyLimitKey: string, - ckLimitsKey: string, queueName: string, messageId: string, messageData: string, @@ -6649,7 +6597,6 @@ declare module "@internal/redis" { runningCounterKey: string, groupConcurrencyKey: string, totalConcurrencyLimitKey: string, - ckLimitsKey: string, ckWildcardName: string, currentTime: string, defaultEnvConcurrencyLimit: string, diff --git a/internal-packages/run-engine/src/run-queue/keyProducer.ts b/internal-packages/run-engine/src/run-queue/keyProducer.ts index 120e04f8c38..98028f5af7b 100644 --- a/internal-packages/run-engine/src/run-queue/keyProducer.ts +++ b/internal-packages/run-engine/src/run-queue/keyProducer.ts @@ -366,14 +366,6 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer { return `${this.baseQueueKeyFromQueue(queue)}:${constants.TOTAL_CONCURRENCY_LIMIT_PART}`; } - queueCkLimitsKey(env: RunQueueKeyProducerEnvironment, queue: string): string { - return `${this.queueKey(env, queue)}:ckLimits`; - } - - queueCkLimitsKeyFromQueue(queue: string): string { - return `${this.baseQueueKeyFromQueue(queue)}:ckLimits`; - } - isCkWildcard(queue: string): boolean { return queue.endsWith(":ck:*"); } diff --git a/internal-packages/run-engine/src/run-queue/types.ts b/internal-packages/run-engine/src/run-queue/types.ts index 2961b642314..2cbfe40c775 100644 --- a/internal-packages/run-engine/src/run-queue/types.ts +++ b/internal-packages/run-engine/src/run-queue/types.ts @@ -111,9 +111,6 @@ export interface RunQueueKeyProducer { queueTotalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string; queueTotalConcurrencyLimitKeyFromQueue(queue: string): string; - queueCkLimitsKey(env: RunQueueKeyProducerEnvironment, queue: string): string; - queueCkLimitsKeyFromQueue(queue: string): string; - //env oncurrency envCurrentConcurrencyKey(env: EnvDescriptor): string; envCurrentConcurrencyKey(env: RunQueueKeyProducerEnvironment): string; diff --git a/knip.json b/knip.json index c8e7f4027dd..84456756ca1 100644 --- a/knip.json +++ b/knip.json @@ -27,9 +27,6 @@ ], "ignoreDependencies": ["@sentry/cli", "assert", "util"] }, - "internal-packages/run-engine": { - "ignore": ["src/run-queue/index.ts"] - }, "internal-packages/dashboard-agent": { "entry": ["trigger.config.ts", "src/investigation-sweep.ts", "src/maintenance.ts"], "ignoreBinaries": ["rg"] diff --git a/packages/core/src/v3/schemas/queues.ts b/packages/core/src/v3/schemas/queues.ts index 34a47b34e3e..cf604c3c298 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 combined concurrency cap across all concurrencyKey values of the queue */ + combined: z + .object({ + /** The current combined concurrency limit as declared or overridden (null = no cap). Enforcement clamps it to the environment concurrency limit at admit time. */ + current: z.number().nullable(), + /** The declared combined limit an override reverts to on reset */ + base: z.number().nullable(), + /** The overridden combined limit, when an override is active */ + override: z.number().nullable(), + /** When the combined override was applied */ + overriddenAt: z.coerce.date().nullable(), + /** Runs currently in flight across all concurrencyKey values */ + running: z.number().nullable(), + }) + .optional(), }) .optional(), });