fix(run-engine,webapp): abort unsealed batches so blocked parents resume

batchTriggerAndWait blocks the parent on a waitpoint as soon as the batch is
created, but the batch is only sealed once all its items finish streaming. If
the item stream fails completely (rate limit, request timeout, crash), the
batch stayed PENDING with no runs and the parent waited on its waitpoint
forever.

A seal-timeout reaper, scheduled when the batch is created, now aborts a batch
that is still unsealed after BATCH_SEAL_TIMEOUT_MS (default 30 minutes) and
completes the parent's waitpoint with an error so it resumes. The reaper is
idempotent and no-ops if the stream sealed the batch in the meantime.

Author: matt-aitken

.server-changes/recover-stranded-batch-trigger-and-wait.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Recover `batchTriggerAndWait` parents that previously hung forever when a batch's item stream never completed. Batches left unsealed past a timeout are now aborted and the waiting parent resumes with an error instead of waiting indefinitely.

apps/webapp/app/env.server.ts

@@ -768,6 +768,13 @@ const EnvironmentSchema = z
     // in-flight memory per request ≈ this × STREAMING_BATCH_ITEM_MAXIMUM_SIZE,
     // so raise with care. Set to 1 for fully sequential ingestion.
     STREAMING_BATCH_INGEST_CONCURRENCY: z.coerce.number().int().positive().default(10),
+    // Seal-timeout reaper: if a batch's Phase 2 item stream never seals the batch
+    // (rate-limit, request timeout, crash), abort it after this delay and resume any
+    // blocked parent (batchTriggerAndWait) with an error instead of hanging forever.
+    // Must exceed the worst-case legitimate time-to-seal: the SDK retries the stream
+    // up to maxAttempts (5) times, each attempt bounded by the server request timeout
+    // (~300s), so the floor is ~5 × requestTimeout. Default 30m leaves headroom.
+    BATCH_SEAL_TIMEOUT_MS: z.coerce.number().int().positive().default(1_800_000),
     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"),

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

@@ -4,6 +4,7 @@ import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic";
 import { type BatchTaskRun, Prisma } from "@trigger.dev/database";
 import { Evt } from "evt";
 import { prisma, type PrismaClientOrTransaction } from "~/db.server";
