restructure the batch queue callbacks to prevent circular import

Author: ericallam

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/routes/api.v3.batches.$batchId.items.ts

@@ -58,16 +58,6 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: authResult.error }, { status: 401 });
   }
 
-  // Verify BatchQueue is enabled
-  if (!engine.isBatchQueueEnabled()) {
-    return json(
-      {
-        error: "Streaming batch API is not available. BatchQueue is not enabled.",
-      },
-      { status: 503 }
-    );
-  }
-
   // Get the request body stream
   const body = request.body;
   if (!body) {

apps/webapp/app/routes/api.v3.batches.ts

@@ -59,16 +59,6 @@ const { action, loader } = createActionApiRoute(
       );
     }
 
-    // Verify BatchQueue is enabled
-    if (!engine.isBatchQueueEnabled()) {
-      return json(
-        {
-          error: "Streaming batch API is not available. BatchQueue is not enabled.",
-        },
-        { status: 503 }
-      );
-    }
-
     const {
       "trigger-version": triggerVersion,
       "x-trigger-span-parent-as-link": spanParentAsLink,

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

@@ -1,15 +1,15 @@
+import type { InitializeBatchOptions } from "@internal/run-engine";
 import { type CreateBatchRequestBody, type CreateBatchResponse } from "@trigger.dev/core/v3";
 import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic";
 import { type BatchTaskRun, Prisma } from "@trigger.dev/database";
-import type { InitializeBatchOptions } from "@internal/run-engine";
 import { Evt } from "evt";
 import { prisma, type PrismaClientOrTransaction } from "~/db.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { DefaultQueueManager } from "../concerns/queues.server";
-import { DefaultTriggerTaskValidator } from "../validators/triggerTaskValidator";
 import { ServiceValidationError, WithRunEngine } from "../../v3/services/baseService.server";
 import { BatchRateLimitExceededError, getBatchLimits } from "../concerns/batchLimits.server";
+import { DefaultQueueManager } from "../concerns/queues.server";
+import { DefaultTriggerTaskValidator } from "../validators/triggerTaskValidator";
 
 export type CreateBatchServiceOptions = {
   triggerVersion?: string;

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

@@ -1,16 +1,12 @@
-import { RunEngine, type CompleteBatchResult } from "@internal/run-engine";
-import { BatchTaskRunStatus, Prisma } from "@trigger.dev/database";
-import { SpanKind } from "@opentelemetry/api";
+import { RunEngine } from "@internal/run-engine";
 import { $replica, prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { createBatchGlobalRateLimiter } from "~/runEngine/concerns/batchGlobalRateLimiter.server";
+import { logger } from "~/services/logger.server";
 import { defaultMachine, getCurrentPlan } from "~/services/platform.v3.server";
 import { singleton } from "~/utils/singleton";
 import { allMachines } from "./machinePresets.server";
 import { meter, tracer } from "./tracer.server";
-import { logger } from "~/services/logger.server";
-import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { TriggerTaskService } from "./services/triggerTask.server";
 
 export const engine = singleton("RunEngine", createRunEngine);
 
@@ -188,202 +184,5 @@ function createRunEngine() {
     },
   });
 
-  // Set up BatchQueue callbacks if enabled
-  if (engine.isBatchQueueEnabled()) {
-    setupBatchQueueCallbacks(engine);
-  }
-
   return engine;
 }
