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

apps/webapp/app/env.server.ts

@@ -540,6 +540,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),
 
@@ -596,6 +599,16 @@ 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),
+
+    /** 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;
   };
   features: {
     hasStagingEnvironment: FeatureInfo;
@@ -84,11 +86,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
@@ -167,6 +171,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;
@@ -206,72 +234,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",
@@ -282,19 +310,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,
         },
       },
       features: {

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);
 
   return (
     <div className="flex flex-col gap-3">
@@ -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} />