+import { env } from "~/env.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { ServiceValidationError, WithRunEngine } from "../../v3/services/baseService.server";
@@ -149,6 +150,14 @@ export class CreateBatchService extends WithRunEngine {
 
           await this._engine.initializeBatch(initOptions);
 
+          // Guard the 2-phase gap: if Phase 2 never seals this batch, the reaper
+          // aborts it after the timeout and resumes any blocked parent with an
+          // error instead of leaving it suspended forever.
+          await this._engine.scheduleExpireBatch({
+            batchId: batch.id,
+            availableAt: new Date(Date.now() + env.BATCH_SEAL_TIMEOUT_MS),
+          });
+
           logger.info("Batch created", {
             batchId: friendlyId,
             runCount: body.runCount,

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

@@ -263,6 +263,9 @@ export class RunEngine {
         tryCompleteBatch: async ({ payload }) => {
           await this.batchSystem.performCompleteBatch({ batchId: payload.batchId });
         },
+        expireBatch: async ({ payload }) => {
+          await this.batchSystem.expireBatch({ batchId: payload.batchId });
+        },
         continueRunIfUnblocked: async ({ payload }) => {
           await this.waitpointSystem.continueRunIfUnblocked({
             runId: payload.runId,
@@ -1640,6 +1643,29 @@ export class RunEngine {
     return this.batchSystem.scheduleCompleteBatch({ batchId });
   }
 
+  /**
+   * Terminally fail a batch whose Phase 2 item stream never sealed, resolving the
+   * parent's batchTriggerAndWait waitpoint with an error so the parent resumes
+   * instead of hanging forever.
+   */
+  async expireBatch({ batchId }: { batchId: string }): Promise<void> {
+    return this.batchSystem.expireBatch({ batchId });
+  }
+
+  /**
+   * Schedule the seal-timeout reaper for a batch. If the batch hasn't sealed by
+   * `availableAt`, {@link expireBatch} terminally fails it and resumes the parent.
+   */
+  async scheduleExpireBatch({
+    batchId,
+    availableAt,
+  }: {
+    batchId: string;
+    availableAt: Date;
+  }): Promise<void> {
+    return this.batchSystem.scheduleExpireBatch({ batchId, availableAt });
+  }
+
   // ============================================================================
   // BatchQueue methods (DRR-based batch processing)
   // ============================================================================

internal-packages/run-engine/src/engine/systems/batchSystem.ts

@@ -1,4 +1,5 @@
 import { startSpan } from "@internal/tracing";
+import { TaskRunError } from "@trigger.dev/core/v3/schemas";
 import { isFinalRunStatus } from "../statuses.js";
 import { SystemResources } from "./systems.js";
 import { WaitpointSystem } from "./waitpointSystem.js";
@@ -32,6 +33,98 @@ export class BatchSystem {
     await this.#tryCompleteBatch({ batchId });
   }
 
+  public async scheduleExpireBatch({
+    batchId,
+    availableAt,
+  }: {
+    batchId: string;
+    availableAt: Date;
+  }): Promise<void> {
+    await this.$.worker.enqueue({
+      // Stable id dedupes repeated schedules for the same batch.
+      id: `expireBatch:${batchId}`,
+      job: "expireBatch",
+      payload: { batchId },
+      availableAt,
+    });
+  }
+
+  /**
+   * Terminally fail a batch whose Phase 2 item stream never sealed it, and resolve
+   * the parent's batchTriggerAndWait waitpoint with an error so the parent resumes
+   * with a failure instead of hanging forever.
+   *
+   * Idempotent and race-safe: if the stream sealed the batch (or it otherwise
+   * progressed past an unsealed PENDING state) in the meantime, this is a no-op.
+   */
+  public async expireBatch({ batchId }: { batchId: string }): Promise<void> {
+    return startSpan(this.$.tracer, "expireBatch", async (span) => {
+      span.setAttribute("batchId", batchId);
+
+      const batch = await this.$.prisma.batchTaskRun.findFirst({
+        select: { status: true, sealed: true },
+        where: { id: batchId },
+      });
+
+      if (!batch) {
+        this.$.logger.debug("expireBatch: batch doesn't exist", { batchId });
+        return;
+      }
+
+      // The stream sealed the batch, or it already progressed — nothing to fail.
+      if (batch.sealed || batch.status !== "PENDING") {
+        this.$.logger.debug("expireBatch: batch already sealed or no longer PENDING", {
+          batchId,
+          status: batch.status,
+          sealed: batch.sealed,
+        });
+        return;
+      }
+
+      // Conditional update guards against racing a late seal — whichever loses no-ops.
+      const aborted = await this.$.prisma.batchTaskRun.updateMany({
+        where: { id: batchId, sealed: false, status: "PENDING" },
+        data: {
+          status: "ABORTED",
+          completedAt: new Date(),
+          processingCompletedAt: new Date(),
+        },
+      });
+
+      if (aborted.count === 0) {
+        this.$.logger.debug("expireBatch: lost race to seal, no-op", { batchId });
+        return;
+      }
+
+      // Only batchTriggerAndWait blocks a parent, so only it has a waitpoint to resolve.
+      const waitpoint = await this.$.prisma.waitpoint.findFirst({
+        where: { completedByBatchId: batchId },
+      });
+
+      if (!waitpoint) {
+        this.$.logger.debug("expireBatch: no waitpoint to resolve (fire-and-forget batch)", {
+          batchId,
+        });
+        return;
+      }
+
+      const error: TaskRunError = {
+        type: "STRING_ERROR",
+        raw: "Batch items could not be streamed before the batch timed out",
+      };
+
+      await this.waitpointSystem.completeWaitpoint({
+        id: waitpoint.id,
+        output: { value: JSON.stringify(error), isError: true },
+      });
+
+      this.$.logger.warn("expireBatch: aborted unsealed batch and resumed parent with error", {
+        batchId,
+        waitpointId: waitpoint.id,
+      });
+    });
+  }
+
   /**
    * Checks to see if all runs for a BatchTaskRun are completed, if they are then update the status.
    * This isn't used operationally, but it's used for the Batches dashboard page.