-
-/**
- * Normalize the payload from BatchQueue.
- *
- * Handles different payload types:
- * - "application/store": Already offloaded to R2, payload is the path - pass through as-is
- * - "application/json": May be a pre-serialized JSON string - parse to avoid double-stringification
- * - Other types: Pass through as-is
- *
- * @param payload - The raw payload from the batch item
- * @param payloadType - The payload type (e.g., "application/json", "application/store")
- */
-function normalizePayload(payload: unknown, payloadType?: string): unknown {
-  // For non-JSON payloads (including application/store for R2-offloaded payloads),
-  // return as-is - no normalization needed
-  if (payloadType !== "application/json" && payloadType !== undefined) {
-    return payload;
-  }
-
-  // For JSON payloads, if payload is a string, try to parse it
-  // This handles pre-serialized JSON from the SDK
-  if (typeof payload === "string") {
-    try {
-      return JSON.parse(payload);
-    } catch {
-      // If it's not valid JSON, return as-is
-      return payload;
-    }
-  }
-
-  return payload;
-}
-
-/**
- * Set up the BatchQueue processing callbacks.
- * These handle creating runs from batch items and completing batches.
- *
- * Payload handling:
- * - If payloadType is "application/store", the payload is an R2 path (already offloaded)
- * - DefaultPayloadProcessor in TriggerTaskService will pass it through without re-offloading
- * - The run engine will download from R2 when the task executes
- */
-function setupBatchQueueCallbacks(engine: RunEngine) {
-  // Item processing callback - creates a run for each batch item
-  engine.setBatchProcessItemCallback(async ({ batchId, friendlyId, itemIndex, item, meta }) => {
-    return tracer.startActiveSpan(
-      "batch.processItem",
-      {
-        kind: SpanKind.INTERNAL,
-        attributes: {
-          "batch.id": friendlyId,
-          "batch.item_index": itemIndex,
-          "batch.task": item.task,
-          "batch.environment_id": meta.environmentId,
-          "batch.parent_run_id": meta.parentRunId ?? "",
-        },
-      },
-      async (span) => {
-        try {
-          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,
-            {
-              payload,
-              options: {
-                ...(item.options as Record<string, unknown>),
-                payloadType: item.payloadType,
-                parentRunId: meta.parentRunId,
-                resumeParentOnCompletion: meta.resumeParentOnCompletion,
-                parentBatch: batchId,
-              },
-            },
-            {
-              triggerVersion: meta.triggerVersion,
-              traceContext: meta.traceContext as Record<string, unknown> | undefined,
-              spanParentAsLink: meta.spanParentAsLink,
-              batchId,
-              batchIndex: itemIndex,
-              skipChecks: true, // Already validated at batch level
-              realtimeStreamsVersion: meta.realtimeStreamsVersion,
-            },
-            "V2"
-          );
-
-          if (result) {
-            span.setAttribute("batch.result.run_id", result.run.friendlyId);
-            span.end();
-            return { success: true as const, runId: result.run.friendlyId };
-          } else {
-            span.setAttribute("batch.result.error", "TriggerTaskService returned undefined");
-            span.end();
-            return {
-              success: false as const,
-              error: "TriggerTaskService returned undefined",
-              errorCode: "TRIGGER_FAILED",
-            };
-          }
-        } catch (error) {
-          span.setAttribute(
-            "batch.result.error",
-            error instanceof Error ? error.message : String(error)
-          );
-          span.recordException(error instanceof Error ? error : new Error(String(error)));
-          span.end();
-          return {
-            success: false as const,
-            error: error instanceof Error ? error.message : String(error),
-            errorCode: "TRIGGER_ERROR",
-          };
-        }
-      }
-    );
-  });
-
-  // Batch completion callback - updates Postgres with results
-  engine.setBatchCompletionCallback(async (result: CompleteBatchResult) => {
-    const { batchId, runIds, successfulRunCount, failedRunCount, failures } = result;
-
-    // Determine final status
-    let status: BatchTaskRunStatus;
-    if (failedRunCount > 0 && successfulRunCount === 0) {
-      status = "ABORTED";
-    } else if (failedRunCount > 0) {
-      status = "PARTIAL_FAILED";
-    } else {
-      status = "PENDING"; // All runs created, waiting for completion
-    }
-
-    try {
-      // Update BatchTaskRun
-      await prisma.batchTaskRun.update({
-        where: { id: batchId },
-        data: {
-          status,
-          runIds,
-          successfulRunCount,
-          failedRunCount,
-          completedAt: status === "ABORTED" ? new Date() : undefined,
-          processingCompletedAt: new Date(),
-        },
-      });
-
-      // Create error records if there were failures
-      if (failures.length > 0) {
-        for (const failure of failures) {
-          await prisma.batchTaskRunError.create({
-            data: {
-              batchTaskRunId: batchId,
-              index: failure.index,
-              taskIdentifier: failure.taskIdentifier,
-              payload: failure.payload,
-              options: failure.options as Prisma.InputJsonValue | undefined,
-              error: failure.error,
-              errorCode: failure.errorCode,
-            },
-          });
-        }
-      }
-
-      // Try to complete the batch (handles waitpoint completion if all runs are done)
-      if (status !== "ABORTED") {
-        await engine.tryCompleteBatch({ batchId });
-      }
-
-      logger.info("Batch completion handled", {
-        batchId,
-        status,
-        successfulRunCount,
-        failedRunCount,
-      });
-    } catch (error) {
-      logger.error("Failed to handle batch completion", {
-        batchId,
-        error: error instanceof Error ? error.message : String(error),
-      });
-    }
-  });
-
-  logger.info("BatchQueue callbacks configured");
-}