feat(engine): Batch trigger reloaded (#2779)

New batch trigger system with larger payloads, streaming ingestion,
larger batch sizes, and a fair processing system.

This PR introduces a new `FairQueue` abstraction inspired by our own
`RunQueue` that enables multi-tenant fair queueing with concurrency
limits. The new `BatchQueue` is built on top of the `FairQueue`, and
handles processing Batch triggers in a fair manner with per-environment
concurrency limits defined per-org. Additionally, there is a global
concurrency limit to prevent the BatchQueue system from creating too
many runs too quickly, which can cause downstream issues.

For this new BatchQueue system we have a completely new batch trigger
creation and ingestion system. Previously this was a single endpoint
with a single JSON body that defined details about the batch as well as
all the items in the batch.

We're introducing a two-phase batch trigger ingestion system. In the
first phase, the BatchTaskRun record is created (and possibly rate
limited). The second phase is another endpoint that accepts an NDJSON
body with each line being a single item/run with payload and options.

At ingestion time all items are added to a queue, in order, and then
processed by the BatchQueue system.

## New batch trigger rate limits

This PR implements a new batch trigger specific rate limit, configured
on the `Organization.batchRateLimitConfig` column, and defaults using
these environment variables:

- `BATCH_RATE_LIMIT_REFILL_RATE` defaults to 10
- `BATCH_RATE_LIMIT_REFILL_INTERVAL` the duration interval, defaults to
`"10s"`
- `BATCH_RATE_LIMIT_MAX` defaults to 1200

This rate limiter is scoped to the environment ID and controls how many
runs can be submitted via batch triggers per interval. The SDK handles
the retrying side.

## Batch queue concurrency limits

The new column `Organization.batchQueueConcurrencyConfig` now defines an
org specific `processingConcurrency` value, with a backup of the env var
`BATCH_CONCURRENCY_LIMIT_DEFAULT` which defaults to 10. This controls
how many batch queue items are processed concurrently per environment.

There is also a global rate limit for the batch queue set via the
`BATCH_QUEUE_GLOBAL_RATE_LIMIT` which defaults to being disabled. If
set, the entire batch queue system won't process more than
`BATCH_QUEUE_GLOBAL_RATE_LIMIT` items per second. This allows
controlling the maximum number of runs created per second via batch
triggers.

## Batch trigger settings

- `STREAMING_BATCH_MAX_ITEMS` controls the maximum number of items in a
single batch
- `STREAMING_BATCH_ITEM_MAXIMUM_SIZE` controls the maximum size of each
item in a batch
- `BATCH_CONCURRENCY_DEFAULT_CONCURRENCY` controls the default
environment concurrency
- `BATCH_QUEUE_DRR_QUANTUM` how many credits each environment gets each
round for the DRR scheduler
- `BATCH_QUEUE_MAX_DEFICIT` the maximum deficit for the DRR scheduler
- `BATCH_QUEUE_CONSUMER_COUNT` how many queue consumers to run
- `BATCH_QUEUE_CONSUMER_INTERVAL_MS` how frequently they poll for items
in the queue

### Configuration Recommendations by Use Case

**High-throughput priority (fairness acceptable at 0.98+):**

```env
BATCH_QUEUE_DRR_QUANTUM=25
BATCH_QUEUE_MAX_DEFICIT=100
BATCH_QUEUE_CONSUMER_COUNT=10
BATCH_QUEUE_CONSUMER_INTERVAL_MS=50
BATCH_CONCURRENCY_DEFAULT_CONCURRENCY=25
```

**Strict fairness priority (throughput can be lower):**

```env
BATCH_QUEUE_DRR_QUANTUM=5
BATCH_QUEUE_MAX_DEFICIT=25
BATCH_QUEUE_CONSUMER_COUNT=3
BATCH_QUEUE_CONSUMER_INTERVAL_MS=100
BATCH_CONCURRENCY_DEFAULT_CONCURRENCY=5
```

Author: ericallam

.changeset/fluffy-crews-rhyme.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+feat: Support for new batch trigger system

.env.example

@@ -85,4 +85,10 @@ POSTHOG_PROJECT_KEY=
 # These control the server-side internal telemetry
 # INTERNAL_OTEL_TRACE_EXPORTER_URL=<URL to send traces to>
 # INTERNAL_OTEL_TRACE_LOGGING_ENABLED=1
-# INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED=0,
+# INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED=0
+
+# Enable local observability stack (requires `pnpm run docker` to start otel-collector)
+# Uncomment these to send metrics to the local Prometheus via OTEL Collector:
+# INTERNAL_OTEL_METRIC_EXPORTER_ENABLED=1
+# INTERNAL_OTEL_METRIC_EXPORTER_URL=http://localhost:4318/v1/metrics
+# INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS=15000

.gitignore

@@ -62,4 +62,5 @@ apps/**/public/build
 /packages/trigger-sdk/src/package.json
 /packages/python/src/package.json
 .claude
-.mcp.log
+.mcp.log
+.cursor/debug.log

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/entry.server.tsx

@@ -1,22 +1,28 @@
-import {
-  createReadableStreamFromReadable,
-  type DataFunctionArgs,
-  type EntryContext,
-} from "@remix-run/node"; // or cloudflare/deno
+import { createReadableStreamFromReadable, type EntryContext } from "@remix-run/node"; // or cloudflare/deno
 import { RemixServer } from "@remix-run/react";
+import { wrapHandleErrorWithSentry } from "@sentry/remix";
 import { parseAcceptLanguage } from "intl-parse-accept-language";
 import isbot from "isbot";
 import { renderToPipeableStream } from "react-dom/server";
 import { PassThrough } from "stream";
 import * as Worker from "~/services/worker.server";
+import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import {
   OperatingSystemContextProvider,
   OperatingSystemPlatform,
 } from "./components/primitives/OperatingSystemProvider";
+import { Prisma } from "./db.server";
+import { env } from "./env.server";
+import { eventLoopMonitor } from "./eventLoopMonitor.server";
+import { logger } from "./services/logger.server";
+import { resourceMonitor } from "./services/resourceMonitor.server";
 import { singleton } from "./utils/singleton";
-import { bootstrap } from "./bootstrap";
-import { wrapHandleErrorWithSentry } from "@sentry/remix";
+import { remoteBuildsEnabled } from "./v3/remoteImageBuilder.server";
+import {
+  registerRunEngineEventBusHandlers,
+  setupBatchQueueCallbacks,
+} from "./v3/runEngineHandlers.server";
 
 const ABORT_DELAY = 30000;
 
@@ -228,19 +234,13 @@ process.on("uncaughtException", (error, origin) => {
 });
 
 singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers);
