feat(webapp): queue allocation view and relevance-ordered queue list

Adds an Allocation tab to the Queues page (behind the queue metrics UI flag): overview cards, a burst-aware capacity bar showing each queue allocation and its live usage in a distinct color, an inline-editable limits table with per-queue locks, load-weighted auto-balance, and a review dialog that bulk-applies limits as overrides through the existing concurrency system.

The queue list now defaults to Busiest ordering (with Backlog and Name options). ClickHouse ranks queues by activity over the last 15 minutes and returns just the requested page of names, so the cost per page is one small aggregate regardless of environment size; idle queues follow in name order and any failure falls back to name ordering. The classic page keeps plain name order.

Author: ericallam

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

@@ -0,0 +1,94 @@
+import { TaskQueueType, type Prisma } from "@trigger.dev/database";
+import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { engine } from "~/v3/runEngine.server";
+import { BasePresenter } from "./basePresenter.server";
+
+const MAX_ALLOCATION_QUEUES = 500;
+
+export type QueueAllocationItem = {
+  id: string;
+  name: string;
+  type: "task" | "custom";
+  running: number;
+  queued: number;
+  paused: boolean;
+  /** Explicit per-queue limit; null means the queue floats up to the env limit. */
+  limit: number | null;
+  overridden: boolean;
+};
+
+export type QueueAllocation = {
+  queues: QueueAllocationItem[];
+  totalQueues: number;
+  truncated: boolean;
+  /** Sum of explicit limits, each clamped to the env limit. */
+  allocated: number;
+  unlimitedCount: number;
+};
+
+/** Every queue in the environment (capped) with live counts, for the allocation view. */
+export class QueueAllocationPresenter extends BasePresenter {
+  public async call({
+    environment,
+  }: {
+    environment: AuthenticatedEnvironment;
+  }): Promise<QueueAllocation> {
+    const where: Prisma.TaskQueueWhereInput = {
+      runtimeEnvironmentId: environment.id,
+      version: "V2",
+    };
+
+    const [totalQueues, queues] = await Promise.all([
+      this._replica.taskQueue.count({ where }),
+      this._replica.taskQueue.findMany({
+        where,
+        select: {
+          friendlyId: true,
+          name: true,
+          type: true,
+          paused: true,
+          concurrencyLimit: true,
+          concurrencyLimitOverriddenAt: true,
+        },
+        orderBy: { orderableName: "asc" },
+        take: MAX_ALLOCATION_QUEUES,
+      }),
+    ]);
+
+    const names = queues.map((q) => q.name);
+    const [queuedByQueue, runningByQueue] = await Promise.all([
+      engine.lengthOfQueues(environment, names),
+      engine.currentConcurrencyOfQueues(environment, names),
+    ]);
+
+    const envLimit = environment.maximumConcurrencyLimit;
+    let allocated = 0;
+    let unlimitedCount = 0;
+
+    const items: QueueAllocationItem[] = queues.map((queue) => {
+      if (queue.concurrencyLimit === null) {
+        unlimitedCount++;
+      } else {
+        allocated += Math.min(queue.concurrencyLimit, envLimit);
+      }
+      return {
+        id: queue.friendlyId,
+        name: queue.name.replace(/^task\//, ""),
+        type: queue.type === TaskQueueType.VIRTUAL ? ("task" as const) : ("custom" as const),
+        running: runningByQueue[queue.name] ?? 0,
+        queued: queuedByQueue[queue.name] ?? 0,
+        paused: queue.paused,
+        limit: queue.concurrencyLimit,
+        overridden: queue.concurrencyLimitOverriddenAt !== null,
+      };
+    });
+
+    return {
+      queues: items,
+      totalQueues,
+      truncated: totalQueues > queues.length,
+      allocated,
+      unlimitedCount,
+    };
+  }
+}

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

@@ -3,6 +3,8 @@ import type { Prisma } from "@trigger.dev/database";
 import { TaskQueueType } from "@trigger.dev/database";
 import { type PrismaClientOrTransaction } from "~/db.server";
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
+import { logger } from "~/services/logger.server";
 import { determineEngineVersion } from "~/v3/engineVersion.server";
 import { engine } from "~/v3/runEngine.server";
 import { BasePresenter } from "./basePresenter.server";
@@ -13,6 +15,12 @@ type QueueListEngine = Pick<RunEngine, "lengthOfQueues" | "currentConcurrencyOfQ
 export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25;
 const MAX_ITEMS_PER_PAGE = 100;
 
+export type QueueListSort = "busiest" | "queued" | "name";
+
+/** Ranking reads recent aggregated gauges, so ordering is a stable snapshot, not a live sort. */
+export const QUEUE_RANKING_WINDOW_MINUTES = 15;
+const MAX_RANKED_QUEUES = 5000;
+
 const typeToDBQueueType: Record<"task" | "custom", TaskQueueType> = {
   task: TaskQueueType.VIRTUAL,
   custom: TaskQueueType.NAMED,
@@ -30,6 +38,38 @@ const queueListSelect = {
   paused: true,
 } satisfies Prisma.TaskQueueSelect;
 
+type QueueListRow = Prisma.TaskQueueGetPayload<{ select: typeof queueListSelect }>;
+
+type QueueListItem = ReturnType<typeof toQueueItem>;
+
+type QueueListPagination =
+  | { mode: "filtered"; currentPage: number; hasMore: boolean }
+  | { mode: "unfiltered"; currentPage: number; totalPages: number; count: number };
+
+// The `?: undefined` markers keep every key reachable across the union, so consumers
+// can destructure before narrowing on `success`.
+export type QueueListResult =
+  | {
+      success: false;
+      code: string;
+      totalQueues: number;
+      hasFilters: boolean;
+      queues?: undefined;
+      pagination?: undefined;
+    }
+  | {
+      success: true;
+      queues: QueueListItem[];
+      pagination: QueueListPagination;
+      totalQueues?: number;
+      hasFilters: boolean;
+      code?: undefined;
+    };
+
+function formatClickhouseDateTime(date: Date): string {
+  return date.toISOString().slice(0, 19).replace("T", " ");
+}
+
 function buildQueueListWhere(
   environmentId: string,
   query: string | undefined,
@@ -70,13 +110,15 @@ export class QueueListPresenter extends BasePresenter {
     query,
     page,
     type,
+    sort = "name",
   }: {
     environment: AuthenticatedEnvironment;
     query?: string;
     page: number;
     perPage?: number;
     type?: "task" | "custom";
-  }) {
+    sort?: QueueListSort;
+  }): Promise<QueueListResult> {
     const hasFilters = Boolean(query?.trim()) || type !== undefined;
 
     const engineVersion = await determineEngineVersion({ environment });
@@ -110,6 +152,18 @@ export class QueueListPresenter extends BasePresenter {
       };
     }
 
+    if (sort !== "name") {
+      // Ranking is additive: any failure or unsupported input falls back to name order.
+      try {
+        const ranked = await this.getRankedQueues(environment, query, page, type, sort);
+        if (ranked) {
+          return ranked;
+        }
+      } catch (error) {
+        logger.warn("Queue ranking unavailable, falling back to name order", { error });
+      }
+    }
+
     if (hasFilters) {
       const { queues, hasMore } = await this.getFilteredQueues(environment, query, page, type);
 
@@ -143,6 +197,123 @@ export class QueueListPresenter extends BasePresenter {
     };
   }
 
+  /**
+   * ClickHouse ranks queues by recent activity and returns the requested page of names;
+   * queues with no recent metrics follow in name order. Null when ranking does not apply.
+   */
+  private async getRankedQueues(
+    environment: AuthenticatedEnvironment,
+    query: string | undefined,
+    page: number,
+    type: "task" | "custom" | undefined,
+    sort: Exclude<QueueListSort, "name">
+  ) {
+    if (type !== undefined) {
+      return null;
+    }
+
+    const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
+      environment.organizationId,
+      "query"
+    );
+
+    const rankingArgs = {
+      organizationId: environment.organizationId,
+      projectId: environment.projectId,
+      environmentId: environment.id,
+      startTime: formatClickhouseDateTime(
+        new Date(Date.now() - QUEUE_RANKING_WINDOW_MINUTES * 60 * 1000)
+      ),
+      nameContains: query?.trim() ?? "",
+    };
+
+    const [countError, countRows] = await clickhouse.queueMetrics.rankingCount(rankingArgs);
+    if (countError) {
+      throw countError;
+    }
+    const ranked = countRows?.[0]?.ranked ?? 0;
+    if (ranked > MAX_RANKED_QUEUES) {
+      return null;
+    }
+
+    const where = buildQueueListWhere(environment.id, query, type);
+    const totalQueues = await this._replica.taskQueue.count({ where });
+    const offset = (page - 1) * this.perPage;
+
+    let rankedPageQueues: QueueListRow[] = [];
+    if (offset < ranked) {
+      const [pageError, pageRows] = await clickhouse.queueMetrics.rankingPage({
+        ...rankingArgs,
+        byQueuedOnly: sort === "queued" ? 1 : 0,
+        limit: this.perPage,
+        offset,
+      });
+      if (pageError) {
+        throw pageError;
+      }
+      const rankedNames = (pageRows ?? []).map((row) => row.queue_name);
+      rankedPageQueues = await this.findQueuesByNames(where, rankedNames);
+    }
+
+    // Tail of the page: name-ordered queues that have no recent metrics. Slot math uses the
+    // ClickHouse counts so pages never overlap, even if some ranked names no longer exist.
+    const rankedSlots = Math.min(Math.max(ranked - offset, 0), this.perPage);
+    const tailNeeded = this.perPage - rankedSlots;
+    let tailQueues: QueueListRow[] = [];
+    if (tailNeeded > 0) {
+      let excludedNames: string[] = [];
+      if (ranked > 0) {
+        const [allError, allRows] = await clickhouse.queueMetrics.rankingPage({
+          ...rankingArgs,
+          byQueuedOnly: 0,
+          limit: MAX_RANKED_QUEUES,
+          offset: 0,
+        });
+        if (allError) {
+          throw allError;
+        }
+        excludedNames = (allRows ?? []).map((row) => row.queue_name);
+      }
+      tailQueues = await this._replica.taskQueue.findMany({
+        where: { ...where, name: { notIn: excludedNames } },
+        select: queueListSelect,
+        orderBy: {
+          orderableName: "asc",
+        },
+        skip: Math.max(0, offset - ranked),
+        take: tailNeeded,
+      });
+    }
+
+    return {
+      success: true as const,
+      queues: await this.enrichQueues(environment, [...rankedPageQueues, ...tailQueues]),
+      pagination: {
+        mode: "unfiltered" as const,
+        currentPage: page,
+        totalPages: Math.max(1, Math.ceil(totalQueues / this.perPage)),
+        count: totalQueues,
+      },
+      totalQueues,
+      hasFilters: Boolean(query?.trim()) || type !== undefined,
+    };
+  }
+
+  private async findQueuesByNames(
+    where: Prisma.TaskQueueWhereInput,
+    names: string[]
+  ): Promise<QueueListRow[]> {
+    if (names.length === 0) {
+      return [];
+    }
+    const queues = await this._replica.taskQueue.findMany({
+      where: { ...where, name: { in: names } },
+      select: queueListSelect,
+    });
+    const byName = new Map(queues.map((queue) => [queue.name, queue]));
+    return names.flatMap((name) => byName.get(name) ?? []);
+  }
+
   private async getFilteredQueues(
     environment: AuthenticatedEnvironment,
     query: string | undefined,