A couple of devin improvements and adding an in memory cache for the env queue size check

Author: ericallam

apps/webapp/app/env.server.ts

@@ -533,8 +533,10 @@ const EnvironmentSchema = z
     BATCH_TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(1_000_000), // 1MB
     TASK_RUN_METADATA_MAXIMUM_SIZE: z.coerce.number().int().default(262_144), // 256KB
 
-    MAXIMUM_DEV_QUEUE_SIZE: z.coerce.number().int().optional(),
+    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_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),
 

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

@@ -1,7 +1,7 @@
-import { env } from "~/env.server";
 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 = {
@@ -41,10 +41,7 @@ export class EnvironmentQueuePresenter extends BasePresenter {
       throw new Error("Organization not found");
     }
 
-    const queueSizeLimit =
-      environment.type === "DEVELOPMENT"
-        ? (organization.maximumDevQueueSize ?? env.MAXIMUM_DEV_QUEUE_SIZE ?? null)
-        : (organization.maximumDeployedQueueSize ?? env.MAXIMUM_DEPLOYED_QUEUE_SIZE ?? null);
+    const queueSizeLimit = getQueueSizeLimit(environment.type, organization);
 
     return {
       running,

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";
@@ -13,6 +14,7 @@ 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", () =>
@@ -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
@@ -168,13 +172,11 @@ export class LimitsPresenter extends BasePresenter {
     );
 
     // 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,
-        type: true,
-        organizationId: true,
-        projectId: true,
         maximumConcurrencyLimit: true,
         concurrencyLimitBurstFactor: true,
       },
@@ -184,11 +186,11 @@ export class LimitsPresenter extends BasePresenter {
     if (runtimeEnv) {
       const engineEnv = {
         id: runtimeEnv.id,
-        type: runtimeEnv.type,
+        type: environmentType,
         maximumConcurrencyLimit: runtimeEnv.maximumConcurrencyLimit,
         concurrencyLimitBurstFactor: runtimeEnv.concurrencyLimitBurstFactor,
-        organization: { id: runtimeEnv.organizationId },
-        project: { id: runtimeEnv.projectId },
+        organization: { id: organizationId },
+        project: { id: projectId },
       };
       currentQueueSize = (await engine.lengthOfEnvQueue(engineEnv)) ?? 0;
     }
@@ -311,21 +313,9 @@ export class LimitsPresenter extends BasePresenter {
         queueSize: {
           name: "Max queued runs",
           description: "Maximum pending runs across all queues in this environment",
-          limit:
-            runtimeEnv?.type === "DEVELOPMENT"
-              ? (organization.maximumDevQueueSize ?? env.MAXIMUM_DEV_QUEUE_SIZE ?? null)
-              : (organization.maximumDeployedQueueSize ?? env.MAXIMUM_DEPLOYED_QUEUE_SIZE ?? null),
+          limit: getQueueSizeLimit(environmentType, organization),
           currentUsage: currentQueueSize,
-          // "plan" = org has a value (typically set by billing sync)
-          // "default" = no org value, using env var fallback
-          source:
-            runtimeEnv?.type === "DEVELOPMENT"
-              ? organization.maximumDevQueueSize
-                ? "plan"
-                : "default"
-              : organization.maximumDeployedQueueSize
-                ? "plan"
-                : "default",
+          source: getQueueSizeLimitSource(environmentType, organization),
         },
       },
       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,
     })
   );

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

