handle batch failures more reliably

Author: ericallam

apps/webapp/app/env.server.ts

@@ -941,8 +941,8 @@ const EnvironmentSchema = z
     BATCH_TRIGGER_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
     // BatchQueue DRR settings (Run Engine v2)
-    BATCH_QUEUE_DRR_QUANTUM: z.coerce.number().int().optional(),
-    BATCH_QUEUE_MAX_DEFICIT: z.coerce.number().int().optional(),
+    BATCH_QUEUE_DRR_QUANTUM: z.coerce.number().int().default(5),
+    BATCH_QUEUE_MAX_DEFICIT: z.coerce.number().int().default(50),
     BATCH_QUEUE_CONSUMER_COUNT: z.coerce.number().int().optional(),
     BATCH_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().optional(),
     // Global rate limit: max items processed per second across all consumers

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

@@ -4,6 +4,7 @@ import { z } from "zod";
 import { env } from "~/env.server";
 import { RateLimiterConfig } from "~/services/authorizationRateLimitMiddleware.server";
 import { createRedisRateLimitClient, Duration, RateLimiter } from "~/services/rateLimiter.server";
+import { singleton } from "~/utils/singleton";
 
 const BatchLimitsConfig = z.object({
   processingConcurrency: z.number().int().default(env.BATCH_CONCURRENCY_LIMIT_DEFAULT),
@@ -14,7 +15,9 @@ const BatchLimitsConfig = z.object({
  */
 export type BatchLimitsConfig = z.infer<typeof BatchLimitsConfig>;
 
-function createOrganizationRateLimiter(organization: Organization): RateLimiter {
+const batchLimitsRedisClient = singleton("batchLimitsRedisClient", createBatchLimitsRedisClient);
+
+function createBatchLimitsRedisClient() {
   const redisClient = createRedisRateLimitClient({
     port: env.RATE_LIMIT_REDIS_PORT,
     host: env.RATE_LIMIT_REDIS_HOST,
@@ -24,6 +27,10 @@ function createOrganizationRateLimiter(organization: Organization): RateLimiter
     clusterMode: env.RATE_LIMIT_REDIS_CLUSTER_MODE_ENABLED === "1",
   });
 
+  return redisClient;
+}
+
+function createOrganizationRateLimiter(organization: Organization): RateLimiter {
   const limiterConfig = resolveBatchRateLimitConfig(organization.batchRateLimitConfig);
 
   const limiter =
@@ -38,7 +45,7 @@ function createOrganizationRateLimiter(organization: Organization): RateLimiter
       : Ratelimit.slidingWindow(limiterConfig.tokens, limiterConfig.window);
 
   return new RateLimiter({
-    redisClient,
+    redisClient: batchLimitsRedisClient,
     keyPrefix: "ratelimit:batch",
     limiter,
     logSuccess: false,

apps/webapp/app/runEngine/services/streamBatchItems.server.ts

@@ -273,9 +273,7 @@ export function createNdjsonParserStream(
  * Convert a ReadableStream into an AsyncIterable.
  * Useful for processing streams with for-await-of loops.
  */
-export async function* streamToAsyncIterable<T>(
-  stream: ReadableStream<T>
-): AsyncIterable<T> {
+export async function* streamToAsyncIterable<T>(stream: ReadableStream<T>): AsyncIterable<T> {
   const reader = stream.getReader();
   try {
     while (true) {
@@ -287,4 +285,3 @@ export async function* streamToAsyncIterable<T>(
     reader.releaseLock();
   }
 }
-

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -6,7 +6,7 @@ import { RunId } from "@trigger.dev/core/v3/isomorphic";
 import { BatchTaskRunStatus, Prisma } from "@trigger.dev/database";
 import { $replica, prisma } from "~/db.server";
 import { env } from "~/env.server";
-import { findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
+import { findEnvironmentById, findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
@@ -659,21 +659,26 @@ export function setupBatchQueueCallbacks() {
       },
       async (span) => {
         try {
+          const environment = await findEnvironmentById(meta.environmentId);
+
+          if (!environment) {
+            span.setAttribute("batch.result.error", "Environment not found");
+            span.end();
+            return {
+              success: false as const,
+              error: "Environment not found",
+              errorCode: "ENVIRONMENT_NOT_FOUND",
+            };
+          }
+
           const triggerTaskService = new TriggerTaskService();
 
           // Normalize payload - for application/store (R2 paths), this passes through as-is
           const payload = normalizePayload(item.payload, item.payloadType);
 
           const result = await triggerTaskService.call(
             item.task,
-            {
-              id: meta.environmentId,
-              type: meta.environmentType,
-              organizationId: meta.organizationId,
-              projectId: meta.projectId,
-              organization: { id: meta.organizationId },
-              project: { id: meta.projectId },
-            } as AuthenticatedEnvironment,
+            environment,
             {
               payload,
               options: {

internal-packages/BATCH_TRIGGER_LIMITS.md

@@ -61,9 +61,7 @@ export class BatchQueue {
   private itemQueueTimeHistogram?: Histogram;
 
   constructor(private options: BatchQueueOptions) {
-    this.logger = options.logger
-      ? new Logger("BatchQueue", "info")
-      : new Logger("BatchQueue", "info");
+    this.logger = options.logger ?? new Logger("BatchQueue", options.logLevel ?? "info");
     this.defaultConcurrency = options.defaultConcurrency ?? 10;
 
     // Initialize metrics if meter is provided

internal-packages/run-engine/src/batch-queue/index.ts

@@ -1,5 +1,8 @@
 import { z } from "zod";
 import { RuntimeEnvironmentType } from "@trigger.dev/database";
+import { Logger, LogLevel } from "@trigger.dev/core/logger";
+import { GlobalRateLimiter } from "@trigger.dev/redis-worker";
+import { Meter, Tracer } from "@internal/tracing";
 
 // ============================================================================
 // Batch Item Schemas
@@ -208,18 +211,14 @@ export type BatchQueueOptions = {
    * Optional global rate limiter to limit processing across all consumers.
    * When configured, limits the max items/second processed globally.
    */
-  globalRateLimiter?: import("@trigger.dev/redis-worker").GlobalRateLimiter;
+  globalRateLimiter?: GlobalRateLimiter;
   /** Logger instance */
-  logger?: {
-    debug: (message: string, context?: Record<string, unknown>) => void;
-    info: (message: string, context?: Record<string, unknown>) => void;
-    warn: (message: string, context?: Record<string, unknown>) => void;
-    error: (message: string, context?: Record<string, unknown>) => void;
-  };
+  logger?: Logger;
+  logLevel?: LogLevel;
   /** OpenTelemetry tracer for distributed tracing */
-  tracer?: import("@internal/tracing").Tracer;
+  tracer?: Tracer;
   /** OpenTelemetry meter for metrics */
-  meter?: import("@internal/tracing").Meter;
+  meter?: Meter;
 };
 
 /**