WIP

Author: ericallam

apps/webapp/app/components/runs/v3/BatchStatus.tsx

@@ -1,17 +1,27 @@
-import { CheckCircleIcon, XCircleIcon } from "@heroicons/react/20/solid";
+import {
+  CheckCircleIcon,
+  ExclamationTriangleIcon,
+  XCircleIcon,
+} from "@heroicons/react/20/solid";
 import type { BatchTaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
 
-export const allBatchStatuses = ["PENDING", "COMPLETED", "ABORTED"] as const satisfies Readonly<
-  Array<BatchTaskRunStatus>
->;
+export const allBatchStatuses = [
+  "PROCESSING",
+  "PENDING",
+  "COMPLETED",
+  "PARTIAL_FAILED",
+  "ABORTED",
+] as const satisfies Readonly<Array<BatchTaskRunStatus>>;
 
 const descriptions: Record<BatchTaskRunStatus, string> = {
+  PROCESSING: "The batch is being processed and runs are being created.",
   PENDING: "The batch has child runs that have not yet completed.",
   COMPLETED: "All the batch child runs have finished.",
-  ABORTED: "The batch was aborted because some child tasks could not be triggered.",
+  PARTIAL_FAILED: "Some runs failed to be created. Successfully created runs are still executing.",
+  ABORTED: "The batch was aborted because child tasks could not be triggered.",
 };
 
 export function descriptionForBatchStatus(status: BatchTaskRunStatus): string {
@@ -47,10 +57,14 @@ export function BatchStatusIcon({
   className: string;
 }) {
   switch (status) {
+    case "PROCESSING":
+      return <Spinner className={cn(batchStatusColor(status), className)} />;
     case "PENDING":
       return <Spinner className={cn(batchStatusColor(status), className)} />;
     case "COMPLETED":
       return <CheckCircleIcon className={cn(batchStatusColor(status), className)} />;
+    case "PARTIAL_FAILED":
+      return <ExclamationTriangleIcon className={cn(batchStatusColor(status), className)} />;
     case "ABORTED":
       return <XCircleIcon className={cn(batchStatusColor(status), className)} />;
     default: {
@@ -61,10 +75,14 @@ export function BatchStatusIcon({
 
 export function batchStatusColor(status: BatchTaskRunStatus): string {
   switch (status) {
+    case "PROCESSING":
+      return "text-blue-500";
     case "PENDING":
       return "text-pending";
     case "COMPLETED":
       return "text-success";
+    case "PARTIAL_FAILED":
+      return "text-warning";
     case "ABORTED":
       return "text-error";
     default: {
@@ -75,10 +93,14 @@ export function batchStatusColor(status: BatchTaskRunStatus): string {
 
 export function batchStatusTitle(status: BatchTaskRunStatus): string {
   switch (status) {
+    case "PROCESSING":
+      return "Processing";
     case "PENDING":
       return "In progress";
     case "COMPLETED":
       return "Completed";
+    case "PARTIAL_FAILED":
+      return "Partial failure";
     case "ABORTED":
       return "Aborted";
     default: {

apps/webapp/app/env.server.ts

@@ -931,6 +931,12 @@ const EnvironmentSchema = z
       .default(process.env.REDIS_TLS_DISABLED ?? "false"),
     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_CONSUMER_COUNT: z.coerce.number().int().optional(),
+    BATCH_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().optional(),
+
     ADMIN_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
     ADMIN_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
     ADMIN_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),

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

@@ -195,7 +195,7 @@ WHERE
           throw new Error(`Environment not found for Batch ${batch.id}`);
         }
 
-        const hasFinished = batch.status !== "PENDING";
+        const hasFinished = batch.status !== "PENDING" && batch.status !== "PROCESSING";
 
         return {
           id: batch.id,

apps/webapp/app/routes/api.v1.batches.$batchId.ts

@@ -18,6 +18,9 @@ export const loader = createLoaderApiRoute(
           friendlyId: params.batchId,
           runtimeEnvironmentId: auth.environment.id,
         },
+        include: {
+          errors: true,
+        },
       });
     },
     authorization: {
@@ -35,6 +38,18 @@ export const loader = createLoaderApiRoute(
       updatedAt: batch.updatedAt,
       runCount: batch.runCount,
       runs: batch.runIds,
+      // Include error details for PARTIAL_FAILED batches
+      successfulRunCount: batch.successfulRunCount ?? undefined,
+      failedRunCount: batch.failedRunCount ?? undefined,
+      errors:
+        batch.errors.length > 0
+          ? batch.errors.map((err) => ({
+              index: err.index,
+              taskIdentifier: err.taskIdentifier,
+              error: err.error,
+              errorCode: err.errorCode ?? undefined,
+            }))
+          : undefined,
     });
   }
 );

apps/webapp/app/routes/api.v2.batches.$batchId.ts

@@ -18,6 +18,9 @@ export const loader = createLoaderApiRoute(
           friendlyId: params.batchId,
           runtimeEnvironmentId: auth.environment.id,
         },
+        include: {
+          errors: true,
+        },
       });
     },
     authorization: {
@@ -35,6 +38,18 @@ export const loader = createLoaderApiRoute(
       updatedAt: batch.updatedAt,
       runCount: batch.runCount,
       runs: batch.runIds,
+      // Include error details for PARTIAL_FAILED batches
+      successfulRunCount: batch.successfulRunCount ?? undefined,
+      failedRunCount: batch.failedRunCount ?? undefined,
+      errors:
+        batch.errors.length > 0
+          ? batch.errors.map((err) => ({
+              index: err.index,
+              taskIdentifier: err.taskIdentifier,
+              error: err.error,
+              errorCode: err.errorCode ?? undefined,
+            }))
+          : undefined,
     });
   }
 );

apps/webapp/app/routes/api.v2.tasks.batch.ts

@@ -7,6 +7,7 @@ import {
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { RunEngineBatchTriggerService } from "~/runEngine/services/batchTrigger.server";
+import { RunEngineBatchTriggerServiceV2 } from "~/runEngine/services/batchTriggerV2.server";
 import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
@@ -20,6 +21,7 @@ import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
+import { engine } from "~/v3/runEngine.server";
 
 const { action, loader } = createActionApiRoute(
   {
@@ -110,6 +112,68 @@ const { action, loader } = createActionApiRoute(
       ? { traceparent, tracestate }
       : { external: { traceparent, tracestate } };
 
+    // Use the new V2 service with DRR-based BatchQueue if enabled, otherwise fall back to V1
+    const useBatchQueueV2 = engine.isBatchQueueEnabled();
+
+    logger.debug("Batch trigger service selection", {
+      useBatchQueueV2,
+      batchProcessingStrategy,
+    });
+
+    if (useBatchQueueV2) {
+      // V2: Uses Redis-based BatchQueue with DRR fair scheduling
+      const serviceV2 = new RunEngineBatchTriggerServiceV2();
+
+      serviceV2.onBatchTaskRunCreated.attachOnce(async (batch) => {
+        await saveRequestIdempotency(requestIdempotencyKey, "batch-trigger", batch.id);
+      });
+
+      try {
+        const batch = await serviceV2.call(authentication.environment, body, {
+          triggerVersion: triggerVersion ?? undefined,
+          traceContext,
+          spanParentAsLink: spanParentAsLink === 1,
+          oneTimeUseToken,
+          realtimeStreamsVersion: determineRealtimeStreamsVersion(
+            realtimeStreamsVersion ?? undefined
+          ),
+          idempotencyKey: requestIdempotencyKey ?? undefined,
+        });
+
+        const $responseHeaders = await responseHeaders(
+          batch,
+          authentication.environment,
+          triggerClient
+        );
+
+        return json(batch, {
+          status: 202,
+          headers: $responseHeaders,
+        });
+      } catch (error) {
+        logger.error("Batch trigger error (V2)", {
+          error: {
+            message: (error as Error).message,
+            stack: (error as Error).stack,
+          },
+        });
+
+        if (error instanceof ServiceValidationError) {
+          return json({ error: error.message }, { status: 422 });
+        } else if (error instanceof OutOfEntitlementError) {
+          return json({ error: error.message }, { status: 422 });
+        } else if (error instanceof Error) {
+          return json(
+            { error: error.message },
+            { status: 500, headers: { "x-should-retry": "false" } }
+          );
+        }
+
+        return json({ error: "Something went wrong" }, { status: 500 });
+      }
+    }
+
+    // V1: Legacy batch trigger service
     const service = new RunEngineBatchTriggerService(batchProcessingStrategy ?? undefined);
 
     service.onBatchTaskRunCreated.attachOnce(async (batch) => {