+singleton("SetupBatchQueueCallbacks", setupBatchQueueCallbacks);
 
 export { apiRateLimiter } from "./services/apiRateLimit.server";
 export { engineRateLimiter } from "./services/engineRateLimit.server";
+export { runWithHttpContext } from "./services/httpAsyncStorage.server";
 export { socketIo } from "./v3/handleSocketIo.server";
 export { wss } from "./v3/handleWebsockets.server";
-export { runWithHttpContext } from "./services/httpAsyncStorage.server";
-import { eventLoopMonitor } from "./eventLoopMonitor.server";
-import { env } from "./env.server";
-import { logger } from "./services/logger.server";
-import { Prisma } from "./db.server";
-import { registerRunEngineEventBusHandlers } from "./v3/runEngineHandlers.server";
-import { remoteBuildsEnabled } from "./v3/remoteImageBuilder.server";
-import { resourceMonitor } from "./services/resourceMonitor.server";
 
 if (env.EVENT_LOOP_MONITOR_ENABLED === "1") {
   eventLoopMonitor.enable();

apps/webapp/app/env.server.ts

@@ -528,6 +528,7 @@ const EnvironmentSchema = z
     MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
     MAXIMUM_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(10_000),
     TASK_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().default(524_288), // 512KB
+    BATCH_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().optional(), // Defaults to TASK_PAYLOAD_OFFLOAD_THRESHOLD if not set
     TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728), // 3MB
     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
@@ -537,6 +538,14 @@ const EnvironmentSchema = z
     MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
     MAX_BATCH_AND_WAIT_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
 
+    // 2-phase batch API settings
+    STREAMING_BATCH_MAX_ITEMS: z.coerce.number().int().default(1_000), // Max items in streaming batch
+    STREAMING_BATCH_ITEM_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728),
+    BATCH_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(100),
+    BATCH_RATE_LIMIT_MAX: z.coerce.number().int().default(1200),
+    BATCH_RATE_LIMIT_REFILL_INTERVAL: z.string().default("10s"),
+    BATCH_CONCURRENCY_LIMIT_DEFAULT: z.coerce.number().int().default(10),
+
     REALTIME_STREAM_VERSION: z.enum(["v1", "v2"]).default("v1"),
     REALTIME_STREAM_MAX_LENGTH: z.coerce.number().int().default(1000),
     REALTIME_STREAM_TTL: z.coerce
