Add queue length limits at the queue level, lazy waitpoint creation, new ttl system

Author: ericallam

apps/webapp/app/env.server.ts

@@ -535,7 +535,7 @@ const EnvironmentSchema = z
 
     MAXIMUM_DEV_QUEUE_SIZE: z.coerce.number().int().optional().default(500),
     MAXIMUM_DEPLOYED_QUEUE_SIZE: z.coerce.number().int().optional(),
-    QUEUE_SIZE_CACHE_TTL_MS: z.coerce.number().int().optional().default(30_000), // 30 seconds
+    QUEUE_SIZE_CACHE_TTL_MS: z.coerce.number().int().optional().default(1_000), // 1 second
     QUEUE_SIZE_CACHE_MAX_SIZE: z.coerce.number().int().optional().default(5_000),
     MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
     MAX_BATCH_AND_WAIT_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
@@ -593,6 +593,12 @@ const EnvironmentSchema = z
     RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER_IN_MS: z.coerce.number().int().optional(),
     RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER_IN_MS: z.coerce.number().int().optional(),
 
+    // TTL System settings for automatic run expiration
+    RUN_ENGINE_TTL_SYSTEM_DISABLED: BoolEnv.default(false),
+    RUN_ENGINE_TTL_SYSTEM_SHARD_COUNT: z.coerce.number().int().optional(),
+    RUN_ENGINE_TTL_SYSTEM_POLL_INTERVAL_MS: z.coerce.number().int().default(1_000),
+    RUN_ENGINE_TTL_SYSTEM_BATCH_SIZE: z.coerce.number().int().default(100),
+
     RUN_ENGINE_RUN_LOCK_DURATION: z.coerce.number().int().default(5000),
     RUN_ENGINE_RUN_LOCK_AUTOMATIC_EXTENSION_THRESHOLD: z.coerce.number().int().default(1000),
     RUN_ENGINE_RUN_LOCK_MAX_RETRIES: z.coerce.number().int().default(10),

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