@@ -347,25 +347,11 @@ export default function Page() {
               title="Queued"
               value={environment.queued}
               suffix={
-                environment.queueSizeLimit ? (
-                  <span className="flex items-center gap-1">
-                    <span className="text-text-dimmed">/</span>
-                    <span
-                      className={getQueueUsageColorClass(
-                        environment.queued,
-                        environment.queueSizeLimit
-                      )}
-                    >
-                      {formatNumberCompact(environment.queueSizeLimit)}
-                    </span>
-                    <InfoIconTooltip
-                      content="Maximum pending runs across all queues in this environment"
-                      contentClassName="max-w-xs"
-                    />
-                  </span>
-                ) : env.paused && environment.queued > 0 ? (
-                  "paused"
-                ) : undefined
+                <QueuedSuffix
+                  queued={environment.queued}
+                  queueSizeLimit={environment.queueSizeLimit}
+                  isPaused={env.paused}
+                />
               }
               animate
               accessory={
@@ -1150,3 +1136,45 @@ function getQueueUsageColorClass(current: number, limit: number | null): string
   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/queues.server.ts

@@ -15,6 +15,22 @@ import type { RunEngine } from "~/v3/runEngine.server";
 import { env } from "~/env.server";
 import { tryCatch } from "@trigger.dev/core/v3";
 import { ServiceValidationError } from "~/v3/services/common.server";
+import { createCache, createLRUMemoryStore, DefaultStatefulContext, Namespace } from "@internal/cache";
+import { singleton } from "~/utils/singleton";
+
+// LRU cache for environment queue sizes to reduce Redis calls
+const queueSizeCache = singleton("queueSizeCache", () => {
+  const ctx = new DefaultStatefulContext();
+  const memory = createLRUMemoryStore(env.QUEUE_SIZE_CACHE_MAX_SIZE, "queue-size-cache");
+
+  return createCache({
+    queueSize: new Namespace<number>(ctx, {
+      stores: [memory],
+      fresh: env.QUEUE_SIZE_CACHE_TTL_MS,
+      stale: env.QUEUE_SIZE_CACHE_TTL_MS + 1000,
+    }),
+  });
+});
 
 /**
  * Extract the queue name from a queue option that may be:
@@ -49,7 +65,7 @@ export class DefaultQueueManager implements QueueManager {
   constructor(
     private readonly prisma: PrismaClientOrTransaction,
     private readonly engine: RunEngine
-  ) {}
+  ) { }
 
   async resolveQueueProperties(
     request: TriggerTaskRequest,
@@ -75,8 +91,7 @@ export class DefaultQueueManager implements QueueManager {
 
         if (!specifiedQueue) {
           throw new ServiceValidationError(
-            `Specified queue '${specifiedQueueName}' not found or not associated with locked version '${
-              lockedBackgroundWorker.version ?? "<unknown>"
+            `Specified queue '${specifiedQueueName}' not found or not associated with locked version '${lockedBackgroundWorker.version ?? "<unknown>"
             }'.`
           );
         }
@@ -98,8 +113,7 @@ export class DefaultQueueManager implements QueueManager {
 
         if (!lockedTask) {
           throw new ServiceValidationError(
-            `Task '${request.taskId}' not found on locked version '${
-              lockedBackgroundWorker.version ?? "<unknown>"
+            `Task '${request.taskId}' not found on locked version '${lockedBackgroundWorker.version ?? "<unknown>"
             }'.`
           );
         }
@@ -113,8 +127,7 @@ export class DefaultQueueManager implements QueueManager {
             version: lockedBackgroundWorker.version,
           });
           throw new ServiceValidationError(
-            `Default queue configuration for task '${request.taskId}' missing on locked version '${
-              lockedBackgroundWorker.version ?? "<unknown>"
+            `Default queue configuration for task '${request.taskId}' missing on locked version '${lockedBackgroundWorker.version ?? "<unknown>"
             }'.`
           );
         }
@@ -282,7 +295,7 @@ async function guardQueueSizeLimitsForEnv(
     return { isWithinLimits: true };
   }
 
-  const queueSize = await engine.lengthOfEnvQueue(environment);
+  const queueSize = await getCachedQueueSize(engine, environment);
   const projectedSize = queueSize + itemsToAdd;
 
   return {
@@ -291,3 +304,14 @@ async function guardQueueSizeLimitsForEnv(
     queueSize,
   };
 }
+
+async function getCachedQueueSize(
+  engine: RunEngine,
+  environment: AuthenticatedEnvironment
+): Promise<number> {
+  const result = await queueSizeCache.queueSize.swr(environment.id, async () => {
+    return engine.lengthOfEnvQueue(environment);
+  });
+
+  return result.val ?? 0;
+}

apps/webapp/app/v3/utils/queueLimits.server.ts

@@ -0,0 +1,51 @@
+import { RuntimeEnvironmentType } from "@trigger.dev/database";
+import { env } from "~/env.server";
+
+/**
+ * Organization fields needed for queue limit calculation.
+ */
+export type QueueLimitOrganization = {
+  maximumDevQueueSize: number | null;
+  maximumDeployedQueueSize: number | null;
+};
+
+/**
+ * Calculates the queue size limit for an environment based on its type and organization settings.
+ *
+ * Resolution order:
+ * 1. Organization-level override (set by billing sync or admin)
+ * 2. Environment variable fallback
+ * 3. null if neither is set
+ *
+ * @param environmentType - The type of the runtime environment
+ * @param organization - Organization with queue limit fields
+ * @returns The queue size limit, or null if unlimited
+ */
+export function getQueueSizeLimit(
+  environmentType: RuntimeEnvironmentType,
+  organization: QueueLimitOrganization
+): number | null {
+  if (environmentType === "DEVELOPMENT") {
+    return organization.maximumDevQueueSize ?? env.MAXIMUM_DEV_QUEUE_SIZE ?? null;
+  }
+
+  return organization.maximumDeployedQueueSize ?? env.MAXIMUM_DEPLOYED_QUEUE_SIZE ?? null;
+}
+
+/**
+ * Determines the source of the queue size limit for display purposes.
+ *
+ * @param environmentType - The type of the runtime environment
+ * @param organization - Organization with queue limit fields
+ * @returns "plan" if org has a value (typically set by billing), "default" if using env var fallback
+ */
+export function getQueueSizeLimitSource(
+  environmentType: RuntimeEnvironmentType,
+  organization: QueueLimitOrganization
+): "plan" | "default" {
+  if (environmentType === "DEVELOPMENT") {
+    return organization.maximumDevQueueSize !== null ? "plan" : "default";
+  }
+
+  return organization.maximumDeployedQueueSize !== null ? "plan" : "default";
+}