@@ -931,6 +940,17 @@ 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().default(25),
+    BATCH_QUEUE_MAX_DEFICIT: z.coerce.number().int().default(100),
+    BATCH_QUEUE_CONSUMER_COUNT: z.coerce.number().int().default(3),
+    BATCH_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(50),
+    // Global rate limit: max items processed per second across all consumers
+    // If not set, no global rate limiting is applied
+    BATCH_QUEUE_GLOBAL_RATE_LIMIT: z.coerce.number().int().positive().optional(),
+    // Processing concurrency: max concurrent batch items being processed per environment
+    BATCH_CONCURRENCY_DEFAULT_CONCURRENCY: z.coerce.number().int().default(1),
+
     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/presenters/v3/BatchPresenter.server.ts

@@ -0,0 +1,122 @@
+import { type BatchTaskRunStatus } from "@trigger.dev/database";
+import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
+import { engine } from "~/v3/runEngine.server";
+import { BasePresenter } from "./basePresenter.server";
+
+type BatchPresenterOptions = {
+  environmentId: string;
+  batchId: string;
+  userId?: string;
+};
+
+export type BatchPresenterData = Awaited<ReturnType<BatchPresenter["call"]>>;
+
+export class BatchPresenter extends BasePresenter {
+  public async call({ environmentId, batchId, userId }: BatchPresenterOptions) {
+    const batch = await this._replica.batchTaskRun.findFirst({
+      select: {
+        id: true,
+        friendlyId: true,
+        status: true,
+        runCount: true,
+        batchVersion: true,
+        createdAt: true,
+        updatedAt: true,
+        completedAt: true,
+        processingStartedAt: true,
+        processingCompletedAt: true,
+        successfulRunCount: true,
+        failedRunCount: true,
+        idempotencyKey: true,
+        runtimeEnvironment: {
+          select: {
+            id: true,
+            type: true,
+            slug: true,
+            orgMember: {
+              select: {
+                user: {
+                  select: {
+                    id: true,
+                    name: true,
+                    displayName: true,
+                  },
+                },
+              },
+            },
+          },
+        },
+        errors: {
+          select: {
+            id: true,
+            index: true,
+            taskIdentifier: true,
+            error: true,
+            errorCode: true,
+            createdAt: true,
+          },
+          orderBy: {
+            index: "asc",
+          },
+        },
+      },
+      where: {
+        runtimeEnvironmentId: environmentId,
+        friendlyId: batchId,
+      },
+    });
+
+    if (!batch) {
+      throw new Error("Batch not found");
+    }
+
+    const hasFinished = batch.status !== "PENDING" && batch.status !== "PROCESSING";
+    const isV2 = batch.batchVersion === "runengine:v2";
+
+    // For v2 batches in PROCESSING state, get live progress from Redis
+    // This provides real-time updates without waiting for the batch to complete
+    let liveSuccessCount = batch.successfulRunCount ?? 0;
+    let liveFailureCount = batch.failedRunCount ?? 0;
+
+    if (isV2 && batch.status === "PROCESSING") {
+      const liveProgress = await engine.getBatchQueueProgress(batch.id);
+      if (liveProgress) {
+        liveSuccessCount = liveProgress.successCount;
+        liveFailureCount = liveProgress.failureCount;
+      }
+    }
+
+    return {
+      id: batch.id,
+      friendlyId: batch.friendlyId,
+      status: batch.status as BatchTaskRunStatus,
+      runCount: batch.runCount,
+      batchVersion: batch.batchVersion,
+      isV2,
+      createdAt: batch.createdAt.toISOString(),
+      updatedAt: batch.updatedAt.toISOString(),
+      completedAt: batch.completedAt?.toISOString(),
+      processingStartedAt: batch.processingStartedAt?.toISOString(),
+      processingCompletedAt: batch.processingCompletedAt?.toISOString(),
+      finishedAt: batch.completedAt
+        ? batch.completedAt.toISOString()
+        : hasFinished
+          ? batch.updatedAt.toISOString()
+          : undefined,
+      hasFinished,
+      successfulRunCount: liveSuccessCount,
+      failedRunCount: liveFailureCount,
+      idempotencyKey: batch.idempotencyKey,
+      environment: displayableEnvironment(batch.runtimeEnvironment, userId),
+      errors: batch.errors.map((error) => ({
+        id: error.id,
+        index: error.index,
+        taskIdentifier: error.taskIdentifier,
+        error: error.error,
+        errorCode: error.errorCode,
+        createdAt: error.createdAt.toISOString(),
+      })),
+    };
+  }
+}
+