@@ -312,7 +312,7 @@ export class LimitsPresenter extends BasePresenter {
         },
         queueSize: {
           name: "Max queued runs",
-          description: "Maximum pending runs across all queues in this environment",
+          description: "Maximum pending runs per individual queue in this environment",
           limit: getQueueSizeLimit(environmentType, organization),
           currentUsage: currentQueueSize,
           source: getQueueSizeLimitSource(environmentType, organization),

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.limits/route.tsx

@@ -556,9 +556,12 @@ function QuotaRow({
   billingPath: string;
 }) {
   // For log retention, we don't show current usage as it's a duration, not a count
+  // For queue size, we don't show current usage as the limit is per-queue, not environment-wide
   const isRetentionQuota = quota.name === "Log retention";
+  const isQueueSizeQuota = quota.name === "Max queued runs";
+  const hideCurrentUsage = isRetentionQuota || isQueueSizeQuota;
   const percentage =
-    !isRetentionQuota && quota.limit && quota.limit > 0 ? quota.currentUsage / quota.limit : null;
+    !hideCurrentUsage && quota.limit && quota.limit > 0 ? quota.currentUsage / quota.limit : null;
 
   // Special handling for Log retention
   if (quota.name === "Log retention") {
@@ -657,10 +660,10 @@ function QuotaRow({
         alignment="right"
         className={cn(
           "tabular-nums",
-          isRetentionQuota ? "text-text-dimmed" : getUsageColorClass(percentage, "usage")
+          hideCurrentUsage ? "text-text-dimmed" : getUsageColorClass(percentage, "usage")
         )}
       >
-        {isRetentionQuota ? "–" : formatNumber(quota.currentUsage)}
+        {hideCurrentUsage ? "–" : formatNumber(quota.currentUsage)}
       </TableCell>
       <TableCell alignment="right">
         <SourceBadge source={quota.source} />

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx

@@ -68,7 +68,6 @@ import { EnvironmentQueuePresenter } from "~/presenters/v3/EnvironmentQueuePrese
 import { QueueListPresenter } from "~/presenters/v3/QueueListPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { cn } from "~/utils/cn";
-import { formatNumberCompact } from "~/utils/numberFormatter";
 import {
   concurrencyPath,
   docsPath,
@@ -346,13 +345,7 @@ export default function Page() {
             <BigNumber
               title="Queued"
               value={environment.queued}
-              suffix={
-                <QueuedSuffix
-                  queued={environment.queued}
-                  queueSizeLimit={environment.queueSizeLimit}
-                  isPaused={env.paused}
-                />
-              }
+              suffix={env.paused ? <span className="text-warning">paused</span> : undefined}
               animate
               accessory={
                 <div className="flex items-start gap-1">
@@ -519,7 +512,10 @@ export default function Page() {
                   {queues.length > 0 ? (
                     queues.map((queue) => {
                       const limit = queue.concurrencyLimit ?? environment.concurrencyLimit;
-                      const isAtLimit = queue.running >= limit;
+                      const isAtConcurrencyLimit = queue.running >= limit;
+                      const isAtQueueLimit =
+                        environment.queueSizeLimit !== null &&
+                        queue.queued >= environment.queueSizeLimit;
                       const queueFilterableName = `${queue.type === "task" ? "task/" : ""}${
                         queue.name
                       }`;
@@ -545,7 +541,12 @@ export default function Page() {
                                   Paused
                                 </Badge>
                               ) : null}
-                              {isAtLimit ? (
+                              {isAtQueueLimit ? (
+                                <Badge variant="extra-small" className="text-error">
+                                  At queue limit
+                                </Badge>
+                              ) : null}
+                              {isAtConcurrencyLimit ? (
                                 <Badge variant="extra-small" className="text-warning">
                                   At concurrency limit
                                 </Badge>
@@ -556,7 +557,8 @@ export default function Page() {
                             alignment="right"
                             className={cn(
                               "w-[1%] pl-16 tabular-nums",
-                              queue.paused ? "opacity-50" : undefined
+                              queue.paused ? "opacity-50" : undefined,
+                              isAtQueueLimit && "text-error"
                             )}
                           >
                             {queue.queued}
@@ -567,7 +569,7 @@ export default function Page() {
                               "w-[1%] pl-16 tabular-nums",
                               queue.paused ? "opacity-50" : undefined,
                               queue.running > 0 && "text-text-bright",
-                              isAtLimit && "text-warning"
+                              isAtConcurrencyLimit && "text-warning"
                             )}
                           >
                             {queue.running}
@@ -587,7 +589,7 @@ export default function Page() {
                             className={cn(
                               "w-[1%] pl-16",
                               queue.paused ? "opacity-50" : undefined,
-                              isAtLimit && "text-warning",
+                              isAtConcurrencyLimit && "text-warning",
                               queue.concurrency?.overriddenAt && "font-medium text-text-bright"
                             )}
                           >
@@ -1128,53 +1130,3 @@ function BurstFactorTooltip({
     />
   );
 }
-
-function getQueueUsageColorClass(current: number, limit: number | null): string | undefined {
-  if (!limit) return undefined;
-  const percentage = current / limit;
-  if (percentage >= 1) return "text-error";
-  if (percentage >= 0.9) return "text-warning";
-  return undefined;
-}
-
-/**
- * Renders the suffix for the Queued BigNumber, showing:
- * - The limit with usage color and tooltip (if queueSizeLimit is set)
- * - "paused" text (if environment is paused)
- * - Both indicators when applicable
- */
-function QueuedSuffix({
-  queued,
-  queueSizeLimit,
-  isPaused,
-}: {
-  queued: number;
-  queueSizeLimit: number | null;
-  isPaused: boolean;
-}) {
-  const showLimit = queueSizeLimit !== null;
-
-  if (!showLimit && !isPaused) {
-    return null;
-  }
-
-  return (
-    <span className="flex items-center gap-1">
-      {showLimit && (
-        <>
-          <span className="text-text-dimmed">/</span>
-          <span className={getQueueUsageColorClass(queued, queueSizeLimit)}>
-            {formatNumberCompact(queueSizeLimit)}
-          </span>
-          <InfoIconTooltip
-            content="Maximum pending runs across all queues in this environment"
-            contentClassName="max-w-xs"
-          />
-        </>
-      )}
-      {isPaused && (
-        <span className="text-warning">{showLimit ? "(paused)" : "paused"}</span>
-      )}
-    </span>
-  );
-}

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -79,11 +79,26 @@ export class IdempotencyKeyConcern {
       }
 
       // We have an idempotent run, so we return it
-      const associatedWaitpoint = existingRun.associatedWaitpoint;
       const parentRunId = request.body.options?.parentRunId;
       const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion;
+
       //We're using `andWait` so we need to block the parent run with a waitpoint
-      if (associatedWaitpoint && resumeParentOnCompletion && parentRunId) {
+      if (resumeParentOnCompletion && parentRunId) {
+        // Get or create waitpoint lazily (existing run may not have one if it was standalone)
+        let associatedWaitpoint = existingRun.associatedWaitpoint;
+        if (!associatedWaitpoint) {
+          associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({
+            runId: existingRun.id,
+            projectId: request.environment.projectId,
+            environmentId: request.environment.id,
+          });
+        }
+
+        // If run already completed, return without blocking
+        if (!associatedWaitpoint) {
+          return { isCached: true, run: existingRun };
+        }
+
         await this.traceEventConcern.traceIdempotentRun(
           request,
           parentStore,
@@ -98,13 +113,13 @@ export class IdempotencyKeyConcern {
               request.options?.parentAsLinkType === "replay"
                 ? event.spanId
                 : event.traceparent?.spanId
-                ? `${event.traceparent.spanId}:${event.spanId}`
-                : event.spanId;
+                  ? `${event.traceparent.spanId}:${event.spanId}`
+                  : event.spanId;
 
             //block run with waitpoint
             await this.engine.blockRunWithWaitpoint({
               runId: RunId.fromFriendlyId(parentRunId),
-              waitpoints: associatedWaitpoint.id,
+              waitpoints: associatedWaitpoint!.id,
               spanIdToComplete: spanId,
               batch: request.options?.batchId
                 ? {