feat(server): New TTL system, enforce max queue length limits, lazy waitpoint creation (#2980)

This PR implements a new run TTL system and queue size limits to prevent
unbounded queue growth which should help prevent situations where queues
enter a "death spiral" where the queue will never be able to catch up.

The main/correct way to battle this situation is to enforce a maximum
TTL on all runs (e.g. up to 14 days) where runs that have been queued
for that maximum TTL will get auto-expired, making room for newer runs
to execute. This required creating a new TTL system that can handle
higher workloads and is now deeply integrated into the RunQueue. When
runs are enqueued with a TTL, they are added to their normal queue as
well as to the TTL queue. When runs are dequeued, they are removed from
both their normal queue and the TTL queue. If runs are dequeued by the
TTL system, they are removed from their normal queue. Both these
dequeues happen automatically so there is no race condition.

The TTL expiration system is also made reliable by expiring runs via a
Redis worker, which is enqueued to atomically inside the TTL dequeue lua
script.

### Optional associated waitpoints

Additionally, this PR implements an optimization where runs that aren't
triggered with a dependent parent run will no longer create an
associated waitpoint. Associated waitpoints are then lazily created if a
dependent run wants to wait for the child run post-facto (via debounce
or idempotency), which is a rare situation but is possible. This means
fewer waitpoint creations but also fewer waitpoint completions for runs
with no dependencies.

### Environment Queue Limits

Prevents any single queue growing too large by enforcing queue size
limits at trigger time.

- Queue size checks happen at trigger time - runs are rejected if queue
would exceed limit
- Dashboard UI shows queue limits on both the Queues page and a new
Limits page
- In-memory caching for queue size checks to reduce Redis load

### Batch trigger fixes

Currently when a batch item cannot be created for whatever reason (e.g.
queue limits) the run will never get created, which means a stalled run
if using `batchTriggerAndWait`. We've updated the system to handle this
differently: now when a batch item cannot be triggered and converted
into a run, we will eventually (after retrying 8 times up to 30s) we
will create a "pre-failed" run with the error details, correctly
resolving the batchTriggerAndWait.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -547,6 +547,9 @@ const EnvironmentSchema = z
 
     MAXIMUM_DEV_QUEUE_SIZE: z.coerce.number().int().optional(),
     MAXIMUM_DEPLOYED_QUEUE_SIZE: z.coerce.number().int().optional(),
+    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),
+    QUEUE_SIZE_CACHE_ENABLED: z.coerce.number().int().optional().default(1),
     MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
     MAX_BATCH_AND_WAIT_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
 
@@ -603,6 +606,19 @@ 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_TTL_WORKER_CONCURRENCY: z.coerce.number().int().default(1),
+    RUN_ENGINE_TTL_WORKER_BATCH_MAX_SIZE: z.coerce.number().int().default(50),
+    RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS: z.coerce.number().int().default(5_000),
+
+    /** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
+     *  will use this as their TTL, and runs with a TTL larger than this will be clamped. */
+    RUN_ENGINE_DEFAULT_MAX_TTL: z.string().optional(),
+
     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/EnvironmentQueuePresenter.server.ts

@@ -1,6 +1,7 @@
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { engine } from "~/v3/runEngine.server";
+import { getQueueSizeLimit } from "~/v3/utils/queueLimits.server";
 import { BasePresenter } from "./basePresenter.server";
 
 export type Environment = {
@@ -9,6 +10,7 @@ export type Environment = {
   concurrencyLimit: number;
   burstFactor: number;
   runsEnabled: boolean;
+  queueSizeLimit: number | null;
 };
 
 export class EnvironmentQueuePresenter extends BasePresenter {
@@ -30,19 +32,24 @@ export class EnvironmentQueuePresenter extends BasePresenter {
       },
       select: {
         runsEnabled: true,
+        maximumDevQueueSize: true,
+        maximumDeployedQueueSize: true,
       },
     });
 
     if (!organization) {
       throw new Error("Organization not found");
     }
 
+    const queueSizeLimit = getQueueSizeLimit(environment.type, organization);
+
     return {
       running,
       queued,
       concurrencyLimit: environment.maximumConcurrencyLimit,
       burstFactor: environment.concurrencyLimitBurstFactor.toNumber(),
       runsEnabled: environment.type === "DEVELOPMENT" || organization.runsEnabled,
+      queueSizeLimit,
     };
   }
 }

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

@@ -1,4 +1,5 @@
 import { Ratelimit } from "@upstash/ratelimit";
