diff --git a/.server-changes/paginate-concurrency-keys-table.md b/.server-changes/paginate-concurrency-keys-table.md new file mode 100644 index 0000000000..da13bfe169 --- /dev/null +++ b/.server-changes/paginate-concurrency-keys-table.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +The concurrency keys table on a queue's page is now paginated, so queues with thousands of keys can page through all of them instead of only showing the top 50. diff --git a/.server-changes/queue-metrics-dashboard.md b/.server-changes/queue-metrics-dashboard.md index 8574983170..74245fefe6 100644 --- a/.server-changes/queue-metrics-dashboard.md +++ b/.server-changes/queue-metrics-dashboard.md @@ -3,4 +3,4 @@ area: webapp type: feature --- -Queue metrics and health on the Queues page: per-queue depth, throughput, concurrency, throttling, and scheduling-delay charts, plus a per-queue detail view, live queue stats and a backlog chart on the task detail page, and a waiting-in-queue explainer in the run inspector for runs that have not started yet. Off by default; enabled per organization. +New Queue metrics & health dashboard (per-org opt-in): per-queue depth, throughput, concurrency, throttling and scheduling-delay charts, a per-queue detail view, and live queue stats. Queue concurrency limits set above the environment limit are now rejected instead of being silently capped. diff --git a/.server-changes/queue-override-reject-above-limit.md b/.server-changes/queue-override-reject-above-limit.md deleted file mode 100644 index 63d38690ad..0000000000 --- a/.server-changes/queue-override-reject-above-limit.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: breaking ---- - -Setting a queue's concurrency limit above the environment limit is now rejected with a clear error instead of being silently capped. Existing overrides are unaffected. diff --git a/.server-changes/queue-pages-live-and-clarity.md b/.server-changes/queue-pages-live-and-clarity.md deleted file mode 100644 index 97785debe6..0000000000 --- a/.server-changes/queue-pages-live-and-clarity.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: improvement ---- - -Queue pages now refresh their live numbers automatically, and the charts are clearer: better axis labels, hover explanations for the headline stats, and the busiest concurrency key named on the chart. diff --git a/apps/webapp/app/components/layout/MetricsLayout.tsx b/apps/webapp/app/components/layout/MetricsLayout.tsx index 6ae40778af..12d0fcd743 100644 --- a/apps/webapp/app/components/layout/MetricsLayout.tsx +++ b/apps/webapp/app/components/layout/MetricsLayout.tsx @@ -178,7 +178,7 @@ function MetricsLayoutMain({ children, scroll }: { children: ReactNode; scroll:
@@ -275,9 +275,21 @@ function MetricsLayoutRoot({ * standard page insets. Compose left/right clusters as child divs — `justify-between` spreads them * (a single child sits at the start). */ -function MetricsLayoutFilters({ children }: { children: ReactNode }) { +function MetricsLayoutFilters({ + children, + className, +}: { + children: ReactNode; + /** Override the baked horizontal padding (the two queue pages want slightly different insets). */ + className?: string; +}) { return ( -
+
{children}
); @@ -306,7 +318,7 @@ function MetricsLayoutGrid({ return (
{children}
; + return
{children}
; } export const MetricsLayout = { diff --git a/apps/webapp/app/components/metrics/BigNumber.tsx b/apps/webapp/app/components/metrics/BigNumber.tsx index f9e0ecdb5c..5843e6229d 100644 --- a/apps/webapp/app/components/metrics/BigNumber.tsx +++ b/apps/webapp/app/components/metrics/BigNumber.tsx @@ -40,7 +40,7 @@ export function BigNumber({ typeof compactThreshold === "number" && v !== undefined && v >= compactThreshold; return ( -
+
{title} {accessory &&
{accessory}
} @@ -56,7 +56,7 @@ export function BigNumber({ ) : formattedValue !== undefined ? (
{formattedValue} - {suffix &&
{suffix}
} + {suffix &&
{suffix}
}
) : v !== undefined ? (
@@ -70,7 +70,7 @@ export function BigNumber({ ) : ( formatNumber(v) )} - {suffix &&
{suffix}
} + {suffix &&
{suffix}
}
) : ( "–" diff --git a/apps/webapp/app/components/metrics/MiniLineChart.tsx b/apps/webapp/app/components/metrics/MiniLineChart.tsx index 9ad221923a..c36e91d0b9 100644 --- a/apps/webapp/app/components/metrics/MiniLineChart.tsx +++ b/apps/webapp/app/components/metrics/MiniLineChart.tsx @@ -138,7 +138,7 @@ export function MiniLineChart({ type="monotone" dataKey="count" stroke={color} - strokeWidth={1.5} + strokeWidth={1} dot={false} activeDot={{ r: 2.5, fill: color, strokeWidth: 0 }} isAnimationActive={false} @@ -148,7 +148,7 @@ export function MiniLineChart({ type="monotone" dataKey="throttledOverlay" stroke="var(--color-warning)" - strokeWidth={1.5} + strokeWidth={1} dot={false} activeDot={{ r: 2.5, fill: "var(--color-warning)", strokeWidth: 0 }} connectNulls={false} diff --git a/apps/webapp/app/components/primitives/SegmentedControl.tsx b/apps/webapp/app/components/primitives/SegmentedControl.tsx index 2f749215c2..de9a84b5c0 100644 --- a/apps/webapp/app/components/primitives/SegmentedControl.tsx +++ b/apps/webapp/app/components/primitives/SegmentedControl.tsx @@ -76,6 +76,8 @@ type SegmentedControlProps = { variant?: VariantType; fullWidth?: boolean; onChange?: (value: string) => void; + /** Override the control's outer styling (e.g. a height that matches an adjacent input). */ + className?: string; }; export default function SegmentedControl({ @@ -86,6 +88,7 @@ export default function SegmentedControl({ variant = "secondary/medium", fullWidth, onChange, + className, }: SegmentedControlProps) { const variantStyle = variants[variant]; const _isPrimary = variant.startsWith("primary"); @@ -95,7 +98,8 @@ export default function SegmentedControl({ className={cn( "flex rounded text-text-bright", variantStyle.base, - fullWidth ? "w-full" : "w-fit" + fullWidth ? "w-full" : "w-fit", + className )} > ) : null; @@ -271,40 +274,19 @@ export const TableHeaderCell = forwardRef {sortable ? ( - // Order is always title → info icon → sort arrows. The info trigger is itself a - {tooltip ? ( - <> - {tooltipNode} - - - ) : null}
) : tooltip ? (
@@ -407,19 +389,27 @@ export const TableCell = forwardRef( // With leading/trailing content, the link is content-sized and the adornments sit beside // it (still inside the td) so interactive triggers never nest inside the . leadingContent || trailingContent ? ( -
- {leadingContent} + // Stretched link: the covers the whole cell via an inset ::before overlay, so the + // entire cell is clickable — not just the text. The interactive adornments (tooltip + // icons, badge buttons) sit above the overlay (relative z-10) and stay clickable, and + // never nest inside the . +
+ {leadingContent ? ( + {leadingContent} + ) : null} {children} - {trailingContent} + {trailingContent ? ( + {trailingContent} + ) : null}
) : ( diff --git a/apps/webapp/app/components/primitives/charts/Chart.tsx b/apps/webapp/app/components/primitives/charts/Chart.tsx index a742fbbde4..932eaa516b 100644 --- a/apps/webapp/app/components/primitives/charts/Chart.tsx +++ b/apps/webapp/app/components/primitives/charts/Chart.tsx @@ -176,7 +176,10 @@ const ChartTooltipContent = React.forwardRef< {payload.map((item, index) => { const key = `${nameKey || item.name || item.dataKey || "value"}`; const itemConfig = getPayloadConfigFromPayload(config, item, key); - const indicatorColor = color || item.payload.fill || item.color; + // Prefer the series' configured colour over item.color: a threshold/gradient line's + // recharts colour is a `url(#…)` gradient ref, which is invalid as a CSS background and + // renders no swatch. The config colour is the intended solid series colour. + const indicatorColor = color || itemConfig?.color || item.payload.fill || item.color; return (
diff --git a/apps/webapp/app/components/primitives/charts/ChartCard.tsx b/apps/webapp/app/components/primitives/charts/ChartCard.tsx index c008db8069..69464cf340 100644 --- a/apps/webapp/app/components/primitives/charts/ChartCard.tsx +++ b/apps/webapp/app/components/primitives/charts/ChartCard.tsx @@ -88,7 +88,9 @@ export function ChartCard({ {maximizable && ( - {title} + {/* In fullscreen, space the title's legend (the flex-col title node) further from the + title — gap-6 instead of the card's gap-1. */} + {title}
{parentSync ? ( diff --git a/apps/webapp/app/components/primitives/charts/ChartLine.tsx b/apps/webapp/app/components/primitives/charts/ChartLine.tsx index 6adcf0484a..000d74b91c 100644 --- a/apps/webapp/app/components/primitives/charts/ChartLine.tsx +++ b/apps/webapp/app/components/primitives/charts/ChartLine.tsx @@ -118,11 +118,12 @@ export type ChartLineRendererProps = { * pinned so the gradient split lines up exactly with the plotted values and reference lines. * Single-series (non-stacked) line charts only. * - * Prefer {@link warningOverlay} when the threshold sits far below the data maximum: a gradient - * split whose boundary collapses onto the baseline paints zero/near-zero buckets the warning - * colour. The overlay retraces per-bucket instead, so under-threshold buckets always stay blue. + * The gradient offset is derived from the plotted line's own value range (objectBoundingBox maps + * 0..1 to the path's bounding box, not the y-axis), so the colour change lands exactly at the + * threshold value however the domain is padded for reference lines. `series` targets which line + * the gradient applies to (others keep their own colour); defaults to the first series. */ - thresholdStroke?: { value: number; aboveColor: string }; + thresholdStroke?: { value: number; aboveColor: string; series?: string }; /** * Per-bucket warning recolour: a series is retraced in the warning colour only across buckets * where it crosses a limit — either strictly above a constant `threshold` (single-series case, @@ -255,18 +256,13 @@ export function ChartLineRenderer({ return ; } - // Get the x-axis ticks based on tooltip state - const xAxisTicks = - highlight.tooltipActive && data.length > 2 - ? [data[0]?.[dataKey], data[data.length - 1]?.[dataKey]] - : undefined; - const xAxisConfig = { dataKey, tickLine: false, axisLine: false, tickMargin: 10, - ticks: xAxisTicks, + // Keep every x-axis label visible at all times, including on hover. Previously the axis + // collapsed to just the first + last tick while the tooltip was active. interval: "preserveStartEnd" as const, tick: { fill: "var(--color-text-dimmed)", @@ -278,24 +274,36 @@ export function ChartLineRenderer({ // A threshold stroke needs an exact, fixed y-domain so the gradient split aligns with the // plotted values and the reference lines. Compute it from the data + reference/threshold ys. - let thresholdDomain: [number, number] | undefined; + let thresholdActive = false; let thresholdOffset = 0; if (thresholdStroke && !stacked) { - let dataMax = thresholdStroke.value; + thresholdActive = true; + // The gradient is objectBoundingBox — its 0..1 maps to the plotted line's own bounding box + // (lineMax at the top, lineMin at the bottom), NOT the y-axis. So derive the split from the + // target line's value range: offset = (lineMax - threshold) / (lineMax - lineMin) lands the + // colour change exactly at the threshold value's pixel, whatever the axis domain is. That means + // we don't pin the domain (which coarsened the ticks) — it auto-scales as usual. + const gradientKey = thresholdStroke.series ?? visibleSeries[0]; + let lineMin = Infinity; + let lineMax = -Infinity; for (const row of data) { - for (const key of visibleSeries) { - const v = Number(row[key]); - if (Number.isFinite(v) && v > dataMax) dataMax = v; + const v = Number(row[gradientKey]); + if (Number.isFinite(v)) { + if (v < lineMin) lineMin = v; + if (v > lineMax) lineMax = v; } } - for (const line of referenceLines ?? []) { - if (Number.isFinite(line.y) && line.y > dataMax) dataMax = line.y; + if (!Number.isFinite(lineMin)) { + lineMin = 0; + lineMax = thresholdStroke.value; } - const domainMax = dataMax > 0 ? dataMax * 1.1 : 1; - thresholdDomain = [0, domainMax]; - // Gradient runs top (offset 0 = domainMax) to bottom (offset 1 = 0); the split sits where - // the threshold value falls within the domain. - thresholdOffset = Math.min(1, Math.max(0, (domainMax - thresholdStroke.value) / domainMax)); + const range = lineMax - lineMin; + thresholdOffset = + range > 0 + ? Math.min(1, Math.max(0, (lineMax - thresholdStroke.value) / range)) + : lineMax >= thresholdStroke.value + ? 0 + : 1; } // Per-bucket warning overlay: single-series line charts only. Retrace the primary series in the @@ -344,7 +352,6 @@ export function ChartLineRenderer({ style: { fontVariantNumeric: "tabular-nums" }, }, tickFormatter: yAxisTickFormatter, - ...(thresholdDomain ? { domain: thresholdDomain } : {}), ...yAxisPropsProp, }; @@ -493,7 +500,7 @@ export function ChartLineRenderer({ {/* When legend is shown below, render tooltip with cursor only (no content popup) */} - {thresholdStroke && thresholdDomain ? ( + {thresholdActive ? ( - - + + ) : null} @@ -541,7 +551,7 @@ export function ChartLineRenderer({ {/* When legend is shown below, render tooltip with cursor only (no content popup) */} {/* Note: Legend is now rendered by ChartRoot outside the chart container */} {referenceOverlays} - {visibleSeries.map((key) => ( - ( - - ) - : { r: 4, fill: config[key]?.color, strokeWidth: 0 } - } - isAnimationActive={false} - /> - ))} + {visibleSeries.map((key) => { + // The gradient stroke only applies to the threshold's target series (default: the first); + // other series (e.g. the grey limit line) keep their own colour. + const gradientLine = + thresholdActive && (thresholdStroke!.series == null || thresholdStroke!.series === key); + return ( + ( + + ) + : { r: 4, fill: config[key]?.color, strokeWidth: 0 } + } + isAnimationActive={false} + /> + ); + })} {overlayActive && ( // Drawn after the base line so the warning colour sits on top. connectNulls={false} keeps // the mask to over-threshold stretches; excluded from the legend and (above) the tooltip. // Its active dot inherits the warning colour so the hover dot is yellow over yellow. - // Slightly wider than the base line so it fully covers it — otherwise the base colour peeks - // out along the edges and the warning stretch reads as an outlined line. + // Same 1px width as the base line so the warning stretch matches the other lines — it traces + // the same points over its over-threshold buckets, so it covers the base exactly. @@ -63,8 +66,23 @@ export function QueuePauseResumeButton({ fullWidth={fullWidth} textAlignLeft={fullWidth} aria-label={label} + className={ + withQueueName + ? queue.paused + ? "border-success/60 text-success [&_span]:text-success hover:border-success" + : "border-warning/60 text-warning [&_span]:text-warning hover:border-warning" + : undefined + } > - {iconOnly ? undefined : queue.paused ? "Resume" : "Pause"} + {iconOnly + ? undefined + : withQueueName + ? queue.paused + ? "Resume this queue…" + : "Pause this queue…" + : queue.paused + ? "Resume" + : "Pause"}
@@ -116,7 +134,7 @@ export function QueuePauseResumeButton({ } cancelButton={ - @@ -209,24 +227,38 @@ export function QueueOverrideConcurrencyButton({ + ) : trigger === "button" ? ( + + + +
+ + + +
+ + + Give this queue its own concurrency limit instead of the environment default. Set it + as a number or a percentage of the environment limit. + + + ) : ( - {trigger === "button" ? ( - - ) : ( - - )} + )} @@ -235,7 +267,7 @@ export function QueueOverrideConcurrencyButton({
{isOverridden ? ( - + This queue's concurrency limit is currently overridden to {currentLimit}. {typeof queue.concurrency?.base === "number" && ` The original limit set in code was ${queue.concurrency.base}.`}{" "} @@ -246,7 +278,7 @@ export function QueueOverrideConcurrencyButton({ . ) : ( - + Override this queue's concurrency limit. The current limit is {currentLimit}, which is set {queue.concurrencyLimit !== null ? "in code" : "by the environment"}. @@ -255,13 +287,43 @@ export function QueueOverrideConcurrencyButton({ -
- + +
+
+ {mode === "percent" ? ( + setPercent(e.target.value)} + placeholder="100" + autoFocus + accessory={%} + /> + ) : ( + setConcurrencyLimit(e.target.value)} + placeholder={currentLimit.toString()} + autoFocus + /> + )} +
{mode === "percent" ? ( - <> - setPercent(e.target.value)} - placeholder="100" - autoFocus - /> - - {materializedFromPercent !== null - ? `${percentNumber}% = ${materializedFromPercent} concurrent ${ - materializedFromPercent === 1 ? "run" : "runs" - } of the environment's ${environmentConcurrencyLimit}. Recalculates automatically when the environment limit changes.` - : "Enter a percentage between 1 and 100."} - - + + {materializedFromPercent !== null + ? `${percentNumber}% = ${materializedFromPercent} concurrent ${ + materializedFromPercent === 1 ? "run" : "runs" + } of the environment's ${environmentConcurrencyLimit}. Recalculates automatically when the environment limit changes.` + : "Enter a percentage between 1 and 100."} + ) : ( - <> - setConcurrencyLimit(e.target.value)} - placeholder={currentLimit.toString()} - autoFocus - /> - - {limitOverCap - ? `Can't exceed the environment limit of ${environmentConcurrencyLimit}.` - : `Up to the environment limit of ${environmentConcurrencyLimit}.`} - - + + {limitOverCap + ? `Can't exceed the environment limit of ${environmentConcurrencyLimit}.` + : `The most concurrent runs this queue can use at once. It can't exceed the environment limit of ${environmentConcurrencyLimit}.`} + )} @@ -346,7 +381,7 @@ export function QueueOverrideConcurrencyButton({ )} - diff --git a/apps/webapp/app/components/queues/QueueMetricCards.tsx b/apps/webapp/app/components/queues/QueueMetricCards.tsx index b7694eade4..4349ea80c3 100644 --- a/apps/webapp/app/components/queues/QueueMetricCards.tsx +++ b/apps/webapp/app/components/queues/QueueMetricCards.tsx @@ -1,4 +1,4 @@ -import { useMemo, type ReactNode } from "react"; +import { useEffect, useMemo, useState, type ReactNode } from "react"; import { buildActivityTimeAxis } from "~/components/primitives/charts/activityTimeAxis"; import { Chart, @@ -104,6 +104,23 @@ type QueueMetricChartProps = { * are config values that existed all along, so carry the first value backward instead. */ carryBackfill?: string[]; + /** Show the series legend below the chart (use for multi-series charts). */ + showLegend?: boolean; + /** + * Recolour a series' stroke above a threshold with a gradient split (colour only above the + * line). `value` sets a constant threshold; `valueFromSeries` reads a (roughly constant) + * threshold off another series — e.g. the concurrency limit. `series` targets which line is + * recoloured; the others keep their own colour. + */ + thresholdStroke?: { + aboveColor: string; + series?: string; + value?: number; + valueFromSeries?: string; + }; + /** Reports whether the chart has data to plot (false once it settles on the "no activity" state), + * so a wrapping card can hide the legend to match. */ + onHasDataChange?: (hasData: boolean) => void; }; // Bare chart (no card chrome) so it can live inside a shared card, e.g. a tabbed panel. @@ -118,6 +135,8 @@ export function QueueMetricChart({ defaultPeriod, warningOverlay, carryBackfill, + thresholdStroke, + onHasDataChange, }: QueueMetricChartProps) { const { rows, showLoading, failed } = useQueueMetric(query, { ids, @@ -163,8 +182,34 @@ export function QueueMetricChart({ [data] ); + // Resolve the threshold value: a constant, or the max of another series (e.g. the limit line, + // which is effectively constant). A gradient split then colours the target series only above it. + // `valueFromSeries` targets integer-count series (concurrency limit), so split half a unit below + // the limit — that way the line renders warning *at or above* the limit (saturated), matching + // "turns yellow at the limit", rather than only when it strictly exceeds it. + const resolvedThresholdStroke = useMemo(() => { + if (!thresholdStroke) return undefined; + let value = thresholdStroke.value; + if (value == null && thresholdStroke.valueFromSeries) { + let max = -Infinity; + for (const p of data) { + const v = Number(p[thresholdStroke.valueFromSeries]); + if (Number.isFinite(v) && v > max) max = v; + } + value = max > 0 ? max - 0.5 : undefined; + } + if (value == null || !Number.isFinite(value)) return undefined; + return { value, aboveColor: thresholdStroke.aboveColor, series: thresholdStroke.series }; + }, [thresholdStroke, data]); + const state: ChartState = showLoading ? "loading" : failed ? "invalid" : undefined; + // Report data presence so a wrapping card can hide its legend when the chart settles on the + // "no activity" state. Only report once loaded, so the legend stays put while loading. + useEffect(() => { + if (!showLoading) onHasDataChange?.(!failed && data.length > 0); + }, [showLoading, failed, data.length, onHasDataChange]); + return ( ); @@ -191,6 +237,7 @@ export function QueueMetricChartCard({ info, titleAccessory, className, + extraLegend, ...chart }: QueueMetricChartProps & { title: string; @@ -198,23 +245,58 @@ export function QueueMetricChartCard({ /** Extra content rendered after the info icon inside the title row (e.g. a live readout). */ titleAccessory?: ReactNode; className?: string; + /** Extra legend entries appended after the series — e.g. a warning state that isn't its own + * series (the orange "over threshold" colour). */ + extraLegend?: Array<{ color: string; label: string }>; }) { + // Hide the legend once the chart settles on the "no activity" state (reported by the chart). + const [hasData, setHasData] = useState(true); return (
+ + {title} - {info ? : null} + {info ? ( + + ) : null} {titleAccessory} - ) : ( - title - ) + {/* Inline legend below the title (swatch + label per series), matching the list-page + charts — instead of the Chart.Root legend with per-series totals. */} + {chart.showLegend && + hasData && + (chart.series.length > 0 || (extraLegend?.length ?? 0) > 0) ? ( + + {chart.series.map((s) => ( + + + {s.label} + + ))} + {extraLegend?.map((e) => ( + + + {e.label} + + ))} + + ) : null} + } > - +
); @@ -322,7 +404,7 @@ export function QueueSparklineStat({ return (
-
+
{title} {info || (data.length > 0 && peak > 0) ? ( } - contentClassName="max-w-xs" + contentClassName="max-w-[230px]" + disableHoverableContent /> ) : null}
diff --git a/apps/webapp/app/hooks/useMetricResourceQuery.ts b/apps/webapp/app/hooks/useMetricResourceQuery.ts index 2a2ee413af..53b7c03af8 100644 --- a/apps/webapp/app/hooks/useMetricResourceQuery.ts +++ b/apps/webapp/app/hooks/useMetricResourceQuery.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useRef, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useInterval } from "./useInterval"; export type MetricResourceRow = Record; @@ -24,16 +24,32 @@ export type MetricResourceQueryOptions = { refreshIntervalMs?: number; }; +// Module-level cache of the last successful rows per query signature. Lets a remounted chart +// (switching tabs, or navigating back to the queues list) paint its previous data immediately +// instead of flashing a loading skeleton every time, while it revalidates in the background. +// Bounded so it can't grow without limit over a long session. +const responseCache = new Map(); +const RESPONSE_CACHE_MAX = 200; + +function cacheSet(key: string, rows: MetricResourceRow[]) { + // Re-insert so the key becomes the most-recently-used (Map preserves insertion order). + responseCache.delete(key); + responseCache.set(key, rows); + if (responseCache.size > RESPONSE_CACHE_MAX) { + const oldest = responseCache.keys().next().value; + if (oldest !== undefined) responseCache.delete(oldest); + } +} + /** * Client-fetch a TRQL query from the metric resource route (like the dashboard * widgets): own loading state, interval + on-focus refresh, abort on change/unmount. + * + * Successful results are cached per query signature, so a chart that remounts (tab switch, or + * back-navigation to the queues list) shows its last data immediately and revalidates in the + * background rather than flashing a loading skeleton. */ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryOptions) { - const [rows, setRows] = useState(null); - const [isLoading, setIsLoading] = useState(true); - const [failed, setFailed] = useState(false); - const abortRef = useRef(null); - const { organizationId, projectId, @@ -44,11 +60,37 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO } = opts; const { period, from, to } = opts.timeRange; const queuesKey = opts.queues && opts.queues.length > 0 ? opts.queues.join(",") : undefined; + const resolvedPeriod = period ?? (from || to ? null : defaultPeriod); + const cacheKey = useMemo( + () => + [ + organizationId, + projectId, + environmentId, + resolvedPeriod ?? "", + from ?? "", + to ?? "", + fillGaps ? 1 : 0, + queuesKey ?? "", + query, + ].join("|"), + [organizationId, projectId, environmentId, resolvedPeriod, from, to, fillGaps, queuesKey, query] + ); + + const [rows, setRows] = useState( + () => responseCache.get(cacheKey) ?? null + ); + const [isLoading, setIsLoading] = useState(true); + const [failed, setFailed] = useState(false); + const abortRef = useRef(null); const load = useCallback(() => { abortRef.current?.abort(); const controller = new AbortController(); abortRef.current = controller; + // Paint cached rows immediately (no skeleton) on remount / key change while we revalidate. + const cached = responseCache.get(cacheKey); + if (cached) setRows(cached); setIsLoading(true); fetch("/resources/metric", { method: "POST", @@ -56,7 +98,7 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO body: JSON.stringify({ query, scope: "environment", - period: period ?? (from || to ? null : defaultPeriod), + period: resolvedPeriod, from, to, fillGaps: !!fillGaps, @@ -71,6 +113,7 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO .then((data) => { if (controller.signal.aborted) return; if (data.success) { + cacheSet(cacheKey, data.data.rows); setRows(data.data.rows); setFailed(false); } else { @@ -86,11 +129,11 @@ export function useMetricResourceQuery(query: string, opts: MetricResourceQueryO } }); }, [ + cacheKey, query, - period, + resolvedPeriod, from, to, - defaultPeriod, fillGaps, organizationId, projectId, 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 1a9ef75c27..45c395b54f 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 @@ -10,7 +10,8 @@ import { DialogClose } from "@radix-ui/react-dialog"; import { Form, useNavigation, type MetaFunction } from "@remix-run/react"; import { type ActionFunctionArgs, type LoaderFunctionArgs } from "@remix-run/server-runtime"; import type { RuntimeEnvironmentType } from "@trigger.dev/database"; -import { useEffect, useMemo, useState } from "react"; +import { useEffect, useMemo, useState, type ReactNode } from "react"; +import { QueuesIcon } from "~/assets/icons/QueuesIcon"; import { typedjson, useTypedLoaderData } from "remix-typedjson"; import { z } from "zod"; import { ConcurrencyIcon } from "~/assets/icons/ConcurrencyIcon"; @@ -42,7 +43,6 @@ import { TableHeaderCell, TableRow, } from "~/components/primitives/Table"; -import { useTableSort, type SortColumn } from "~/components/primitives/useTableSort"; import { InfoIconTooltip, SimpleTooltip, @@ -121,6 +121,10 @@ const QUEUE_METRICS_DEFAULT_PERIOD = "1d"; const QUEUE_LIVE_BLOCKS_PERIOD = "15m"; const QUEUE_LIVE_BLOCKS_QUERY = "SELECT timeBucket() AS t, max(max_env_queued) AS env_queued, max(max_env_running) AS env_running FROM env_metrics GROUP BY t ORDER BY t"; +// Trust the ClickHouse gauge only while its newest bucket is this recent; otherwise fall back to +// the loader's Redis-exact live values (matches LIVE_GAUGE_FRESH_MS on the queue detail page / run +// inspector). +const LIVE_GAUGE_FRESH_MS = 90_000; export const meta: MetaFunction = () => { return [ @@ -408,11 +412,21 @@ function QueuesWithMetricsView() { }); const lastLiveBlockRow = liveBlockRows.length > 0 ? liveBlockRows[liveBlockRows.length - 1] : null; - const envQueuedLive = lastLiveBlockRow - ? tileNumber(lastLiveBlockRow.env_queued) + // Only trust the gauge while its newest bucket is fresh. A row painted from the hook's cache on + // client-side nav-back (responseCache), or a quiet env whose latest bucket is minutes old, must + // not override the loader's Redis-exact live values with a stale count. + const lastLiveBucketMs = lastLiveBlockRow ? tileTimeToMs(lastLiveBlockRow.t) : NaN; + const freshLiveBlockRow = + lastLiveBlockRow && + Number.isFinite(lastLiveBucketMs) && + Date.now() - lastLiveBucketMs < LIVE_GAUGE_FRESH_MS + ? lastLiveBlockRow + : null; + const envQueuedLive = freshLiveBlockRow + ? tileNumber(freshLiveBlockRow.env_queued) : environment.queued; - const envRunningLive = lastLiveBlockRow - ? tileNumber(lastLiveBlockRow.env_running) + const envRunningLive = freshLiveBlockRow + ? tileNumber(freshLiveBlockRow.env_running) : environment.running; // Allocation summary tiles. The presenter computes the env-wide allocated total (sum of @@ -433,43 +447,6 @@ function QueuesWithMetricsView() { // Client-side, header-click sorting over the current page's rows. Server pagination and the // default busiest order are unchanged; clearing a sort returns to that server order. const queueRows = queues ?? []; - type QueueRow = (typeof queueRows)[number]; - const sortColumns = useMemo[]>( - () => [ - { key: "name", type: "alpha", value: (q) => q.name }, - { key: "queued", type: "number", value: (q) => q.queued }, - { key: "running", type: "number", value: (q) => q.running }, - { - key: "limit", - type: "number", - value: (q) => q.concurrencyLimit ?? environment.concurrencyLimit, - }, - { key: "limitedBy", type: "alpha", value: (q) => queueLimitedByLabel(q) }, - { - key: "health", - type: "alpha", - value: (q) => - queueHealthLabel({ - paused: q.paused, - running: q.running, - queued: q.queued, - limit: q.concurrencyLimit ?? environment.concurrencyLimit, - }), - }, - { - key: "delayP95", - type: "number", - value: (q) => metricsByQueue[queueMetricsKey(q)]?.p95WaitMs ?? null, - }, - { - key: "backlog", - type: "number", - value: (q) => metricsByQueue[queueMetricsKey(q)]?.peakQueued ?? null, - }, - ], - [environment.concurrencyLimit, metricsByQueue] - ); - const { sortedRows: sortedQueues, getSortProps } = useTableSort(queueRows, sortColumns); return ( @@ -490,24 +467,28 @@ function QueuesWithMetricsView() { {/* Filters — pinned bar directly under the NavBar. Left cluster = search + period; right cluster = pagination. */} {success ? ( - -
+ +
+
+
+ {environment.runsEnabled && + env.pauseSource !== ENVIRONMENT_PAUSE_SOURCE_BILLING_LIMIT ? ( + + ) : null} +
- ) : null} @@ -524,15 +505,12 @@ function QueuesWithMetricsView() { suffix={env.paused ? paused : undefined} animate accessory={ -
- {environment.runsEnabled && - env.pauseSource !== ENVIRONMENT_PAUSE_SOURCE_BILLING_LIMIT ? ( - - ) : null} + -
+ } valueClassName={env.paused ? "text-warning tabular-nums" : "tabular-nums"} compactThreshold={1000000} @@ -561,23 +539,26 @@ function QueuesWithMetricsView() { ) : undefined } accessory={ - + + + } compactThreshold={1000000} /> + Allocated {allocation && overAllocated ? ( + > + Increase limit + ) : ( + > + Increase limit + ) ) : undefined } @@ -660,23 +643,23 @@ function QueuesWithMetricsView() { ] : undefined } - // Saturation and p95 "step over the line": a per-bucket overlay retraces only - // the over-threshold stretches in warning colour, so under-threshold values stay - // blue. (A gradient split can't do this reliably — an SVG objectBoundingBox - // gradient tracks the line's own bbox, not the y-axis, so a low/flat line reads - // as entirely warning-coloured.) - // All thresholded lines colour warning where they step over the threshold: the - // per-bucket overlay retraces only the over-threshold stretches, so the colour - // change tracks the axis crossing. - warningOverlay={ + // Saturation recolours the line above its 100% limit with a gradient split, so + // only the portion over the line is orange (the offset is derived from the line's + // own value range, so the split lands exactly at 100% regardless of domain + // padding). p95 and throttled use a per-bucket overlay: it retraces only the + // over-threshold stretches, so under-threshold buckets stay blue. + thresholdStroke={ tile.id === "saturation" - ? { threshold: 100 } - : tile.id === "p95" - ? { threshold: 60_000 } - : tile.id === "throttled" - ? // Integer counts: threshold 0 warns once a bucket has ≥1 throttle. - { threshold: 0 } - : undefined + ? { value: 100, aboveColor: "var(--color-warning)" } + : undefined + } + warningOverlay={ + tile.id === "p95" + ? { threshold: 60_000 } + : tile.id === "throttled" + ? // Integer counts: threshold 0 warns once a bucket has ≥1 throttle. + { threshold: 0 } + : undefined } /> ))} @@ -691,51 +674,50 @@ function QueuesWithMetricsView() { - Name - - Queued - - - Running - - - Limit - + Name + Queued + Running + Limit +

- Environment — the environment's - limit of {environment.concurrencyLimit}. + Environment: uses the + environment limit of {environment.concurrencyLimit}.

- User — a limit set in your code. + User: a limit you set in your + code.

- Override — set manually from the - dashboard or API. + Override: a limit you set here + or via the API.

} > Limited by - - Health - + Health Delay p95 + How many runs were waiting, over the selected time. marks + where the queue was throttled. + + } > Backlog @@ -745,8 +727,8 @@ function QueuesWithMetricsView() { - {sortedQueues.length > 0 ? ( - sortedQueues.map((queue) => { + {queueRows.length > 0 ? ( + queueRows.map((queue) => { const limit = queue.concurrencyLimit ?? environment.concurrencyLimit; const isAtConcurrencyLimit = queue.running >= limit; const isAtQueueLimit = @@ -777,7 +759,7 @@ function QueuesWithMetricsView() { )} /> ) : ( - } content="At concurrency limit: this queue is running as many runs as its limit allows; new runs wait in the backlog." - className="max-w-xs" + className="max-w-[230px]" disableHoverableContent /> ) : null @@ -855,8 +837,8 @@ function QueuesWithMetricsView() { > {queue.concurrencyLimitOverridePercent !== null ? ( <> - {limit}{" "} - + {limit} + ({formatOverridePercent(queue.concurrencyLimitOverridePercent)}%) @@ -865,31 +847,37 @@ function QueuesWithMetricsView() { )} , and the label itself stays the link. + trailingContent={ + queue.concurrency?.overriddenAt ? ( + + ) : undefined + } > - {queue.concurrency?.overriddenAt ? ( - Override} - content={ - queue.concurrencyLimitOverridePercent !== null - ? `Overridden at ${formatOverridePercent( - queue.concurrencyLimitOverridePercent - )}% of the environment limit.` - : `This queue's concurrency limit has been manually overridden to ${limit}.` - } - className="max-w-xs" - disableHoverableContent - /> - ) : queue.concurrencyLimit ? ( - "User" - ) : ( - "Environment" - )} + {queue.concurrency?.overriddenAt + ? "Override" + : queue.concurrencyLimit + ? "User" + : "Environment"} {env.paused - ? `Resume processing runs in ${environmentFullTitle(env)}` - : `Pause processing runs in ${environmentFullTitle(env)}`} + ? `Resumes ${environmentFullTitle(env)} so its runs can be dequeued again.` + : `Pauses all runs from being dequeued in ${environmentFullTitle(env)}. Any executing runs will continue to run.`} @@ -1114,7 +1111,7 @@ function EnvironmentPauseResumeButton({ } cancelButton={ - @@ -1179,20 +1176,35 @@ type MetricTileRow = Record; /** One charted point per time bucket, already aggregated across the visible queue set. */ type TilePoint = { bucket: number; value: number }; +// Inline colour swatch matching the chart's warning ("yellow") line — used in tooltip copy that +// refers to that colour instead of naming it, so the swatch always matches the chart. +function WarningSwatch() { + return ( + + ); +} + type QueueHeaderTile = { id: string; label: string; /** Info-icon copy explaining what the chart shows, rendered next to the card title. */ - description: string; + description: ReactNode; color: string; + /** Optional inline legend rendered below the card title: a fixed set of {colored square, label} + * entries, for charts where a colour (e.g. the orange warning line) needs explaining. */ + legend?: Array<{ color: string; label: string }>; query: string; /** Formats a single bucket's value in the chart tooltip. */ formatValue?: (value: number) => string; /** Formats the y-axis tick labels. Without it the axis shows raw numbers (bad for durations * in ms or percent scales). Passed through to Chart.Line's yAxisProps.tickFormatter. */ formatAxis?: (value: number) => string; - /** Hover tooltip explaining the headline readout next to the title (e.g. what "9% of window" - * means). Without it the readout has no tooltip. */ + /** Hover tooltip explaining the headline readout next to the title (e.g. what "9% of current + * period" means). Without it the readout has no tooltip. */ totalTooltip?: string; // Rows can be one-per-bucket (p95, throttled: aggregated across the set in ClickHouse) or // one-per-(bucket, queue) (saturation, backlog: summed across the set here, since summing a @@ -1245,8 +1257,17 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ { id: "saturation", label: "Env saturation", - description: "Running as a share of the environment limit. Yellow over 100% (burst headroom).", + description: ( + <> + How much of the environment's concurrency these queues are using. Turns {" "} + above 100%, when they're into burst capacity. + + ), color: "var(--color-queues)", + legend: [ + { color: "var(--color-queues)", label: "Saturation" }, + { color: "var(--color-warning)", label: "Over limit" }, + ], // Numerator: running summed across the visible set. Denominator: the env-wide limit (same for // every queue in a bucket), so the line reads as the set's share of the environment capacity. query: `SELECT timeBucket() AS t,\n queue,\n max(max_running) AS running,\n max(max_env_limit) AS env_limit\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`, @@ -1266,7 +1287,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ { id: "backlog", label: "Backlog", - description: "Runs waiting across these queues over time.", + description: "How many runs are waiting across these queues, over time.", color: "var(--color-queues)", query: `SELECT timeBucket() AS t,\n queue,\n max(max_queued) AS queued\nFROM queue_metrics\nGROUP BY t, queue\nORDER BY t`, derive: (rows) => { @@ -1281,9 +1302,18 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ { id: "p95", label: "Scheduling delay p95", - description: "p95 wait from eligible to dequeued. Yellow over 1 min.", + description: ( + <> + How long runs wait before they start (95% start faster than this). Turns {" "} + above 1 minute. + + ), totalTooltip: "The worst p95 in the selected window.", color: "var(--color-queues)", + legend: [ + { color: "var(--color-queues)", label: "p95" }, + { color: "var(--color-warning)", label: "Over 1 min" }, + ], // quantilesMerge over the set's rows in a bucket is the true p95 across the union of samples // (merging quantile states is valid; averaging per-queue percentiles would not be). query: `SELECT timeBucket() AS t,\n round(quantilesMerge(0.5, 0.9, 0.95, 0.99)(wait_quantiles)[3]) AS p95\nFROM queue_metrics\nGROUP BY t\nORDER BY t`, @@ -1303,9 +1333,10 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ { id: "throttled", label: "Throttled", - description: "Times dequeuing was blocked by a limit.", + description: "How often runs were held back by a limit.", totalTooltip: "The share of the selected window with at least one blocked dequeue.", color: "var(--color-queues)", + legend: [{ color: "var(--color-warning)", label: "Throttled" }], query: `SELECT timeBucket() AS t,\n sum(throttled_count) AS throttled\nFROM queue_metrics\nGROUP BY t\nORDER BY t`, derive: (rows) => { const points = rows.map((r) => ({ @@ -1321,7 +1352,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [ return { points, total: pct, - formatTotal: (v) => `${v}% of window`, + formatTotal: (v) => `${v}% of current period`, totalClassName: pct > 0 ? "text-warning" : undefined, }; }, @@ -1417,35 +1448,60 @@ function QueueEnvMetricChart({ return ( - - {tile.label} - - - {peak != null ? ( - tile.totalTooltip && !showLoading ? ( - - {peak} - - } - content={tile.totalTooltip} - className="max-w-xs" + + + + {tile.label} + - ) : ( - - {peak} - - ) + + {peak != null ? ( + tile.totalTooltip && !showLoading ? ( + + {peak} + + } + content={tile.totalTooltip} + className="max-w-[230px]" + disableHoverableContent + /> + ) : ( + + {peak} + + ) + ) : null} + + {tile.legend && (showLoading || hasData) ? ( + + {tile.legend.map((item) => ( + + + {item.label} + + ))} + ) : null} } @@ -1531,16 +1587,6 @@ function QueueHealthBadge(health: QueueHealth) { ); } -// The label rendered in the "Limited by" cell, also used to sort that column. -function queueLimitedByLabel(queue: { - concurrency?: { overriddenAt?: Date | string | null } | null; - concurrencyLimit?: number | null; -}): "Override" | "User" | "Environment" { - if (queue.concurrency?.overriddenAt) return "Override"; - if (queue.concurrencyLimit) return "User"; - return "Environment"; -} - // The `queue_metrics`-prefixed key a queue is stored under (task queues are prefixed `task/`). function queueMetricsKey(queue: { type: string; name: string }): string { return `${queue.type === "task" ? "task/" : ""}${queue.name}`; @@ -1785,7 +1831,7 @@ function ClassicQueuesView() { } content="This queue's concurrency limit has been manually overridden from the dashboard or API." - className="max-w-xs" + className="max-w-[230px]" disableHoverableContent /> ) : null} @@ -1969,7 +2015,7 @@ function BurstFactorTooltip({ }, but you can burst up to ${ environment.burstFactor * environment.concurrencyLimit } when across multiple queues/tasks.`} - contentClassName="max-w-xs" + contentClassName="max-w-[230px]" /> ); } 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 a2792b4f4d..8dc0c2c24f 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 @@ -1,11 +1,12 @@ import { type MetaFunction } from "@remix-run/react"; import { type ActionFunctionArgs, type LoaderFunctionArgs } from "@remix-run/server-runtime"; -import { useMemo, type ReactNode } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react"; import type { QueueItem } from "@trigger.dev/core/v3/schemas"; import { typedjson, useTypedLoaderData } from "remix-typedjson"; import { z } from "zod"; import { MainCenteredContainer, PageContainer } from "~/components/layout/AppLayout"; import { MetricsLayout } from "~/components/layout/MetricsLayout"; +import { AnimatedOrgBannerBar } from "~/components/billing/AnimatedOrgBannerBar"; import { BigNumber } from "~/components/metrics/BigNumber"; import { Header3 } from "~/components/primitives/Headers"; import { NavBar, PageTitle } from "~/components/primitives/PageHeader"; @@ -48,7 +49,12 @@ import { SearchInput } from "~/components/primitives/SearchInput"; import { engine } from "~/v3/runEngine.server"; import { TimeFilter } from "~/components/runs/v3/SharedFilters"; import { useSearchParams } from "~/hooks/useSearchParam"; -import { useTableSort, type SortColumn } from "~/components/primitives/useTableSort"; +import { useInterval } from "~/hooks/useInterval"; +import { PaginationControls } from "~/components/primitives/Pagination"; +import type { + ConcurrencyKeyRow, + ConcurrencyKeysResponse, +} from "~/routes/resources.queues.concurrency-keys"; import { useCurrentPlan } from "../_app.orgs.$organizationSlug/route"; import { canAccessQueueMetricsUi } from "~/v3/canAccessQueueMetricsUi.server"; import { requireUserId } from "~/services/session.server"; @@ -240,23 +246,58 @@ export default function Page() { + {/* Paused-queue banner — mirrors the environment-paused banner (OrgBanner) at the top of + the page when this individual queue is paused. */} + + {`"${queue.name}" queue paused. No new runs will be dequeued and executed.`} + {/* Filters — search (concurrency keys) + time filter in one left cluster, above everything, like the Queues list. The time filter scopes the tab charts; search filters the keys table. The bar is pinned by the layout while the page scrolls. */} - -
+ +
+ + replace({ view: undefined, key: undefined })} + > + Overview + + replace({ view: "keys" })} + > + Concurrency keys + + +
+
{view === "keys" && hasKeys ? ( - + ) : null} + +
@@ -275,23 +316,6 @@ export default function Page() { {/* Tabs + charts share the padded (inset) column. Both tabs always render; the keys tab shows an empty state when the queue has no concurrency keys. */} - - replace({ view: undefined, key: undefined })} - > - Overview - - replace({ view: "keys" })} - > - Concurrency keys - - - {view === "keys" ? ( hasKeys ? ( - + {selectedKey ? ( @@ -340,6 +358,17 @@ export default function Page() { ); } +// Inline colour swatch for tooltip copy — matches the chart legend swatch (rounded-[2px]) and is +// nudged up 1px so it sits on the text baseline. +function ColorSwatch({ color }: { color: string }) { + return ( + + ); +} + function OverviewCharts({ ids, timeRange, @@ -355,7 +384,14 @@ function OverviewCharts({
+ How many runs are executing at once () versus + the queue's limit ( + ). Turns color when it reaches the limit. + + } + showLegend className="aspect-[2/1]" query={`SELECT timeBucket() AS t, max(max_running) AS running, max(max_limit) AS limit\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} fillGaps @@ -367,15 +403,21 @@ function OverviewCharts({ { key: "limit", label: "Limit", color: COLORS.limit }, { key: "running", label: "Running", color: COLORS.running }, ]} - // Recolour Running warning where it reaches the limit (saturated), matching the tooltip. - warningOverlay={{ series: "running", atOrAbove: "limit" }} + // Recolour Running above the limit line with a gradient split, so it's orange only where + // it's actually over the limit — not on the way up. The threshold reads off the (roughly + // constant) limit series. + thresholdStroke={{ + series: "running", + valueFromSeries: "limit", + aboveColor: "var(--color-warning)", + }} // The limit is a config value emitted only while the queue is active; back-fill its // leading zeros so the reference line doesn't start with a false 0→limit step. carryBackfill={["limit"]} /> + Runs arriving ( Enqueued) versus starting ( + Started). Turns{" "} + color when Started falls behind. + + } + showLegend + extraLegend={[{ color: "var(--color-warning)", label: "Falling behind" }]} className="aspect-[2/1]" query={`SELECT timeBucket() AS t,\n deltaSumTimestampMerge(enqueue_delta) AS enqueued,\n deltaSumTimestampMerge(started_delta) AS started\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} fillGaps @@ -403,7 +453,8 @@ function OverviewCharts({ /> + How often runs were held back by a limit ({" "} + color). + + } + className="aspect-[2/1] sm:col-span-2 sm:aspect-[4/1]" query={`SELECT timeBucket() AS t, sum(throttled_count) AS throttled\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} fillGaps ids={ids} @@ -592,7 +648,7 @@ function chartTitleWithInfo(title: string, info: string) { return ( {title} - + ); } @@ -704,143 +760,184 @@ function GroupedKeySeries({ ); } -type KeyRangeStats = { started: number; peakBacklog: number; meanWaitMs: number }; +// One page of the paginated per-key table. The ClickHouse tier is the authority (ranked by peak +// backlog over the window, with the total on every row so page + count are a single scan); each +// page's keys are enriched with live "now" counts from Redis server-side. Fetched per page rather +// than capped at 50, so high-cardinality queues (tens of thousands of keys) page through instead +// of silently truncating. See resources.queues.concurrency-keys. +function useConcurrencyKeys(opts: { + ids: Ids; + timeRange: TimeRangeParams; + queueName: string; + search: string; + page: number; +}) { + const { ids, timeRange, queueName, search, page } = opts; + const [data, setData] = useState(null); + const [isLoading, setIsLoading] = useState(true); + const abortRef = useRef(null); + + const body = useMemo( + () => + JSON.stringify({ + organizationId: ids.organizationId, + projectId: ids.projectId, + environmentId: ids.environmentId, + queueName, + period: timeRange.period, + from: timeRange.from, + to: timeRange.to, + search, + page, + }), + [ + ids.organizationId, + ids.projectId, + ids.environmentId, + queueName, + timeRange.period, + timeRange.from, + timeRange.to, + search, + page, + ] + ); -// Live breakdown (queued/running now, oldest wait) merged with per-key range stats from -// the history tier; keys with history but no live backlog still appear. Clicking a key -// pins the drill-down charts via the `key` search param. + const load = useCallback(() => { + abortRef.current?.abort(); + const controller = new AbortController(); + abortRef.current = controller; + setIsLoading(true); + fetch("/resources/queues/concurrency-keys", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body, + signal: controller.signal, + }) + .then((res) => res.json() as Promise) + .then((res) => { + if (controller.signal.aborted) return; + setData(res); + setIsLoading(false); + }) + .catch((error) => { + if (error instanceof DOMException && error.name === "AbortError") return; + if (!controller.signal.aborted) { + setData({ success: false, error: error?.message ?? "Network error" }); + setIsLoading(false); + } + }); + }, [body]); + + useEffect(() => { + load(); + return () => abortRef.current?.abort(); + }, [load]); + + // Keep the live "now" counts fresh without a manual reload. + useInterval({ + interval: 30_000, + onLoad: false, + onFocus: true, + pauseWhenHidden: true, + callback: load, + }); + + return { data, isLoading }; +} + +// Paginated per-key table: which keys hold the backlog / do the work. Clicking a key pins the +// drill-down charts via the `key` search param. function KeyStatsTable({ - breakdown, - loadedAt, ids, timeRange, queueName, }: { - breakdown: CkBreakdown; - loadedAt: number; ids: Ids; timeRange: TimeRangeParams; queueName: string; }) { const { value, replace, del } = useSearchParams(); const selectedKey = value("key"); - - const { rows, showLoading } = useQueueMetric( - `SELECT concurrency_key,\n deltaSumTimestampMerge(started_delta) AS started,\n max(max_queued) AS peak_backlog,\n if(sum(wait_ms_count) > 0, round(sum(wait_ms_sum) / sum(wait_ms_count)), 0) AS mean_wait\nFROM queue_metrics_by_key\nGROUP BY concurrency_key\nORDER BY peak_backlog DESC\nLIMIT 50`, - { ids, timeRange, queueName } - ); - - const merged = useMemo(() => { - const range = new Map(); - for (const r of rows) { - range.set(String(r.concurrency_key), { - started: toNumber(r.started), - peakBacklog: toNumber(r.peak_backlog), - meanWaitMs: toNumber(r.mean_wait), - }); + const search = value("query")?.trim() ?? ""; + const page = Math.max(1, Number(value("page")) || 1); + + const { data, isLoading } = useConcurrencyKeys({ ids, timeRange, queueName, search, page }); + + const rows: ConcurrencyKeyRow[] = data?.success ? data.rows : []; + const total = data?.success ? data.total : 0; + const perPage = data?.success ? data.perPage : 25; + const totalPages = Math.max(1, Math.ceil(total / perPage)); + // Only show a skeleton before the first response; keep prior rows visible while revalidating. + const showLoading = isLoading && !data; + + // Recover from a page past the end: narrowing the time range (or a stale bookmarked URL) can + // leave `page` beyond the result set, which comes back empty with the pagination control — shown + // only when there's >1 page — hidden, stranding the reader. Once a response settles empty on a + // page > 1, snap back to page 1 (whose data is fetched fresh) so there's always a way out. + useEffect(() => { + if (!isLoading && data?.success && data.rows.length === 0 && page > 1) { + del("page"); } - const liveKeys = new Set(breakdown.keys.map((k) => k.concurrencyKey)); - const live = breakdown.keys.map((k) => ({ - key: k.concurrencyKey, - queued: k.queued, - running: k.running, - oldestWaitMs: Math.max(0, loadedAt - k.oldestEnqueuedAt), - range: range.get(k.concurrencyKey), - })); - const historyOnly = [...range.entries()] - .filter(([key]) => !liveKeys.has(key)) - .map(([key, stats]) => ({ - key, - queued: 0, - running: 0, - oldestWaitMs: null as number | null, - range: stats, - })); - return [...live, ...historyOnly].slice(0, 50); - }, [rows, breakdown, loadedAt]); - - // The top-bar search filters the key list by substring (case-insensitive), the same idea as the - // Queues page search over queue names. - const query = value("query")?.trim().toLowerCase(); - const filtered = useMemo( - () => (query ? merged.filter((r) => r.key.toLowerCase().includes(query)) : merged), - [merged, query] - ); - - const sortColumns = useMemo[]>( - () => [ - { key: "key", type: "alpha", value: (r) => r.key }, - { key: "queued", type: "number", value: (r) => r.queued }, - { key: "running", type: "number", value: (r) => r.running }, - { key: "oldestWait", type: "number", value: (r) => r.oldestWaitMs }, - { key: "started", type: "number", value: (r) => r.range?.started }, - { key: "peakBacklog", type: "number", value: (r) => r.range?.peakBacklog }, - { key: "meanWait", type: "number", value: (r) => r.range?.meanWaitMs }, - ], - [] - ); - const { sortedRows, getSortProps } = useTableSort(filtered, sortColumns); - - if (merged.length === 0) return null; + }, [isLoading, data, page, del]); return ( - // Full-bleed, edge-to-edge like the Queues list table: a top border, no rounded side box. -
- - - Key - - Queued now - - - Running now - - - Oldest wait - - - Started - - - Peak backlog - - - Mean delay - - - - - {sortedRows.length === 0 ? ( - - No keys match “{query}” - - ) : null} - {sortedRows.map((row) => ( - (selectedKey === row.key ? del("key") : replace({ key: row.key }))} - > - {row.key} - {row.queued.toLocaleString()} - {row.running.toLocaleString()} - - {row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)} - - - {row.range ? row.range.started.toLocaleString() : showLoading ? "…" : "–"} - - - {row.range ? row.range.peakBacklog.toLocaleString() : showLoading ? "…" : "–"} - - - {row.range && row.range.meanWaitMs > 0 ? formatWaitMs(row.range.meanWaitMs) : "–"} - +
+ {/* Title bar above the table, shown only when there's more than one page: the section title + on the left, prev/next pagination on the right. Hidden entirely for a single page. */} + {totalPages > 1 ? ( +
+ Concurrency keys + +
+ ) : null} + {/* Full-bleed, edge-to-edge like the Queues list table: a top border, no rounded side box. */} +
+ + + Key + Queued now + Running now + Oldest wait + Started + Peak backlog + Mean delay - ))} - -
+ + + {showLoading ? ( + + Loading… + + ) : rows.length === 0 ? ( + + {search ? `No keys match “${search}”` : "No concurrency keys"} + + ) : ( + rows.map((row) => ( + (selectedKey === row.key ? del("key") : replace({ key: row.key }))} + > + {row.key} + {row.queued.toLocaleString()} + {row.running.toLocaleString()} + + {row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)} + + {row.started.toLocaleString()} + {row.peakBacklog.toLocaleString()} + + {row.meanWaitMs > 0 ? formatWaitMs(row.meanWaitMs) : "–"} + + + )) + )} + + +
); } @@ -862,7 +959,12 @@ function KeyDrilldown({
+ This key: waiting (Queued, color) vs running ( + color). + + } className="aspect-[2/1]" query={`SELECT timeBucket() AS t, max(max_queued) AS queued, max(max_running) AS running\nFROM queue_metrics_by_key\nWHERE ${pin}\nGROUP BY t\nORDER BY t`} fillGaps @@ -983,25 +1085,7 @@ function QueueStats({ return ( - - - -
- } - /> + 0 ? `peak ${formatNumberCompact(peakQueued)}` : undefined} suffixClassName="text-text-dimmed" accessory={ - + + + } /> 0 && running >= limit; const pct = limit && limit > 0 ? Math.min(100, Math.round((running / limit) * 100)) : 0; return ( -
+
Concurrency @@ -1082,7 +1169,12 @@ function ConcurrencyBlock({ / {limit !== null ? limit.toLocaleString() : "∞"} {limit !== null && limit > 0 && ( - + {/* Separator so the limit and the percentage don't read as one number (e.g. "/ 25" + "44%" mashing into "2544%"). */} · diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx index 9ab388718b..3527efdbf4 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx @@ -1296,8 +1296,10 @@ function WaitingInQueueBlock({ : null; return ( -
-
+
+ {/* Responsive to the side panel width (container query, not viewport): 3-up while there's + room, stacking to a single column only once the panel gets narrow. */} +
@@ -1323,7 +1325,10 @@ function WaitingInQueueBlock({ : "of ∞" } /> - +
diff --git a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts new file mode 100644 index 0000000000..6ae479789f --- /dev/null +++ b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts @@ -0,0 +1,180 @@ +import { type ActionFunctionArgs, json } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { timeFilterFromTo } from "~/components/runs/v3/SharedFilters"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; +import { findEnvironmentById, hasAccessToEnvironment } from "~/models/runtimeEnvironment.server"; +import { requireUserId } from "~/services/session.server"; +import { canAccessQueueMetricsUi } from "~/v3/canAccessQueueMetricsUi.server"; +import { engine } from "~/v3/runEngine.server"; + +// One page of a queue's concurrency keys. The ClickHouse tier (queue_metrics_ck_v1) is the +// paginated authority — ranked by peak backlog over the window with the total on every row +// (single scan) — and the ≤PER_PAGE keys on the page are enriched with live "now" counts from +// Redis (O(page), independent of total key cardinality). This replaces the old top-50 cap. +export const CONCURRENCY_KEYS_PER_PAGE = 25; + +// Matches QUEUE_METRICS_DEFAULT_PERIOD (the detail page's TimeFilter default). +const DEFAULT_PERIOD = "1d"; + +const Body = z.object({ + organizationId: z.string(), + projectId: z.string(), + environmentId: z.string(), + /** The queue's full name (e.g. `task/my-task` or a custom queue name). */ + queueName: z.string(), + period: z.string().nullish(), + from: z.string().nullish(), + to: z.string().nullish(), + /** Case-insensitive substring filter on the key. */ + search: z.string().nullish(), + page: z.number().int().min(1).default(1), +}); + +export type ConcurrencyKeyRow = { + key: string; + queued: number; + running: number; + oldestWaitMs: number | null; + started: number; + peakBacklog: number; + peakRunning: number; + meanWaitMs: number; +}; + +export type ConcurrencyKeysResponse = + | { + success: true; + rows: ConcurrencyKeyRow[]; + total: number; + page: number; + perPage: number; + loadedAt: number; + } + | { success: false; error: string }; + +// Snap the window to a minute grid so repeated loads within a bucket produce identical query +// params and share ClickHouse query-cache entries (same trick as the queue-list ranking). +function formatClickhouseDateTime(date: Date): string { + return date.toISOString().slice(0, 19).replace("T", " "); +} +function floorToMinute(ms: number): number { + return Math.floor(ms / 60_000) * 60_000; +} +function ceilToMinute(ms: number): number { + return Math.ceil(ms / 60_000) * 60_000; +} + +export const action = async ({ request }: ActionFunctionArgs) => { + const userId = await requireUserId(request); + + const submission = Body.safeParse(await request.json()); + if (!submission.success) { + return json( + { success: false, error: "Invalid input" }, + { status: 400 } + ); + } + const { organizationId, projectId, environmentId, queueName, period, from, to, search, page } = + submission.data; + + const hasAccess = await hasAccessToEnvironment({ + environmentId, + projectId, + organizationId, + userId, + }); + if (!hasAccess) { + return json( + { success: false, error: "You don't have permission for this resource" }, + { status: 403 } + ); + } + + // Needed as the tenant scope for the live Redis lookup (organization/project/environment ids). + const environment = await findEnvironmentById(environmentId); + if (!environment) { + return json( + { success: false, error: "Environment not found" }, + { status: 404 } + ); + } + + // Gate on the per-org Queue Metrics UI flag, matching the queue detail page and run inspector, so + // this endpoint's data isn't reachable for orgs that can't see the UI. 404 (not 403) to hide it. + if ( + !(await canAccessQueueMetricsUi({ + userId, + organizationSlug: environment.organization.slug, + })) + ) { + return json({ success: false, error: "Not found" }, { status: 404 }); + } + + const range = timeFilterFromTo({ + period: period ?? undefined, + from: from ?? undefined, + to: to ?? undefined, + defaultPeriod: DEFAULT_PERIOD, + }); + const startTime = formatClickhouseDateTime(new Date(floorToMinute(range.from.getTime()))); + const endTime = formatClickhouseDateTime(new Date(ceilToMinute(range.to.getTime()))); + + try { + const clickhouse = await clickhouseFactory.getClickhouseForOrganization( + organizationId, + "query" + ); + + const [rankingError, rankingRows] = await clickhouse.queueMetrics.concurrencyKeyRanking({ + organizationId, + projectId, + environmentId, + queueName, + startTime, + endTime, + nameContains: search?.trim() ?? "", + limit: CONCURRENCY_KEYS_PER_PAGE, + offset: (page - 1) * CONCURRENCY_KEYS_PER_PAGE, + }); + if (rankingError) { + throw rankingError; + } + + 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); + const loadedAt = Date.now(); + + const rows: ConcurrencyKeyRow[] = (rankingRows ?? []).map((r) => { + const l = live.get(r.concurrency_key); + const oldestWaitMs = + l && l.oldestEnqueuedAt != null ? Math.max(0, loadedAt - l.oldestEnqueuedAt) : null; + return { + key: r.concurrency_key, + queued: l?.queued ?? 0, + running: l?.running ?? 0, + oldestWaitMs, + started: r.started, + peakBacklog: r.peak_backlog, + peakRunning: r.peak_running, + meanWaitMs: r.mean_wait_ms, + }; + }); + + return json({ + success: true, + rows, + total, + page, + perPage: CONCURRENCY_KEYS_PER_PAGE, + loadedAt, + }); + } catch (error) { + return json( + { success: false, error: error instanceof Error ? error.message : "Query failed" }, + { status: 400 } + ); + } +}; diff --git a/apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts b/apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts index 23d20e108f..59da5d9b47 100644 --- a/apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts +++ b/apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts @@ -62,12 +62,20 @@ if (workloadCreatedAtGateEnabled && !workloadTokenCutoff) { type WorkloadGateAction = "start" | "complete" | "continue" | "snapshots_since"; -const workloadAuthGateCounter = new Counter({ - name: "workload_auth_gate_total", - help: "Deployment token authorization outcomes on worker actions", - labelNames: ["outcome", "action"] as const, - registers: [metricsRegister], -}); +// Wrapped in singleton() so a dev HMR re-eval of this module reuses the existing counter instead +// of calling `new Counter` again — prom-client throws "already registered" on the second +// registration, which crashes the dev server. Matches authenticatedWorkerInstanceCache above; a +// no-op in prod (the module evaluates once). +const workloadAuthGateCounter = singleton( + "workloadAuthGateCounter", + () => + new Counter({ + name: "workload_auth_gate_total", + help: "Deployment token authorization outcomes on worker actions", + labelNames: ["outcome", "action"] as const, + registers: [metricsRegister], + }) +); function createAuthenticatedWorkerInstanceCache() { return createCache({ diff --git a/internal-packages/clickhouse/src/index.ts b/internal-packages/clickhouse/src/index.ts index 1481be3653..49d2d7c02b 100644 --- a/internal-packages/clickhouse/src/index.ts +++ b/internal-packages/clickhouse/src/index.ts @@ -40,6 +40,7 @@ import { getQueueRanking, getQueueRankingNames, getQueueRankingCount, + getConcurrencyKeyRanking, } from "./queueMetrics.js"; import { getSessionTagsQueryBuilder, @@ -279,6 +280,7 @@ export class ClickHouse { ranking: getQueueRanking(this.reader), rankingNames: getQueueRankingNames(this.reader), rankingCount: getQueueRankingCount(this.reader), + concurrencyKeyRanking: getConcurrencyKeyRanking(this.reader), }; } diff --git a/internal-packages/clickhouse/src/queueMetrics.ts b/internal-packages/clickhouse/src/queueMetrics.ts index d86645e515..39576b4a0a 100644 --- a/internal-packages/clickhouse/src/queueMetrics.ts +++ b/internal-packages/clickhouse/src/queueMetrics.ts @@ -219,4 +219,76 @@ export function getQueueRankingCount(reader: ClickhouseReader) { }); } +// --- Per-concurrency-key ranking (the queue detail "Concurrency keys" table) --- + +const ConcurrencyKeyRankingParams = z.object({ + organizationId: z.string(), + projectId: z.string(), + environmentId: z.string(), + queueName: z.string(), + startTime: z.string(), + endTime: z.string(), + /** Case-insensitive substring filter on the key ('' = no filter). */ + nameContains: z.string(), + limit: z.number(), + offset: z.number(), +}); + +const ConcurrencyKeyRankingRow = z.object({ + concurrency_key: z.string(), + started: z.coerce.number(), + peak_backlog: z.coerce.number(), + peak_running: z.coerce.number(), + mean_wait_ms: z.coerce.number(), + ranked_total: z.coerce.number(), +}); + +// The per-key table (queue_metrics_ck_v1) is activity-bound and its ORDER BY starts with the +// tenant + queue, so filtering to one queue prunes to a contiguous index range — the aggregate +// is bounded by real activity, never by total key cardinality. There is no per-key 5m rollup, +// so this reads the 10s tier directly (the pre-existing LIMIT-50 query did the same). +const CK_RANKING_WHERE = `organization_id = {organizationId: String} + AND project_id = {projectId: String} + AND environment_id = {environmentId: String} + AND queue_name = {queueName: String} + AND bucket_start >= {startTime: DateTime} + AND bucket_start < {endTime: DateTime} + AND ({nameContains: String} = '' OR positionCaseInsensitive(concurrency_key, {nameContains: String}) > 0)`; + +/** + * One page of a queue's concurrency keys ranked by peak backlog over the window, with the total + * ranked-key count on every row (window function) so page + count cost a single scan — the same + * shape as getQueueRanking. The `concurrency_key ASC` tiebreak makes OFFSET paging stable across + * keys that share a peak. Range stats (started/peak_backlog/peak_running/mean wait) come back on + * the same rows; live "now" counts are enriched per page from Redis by the caller. + */ +export function getConcurrencyKeyRanking(reader: ClickhouseReader) { + return reader.query({ + name: "getConcurrencyKeyRanking", + query: `SELECT + concurrency_key, + started, + peak_backlog, + peak_running, + mean_wait_ms, + count() OVER () AS ranked_total + FROM ( + SELECT + concurrency_key, + deltaSumTimestampMerge(started_delta) AS started, + max(max_queued) AS peak_backlog, + max(max_running) AS peak_running, + if(sum(wait_ms_count) > 0, round(sum(wait_ms_sum) / sum(wait_ms_count)), 0) AS mean_wait_ms + FROM trigger_dev.queue_metrics_ck_v1 + WHERE ${CK_RANKING_WHERE} + GROUP BY concurrency_key + ORDER BY peak_backlog DESC, concurrency_key ASC + ) + LIMIT {limit: UInt32} OFFSET {offset: UInt32}`, + params: ConcurrencyKeyRankingParams, + schema: ConcurrencyKeyRankingRow, + settings: QUEUE_METRICS_CACHE_SETTINGS, + }); +} + // (per-queue detail series is now fetched via TRQL + fillGaps from the metric resource route) diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index ce5a55b215..e7853ee83c 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -1660,6 +1660,14 @@ export class RunEngine { return this.runQueue.concurrencyKeyBreakdown(environment, queue, options); } + async concurrencyKeyLiveStats( + environment: MinimalAuthenticatedEnvironment, + queue: string, + concurrencyKeys: string[] + ) { + return this.runQueue.concurrencyKeyLiveStats(environment, queue, concurrencyKeys); + } + async removeEnvironmentQueuesFromMasterQueue({ runtimeEnvironmentId, organizationId, diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 4e6ca89d84..58225cc505 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -613,6 +613,50 @@ export class RunQueue { return { totalBackloggedKeys, keys }; } + /** + * Live "now" stats for a specific set of concurrency keys — the current page of the paginated + * per-key table. Unlike concurrencyKeyBreakdown (which reads the top of the ckIndex), this + * targets exactly the given keys, so the table can enrich its ClickHouse-ranked page without + * scanning the whole index: O(keys) via one pipeline, independent of total key cardinality. + * Keys with no live backlog come back as zeros with a null oldest-enqueue time. + */ + public async concurrencyKeyLiveStats( + env: MinimalAuthenticatedEnvironment, + queue: string, + concurrencyKeys: string[] + ): Promise> { + const result = new Map< + string, + { queued: number; running: number; oldestEnqueuedAt: number | null } + >(); + if (concurrencyKeys.length === 0) return result; + + const ckIndexKey = this.keys.ckIndexKeyFromQueue(this.keys.queueKey(env, queue)); + + const pipeline = this.redis.pipeline(); + for (const concurrencyKey of concurrencyKeys) { + const member = this.keys.queueKey(env, queue, concurrencyKey); + pipeline.zcard(member); // queued in this key's subqueue + pipeline.scard(this.keys.queueCurrentConcurrencyKeyFromQueue(member)); // running + pipeline.zscore(ckIndexKey, member); // oldest-enqueued score (null once the key drains) + } + const res = await pipeline.exec(); + if (!res) return result; + + concurrencyKeys.forEach((concurrencyKey, i) => { + const queuedResult = res[i * 3]; + const runningResult = res[i * 3 + 1]; + const scoreResult = res[i * 3 + 2]; + const queued = queuedResult && !queuedResult[0] ? ((queuedResult[1] as number) ?? 0) : 0; + const running = runningResult && !runningResult[0] ? ((runningResult[1] as number) ?? 0) : 0; + const rawScore = scoreResult && !scoreResult[0] ? scoreResult[1] : null; + const oldestEnqueuedAt = rawScore != null ? Number(rawScore) : null; + result.set(concurrencyKey, { queued, running, oldestEnqueuedAt }); + }); + + return result; + } + public async lengthOfEnvQueue(env: MinimalAuthenticatedEnvironment) { return this.redis.zcard(this.keys.envQueueKey(env)); }