+import { RuntimeEnvironmentType } from "@trigger.dev/database";
 import { createHash } from "node:crypto";
 import { env } from "~/env.server";
 import { getCurrentPlan } from "~/services/platform.v3.server";
@@ -12,6 +13,8 @@ import { BasePresenter } from "./basePresenter.server";
 import { singleton } from "~/utils/singleton";
 import { logger } from "~/services/logger.server";
 import { CheckScheduleService } from "~/v3/services/checkSchedule.server";
+import { engine } from "~/v3/runEngine.server";
+import { getQueueSizeLimit, getQueueSizeLimitSource } from "~/v3/utils/queueLimits.server";
 
 // Create a singleton Redis client for rate limit queries
 const rateLimitRedisClient = singleton("rateLimitQueryRedisClient", () =>
@@ -66,8 +69,7 @@ export type LimitsResult = {
     logRetentionDays: QuotaInfo | null;
     realtimeConnections: QuotaInfo | null;
     batchProcessingConcurrency: QuotaInfo;
-    devQueueSize: QuotaInfo;
-    deployedQueueSize: QuotaInfo;
+    queueSize: QuotaInfo;
     metricDashboards: QuotaInfo | null;
     metricWidgetsPerDashboard: QuotaInfo | null;
     queryPeriodDays: QuotaInfo | null;
@@ -87,11 +89,13 @@ export class LimitsPresenter extends BasePresenter {
     organizationId,
     projectId,
     environmentId,
+    environmentType,
     environmentApiKey,
   }: {
     organizationId: string;
     projectId: string;
     environmentId: string;
+    environmentType: RuntimeEnvironmentType;
     environmentApiKey: string;
   }): Promise<LimitsResult> {
     // Get organization with all limit-related fields
@@ -175,6 +179,30 @@ export class LimitsPresenter extends BasePresenter {
       batchRateLimitConfig
     );
 
+    // Get current queue size for this environment
+    // We need the runtime environment fields for the engine query
+    const runtimeEnv = await this._replica.runtimeEnvironment.findFirst({
+      where: { id: environmentId },
+      select: {
+        id: true,
+        maximumConcurrencyLimit: true,
+        concurrencyLimitBurstFactor: true,
+      },
+    });
+
+    let currentQueueSize = 0;
+    if (runtimeEnv) {
+      const engineEnv = {
+        id: runtimeEnv.id,
+        type: environmentType,
+        maximumConcurrencyLimit: runtimeEnv.maximumConcurrencyLimit,
+        concurrencyLimitBurstFactor: runtimeEnv.concurrencyLimitBurstFactor,
+        organization: { id: organizationId },
+        project: { id: projectId },
+      };
+      currentQueueSize = (await engine.lengthOfEnvQueue(engineEnv)) ?? 0;
+    }
+
     // Get plan-level limits
     const schedulesLimit = limits?.schedules?.number ?? null;
     const teamMembersLimit = limits?.teamMembers?.number ?? null;
@@ -217,72 +245,72 @@ export class LimitsPresenter extends BasePresenter {
         schedules:
           schedulesLimit !== null
             ? {
-                name: "Schedules",
-                description: "Maximum number of schedules per project",
-                limit: schedulesLimit,
-                currentUsage: scheduleCount,
-                source: "plan",
-                canExceed: limits?.schedules?.canExceed,
-                isUpgradable: true,
-              }
+              name: "Schedules",
+              description: "Maximum number of schedules per project",
+              limit: schedulesLimit,
+              currentUsage: scheduleCount,
+              source: "plan",
+              canExceed: limits?.schedules?.canExceed,
+              isUpgradable: true,
+            }
             : null,
         teamMembers:
           teamMembersLimit !== null
             ? {
-                name: "Team members",
-                description: "Maximum number of team members in this organization",
-                limit: teamMembersLimit,
-                currentUsage: organization._count.members,
-                source: "plan",
-                canExceed: limits?.teamMembers?.canExceed,
-                isUpgradable: true,
-              }
+              name: "Team members",
+              description: "Maximum number of team members in this organization",
+              limit: teamMembersLimit,
+              currentUsage: organization._count.members,
+              source: "plan",
+              canExceed: limits?.teamMembers?.canExceed,
+              isUpgradable: true,
+            }
             : null,
         alerts:
           alertsLimit !== null
             ? {
-                name: "Alert channels",
-                description: "Maximum number of alert channels per project",
-                limit: alertsLimit,
-                currentUsage: alertChannelCount,
-                source: "plan",
-                canExceed: limits?.alerts?.canExceed,
-                isUpgradable: true,
-              }
+              name: "Alert channels",
+              description: "Maximum number of alert channels per project",
+              limit: alertsLimit,
+              currentUsage: alertChannelCount,
+              source: "plan",
+              canExceed: limits?.alerts?.canExceed,
+              isUpgradable: true,
+            }
             : null,
         branches:
           branchesLimit !== null
             ? {
-                name: "Preview branches",
-                description: "Maximum number of active preview branches per project",
-                limit: branchesLimit,
-                currentUsage: activeBranchCount,
-                source: "plan",
-                canExceed: limits?.branches?.canExceed,
-                isUpgradable: true,
-              }
+              name: "Preview branches",
+              description: "Maximum number of active preview branches per project",
+              limit: branchesLimit,
+              currentUsage: activeBranchCount,
+              source: "plan",
+              canExceed: limits?.branches?.canExceed,
+              isUpgradable: true,
+            }
             : null,
         logRetentionDays:
           logRetentionDaysLimit !== null
             ? {
-                name: "Log retention",
-                description: "Number of days logs are retained",
-                limit: logRetentionDaysLimit,
-                currentUsage: 0, // Not applicable - this is a duration, not a count
-                source: "plan",
-              }
+              name: "Log retention",
+              description: "Number of days logs are retained",
+              limit: logRetentionDaysLimit,
+              currentUsage: 0, // Not applicable - this is a duration, not a count
+              source: "plan",
+            }
             : null,
         realtimeConnections:
           realtimeConnectionsLimit !== null
             ? {
-                name: "Realtime connections",
-                description: "Maximum concurrent Realtime connections",
-                limit: realtimeConnectionsLimit,
-                currentUsage: 0, // Would need to query realtime service for this
-                source: "plan",
-                canExceed: limits?.realtimeConcurrentConnections?.canExceed,
-                isUpgradable: true,
-              }
+              name: "Realtime connections",
+              description: "Maximum concurrent Realtime connections",
+              limit: realtimeConnectionsLimit,
+              currentUsage: 0, // Would need to query realtime service for this
+              source: "plan",
+              canExceed: limits?.realtimeConcurrentConnections?.canExceed,
+              isUpgradable: true,
+            }
             : null,
         batchProcessingConcurrency: {
           name: "Batch processing concurrency",
@@ -293,19 +321,13 @@ export class LimitsPresenter extends BasePresenter {
           canExceed: true,
           isUpgradable: true,
         },
-        devQueueSize: {
-          name: "Dev queue size",
-          description: "Maximum pending runs in development environments",
-          limit: organization.maximumDevQueueSize ?? null,
-          currentUsage: 0, // Would need to query Redis for this
-          source: organization.maximumDevQueueSize ? "override" : "default",
-        },
-        deployedQueueSize: {
-          name: "Deployed queue size",
-          description: "Maximum pending runs in deployed environments",
-          limit: organization.maximumDeployedQueueSize ?? null,
-          currentUsage: 0, // Would need to query Redis for this
-          source: organization.maximumDeployedQueueSize ? "override" : "default",
+        queueSize: {
+          name: "Max queued runs",
+          description: "Maximum pending runs per individual queue in this environment",
+          limit: getQueueSizeLimit(environmentType, organization),
+          currentUsage: currentQueueSize,
+          source: getQueueSizeLimitSource(environmentType, organization),
+          isUpgradable: true,
         },
         metricDashboards:
           metricDashboardsLimit !== null

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

@@ -82,6 +82,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       organizationId: project.organizationId,
       projectId: project.id,
       environmentId: environment.id,
+      environmentType: environment.type,
       environmentApiKey: environment.apiKey,
     })
   );
@@ -507,9 +508,8 @@ function QuotasSection({
   // Include batch processing concurrency
   quotaRows.push(quotas.batchProcessingConcurrency);
 
-  // Add queue size quotas if set
-  if (quotas.devQueueSize.limit !== null) quotaRows.push(quotas.devQueueSize);
-  if (quotas.deployedQueueSize.limit !== null) quotaRows.push(quotas.deployedQueueSize);
+  // Add queue size quota if set
+  if (quotas.queueSize.limit !== null) quotaRows.push(quotas.queueSize);
 
   // Metric & query quotas
   if (quotas.metricDashboards) quotaRows.push(quotas.metricDashboards);
@@ -565,8 +565,11 @@ function QuotaRow({
   const isDurationQuota = quota.name === "Log retention" || quota.name === "Query period";
   const isPerItemQuota = quota.name === "Charts per dashboard";
   const isRetentionQuota = isDurationQuota || isPerItemQuota;
+  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 duration-based quotas (Log retention, Query period)
   if (isDurationQuota) {
@@ -667,10 +670,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} />