Release 3.3.6 (#1544)

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

apps/webapp/app/env.server.ts

@@ -243,6 +243,8 @@ const EnvironmentSchema = z.object({
   MAXIMUM_DEV_QUEUE_SIZE: z.coerce.number().int().optional(),
   MAXIMUM_DEPLOYED_QUEUE_SIZE: z.coerce.number().int().optional(),
   MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
+
+  REALTIME_STREAM_VERSION: z.enum(["v1", "v2"]).default("v1"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

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

@@ -1,4 +1,4 @@
-import { ListRunResponse, ListRunResponseItem, RunStatus } from "@trigger.dev/core/v3";
+import { ListRunResponse, ListRunResponseItem, parsePacket, RunStatus } from "@trigger.dev/core/v3";
 import { Project, RuntimeEnvironment, TaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { z } from "zod";
@@ -220,36 +220,46 @@ export class ApiRunListPresenter extends BasePresenter {
 
       const results = await presenter.call(options);
 
-      const data: ListRunResponseItem[] = results.runs.map((run) => {
-        return {
-          id: run.friendlyId,
-          status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status),
-          taskIdentifier: run.taskIdentifier,
-          idempotencyKey: run.idempotencyKey,
-          version: run.version ?? undefined,
-          createdAt: new Date(run.createdAt),
-          updatedAt: new Date(run.updatedAt),
-          startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
-          finishedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
-          delayedUntil: run.delayUntil ? new Date(run.delayUntil) : undefined,
-          isTest: run.isTest,
-          ttl: run.ttl ?? undefined,
-          expiredAt: run.expiredAt ? new Date(run.expiredAt) : undefined,
-          env: {
-            id: run.environment.id,
-            name: run.environment.slug,
-            user: run.environment.userName,
-          },
-          tags: run.tags,
-          costInCents: run.costInCents,
-          baseCostInCents: run.baseCostInCents,
-          durationMs: run.usageDurationMs,
-          depth: run.depth,
-          ...ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(
-            ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status)
-          ),
-        };
-      });
+      logger.debug("RunListPresenter results", { results });
+
+      const data: ListRunResponseItem[] = await Promise.all(
+        results.runs.map(async (run) => {
+          const metadata = await parsePacket({
+            data: run.metadata ?? undefined,
+            dataType: run.metadataType,
+          });
+
+          return {
+            id: run.friendlyId,
+            status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status),
+            taskIdentifier: run.taskIdentifier,
+            idempotencyKey: run.idempotencyKey,
+            version: run.version ?? undefined,
+            createdAt: new Date(run.createdAt),
+            updatedAt: new Date(run.updatedAt),
+            startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
+            finishedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
+            delayedUntil: run.delayUntil ? new Date(run.delayUntil) : undefined,
+            isTest: run.isTest,
+            ttl: run.ttl ?? undefined,
+            expiredAt: run.expiredAt ? new Date(run.expiredAt) : undefined,
+            env: {
+              id: run.environment.id,
+              name: run.environment.slug,
+              user: run.environment.userName,
+            },
+            tags: run.tags,
+            costInCents: run.costInCents,
+            baseCostInCents: run.baseCostInCents,
+            durationMs: run.usageDurationMs,
+            depth: run.depth,
+            metadata,
+            ...ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(
+              ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status)
+            ),
+          };
+        })
+      );
 
       return {
         data,

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

@@ -216,6 +216,8 @@ export class RunListPresenter extends BasePresenter {
         depth: number;
         rootTaskRunId: string | null;
         batchId: string | null;
+        metadata: string | null;
+        metadataType: string;
       }[]
     >`
     SELECT
@@ -241,7 +243,9 @@ export class RunListPresenter extends BasePresenter {
     tr."usageDurationMs" AS "usageDurationMs",
     tr."depth" AS "depth",
     tr."rootTaskRunId" AS "rootTaskRunId",
-    tr."runTags" AS "tags"
+    tr."runTags" AS "tags",
+    tr."metadata" AS "metadata",
+    tr."metadataType" AS "metadataType"
 FROM
     ${sqlDatabaseSchema}."TaskRun" tr
 LEFT JOIN
@@ -374,6 +378,8 @@ WHERE
           tags: run.tags ? run.tags.sort((a, b) => a.localeCompare(b)) : [],
           depth: run.depth,
           rootTaskRunId: run.rootTaskRunId,
+          metadata: run.metadata,
+          metadataType: run.metadataType,
         };
       }),
       pagination: {

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

@@ -215,7 +215,9 @@ export class SpanPresenter extends BasePresenter {
     const span = await eventRepository.getSpan(spanId, run.traceId);
 
     const metadata = run.metadata
-      ? await prettyPrintPacket(run.metadata, run.metadataType, { filteredKeys: ["$$streams"] })
+      ? await prettyPrintPacket(run.metadata, run.metadataType, {
+          filteredKeys: ["$$streams", "$$streamsVersion"],
+        })
       : undefined;
 
     const context = {

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test.tasks.$taskParam/route.tsx

@@ -204,7 +204,7 @@ function StandardTaskForm({ task, runs }: { task: TestTask["task"]; runs: Standa
       );
       e.preventDefault();
     },
-    [currentPayloadJson, currentMetadataJson]
+    [currentPayloadJson, currentMetadataJson, task]
   );
 
   const [form, { environmentId, payload }] = useForm({

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

@@ -9,15 +9,21 @@ import { env } from "~/env.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
-import { BatchTriggerV2Service } from "~/v3/services/batchTriggerV2.server";
+import {
+  BatchProcessingStrategy,
+  BatchTriggerV2Service,
+} from "~/v3/services/batchTriggerV2.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { z } from "zod";
 
 const { action, loader } = createActionApiRoute(
   {
-    headers: HeadersSchema,
+    headers: HeadersSchema.extend({
+      "batch-processing-strategy": BatchProcessingStrategy.nullish(),
+    }),
     body: BatchTriggerTaskV2RequestBody,
     allowJWT: true,
     maxContentLength: env.BATCH_TASK_PAYLOAD_MAXIMUM_SIZE,
@@ -52,6 +58,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-span-parent-as-link": spanParentAsLink,
       "x-trigger-worker": isFromWorker,
       "x-trigger-client": triggerClient,
+      "batch-processing-strategy": batchProcessingStrategy,
       traceparent,
       tracestate,
     } = headers;
@@ -67,6 +74,7 @@ const { action, loader } = createActionApiRoute(
       triggerClient,
       traceparent,
       tracestate,
+      batchProcessingStrategy,
     });
 
     const traceContext =
@@ -79,7 +87,7 @@ const { action, loader } = createActionApiRoute(
       resolveIdempotencyKeyTTL(idempotencyKeyTTL) ??
       new Date(Date.now() + 24 * 60 * 60 * 1000 * 30);
 
-    const service = new BatchTriggerV2Service();
+    const service = new BatchTriggerV2Service(batchProcessingStrategy ?? undefined);
 
     try {
       const batch = await service.call(authentication.environment, body, {

apps/webapp/app/routes/api.v3.runs.$runId.ts

@@ -15,6 +15,7 @@ export const loader = createLoaderApiRoute(
     findResource: (params, auth) => {
       return ApiRetrieveRunPresenter.findRun(params.runId, auth.environment);
     },
+    shouldRetryNotFound: true,
     authorization: {
       action: "read",
       resource: (run) => ({

apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts

@@ -1,7 +1,7 @@
 import { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
-import { realtimeStreams } from "~/services/realtimeStreamsGlobal.server";
+import { v1RealtimeStreams } from "~/services/realtime/v1StreamsGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
@@ -16,7 +16,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return new Response("No body provided", { status: 400 });
   }
 
-  return realtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
+  return v1RealtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
 }
 
 export const loader = createLoaderApiRoute(
@@ -50,7 +50,13 @@ export const loader = createLoaderApiRoute(
       superScopes: ["read:runs", "read:all", "admin"],
     },
   },
-  async ({ params, request, resource: run }) => {
-    return realtimeStreams.streamResponse(run.friendlyId, params.streamId, request.signal);
+  async ({ params, request, resource: run, authentication }) => {
+    return v1RealtimeStreams.streamResponse(
+      request,
+      run.friendlyId,
+      params.streamId,
+      authentication.environment,
+      request.signal
+    );
   }
 );

apps/webapp/app/routes/realtime.v2.streams.$runId.$streamId.ts

@@ -0,0 +1,87 @@
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+import { v2RealtimeStreams } from "~/services/realtime/v2StreamsGlobal.server";
+
+const ParamsSchema = z.object({
+  runId: z.string(),
+  streamId: z.string(),
+});
+
+const { action } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+  },
+  async ({ request, params, authentication }) => {
+    if (!request.body) {
+      return new Response("No body provided", { status: 400 });
+    }
+
+    const run = await $replica.taskRun.findFirst({
+      where: {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      },
+      include: {
+        batch: {
+          select: {
+            friendlyId: true,
+          },
+        },
+      },
+    });
+
+    if (!run) {
+      return new Response("Run not found", { status: 404 });
+    }
+
+    return v2RealtimeStreams.ingestData(request.body, run.id, params.streamId);
+  }
+);
+
+export { action };
+
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) => {
+      return $replica.taskRun.findFirst({
+        where: {
+          friendlyId: params.runId,
+          runtimeEnvironmentId: auth.environment.id,
+        },
+        include: {
+          batch: {
+            select: {
+              friendlyId: true,
+            },
+          },
+        },
+      });
+    },
+    authorization: {
+      action: "read",
+      resource: (run) => ({
+        runs: run.friendlyId,
+        tags: run.runTags,
+        batch: run.batch?.friendlyId,
+        tasks: run.taskIdentifier,
+      }),
+      superScopes: ["read:runs", "read:all", "admin"],
+    },
+  },
+  async ({ params, request, resource: run, authentication }) => {
+    return v2RealtimeStreams.streamResponse(
+      request,
+      run.id,
+      params.streamId,
+      authentication.environment,
+      request.signal
+    );
+  }
+);

apps/webapp/app/services/realtime/databaseRealtimeStreams.server.ts

@@ -0,0 +1,85 @@
+import { PrismaClient } from "@trigger.dev/database";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { logger } from "../logger.server";
+import { RealtimeClient } from "../realtimeClient.server";
+import { StreamIngestor, StreamResponder } from "./types";
+
+export type DatabaseRealtimeStreamsOptions = {
+  prisma: PrismaClient;
+  realtimeClient: RealtimeClient;
+};
+
+// Class implementing both interfaces
+export class DatabaseRealtimeStreams implements StreamIngestor, StreamResponder {
+  constructor(private options: DatabaseRealtimeStreamsOptions) {}
+
+  async streamResponse(
+    request: Request,
+    runId: string,
+    streamId: string,
+    environment: AuthenticatedEnvironment,
+    signal: AbortSignal
+  ): Promise<Response> {
+    return this.options.realtimeClient.streamChunks(
+      request.url,
+      environment,
+      runId,
+      streamId,
+      signal,
+      request.headers.get("x-trigger-electric-version") ?? undefined
+    );
+  }
+
+  async ingestData(
+    stream: ReadableStream<Uint8Array>,
+    runId: string,
+    streamId: string
+  ): Promise<Response> {
+    try {
+      const textStream = stream.pipeThrough(new TextDecoderStream());
+      const reader = textStream.getReader();
+      let sequence = 0;
+
+      while (true) {
+        const { done, value } = await reader.read();
+
+        if (done) {
+          break;
+        }
+
+        logger.debug("[DatabaseRealtimeStreams][ingestData] Reading data", {
+          streamId,
+          runId,
+          value,
+        });
+
+        const chunks = value
+          .split("\n")
+          .filter((chunk) => chunk) // Remove empty lines
+          .map((line) => {
+            return {
+              sequence: sequence++,
+              value: line,
+            };
+          });
+
+        await this.options.prisma.realtimeStreamChunk.createMany({
+          data: chunks.map((chunk) => {
+            return {
+              runId,
+              key: streamId,
+              sequence: chunk.sequence,
+              value: chunk.value,
+            };
+          }),
+        });
+      }
+
+      return new Response(null, { status: 200 });
+    } catch (error) {
+      logger.error("[DatabaseRealtimeStreams][ingestData] Error in ingestData:", { error });
+
+      return new Response(null, { status: 500 });
+    }
+  }
+}

apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts

@@ -1,5 +1,7 @@
 import Redis, { RedisKey, RedisOptions, RedisValue } from "ioredis";
-import { logger } from "./logger.server";
+import { logger } from "../logger.server";
+import { StreamIngestor, StreamResponder } from "./types";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
 
 export type RealtimeStreamsOptions = {
   redis: RedisOptions | undefined;
@@ -7,10 +9,17 @@ export type RealtimeStreamsOptions = {
 
 const END_SENTINEL = "<<CLOSE_STREAM>>";
 
-export class RealtimeStreams {
+// Class implementing both interfaces
+export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
   constructor(private options: RealtimeStreamsOptions) {}
 
-  async streamResponse(runId: string, streamId: string, signal: AbortSignal): Promise<Response> {
+  async streamResponse(
+    request: Request,
+    runId: string,
+    streamId: string,
+    environment: AuthenticatedEnvironment,
+    signal: AbortSignal
+  ): Promise<Response> {
     const redis = new Redis(this.options.redis ?? {});
     const streamKey = `stream:${runId}:${streamId}`;
     let isCleanedUp = false;
@@ -115,11 +124,10 @@ export class RealtimeStreams {
     }
 
     try {
-      // Use TextDecoderStream to simplify text decoding
       const textStream = stream.pipeThrough(new TextDecoderStream());
       const reader = textStream.getReader();
 
-      const batchSize = 10; // Adjust this value based on performance testing
+      const batchSize = 10;
       let batchCommands: Array<[key: RedisKey, ...args: RedisValue[]]> = [];
 
       while (true) {
@@ -131,17 +139,13 @@ export class RealtimeStreams {
 
         logger.debug("[RealtimeStreams][ingestData] Reading data", { streamKey, value });
 
-        // 'value' is a string containing the decoded text
         const lines = value.split("\n");
 
         for (const line of lines) {
           if (line.trim()) {
-            // Avoid unnecessary parsing; assume 'line' is already a JSON string
-            // Add XADD command with MAXLEN option to limit stream size
             batchCommands.push([streamKey, "MAXLEN", "~", "2500", "*", "data", line]);
 
             if (batchCommands.length >= batchSize) {
-              // Send batch using a pipeline
               const pipeline = redis.pipeline();
               for (const args of batchCommands) {
                 pipeline.xadd(...args);
@@ -153,7 +157,6 @@ export class RealtimeStreams {
         }
       }
 
-      // Send any remaining commands
       if (batchCommands.length > 0) {
         const pipeline = redis.pipeline();
         for (const args of batchCommands) {
@@ -162,7 +165,6 @@ export class RealtimeStreams {
         await pipeline.exec();
       }
 
-      // Send the __end message to indicate the end of the stream
       await redis.xadd(streamKey, "MAXLEN", "~", "1000", "*", "data", END_SENTINEL);
 
       return new Response(null, { status: 200 });

apps/webapp/app/services/realtime/types.ts

@@ -0,0 +1,21 @@
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+
+// Interface for stream ingestion
+export interface StreamIngestor {
+  ingestData(
+    stream: ReadableStream<Uint8Array>,
+    runId: string,
+    streamId: string
+  ): Promise<Response>;
+}
+
+// Interface for stream response
+export interface StreamResponder {
+  streamResponse(
+    request: Request,
+    runId: string,
+    streamId: string,
+    environment: AuthenticatedEnvironment,
+    signal: AbortSignal
+  ): Promise<Response>;
+}

apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts

@@ -1,9 +1,9 @@
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
-import { RealtimeStreams } from "./realtimeStreams.server";
+import { RedisRealtimeStreams } from "./redisRealtimeStreams.server";
 
-function initializeRealtimeStreams() {
-  return new RealtimeStreams({
+function initializeRedisRealtimeStreams() {
+  return new RedisRealtimeStreams({
     redis: {
       port: env.REDIS_PORT,
       host: env.REDIS_HOST,
@@ -16,4 +16,4 @@ function initializeRealtimeStreams() {
   });
 }
 
-export const realtimeStreams = singleton("realtimeStreams", initializeRealtimeStreams);
+export const v1RealtimeStreams = singleton("realtimeStreams", initializeRedisRealtimeStreams);

apps/webapp/app/services/realtime/v2StreamsGlobal.server.ts

@@ -0,0 +1,13 @@
+import { prisma } from "~/db.server";
+import { singleton } from "~/utils/singleton";
+import { realtimeClient } from "../realtimeClientGlobal.server";
+import { DatabaseRealtimeStreams } from "./databaseRealtimeStreams.server";
+
+function initializeDatabaseRealtimeStreams() {
+  return new DatabaseRealtimeStreams({
+    prisma,
+    realtimeClient,
+  });
+}
+
+export const v2RealtimeStreams = singleton("dbRealtimeStreams", initializeDatabaseRealtimeStreams);

apps/webapp/app/services/realtimeClient.server.ts

@@ -37,6 +37,23 @@ export class RealtimeClient {
     this.#registerCommands();
   }
 
+  async streamChunks(
+    url: URL | string,
+    environment: RealtimeEnvironment,
+    runId: string,
+    streamId: string,
+    signal?: AbortSignal,
+    clientVersion?: string
+  ) {
+    return this.#streamChunksWhere(
+      url,
+      environment,
+      `"runId"='${runId}' AND "key"='${streamId}'`,
+      signal,
+      clientVersion
+    );
+  }
+
   async streamRun(
     url: URL | string,
     environment: RealtimeEnvironment,
@@ -85,12 +102,12 @@ export class RealtimeClient {
     whereClause: string,
     clientVersion?: string
   ) {
-    const electricUrl = this.#constructElectricUrl(url, whereClause, clientVersion);
+    const electricUrl = this.#constructRunsElectricUrl(url, whereClause, clientVersion);
 
-    return this.#performElectricRequest(electricUrl, environment, clientVersion);
+    return this.#performElectricRequest(electricUrl, environment, undefined, clientVersion);
   }
 
-  #constructElectricUrl(url: URL | string, whereClause: string, clientVersion?: string): URL {
+  #constructRunsElectricUrl(url: URL | string, whereClause: string, clientVersion?: string): URL {
     const $url = new URL(url.toString());
 
     const electricUrl = new URL(`${this.options.electricOrigin}/v1/shape`);
@@ -112,9 +129,44 @@ export class RealtimeClient {
     return electricUrl;
   }
 
+  async #streamChunksWhere(
+    url: URL | string,
+    environment: RealtimeEnvironment,
+    whereClause: string,
+    signal?: AbortSignal,
+    clientVersion?: string
+  ) {
+    const electricUrl = this.#constructChunksElectricUrl(url, whereClause, clientVersion);
+
+    return this.#performElectricRequest(electricUrl, environment, signal, clientVersion);
+  }
+
+  #constructChunksElectricUrl(url: URL | string, whereClause: string, clientVersion?: string): URL {
+    const $url = new URL(url.toString());
+
+    const electricUrl = new URL(`${this.options.electricOrigin}/v1/shape`);
+
+    // Copy over all the url search params to the electric url
+    $url.searchParams.forEach((value, key) => {
+      electricUrl.searchParams.set(key, value);
+    });
+
+    electricUrl.searchParams.set("where", whereClause);
+    electricUrl.searchParams.set("table", `public."RealtimeStreamChunk"`);
+
+    if (!clientVersion) {
+      // If the client version is not provided, that means we're using an older client
+      // This means the client will be sending shape_id instead of handle
+      electricUrl.searchParams.set("handle", electricUrl.searchParams.get("shape_id") ?? "");
+    }
+
+    return electricUrl;
+  }
+
   async #performElectricRequest(
     url: URL,
     environment: RealtimeEnvironment,
+    signal?: AbortSignal,
     clientVersion?: string
   ) {
     const shapeId = extractShapeId(url);
@@ -129,13 +181,13 @@ export class RealtimeClient {
 
     if (!shapeId) {
       // If the shapeId is not present, we're just getting the initial value
-      return longPollingFetch(url.toString(), {}, rewriteResponseHeaders);
+      return longPollingFetch(url.toString(), { signal }, rewriteResponseHeaders);
     }
 
     const isLive = isLiveRequestUrl(url);
 
     if (!isLive) {
-      return longPollingFetch(url.toString(), {}, rewriteResponseHeaders);
+      return longPollingFetch(url.toString(), { signal }, rewriteResponseHeaders);
     }
 
     const requestId = randomUUID();
@@ -177,7 +229,7 @@ export class RealtimeClient {
 
     try {
       // ... (rest of your existing code for the long polling request)
-      const response = await longPollingFetch(url.toString(), {}, rewriteResponseHeaders);
+      const response = await longPollingFetch(url.toString(), { signal }, rewriteResponseHeaders);
 
       // Decrement the counter after the long polling request is complete
       await this.#decrementConcurrency(environment.id, requestId);

apps/webapp/app/services/routeBuilders/apiBuilder.server.ts

@@ -33,6 +33,7 @@ type ApiKeyRouteBuilderOptions<
     params: TParamsSchema extends z.AnyZodObject ? z.infer<TParamsSchema> : undefined,
     authentication: ApiAuthenticationResultSuccess
   ) => Promise<TResource | undefined>;
+  shouldRetryNotFound?: boolean;
   authorization?: {
     action: AuthorizationAction;
     resource: (
@@ -81,6 +82,7 @@ export function createLoaderApiRoute<
       corsStrategy = "none",
       authorization,
       findResource,
+      shouldRetryNotFound,
     } = options;
 
     if (corsStrategy !== "none" && request.method.toUpperCase() === "OPTIONS") {
@@ -162,7 +164,10 @@ export function createLoaderApiRoute<
       if (!resource) {
         return await wrapResponse(
           request,
-          json({ error: "Not found" }, { status: 404 }),
+          json(
+            { error: "Not found" },
+            { status: 404, headers: { "x-should-retry": shouldRetryNotFound ? "true" : "false" } }
+          ),
           corsStrategy !== "none"
         );
       }

apps/webapp/app/services/worker.server.ts

@@ -733,7 +733,7 @@ function getWorkerQueue() {
         priority: 0,
         maxAttempts: 5,
         handler: async (payload, job) => {
-          const service = new BatchTriggerV2Service();
+          const service = new BatchTriggerV2Service(payload.strategy);
 
           await service.processBatchTaskRun(payload);
         },

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -662,12 +662,25 @@ export async function resolveVariablesForEnvironment(runtimeEnvironment: Runtime
     runtimeEnvironment.id
   );
 
+  const overridableTriggerVariables = await resolveOverridableTriggerVariables(runtimeEnvironment);
+
   const builtInVariables =
     runtimeEnvironment.type === "DEVELOPMENT"
       ? await resolveBuiltInDevVariables(runtimeEnvironment)
       : await resolveBuiltInProdVariables(runtimeEnvironment);
 
-  return [...projectSecrets, ...builtInVariables];
+  return [...overridableTriggerVariables, ...projectSecrets, ...builtInVariables];
+}
+
+async function resolveOverridableTriggerVariables(runtimeEnvironment: RuntimeEnvironment) {
+  let result: Array<EnvironmentVariable> = [
+    {
+      key: "TRIGGER_REALTIME_STREAM_VERSION",
+      value: env.REALTIME_STREAM_VERSION,
+    },
+  ];
+
+  return result;
 }
 
 async function resolveBuiltInDevVariables(runtimeEnvironment: RuntimeEnvironment) {

apps/webapp/app/v3/services/batchTriggerV2.server.ts

@@ -6,7 +6,7 @@ import {
   parsePacket,
 } from "@trigger.dev/core/v3";
 import { BatchTaskRun, Prisma, TaskRunAttempt } from "@trigger.dev/database";
-import { $transaction, PrismaClientOrTransaction } from "~/db.server";
+import { $transaction, prisma, PrismaClientOrTransaction } from "~/db.server";
 import { env } from "~/env.server";
 import { batchTaskRunItemStatusForRunStatus } from "~/models/taskRun.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
@@ -25,12 +25,10 @@ import { z } from "zod";
 
 const PROCESSING_BATCH_SIZE = 50;
 const ASYNC_BATCH_PROCESS_SIZE_THRESHOLD = 20;
+const MAX_ATTEMPTS = 10;
 
-const BatchProcessingStrategy = z.enum(["sequential", "parallel"]);
-
-type BatchProcessingStrategy = z.infer<typeof BatchProcessingStrategy>;
-
-const CURRENT_STRATEGY: BatchProcessingStrategy = "parallel";
+export const BatchProcessingStrategy = z.enum(["sequential", "parallel"]);
+export type BatchProcessingStrategy = z.infer<typeof BatchProcessingStrategy>;
 
 export const BatchProcessingOptions = z.object({
   batchId: z.string(),
@@ -52,6 +50,17 @@ export type BatchTriggerTaskServiceOptions = {
 };
 
 export class BatchTriggerV2Service extends BaseService {
+  private _batchProcessingStrategy: BatchProcessingStrategy;
+
+  constructor(
+    batchProcessingStrategy?: BatchProcessingStrategy,
+    protected readonly _prisma: PrismaClientOrTransaction = prisma
+  ) {
+    super(_prisma);
+
+    this._batchProcessingStrategy = batchProcessingStrategy ?? "parallel";
+  }
+
   public async call(
     environment: AuthenticatedEnvironment,
     body: BatchTriggerTaskV2RequestBody,
@@ -452,14 +461,14 @@ export class BatchTriggerV2Service extends BaseService {
           },
         });
 
-        switch (CURRENT_STRATEGY) {
+        switch (this._batchProcessingStrategy) {
           case "sequential": {
             await this.#enqueueBatchTaskRun({
               batchId: batch.id,
               processingId: batchId,
               range: { start: 0, count: PROCESSING_BATCH_SIZE },
               attemptCount: 0,
-              strategy: CURRENT_STRATEGY,
+              strategy: this._batchProcessingStrategy,
             });
 
             break;
@@ -480,7 +489,7 @@ export class BatchTriggerV2Service extends BaseService {
                     processingId: `${index}`,
                     range,
                     attemptCount: 0,
-                    strategy: CURRENT_STRATEGY,
+                    strategy: this._batchProcessingStrategy,
                   },
                   tx
                 )
@@ -539,6 +548,16 @@ export class BatchTriggerV2Service extends BaseService {
 
     const $attemptCount = options.attemptCount + 1;
 
+    // Add early return if max attempts reached
+    if ($attemptCount > MAX_ATTEMPTS) {
+      logger.error("[BatchTriggerV2][processBatchTaskRun] Max attempts reached", {
+        options,
+        attemptCount: $attemptCount,
+      });
+      // You might want to update the batch status to failed here
+      return;
+    }
+
     const batch = await this._prisma.batchTaskRun.findFirst({
       where: { id: options.batchId },
       include: {

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -208,6 +208,15 @@ export async function createBackgroundTasks(
           taskQueue.concurrencyLimit
         );
       } else {
+        logger.debug("CreateBackgroundWorkerService: removing concurrency limit", {
+          workerId: worker.id,
+          taskQueue,
+          orgId: environment.organizationId,
+          projectId: environment.projectId,
+          environmentId: environment.id,
+          concurrencyLimit,
+          taskidentifier: task.id,
+        });
         await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
       }
     } catch (error) {

apps/webapp/app/v3/services/finalizeTaskRun.server.ts

@@ -64,9 +64,13 @@ export class FinalizeTaskRunService extends BaseService {
       completedAt,
     });
 
+    // I moved the error update here for two reasons:
+    // - A single update is more efficient than two
+    // - If the status updates to a final status, realtime will receive that status and then shut down the stream
+    //   before the error is updated, which would cause the error to be lost
     const run = await this._prisma.taskRun.update({
       where: { id },
-      data: { status, expiredAt, completedAt },
+      data: { status, expiredAt, completedAt, error: error ? sanitizeError(error) : undefined },
       ...(include ? { include } : {}),
     });
 
@@ -78,10 +82,6 @@ export class FinalizeTaskRunService extends BaseService {
       await this.finalizeAttempt({ attemptStatus, error, run });
     }
 
-    if (error) {
-      await this.finalizeRunError(run, error);
-    }
-
     try {
       await this.#finalizeBatch(run);
     } catch (finalizeBatchError) {
@@ -211,15 +211,6 @@ export class FinalizeTaskRunService extends BaseService {
     }
   }
 
-  async finalizeRunError(run: TaskRun, error: TaskRunError) {
-    await this._prisma.taskRun.update({
-      where: { id: run.id },
-      data: {
-        error: sanitizeError(error),
-      },
-    });
-  }
-
   async finalizeAttempt({
     attemptStatus,
     error,

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -474,6 +474,16 @@ export class TriggerTaskService extends BaseService {
                           taskQueue.concurrencyLimit
                         );
                       } else {
+                        logger.debug("TriggerTaskService: removing concurrency limit", {
+                          runId: taskRun.id,
+                          friendlyId: taskRun.friendlyId,
+                          taskQueue,
+                          orgId: environment.organizationId,
+                          projectId: environment.projectId,
+                          existingConcurrencyLimit,
+                          concurrencyLimit,
+                          queueOptions: body.options?.queue,
+                        });
                         await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
                       }
                     }

apps/webapp/test/realtimeClient.test.ts

@@ -1,9 +1,9 @@
-import { containerWithElectricTest } from "@internal/testcontainers";
+import { containerWithElectricAndRedisTest } from "@internal/testcontainers";
 import { expect, describe } from "vitest";
 import { RealtimeClient } from "../app/services/realtimeClient.server.js";
 
 describe("RealtimeClient", () => {
-  containerWithElectricTest(
+  containerWithElectricAndRedisTest(
     "Should only track concurrency for live requests",
     { timeout: 30_000 },
     async ({ redis, electricOrigin, prisma }) => {
@@ -139,7 +139,7 @@ describe("RealtimeClient", () => {
     }
   );
 
-  containerWithElectricTest(
+  containerWithElectricAndRedisTest(
     "Should support subscribing to a run tag",
     { timeout: 30_000 },
     async ({ redis, electricOrigin, prisma }) => {
@@ -218,7 +218,7 @@ describe("RealtimeClient", () => {
     }
   );
 
-  containerWithElectricTest(
+  containerWithElectricAndRedisTest(
     "Should adapt for older client versions",
     { timeout: 30_000 },
     async ({ redis, electricOrigin, prisma }) => {

apps/webapp/test/realtimeStreams.test.ts

@@ -1,106 +0,0 @@
-import { redisTest } from "@internal/testcontainers";
-import { describe, expect, vi } from "vitest";
-import { RealtimeStreams } from "../app/services/realtimeStreams.server.js";
-import { convertArrayToReadableStream, convertResponseSSEStreamToArray } from "./utils/streams.js";
-
-vi.setConfig({ testTimeout: 10_000 }); // 5 seconds
-
-// Mock the logger
-vi.mock("./logger.server", () => ({
-  logger: {
-    debug: vi.fn(),
-    error: vi.fn(),
-  },
-}));
-
-describe("RealtimeStreams", () => {
-  redisTest("should stream data from producer to consumer", async ({ redis }) => {
-    const streams = new RealtimeStreams({ redis: redis.options });
-    const runId = "test-run";
-    const streamId = "test-stream";
-
-    // Create a stream of test data
-    const stream = convertArrayToReadableStream(["chunk1", "chunk2", "chunk3"]).pipeThrough(
-      new TextEncoderStream()
-    );
-
-    // Start consuming the stream
-    const abortController = new AbortController();
-    const responsePromise = streams.streamResponse(runId, streamId, abortController.signal);
-
-    // Start ingesting data
-    await streams.ingestData(stream, runId, streamId);
-
-    // Get the response and read the stream
-    const response = await responsePromise;
-    const received = await convertResponseSSEStreamToArray(response);
-
-    expect(received).toEqual(["chunk1", "chunk2", "chunk3"]);
-  });
-
-  redisTest("should handle multiple concurrent streams", async ({ redis }) => {
-    const streams = new RealtimeStreams({ redis: redis.options });
-    const runId = "test-run";
-
-    // Set up two different streams
-    const stream1 = convertArrayToReadableStream(["1a", "1b", "1c"]).pipeThrough(
-      new TextEncoderStream()
-    );
-    const stream2 = convertArrayToReadableStream(["2a", "2b", "2c"]).pipeThrough(
-      new TextEncoderStream()
-    );
-
-    // Start consuming both streams
-    const abortController = new AbortController();
-    const response1Promise = streams.streamResponse(runId, "stream1", abortController.signal);
-    const response2Promise = streams.streamResponse(runId, "stream2", abortController.signal);
-
-    // Ingest data to both streams
-    await Promise.all([
-      streams.ingestData(stream1, runId, "stream1"),
-      streams.ingestData(stream2, runId, "stream2"),
-    ]);
-
-    // Get and verify both responses
-    const [response1, response2] = await Promise.all([response1Promise, response2Promise]);
-    const [received1, received2] = await Promise.all([
-      convertResponseSSEStreamToArray(response1),
-      convertResponseSSEStreamToArray(response2),
-    ]);
-
-    expect(received1).toEqual(["1a", "1b", "1c"]);
-    expect(received2).toEqual(["2a", "2b", "2c"]);
-  });
-
-  redisTest("should handle early consumer abort", async ({ redis }) => {
-    const streams = new RealtimeStreams({ redis: redis.options });
-    const runId = "test-run";
-    const streamId = "test-stream";
-
-    const stream = convertArrayToReadableStream(["chunk1", "chunk2", "chunk3"]).pipeThrough(
-      new TextEncoderStream()
-    );
-
-    // Start consuming but abort early
-    const abortController = new AbortController();
-    const responsePromise = streams.streamResponse(runId, streamId, abortController.signal);
-
-    // Get the response before aborting to ensure stream is properly set up
-    const response = await responsePromise;
-
-    // Start reading the stream
-    const readPromise = convertResponseSSEStreamToArray(response);
-
-    // Abort after a small delay to ensure everything is set up
-    await new Promise((resolve) => setTimeout(resolve, 100));
-    abortController.abort();
-
-    // Start ingesting data after abort
-    await streams.ingestData(stream, runId, streamId);
-
-    // Verify the stream was terminated
-    const received = await readPromise;
-
-    expect(received).toEqual(["chunk1"]);
-  });
-});

docker/Dockerfile.postgres

@@ -0,0 +1,5 @@
+FROM postgres:14
+
+RUN apt-get update \
+    && apt-get install -y postgresql-14-partman \
+    && rm -rf /var/lib/apt/lists/*

docker/docker-compose.yml

@@ -13,7 +13,9 @@ networks:
 services:
   database:
     container_name: database
-    image: postgres:14
+    build:
+      context: .
+      dockerfile: Dockerfile.postgres
     restart: always
     volumes:
       - ${DB_VOLUME:-database-data}:/var/lib/postgresql/data/
@@ -30,6 +32,8 @@ services:
       - listen_addresses=*
       - -c
       - wal_level=logical
+      - -c
+      - shared_preload_libraries=pg_partman_bgw
 
   pgadmin:
     container_name: pgadmin
@@ -61,7 +65,7 @@ services:
       - 6379:6379
 
   electric:
-    image: electricsql/electric:0.8.1
+    image: electricsql/electric:0.9.4
     restart: always
     environment:
       DATABASE_URL: postgresql://postgres:postgres@database:5432/postgres?sslmode=disable

docs/frontend/react-hooks/realtime.mdx

@@ -62,6 +62,31 @@ export function MyComponent({
 }
 ```
 
+You can supply an `onComplete` callback to the `useRealtimeRun` hook to be called when the run is completed or errored. This is useful if you want to perform some action when the run is completed, like navigating to a different page or showing a notification.
+
+```tsx
+import { useRealtimeRun } from "@trigger.dev/react-hooks";
+
+export function MyComponent({
+  runId,
+  publicAccessToken,
+}: {
+  runId: string;
+  publicAccessToken: string;
+}) {
+  const { run, error } = useRealtimeRun(runId, {
+    accessToken: publicAccessToken,
+    onComplete: (run, error) => {
+      console.log("Run completed", run);
+    },
+  });
+
+  if (error) return <div>Error: {error.message}</div>;
+
+  return <div>Run: {run.id}</div>;
+}
+```
+
 See our [Realtime documentation](/realtime) for more information about the type of the run object and more.
 
 ### useRealtimeRunsWithTag

internal-packages/database/prisma/migrations/20241206135145_create_realtime_chunks_table/migration.sql

@@ -0,0 +1,13 @@
+-- CreateTable
+CREATE TABLE "RealtimeStreamChunk" (
+  "id" TEXT NOT NULL,
+  "key" TEXT NOT NULL,
+  "value" TEXT NOT NULL,
+  "sequence" INTEGER NOT NULL,
+  "runId" TEXT NOT NULL,
+  "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  CONSTRAINT "RealtimeStreamChunk_pkey" PRIMARY KEY ("id")
+);
+
+-- Add index on (runID, createdAt) for efficient queries
+CREATE INDEX "RealtimeStreamChunk_runId" ON "RealtimeStreamChunk" ("runId");

internal-packages/database/prisma/migrations/20241208074324_add_created_at_index_to_realtime_stream_chunks/migration.sql

@@ -0,0 +1,5 @@
+-- CreateIndex
+CREATE INDEX "RealtimeStreamChunk_createdAt_idx" ON "RealtimeStreamChunk"("createdAt");
+
+-- RenameIndex
+ALTER INDEX "RealtimeStreamChunk_runId" RENAME TO "RealtimeStreamChunk_runId_idx";

internal-packages/database/prisma/schema.prisma

@@ -2667,3 +2667,19 @@ enum BulkActionItemStatus {
   COMPLETED
   FAILED
 }
+
+model RealtimeStreamChunk {
+  id String @id @default(cuid())
+
+  key   String
+  value String
+
+  sequence Int
+
+  runId String
+
+  createdAt DateTime @default(now())
+
+  @@index([runId])
+  @@index([createdAt])
+}

internal-packages/emails/src/index.tsx

@@ -127,7 +127,7 @@ export class EmailClient {
 
   async #sendEmail({ to, subject, react }: { to: string; subject: string; react: ReactElement }) {
     if (this.#client) {
-      await this.#client.emails.send({
+      const result = await this.#client.emails.send({
         from: this.#from,
         to,
         reply_to: this.#replyTo,
@@ -135,6 +135,13 @@ export class EmailClient {
         react,
       });
 
+      if (result.error) {
+        console.error(
+          `Failed to send email to ${to}, ${subject}. Error ${result.error.name}: ${result.error.message}`
+        );
+        throw new EmailError(result.error);
+      }
+
       return;
     }
 
@@ -147,3 +154,11 @@ ${render(react, {
     `);
   }
 }
+
+//EmailError type where you can set the name and message
+export class EmailError extends Error {
+  constructor({ name, message }: { name: string; message: string }) {
+    super(message);
+    this.name = name;
+  }
+}

internal-packages/testcontainers/src/index.ts

@@ -1,10 +1,10 @@
 import { StartedPostgreSqlContainer } from "@testcontainers/postgresql";
 import { StartedRedisContainer } from "@testcontainers/redis";
-import { Redis } from "ioredis";
-import { test } from "vitest";
 import { PrismaClient } from "@trigger.dev/database";
-import { createPostgresContainer, createRedisContainer, createElectricContainer } from "./utils";
-import { Network, type StartedNetwork, type StartedTestContainer } from "testcontainers";
+import { Redis } from "ioredis";
+import { Network, type StartedNetwork } from "testcontainers";
+import { test } from "vitest";
+import { createElectricContainer, createPostgresContainer, createRedisContainer } from "./utils";
 
 type NetworkContext = { network: StartedNetwork };
 
@@ -20,7 +20,8 @@ type ElectricContext = {
 };
 
 type ContainerContext = NetworkContext & PostgresContext & RedisContext;
-type ContainerWithElectricContext = ContainerContext & ElectricContext;
+type ContainerWithElectricAndRedisContext = ContainerContext & ElectricContext;
+type ContainerWithElectricContext = NetworkContext & PostgresContext & ElectricContext;
 
 type Use<T> = (value: T) => Promise<void>;
 
@@ -97,6 +98,13 @@ export const containerTest = test.extend<ContainerContext>({
 });
 
 export const containerWithElectricTest = test.extend<ContainerWithElectricContext>({
+  network,
+  postgresContainer,
+  prisma,
+  electricOrigin,
+});
+
+export const containerWithElectricAndRedisTest = test.extend<ContainerWithElectricAndRedisContext>({
   network,
   postgresContainer,
   prisma,

internal-packages/testcontainers/src/utils.ts

@@ -55,7 +55,7 @@ export async function createElectricContainer(
     network.getName()
   )}:5432/${postgresContainer.getDatabase()}?sslmode=disable`;
 
-  const container = await new GenericContainer("electricsql/electric:0.8.1")
+  const container = await new GenericContainer("electricsql/electric:0.9.4")
     .withExposedPorts(3000)
     .withNetwork(network)
     .withEnvironment({

packages/build/CHANGELOG.md

@@ -1,5 +1,19 @@
 # @trigger.dev/build
 
+## 3.3.6
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.6`
+
+## 3.3.5
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.5`
+
 ## 3.3.4
 
 ### Patch Changes

packages/build/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/build",
-  "version": "3.3.4",
+  "version": "3.3.6",
   "description": "trigger.dev build extensions",
   "license": "MIT",
   "publishConfig": {
@@ -65,7 +65,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:3.3.4",
+    "@trigger.dev/core": "workspace:3.3.6",
     "pkg-types": "^1.1.3",
     "tinyglobby": "^0.2.2",
     "tsconfck": "3.1.3"

packages/cli-v3/CHANGELOG.md

@@ -1,5 +1,21 @@
 # trigger.dev
 
+## 3.3.6
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.6`
+  - `@trigger.dev/build@3.3.6`
+
+## 3.3.5
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.5`
+  - `@trigger.dev/build@3.3.5`
+
 ## 3.3.4
 
 ### Patch Changes

packages/cli-v3/package.json

@@ -1,6 +1,6 @@
 {
   "name": "trigger.dev",
-  "version": "3.3.4",
+  "version": "3.3.6",
   "description": "A Command-Line Interface for Trigger.dev (v3) projects",
   "type": "module",
   "license": "MIT",
@@ -87,8 +87,8 @@
     "@opentelemetry/sdk-trace-base": "1.25.1",
     "@opentelemetry/sdk-trace-node": "1.25.1",
     "@opentelemetry/semantic-conventions": "1.25.1",
-    "@trigger.dev/build": "workspace:3.3.4",
-    "@trigger.dev/core": "workspace:3.3.4",
+    "@trigger.dev/build": "workspace:3.3.6",
+    "@trigger.dev/core": "workspace:3.3.6",
     "c12": "^1.11.1",
     "chalk": "^5.2.0",
     "cli-table3": "^0.6.3",

packages/cli-v3/src/entryPoints/deploy-run-worker.ts

@@ -105,7 +105,8 @@ const durableClock = new DurableClock();
 clock.setGlobalClock(durableClock);
 const runMetadataManager = new StandardMetadataManager(
   apiClientManager.clientOrThrow(),
-  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
+  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
+  (getEnvVar("TRIGGER_REALTIME_STREAM_VERSION") ?? "v1") as "v1" | "v2"
 );
 runMetadata.setGlobalManager(runMetadataManager);
 const waitUntilManager = new StandardWaitUntilManager();

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -87,7 +87,8 @@ runtime.setGlobalRuntimeManager(devRuntimeManager);
 timeout.setGlobalManager(new UsageTimeoutManager(devUsageManager));
 const runMetadataManager = new StandardMetadataManager(
   apiClientManager.clientOrThrow(),
-  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
+  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
+  (getEnvVar("TRIGGER_REALTIME_STREAM_VERSION") ?? "v1") as "v1" | "v2"
 );
 runMetadata.setGlobalManager(runMetadataManager);
 const waitUntilManager = new StandardWaitUntilManager();

packages/core/CHANGELOG.md

@@ -1,5 +1,17 @@
 # internal-platform
 
+## 3.3.6
+
+### Patch Changes
+
+- Add option to trigger batched items sequentially, and default to parallel triggering which is faster ([#1536](https://github.com/triggerdotdev/trigger.dev/pull/1536))
+
+## 3.3.5
+
+### Patch Changes
+
+- Fix an issue that caused errors when using realtime with a run that is cancelled ([#1533](https://github.com/triggerdotdev/trigger.dev/pull/1533))
+
 ## 3.3.4
 
 ## 3.3.3

packages/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/core",
-  "version": "3.3.4",
+  "version": "3.3.6",
   "description": "Core code used across the Trigger.dev SDK and platform",
   "license": "MIT",
   "publishConfig": {
@@ -182,7 +182,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@electric-sql/client": "0.7.1",
+    "@electric-sql/client": "0.9.0",
     "@google-cloud/precise-date": "^4.0.0",
     "@jsonhero/path": "^1.0.21",
     "@opentelemetry/api": "1.9.0",

packages/core/src/v3/apiClient/index.ts

@@ -74,6 +74,7 @@ export type ClientTriggerOptions = {
 export type ClientBatchTriggerOptions = ClientTriggerOptions & {
   idempotencyKey?: string;
   idempotencyKeyTTL?: string;
+  processingStrategy?: "parallel" | "sequential";
 };
 
 export type TriggerRequestOptions = ZodFetchOptions & {
@@ -138,6 +139,10 @@ export class ApiClient {
     return fetchClient;
   }
 
+  getHeaders() {
+    return this.#getHeaders(false);
+  }
+
   async getRunResult(
     runId: string,
     requestOptions?: ZodFetchOptions
@@ -239,6 +244,7 @@ export class ApiClient {
         headers: this.#getHeaders(clientOptions?.spanParentAsLink ?? false, {
           "idempotency-key": clientOptions?.idempotencyKey,
           "idempotency-key-ttl": clientOptions?.idempotencyKeyTTL,
+          "batch-processing-strategy": clientOptions?.processingStrategy,
         }),
         body: JSON.stringify(body),
       },

packages/core/src/v3/apiClient/runStream.ts

@@ -1,5 +1,11 @@
+import { EventSourceParserStream } from "eventsource-parser/stream";
 import { DeserializedJson } from "../../schemas/json.js";
-import { RunStatus, SubscribeRunRawShape } from "../schemas/api.js";
+import { createJsonErrorObject } from "../errors.js";
+import {
+  RunStatus,
+  SubscribeRealtimeStreamChunkRawShape,
+  SubscribeRunRawShape,
+} from "../schemas/api.js";
 import { SerializedError } from "../schemas/common.js";
 import { AnyRunTypes, AnyTask, InferRunTypes } from "../types/tasks.js";
 import { getEnvVar } from "../utils/getEnv.js";
@@ -10,8 +16,7 @@ import {
 } from "../utils/ioSerialization.js";
 import { ApiError } from "./errors.js";
 import { ApiClient } from "./index.js";
-import { AsyncIterableStream, createAsyncIterableStream, zodShapeStream } from "./stream.js";
-import { EventSourceParserStream } from "eventsource-parser/stream";
+import { AsyncIterableStream, createAsyncIterableReadable, zodShapeStream } from "./stream.js";
 
 export type RunShape<TRunTypes extends AnyRunTypes> = TRunTypes extends AnyRunTypes
   ? {
@@ -77,19 +82,42 @@ export function runShapeStream<TRunTypes extends AnyRunTypes>(
   url: string,
   options?: RunShapeStreamOptions
 ): RunSubscription<TRunTypes> {
-  const $options: RunSubscriptionOptions = {
-    provider: {
-      async onShape(callback) {
-        return zodShapeStream(SubscribeRunRawShape, url, callback, options);
-      },
-    },
-    streamFactory: new SSEStreamSubscriptionFactory(
-      getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
-      {
-        headers: options?.headers,
-        signal: options?.signal,
+  const abortController = new AbortController();
+
+  const version1 = new SSEStreamSubscriptionFactory(
+    getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
+    {
+      headers: options?.headers,
+      signal: abortController.signal,
+    }
+  );
+
+  const version2 = new ElectricStreamSubscriptionFactory(
+    getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
+    {
+      headers: options?.headers,
+      signal: abortController.signal,
+    }
+  );
+
+  // If the user supplied AbortSignal is aborted, we should abort the internal controller
+  options?.signal?.addEventListener(
+    "abort",
+    () => {
+      if (!abortController.signal.aborted) {
+        abortController.abort();
       }
-    ),
+    },
+    { once: true }
+  );
+
+  const $options: RunSubscriptionOptions = {
+    runShapeStream: zodShapeStream(SubscribeRunRawShape, url, {
+      ...options,
+      signal: abortController.signal,
+    }),
+    streamFactory: new VersionedStreamSubscriptionFactory(version1, version2),
+    abortController,
     ...options,
   };
 
@@ -102,7 +130,12 @@ export interface StreamSubscription {
 }
 
 export interface StreamSubscriptionFactory {
-  createSubscription(runId: string, streamKey: string, baseUrl?: string): StreamSubscription;
+  createSubscription(
+    metadata: Record<string, unknown>,
+    runId: string,
+    streamKey: string,
+    baseUrl?: string
+  ): StreamSubscription;
 }
 
 // Real implementation for production
@@ -153,7 +186,12 @@ export class SSEStreamSubscriptionFactory implements StreamSubscriptionFactory {
     private options: { headers?: Record<string, string>; signal?: AbortSignal }
   ) {}
 
-  createSubscription(runId: string, streamKey: string, baseUrl?: string): StreamSubscription {
+  createSubscription(
+    metadata: Record<string, unknown>,
+    runId: string,
+    streamKey: string,
+    baseUrl?: string
+  ): StreamSubscription {
     if (!runId || !streamKey) {
       throw new Error("runId and streamKey are required");
     }
@@ -163,17 +201,89 @@ export class SSEStreamSubscriptionFactory implements StreamSubscriptionFactory {
   }
 }
 
+// Real implementation for production
+export class ElectricStreamSubscription implements StreamSubscription {
+  constructor(
+    private url: string,
+    private options: { headers?: Record<string, string>; signal?: AbortSignal }
+  ) {}
+
+  async subscribe(): Promise<ReadableStream<unknown>> {
+    return zodShapeStream(SubscribeRealtimeStreamChunkRawShape, this.url, this.options).pipeThrough(
+      new TransformStream({
+        transform(chunk, controller) {
+          controller.enqueue(safeParseJSON(chunk.value));
+        },
+      })
+    );
+  }
+}
+
+export class ElectricStreamSubscriptionFactory implements StreamSubscriptionFactory {
+  constructor(
+    private baseUrl: string,
+    private options: { headers?: Record<string, string>; signal?: AbortSignal }
+  ) {}
+
+  createSubscription(
+    metadata: Record<string, unknown>,
+    runId: string,
+    streamKey: string,
+    baseUrl?: string
+  ): StreamSubscription {
+    if (!runId || !streamKey) {
+      throw new Error("runId and streamKey are required");
+    }
+
+    return new ElectricStreamSubscription(
+      `${baseUrl ?? this.baseUrl}/realtime/v2/streams/${runId}/${streamKey}`,
+      this.options
+    );
+  }
+}
+
+export class VersionedStreamSubscriptionFactory implements StreamSubscriptionFactory {
+  constructor(
+    private version1: StreamSubscriptionFactory,
+    private version2: StreamSubscriptionFactory
+  ) {}
+
+  createSubscription(
+    metadata: Record<string, unknown>,
+    runId: string,
+    streamKey: string,
+    baseUrl?: string
+  ): StreamSubscription {
+    if (!runId || !streamKey) {
+      throw new Error("runId and streamKey are required");
+    }
+
+    const version =
+      typeof metadata.$$streamsVersion === "string" ? metadata.$$streamsVersion : "v1";
+
+    if (version === "v1") {
+      return this.version1.createSubscription(metadata, runId, streamKey, baseUrl);
+    }
+
+    if (version === "v2") {
+      return this.version2.createSubscription(metadata, runId, streamKey, baseUrl);
+    }
+
+    throw new Error(`Unknown stream version: ${version}`);
+  }
+}
+
 export interface RunShapeProvider {
   onShape(callback: (shape: SubscribeRunRawShape) => Promise<void>): Promise<() => void>;
 }
 
 export type RunSubscriptionOptions = RunShapeStreamOptions & {
-  provider: RunShapeProvider;
+  runShapeStream: ReadableStream<SubscribeRunRawShape>;
   streamFactory: StreamSubscriptionFactory;
+  abortController: AbortController;
 };
 
 export class RunSubscription<TRunTypes extends AnyRunTypes> {
-  private abortController: AbortController;
   private unsubscribeShape?: () => void;
   private stream: AsyncIterableStream<RunShape<TRunTypes>>;
   private packetCache = new Map<string, any>();
@@ -181,44 +291,37 @@ export class RunSubscription<TRunTypes extends AnyRunTypes> {
   private _isRunComplete = false;
 
   constructor(private options: RunSubscriptionOptions) {
-    this.abortController = new AbortController();
     this._closeOnComplete =
       typeof options.closeOnComplete === "undefined" ? true : options.closeOnComplete;
 
-    const source = new ReadableStream<SubscribeRunRawShape>({
-      start: async (controller) => {
-        this.unsubscribeShape = await this.options.provider.onShape(async (shape) => {
-          controller.enqueue(shape);
+    this.stream = createAsyncIterableReadable(
+      this.options.runShapeStream,
+      {
+        transform: async (chunk, controller) => {
+          const run = await this.transformRunShape(chunk);
 
-          this._isRunComplete = !!shape.completedAt;
+          controller.enqueue(run);
+
+          this._isRunComplete = !!run.finishedAt;
 
           if (
             this._closeOnComplete &&
             this._isRunComplete &&
-            !this.abortController.signal.aborted
+            !this.options.abortController.signal.aborted
           ) {
-            controller.close();
-            this.abortController.abort();
+            console.log("Closing stream because run is complete");
+
+            this.options.abortController.abort();
           }
-        });
+        },
       },
-      cancel: () => {
-        this.unsubscribe();
-      },
-    });
-
-    this.stream = createAsyncIterableStream(source, {
-      transform: async (chunk, controller) => {
-        const run = await this.transformRunShape(chunk);
-
-        controller.enqueue(run);
-      },
-    });
+      this.options.abortController.signal
+    );
   }
 
   unsubscribe(): void {
-    if (!this.abortController.signal.aborted) {
-      this.abortController.abort();
+    if (!this.options.abortController.signal.aborted) {
+      this.options.abortController.abort();
     }
     this.unsubscribeShape?.();
   }
@@ -237,59 +340,68 @@ export class RunSubscription<TRunTypes extends AnyRunTypes> {
     // Keep track of which streams we've already subscribed to
     const activeStreams = new Set<string>();
 
-    return createAsyncIterableStream(this.stream, {
-      transform: async (run, controller) => {
-        controller.enqueue({
-          type: "run",
-          run,
-        });
+    return createAsyncIterableReadable(
+      this.stream,
+      {
+        transform: async (run, controller) => {
+          controller.enqueue({
+            type: "run",
+            run,
+          });
 
-        // Check for stream metadata
-        if (run.metadata && "$$streams" in run.metadata && Array.isArray(run.metadata.$$streams)) {
-          for (const streamKey of run.metadata.$$streams) {
-            if (typeof streamKey !== "string") {
-              continue;
-            }
+          // Check for stream metadata
+          if (
+            run.metadata &&
+            "$$streams" in run.metadata &&
+            Array.isArray(run.metadata.$$streams)
+          ) {
+            for (const streamKey of run.metadata.$$streams) {
+              if (typeof streamKey !== "string") {
+                continue;
+              }
 
-            if (!activeStreams.has(streamKey)) {
-              activeStreams.add(streamKey);
+              if (!activeStreams.has(streamKey)) {
+                activeStreams.add(streamKey);
 
-              const subscription = this.options.streamFactory.createSubscription(
-                run.id,
-                streamKey,
-                this.options.client?.baseUrl
-              );
+                const subscription = this.options.streamFactory.createSubscription(
+                  run.metadata,
+                  run.id,
+                  streamKey,
+                  this.options.client?.baseUrl
+                );
 
-              const stream = await subscription.subscribe();
+                const stream = await subscription.subscribe();
 
-              // Create the pipeline and start it
-              stream
-                .pipeThrough(
-                  new TransformStream({
-                    transform(chunk, controller) {
-                      controller.enqueue({
-                        type: streamKey,
-                        chunk: chunk as TStreams[typeof streamKey],
-                        run,
-                      } as StreamPartResult<RunShape<TRunTypes>, TStreams>);
-                    },
-                  })
-                )
-                .pipeTo(
-                  new WritableStream({
-                    write(chunk) {
-                      controller.enqueue(chunk);
-                    },
-                  })
-                )
-                .catch((error) => {
-                  console.error(`Error in stream ${streamKey}:`, error);
-                });
+                // Create the pipeline and start it
+                stream
+                  .pipeThrough(
+                    new TransformStream({
+                      transform(chunk, controller) {
+                        controller.enqueue({
+                          type: streamKey,
+                          chunk: chunk as TStreams[typeof streamKey],
+                          run,
+                        } as StreamPartResult<RunShape<TRunTypes>, TStreams>);
+                      },
+                    })
+                  )
+                  .pipeTo(
+                    new WritableStream({
+                      write(chunk) {
+                        controller.enqueue(chunk);
+                      },
+                    })
+                  )
+                  .catch((error) => {
+                    console.error(`Error in stream ${streamKey}:`, error);
+                  });
+              }
             }
           }
-        }
+        },
       },
-    });
+      this.options.abortController.signal
+    );
   }
 
   private async transformRunShape(row: SubscribeRunRawShape): Promise<RunShape<TRunTypes>> {
@@ -347,7 +459,7 @@ export class RunSubscription<TRunTypes extends AnyRunTypes> {
       startedAt: row.startedAt ?? undefined,
       delayedUntil: row.delayUntil ?? undefined,
       queuedAt: row.queuedAt ?? undefined,
-      error: row.error ?? undefined,
+      error: row.error ? createJsonErrorObject(row.error) : undefined,
       isTest: row.isTest,
       metadata,
     } as RunShape<TRunTypes>;

packages/core/src/v3/apiClient/stream.ts

@@ -1,5 +1,15 @@
 import { z } from "zod";
-import { ApiError } from "./errors.js";
+import {
+  FetchError,
+  isChangeMessage,
+  isControlMessage,
+  Offset,
+  ShapeStream,
+  type Message,
+  type Row,
+  type ShapeStreamInterface,
+  // @ts-ignore it's safe to import types from the client
+} from "@electric-sql/client";
 
 export type ZodShapeStreamOptions = {
   headers?: Record<string, string>;
@@ -7,14 +17,11 @@ export type ZodShapeStreamOptions = {
   signal?: AbortSignal;
 };
 
-export async function zodShapeStream<TShapeSchema extends z.ZodTypeAny>(
+export function zodShapeStream<TShapeSchema extends z.ZodTypeAny>(
   schema: TShapeSchema,
   url: string,
-  callback: (shape: z.output<TShapeSchema>) => void | Promise<void>,
   options?: ZodShapeStreamOptions
 ) {
-  const { ShapeStream, Shape, FetchError } = await import("@electric-sql/client");
-
   const stream = new ShapeStream<z.input<TShapeSchema>>({
     url,
     headers: {
@@ -25,27 +32,21 @@ export async function zodShapeStream<TShapeSchema extends z.ZodTypeAny>(
     signal: options?.signal,
   });
 
-  try {
-    const shape = new Shape(stream);
+  const readableShape = new ReadableShapeStream(stream);
 
-    const initialRows = await shape.rows;
+  return readableShape.stream.pipeThrough(
+    new TransformStream({
+      async transform(chunk, controller) {
+        const result = schema.safeParse(chunk);
 
-    for (const shapeRow of initialRows) {
-      await callback(schema.parse(shapeRow));
-    }
-
-    return shape.subscribe(async (newShape) => {
-      for (const shapeRow of newShape.rows) {
-        await callback(schema.parse(shapeRow));
-      }
-    });
-  } catch (error) {
-    if (error instanceof FetchError) {
-      throw ApiError.generate(error.status, error.json, error.message, error.headers);
-    } else {
-      throw error;
-    }
-  }
+        if (result.success) {
+          controller.enqueue(result.data);
+        } else {
+          controller.error(new Error(`Unable to parse shape: ${result.error.message}`));
+        }
+      },
+    })
+  );
 }
 
 export type AsyncIterableStream<T> = AsyncIterable<T> & ReadableStream<T>;
@@ -68,3 +69,137 @@ export function createAsyncIterableStream<S, T>(
 
   return transformedStream;
 }
+
+export function createAsyncIterableReadable<S, T>(
+  source: ReadableStream<S>,
+  transformer: Transformer<S, T>,
+  signal: AbortSignal
+): AsyncIterableStream<T> {
+  return new ReadableStream<T>({
+    async start(controller) {
+      const transformedStream = source.pipeThrough(new TransformStream(transformer));
+      const reader = transformedStream.getReader();
+
+      signal.addEventListener("abort", () => {
+        queueMicrotask(() => {
+          reader.cancel();
+          controller.close();
+        });
+      });
+
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) {
+          controller.close();
+          break;
+        }
+
+        controller.enqueue(value);
+      }
+    },
+  }) as AsyncIterableStream<T>;
+}
+
+class ReadableShapeStream<T extends Row<unknown> = Row> {
+  readonly #stream: ShapeStreamInterface<T>;
+  readonly #currentState: Map<string, T> = new Map();
+  readonly #changeStream: AsyncIterableStream<T>;
+  #error: FetchError | false = false;
+
+  constructor(stream: ShapeStreamInterface<T>) {
+    this.#stream = stream;
+
+    // Create the source stream that will receive messages
+    const source = new ReadableStream<Message<T>[]>({
+      start: (controller) => {
+        this.#stream.subscribe(
+          (messages) => controller.enqueue(messages),
+          this.#handleError.bind(this)
+        );
+      },
+    });
+
+    // Create the transformed stream that processes messages and emits complete rows
+    this.#changeStream = createAsyncIterableStream(source, {
+      transform: (messages, controller) => {
+        messages.forEach((message) => {
+          if (isChangeMessage(message)) {
+            switch (message.headers.operation) {
+              case "insert": {
+                this.#currentState.set(message.key, message.value);
+                controller.enqueue(message.value);
+                break;
+              }
+              case "update": {
+                const existingRow = this.#currentState.get(message.key);
+                if (existingRow) {
+                  const updatedRow = {
+                    ...existingRow,
+                    ...message.value,
+                  };
+                  this.#currentState.set(message.key, updatedRow);
+                  controller.enqueue(updatedRow);
+                } else {
+                  this.#currentState.set(message.key, message.value);
+                  controller.enqueue(message.value);
+                }
+                break;
+              }
+            }
+          }
+
+          if (isControlMessage(message)) {
+            switch (message.headers.control) {
+              case "must-refetch":
+                this.#currentState.clear();
+                this.#error = false;
+                break;
+            }
+          }
+        });
+      },
+    });
+  }
+
+  get stream(): AsyncIterableStream<T> {
+    return this.#changeStream;
+  }
+
+  get isUpToDate(): boolean {
+    return this.#stream.isUpToDate;
+  }
+
+  get lastOffset(): Offset {
+    return this.#stream.lastOffset;
+  }
+
+  get handle(): string | undefined {
+    return this.#stream.shapeHandle;
+  }
+
+  get error() {
+    return this.#error;
+  }
+
+  lastSyncedAt(): number | undefined {
+    return this.#stream.lastSyncedAt();
+  }
+
+  lastSynced() {
+    return this.#stream.lastSynced();
+  }
+
+  isLoading() {
+    return this.#stream.isLoading();
+  }
+
+  isConnected(): boolean {
+    return this.#stream.isConnected();
+  }
+
+  #handleError(e: Error): void {
+    if (e instanceof FetchError) {
+      this.#error = e;
+    }
+  }
+}

packages/core/src/v3/runMetadata/manager.ts

@@ -20,7 +20,8 @@ export class StandardMetadataManager implements RunMetadataManager {
 
   constructor(
     private apiClient: ApiClient,
-    private streamsBaseUrl: string
+    private streamsBaseUrl: string,
+    private streamsVersion: "v1" | "v2" = "v1"
   ) {}
 
   public enterWithMetadata(metadata: Record<string, DeserializedJson>): void {
@@ -231,6 +232,7 @@ export class StandardMetadataManager implements RunMetadataManager {
     try {
       // Add the key to the special stream metadata object
       this.appendKey(`$$streams`, key);
+      this.setKey("$$streamsVersion", this.streamsVersion);
 
       await this.flush();
 
@@ -239,7 +241,9 @@ export class StandardMetadataManager implements RunMetadataManager {
         runId: this.runId,
         iterator: $value[Symbol.asyncIterator](),
         baseUrl: this.streamsBaseUrl,
+        headers: this.apiClient.getHeaders(),
         signal,
+        version: this.streamsVersion,
       });
 
       this.activeStreams.set(key, streamInstance);

packages/core/src/v3/runMetadata/metadataStream.ts

@@ -3,7 +3,9 @@ export type MetadataOptions<T> = {
   runId: string;
   key: string;
   iterator: AsyncIterator<T>;
+  headers?: Record<string, string>;
   signal?: AbortSignal;
+  version?: "v1" | "v2";
 };
 
 export class MetadataStream<T> {
@@ -43,7 +45,6 @@ export class MetadataStream<T> {
   private initializeServerStream(): Promise<void | Response> {
     const serverIterator = this.serverIterator;
 
-    // TODO: Why is this only sending stuff to the server at the end of the run?
     const serverStream = new ReadableStream({
       async pull(controller) {
         try {
@@ -62,10 +63,12 @@ export class MetadataStream<T> {
     });
 
     return fetch(
-      `${this.options.baseUrl}/realtime/v1/streams/${this.options.runId}/${this.options.key}`,
+      `${this.options.baseUrl}/realtime/${this.options.version ?? "v1"}/streams/${
+        this.options.runId
+      }/${this.options.key}`,
       {
         method: "POST",
-        headers: {},
+        headers: this.options.headers ?? {},
         body: serverStream,
         // @ts-expect-error
         duplex: "half",

packages/core/src/v3/schemas/api.ts

@@ -1,6 +1,6 @@
 import { z } from "zod";
 import { DeserializedJsonSchema } from "../../schemas/json.js";
-import { SerializedError } from "./common.js";
+import { SerializedError, TaskRunError } from "./common.js";
 import { BackgroundWorkerMetadata } from "./resources.js";
 import { QueueOptions } from "./schemas.js";
 
@@ -708,7 +708,7 @@ export const SubscribeRunRawShape = z.object({
   output: z.string().nullish(),
   outputType: z.string().nullish(),
   runTags: z.array(z.string()).nullish().default([]),
-  error: SerializedError.nullish(),
+  error: TaskRunError.nullish(),
 });
 
 export type SubscribeRunRawShape = z.infer<typeof SubscribeRunRawShape>;
@@ -727,3 +727,16 @@ export const RetrieveBatchResponse = z.object({
 });
 
 export type RetrieveBatchResponse = z.infer<typeof RetrieveBatchResponse>;
+
+export const SubscribeRealtimeStreamChunkRawShape = z.object({
+  id: z.string(),
+  runId: z.string(),
+  sequence: z.number(),
+  key: z.string(),
+  value: z.string(),
+  createdAt: z.coerce.date(),
+});
+
+export type SubscribeRealtimeStreamChunkRawShape = z.infer<
+  typeof SubscribeRealtimeStreamChunkRawShape
+>;

packages/core/src/v3/types/tasks.ts

@@ -592,7 +592,8 @@ export interface Task<TIdentifier extends string, TInput = void, TOutput = any>
    * ```
    */
   batchTriggerAndWait: (
-    items: Array<BatchTriggerAndWaitItem<TInput>>
+    items: Array<BatchTriggerAndWaitItem<TInput>>,
+    options?: BatchTriggerAndWaitOptions
   ) => Promise<BatchResult<TIdentifier, TOutput>>;
 }
 
@@ -781,6 +782,32 @@ export type TriggerAndWaitOptions = Omit<TriggerOptions, "idempotencyKey" | "ide
 export type BatchTriggerOptions = {
   idempotencyKey?: IdempotencyKey | string | string[];
   idempotencyKeyTTL?: string;
+
+  /**
+   * When true, triggers tasks sequentially in batch order. This ensures ordering but may be slower,
+   * especially for large batches.
+   *
+   * When false (default), triggers tasks in parallel for better performance, but order is not guaranteed.
+   *
+   * Note: This only affects the order of run creation, not the actual task execution.
+   *
+   * @default false
+   */
+  triggerSequentially?: boolean;
+};
+
+export type BatchTriggerAndWaitOptions = {
+  /**
+   * When true, triggers tasks sequentially in batch order. This ensures ordering but may be slower,
+   * especially for large batches.
+   *
+   * When false (default), triggers tasks in parallel for better performance, but order is not guaranteed.
+   *
+   * Note: This only affects the order of run creation, not the actual task execution.
+   *
+   * @default false
+   */
+  triggerSequentially?: boolean;
 };
 
 export type TaskMetadataWithFunctions = TaskMetadata & {

packages/core/test/runStream.test.ts

@@ -1,10 +1,8 @@
-import { describe, it, expect } from "vitest";
+import { describe, expect, it } from "vitest";
 import {
-  AnyRunShape,
   RunSubscription,
   StreamSubscription,
   StreamSubscriptionFactory,
-  type RunShapeProvider,
 } from "../src/v3/apiClient/runStream.js";
 import type { SubscribeRunRawShape } from "../src/v3/schemas/api.js";
 
@@ -33,64 +31,54 @@ class TestStreamSubscriptionFactory implements StreamSubscriptionFactory {
     this.streams.set(`${runId}:${streamKey}`, chunks);
   }
 
-  createSubscription(runId: string, streamKey: string): StreamSubscription {
+  createSubscription(
+    metadata: Record<string, unknown>,
+    runId: string,
+    streamKey: string
+  ): StreamSubscription {
     const chunks = this.streams.get(`${runId}:${streamKey}`) ?? [];
     return new TestStreamSubscription(chunks);
   }
 }
 
-// Create a real test provider that uses an array of shapes
-class TestShapeProvider implements RunShapeProvider {
-  private shapes: SubscribeRunRawShape[];
-  private unsubscribed = false;
-
-  constructor(shapes: SubscribeRunRawShape[]) {
-    this.shapes = shapes;
-  }
-
-  async onShape(callback: (shape: SubscribeRunRawShape) => Promise<void>): Promise<() => void> {
-    // Process all shapes immediately
-    for (const shape of this.shapes) {
-      if (this.unsubscribed) break;
-      await callback(shape);
-    }
-
-    return () => {
-      this.unsubscribed = true;
-    };
-  }
+// Remove the RunShapeProvider implementations and replace with stream creators
+function createTestShapeStream(
+  shapes: SubscribeRunRawShape[]
+): ReadableStream<SubscribeRunRawShape> {
+  return new ReadableStream({
+    start: async (controller) => {
+      // Emit all shapes immediately
+      for (const shape of shapes) {
+        controller.enqueue(shape);
+      }
+      controller.close();
+    },
+  });
 }
 
-// Add this new provider that can emit shapes over time
-class DelayedTestShapeProvider implements RunShapeProvider {
-  private shapes: SubscribeRunRawShape[];
-  private unsubscribed = false;
-  private currentShapeIndex = 0;
-
-  constructor(shapes: SubscribeRunRawShape[]) {
-    this.shapes = shapes;
-  }
-
-  async onShape(callback: (shape: SubscribeRunRawShape) => Promise<void>): Promise<() => void> {
-    // Only emit the first shape immediately
-    if (this.shapes.length > 0) {
-      await callback(this.shapes[this.currentShapeIndex++]!);
-    }
-
-    // Set up an interval to emit remaining shapes
-    const interval = setInterval(async () => {
-      if (this.unsubscribed || this.currentShapeIndex >= this.shapes.length) {
-        clearInterval(interval);
-        return;
+function createDelayedTestShapeStream(
+  shapes: SubscribeRunRawShape[]
+): ReadableStream<SubscribeRunRawShape> {
+  return new ReadableStream({
+    start: async (controller) => {
+      // Emit first shape immediately
+      if (shapes.length > 0) {
+        controller.enqueue(shapes[0]);
       }
-      await callback(this.shapes[this.currentShapeIndex++]!);
-    }, 100);
 
-    return () => {
-      this.unsubscribed = true;
-      clearInterval(interval);
-    };
-  }
+      let currentShapeIndex = 1;
+
+      // Emit remaining shapes with delay
+      const interval = setInterval(() => {
+        if (currentShapeIndex >= shapes.length) {
+          clearInterval(interval);
+          controller.close();
+          return;
+        }
+        controller.enqueue(shapes[currentShapeIndex++]!);
+      }, 100);
+    },
+  });
 }
 
 describe("RunSubscription", () => {
@@ -114,9 +102,10 @@ describe("RunSubscription", () => {
     ];
 
     const subscription = new RunSubscription({
-      provider: new TestShapeProvider(shapes),
+      runShapeStream: createTestShapeStream(shapes),
       streamFactory: new TestStreamSubscriptionFactory(),
       closeOnComplete: true,
+      abortController: new AbortController(),
     });
 
     const results = await convertAsyncIterableToArray(subscription);
@@ -153,9 +142,10 @@ describe("RunSubscription", () => {
     ];
 
     const subscription = new RunSubscription({
-      provider: new TestShapeProvider(shapes),
+      runShapeStream: createTestShapeStream(shapes),
       streamFactory: new TestStreamSubscriptionFactory(),
       closeOnComplete: true,
+      abortController: new AbortController(),
     });
 
     const results = await convertAsyncIterableToArray(subscription);
@@ -205,9 +195,10 @@ describe("RunSubscription", () => {
     ];
 
     const subscription = new RunSubscription({
-      provider: new DelayedTestShapeProvider(shapes),
+      runShapeStream: createDelayedTestShapeStream(shapes),
       streamFactory: new TestStreamSubscriptionFactory(),
       closeOnComplete: false,
+      abortController: new AbortController(),
     });
 
     // Collect 2 results
@@ -257,8 +248,9 @@ describe("RunSubscription", () => {
     ];
 
     const subscription = new RunSubscription({
-      provider: new TestShapeProvider(shapes),
+      runShapeStream: createTestShapeStream(shapes),
       streamFactory,
+      abortController: new AbortController(),
     });
 
     const results = await collectNResults(
@@ -289,9 +281,13 @@ describe("RunSubscription", () => {
 
     // Override createSubscription to count calls
     const originalCreate = streamFactory.createSubscription.bind(streamFactory);
-    streamFactory.createSubscription = (runId: string, streamKey: string) => {
+    streamFactory.createSubscription = (
+      metadata: Record<string, unknown>,
+      runId: string,
+      streamKey: string
+    ) => {
       streamCreationCount++;
-      return originalCreate(runId, streamKey);
+      return originalCreate(metadata, runId, streamKey);
     };
 
     // Set up test chunks
@@ -342,8 +338,9 @@ describe("RunSubscription", () => {
     ];
 
     const subscription = new RunSubscription({
-      provider: new TestShapeProvider(shapes),
+      runShapeStream: createTestShapeStream(shapes),
       streamFactory,
+      abortController: new AbortController(),
     });
 
     const results = await collectNResults(
@@ -421,8 +418,9 @@ describe("RunSubscription", () => {
     ];
 
     const subscription = new RunSubscription({
-      provider: new TestShapeProvider(shapes),
+      runShapeStream: createTestShapeStream(shapes),
       streamFactory,
+      abortController: new AbortController(),
     });
 
     const results = await collectNResults(
@@ -467,110 +465,6 @@ describe("RunSubscription", () => {
       run: { id: "run_123" },
     });
   });
-
-  it("should handle streams that appear in different run updates", async () => {
-    const streamFactory = new TestStreamSubscriptionFactory();
-
-    // Set up test chunks for two different streams
-    streamFactory.setStreamChunks("run_123", "openai", [
-      { id: "openai1", content: "Hello" },
-      { id: "openai2", content: "World" },
-    ]);
-    streamFactory.setStreamChunks("run_123", "anthropic", [
-      { id: "claude1", message: "Hi" },
-      { id: "claude2", message: "There" },
-    ]);
-
-    const shapes = [
-      // First run update - only has openai stream
-      {
-        id: "123",
-        friendlyId: "run_123",
-        taskIdentifier: "multi-streaming",
-        status: "EXECUTING",
-        createdAt: new Date(),
-        updatedAt: new Date(),
-        number: 1,
-        usageDurationMs: 100,
-        costInCents: 0,
-        baseCostInCents: 0,
-        isTest: false,
-        runTags: [],
-        metadata: JSON.stringify({
-          $$streams: ["openai"],
-        }),
-        metadataType: "application/json",
-      },
-      // Second run update - adds anthropic stream
-      {
-        id: "123",
-        friendlyId: "run_123",
-        taskIdentifier: "multi-streaming",
-        status: "EXECUTING",
-        createdAt: new Date(),
-        updatedAt: new Date(),
-        number: 1,
-        usageDurationMs: 200,
-        costInCents: 0,
-        baseCostInCents: 0,
-        isTest: false,
-        runTags: [],
-        metadata: JSON.stringify({
-          $$streams: ["openai", "anthropic"],
-        }),
-        metadataType: "application/json",
-      },
-      // Final run update - marks as complete
-      {
-        id: "123",
-        friendlyId: "run_123",
-        taskIdentifier: "multi-streaming",
-        status: "COMPLETED_SUCCESSFULLY",
-        createdAt: new Date(),
-        updatedAt: new Date(),
-        completedAt: new Date(),
-        number: 1,
-        usageDurationMs: 300,
-        costInCents: 0,
-        baseCostInCents: 0,
-        isTest: false,
-        runTags: [],
-        metadata: JSON.stringify({
-          $$streams: ["openai", "anthropic"],
-        }),
-        metadataType: "application/json",
-      },
-    ];
-
-    const subscription = new RunSubscription({
-      provider: new TestShapeProvider(shapes),
-      streamFactory,
-      closeOnComplete: true,
-    });
-
-    const results = await collectNResults(
-      subscription.withStreams<{
-        openai: { id: string; content: string };
-        anthropic: { id: string; message: string };
-      }>(),
-      7 // 3 runs + 2 openai chunks + 2 anthropic chunks
-    );
-
-    expect(results).toHaveLength(7);
-
-    // Verify run updates
-    const runUpdates = results.filter((r) => r.type === "run");
-    expect(runUpdates).toHaveLength(3);
-    expect(runUpdates[2]!.run.status).toBe("COMPLETED");
-
-    // Verify openai chunks
-    const openaiChunks = results.filter((r) => r.type === "openai");
-    expect(openaiChunks).toHaveLength(2);
-
-    // Verify anthropic chunks
-    const anthropicChunks = results.filter((r) => r.type === "anthropic");
-    expect(anthropicChunks).toHaveLength(2);
-  });
 });
 
 export async function convertAsyncIterableToArray<T>(iterable: AsyncIterable<T>): Promise<T[]> {
@@ -603,7 +497,12 @@ async function collectNResults<T>(
     promise,
     new Promise<T[]>((_, reject) =>
       setTimeout(
-        () => reject(new Error(`Timeout waiting for ${count} results after ${timeoutMs}ms`)),
+        () =>
+          reject(
+            new Error(
+              `Timeout waiting for ${count} results after ${timeoutMs}ms, but only had ${results.length}`
+            )
+          ),
         timeoutMs
       )
     ),

packages/react-hooks/CHANGELOG.md

@@ -1,5 +1,20 @@
 # @trigger.dev/react-hooks
 
+## 3.3.6
+
+### Patch Changes
+
+- Realtime streams now powered by electric. Also, this change fixes a realtime bug that was causing too many re-renders, even on records that didn't change ([#1541](https://github.com/triggerdotdev/trigger.dev/pull/1541))
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.6`
+
+## 3.3.5
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.5`
+
 ## 3.3.4
 
 ### Patch Changes

packages/react-hooks/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/react-hooks",
-  "version": "3.3.4",
+  "version": "3.3.6",
   "description": "trigger.dev react hooks",
   "license": "MIT",
   "publishConfig": {
@@ -37,7 +37,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:^3.3.4",
+    "@trigger.dev/core": "workspace:^3.3.6",
     "swr": "^2.2.5"
   },
   "devDependencies": {

packages/react-hooks/src/hooks/useRealtime.ts

@@ -12,6 +12,16 @@ export type UseRealtimeRunOptions = UseApiClientOptions & {
   experimental_throttleInMs?: number;
 };
 
+export type UseRealtimeSingleRunOptions<TTask extends AnyTask = AnyTask> = UseRealtimeRunOptions & {
+  /**
+   * Callback this is called when the run completes, an error occurs, or the subscription is stopped.
+   *
+   * @param {RealtimeRun<TTask>} run - The run object
+   * @param {Error} [err] - The error that occurred
+   */
+  onComplete?: (run: RealtimeRun<TTask>, err?: Error) => void;
+};
+
 export type UseRealtimeRunInstance<TTask extends AnyTask = AnyTask> = {
   run: RealtimeRun<TTask> | undefined;
 
@@ -28,7 +38,7 @@ export type UseRealtimeRunInstance<TTask extends AnyTask = AnyTask> = {
  *
  * @template TTask - The type of the task
  * @param {string} [runId] - The unique identifier of the run to subscribe to
- * @param {UseRealtimeRunOptions} [options] - Configuration options for the subscription
+ * @param {UseRealtimeSingleRunOptions} [options] - Configuration options for the subscription
  * @returns {UseRealtimeRunInstance<TTask>} An object containing the current state of the run, error handling, and control methods
  *
  * @example
@@ -40,7 +50,7 @@ export type UseRealtimeRunInstance<TTask extends AnyTask = AnyTask> = {
 
 export function useRealtimeRun<TTask extends AnyTask>(
   runId?: string,
-  options?: UseRealtimeRunOptions
+  options?: UseRealtimeSingleRunOptions<TTask>
 ): UseRealtimeRunInstance<TTask> {
   const hookId = useId();
   const idKey = options?.id ?? hookId;
@@ -48,17 +58,17 @@ export function useRealtimeRun<TTask extends AnyTask>(
   // Store the streams state in SWR, using the idKey as the key to share states.
   const { data: run, mutate: mutateRun } = useSWR<RealtimeRun<TTask>>([idKey, "run"], null);
 
-  // Keep the latest streams in a ref.
-  const runRef = useRef<RealtimeRun<TTask> | undefined>();
-  useEffect(() => {
-    runRef.current = run;
-  }, [run]);
-
   const { data: error = undefined, mutate: setError } = useSWR<undefined | Error>(
     [idKey, "error"],
     null
   );
 
+  // Add state to track when the subscription is complete
+  const { data: isComplete = false, mutate: setIsComplete } = useSWR<boolean>(
+    [idKey, "complete"],
+    null
+  );
+
   // Abort controller to cancel the current API call.
   const abortControllerRef = useRef<AbortController | null>(null);
 
@@ -93,9 +103,19 @@ export function useRealtimeRun<TTask extends AnyTask>(
       if (abortControllerRef.current) {
         abortControllerRef.current = null;
       }
+
+      // Mark the subscription as complete
+      setIsComplete(true);
     }
   }, [runId, mutateRun, abortControllerRef, apiClient, setError]);
 
+  // Effect to handle onComplete callback
+  useEffect(() => {
+    if (isComplete && options?.onComplete && run) {
+      options.onComplete(run, error);
+    }
+  }, [isComplete, run, error, options?.onComplete]);
+
   useEffect(() => {
     if (typeof options?.enabled === "boolean" && !options.enabled) {
       return;
@@ -157,7 +177,7 @@ export function useRealtimeRunWithStreams<
   TStreams extends Record<string, any> = Record<string, any>,
 >(
   runId?: string,
-  options?: UseRealtimeRunOptions
+  options?: UseRealtimeSingleRunOptions<TTask>
 ): UseRealtimeRunWithStreamsInstance<TTask, TStreams> {
   const hookId = useId();
   const idKey = options?.id ?? hookId;
@@ -182,11 +202,11 @@ export function useRealtimeRunWithStreams<
   // Store the streams state in SWR, using the idKey as the key to share states.
   const { data: run, mutate: mutateRun } = useSWR<RealtimeRun<TTask>>([idKey, "run"], null);
 
-  // Keep the latest streams in a ref.
-  const runRef = useRef<RealtimeRun<TTask> | undefined>();
-  useEffect(() => {
-    runRef.current = run;
-  }, [run]);
+  // Add state to track when the subscription is complete
+  const { data: isComplete = false, mutate: setIsComplete } = useSWR<boolean>(
+    [idKey, "complete"],
+    null
+  );
 
   const { data: error = undefined, mutate: setError } = useSWR<undefined | Error>(
     [idKey, "error"],
@@ -235,9 +255,19 @@ export function useRealtimeRunWithStreams<
       if (abortControllerRef.current) {
         abortControllerRef.current = null;
       }
+
+      // Mark the subscription as complete
+      setIsComplete(true);
     }
   }, [runId, mutateRun, mutateStreams, streamsRef, abortControllerRef, apiClient, setError]);
 
+  // Effect to handle onComplete callback
+  useEffect(() => {
+    if (isComplete && options?.onComplete && run) {
+      options.onComplete(run, error);
+    }
+  }, [isComplete, run, error, options?.onComplete]);
+
   useEffect(() => {
     if (typeof options?.enabled === "boolean" && !options.enabled) {
       return;

packages/rsc/CHANGELOG.md

@@ -1,5 +1,19 @@
 # @trigger.dev/rsc
 
+## 3.3.6
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.6`
+
+## 3.3.5
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.5`
+
 ## 3.3.4
 
 ### Patch Changes

packages/rsc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/rsc",
-  "version": "3.3.4",
+  "version": "3.3.6",
   "description": "trigger.dev rsc",
   "license": "MIT",
   "publishConfig": {
@@ -37,14 +37,14 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:^3.3.4",
+    "@trigger.dev/core": "workspace:^3.3.6",
     "mlly": "^1.7.1",
     "react": "19.0.0-rc.1",
     "react-dom": "19.0.0-rc.1"
   },
   "devDependencies": {
     "@arethetypeswrong/cli": "^0.15.4",
-    "@trigger.dev/build": "workspace:^3.3.4",
+    "@trigger.dev/build": "workspace:^3.3.6",
     "@types/node": "^20.14.14",
     "@types/react": "*",
     "@types/react-dom": "*",

packages/trigger-sdk/CHANGELOG.md

@@ -1,5 +1,21 @@
 # @trigger.dev/sdk
 
+## 3.3.6
+
+### Patch Changes
+
+- Realtime streams now powered by electric. Also, this change fixes a realtime bug that was causing too many re-renders, even on records that didn't change ([#1541](https://github.com/triggerdotdev/trigger.dev/pull/1541))
+- Add option to trigger batched items sequentially, and default to parallel triggering which is faster ([#1536](https://github.com/triggerdotdev/trigger.dev/pull/1536))
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.6`
+
+## 3.3.5
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.5`
+
 ## 3.3.4
 
 ### Patch Changes

packages/trigger-sdk/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/sdk",
-  "version": "3.3.4",
+  "version": "3.3.6",
   "description": "trigger.dev Node.JS SDK",
   "license": "MIT",
   "publishConfig": {
@@ -48,7 +48,7 @@
     "@opentelemetry/api": "1.9.0",
     "@opentelemetry/api-logs": "0.52.1",
     "@opentelemetry/semantic-conventions": "1.25.1",
-    "@trigger.dev/core": "workspace:3.3.4",
+    "@trigger.dev/core": "workspace:3.3.6",
     "chalk": "^5.2.0",
     "cronstrue": "^2.21.0",
     "debug": "^4.3.4",

packages/trigger-sdk/src/v3/shared.ts

@@ -74,6 +74,7 @@ import type {
   TriggerApiRequestOptions,
   TriggerOptions,
   AnyTaskRunResult,
+  BatchTriggerAndWaitOptions,
 } from "@trigger.dev/core/v3";
 
 export type {
@@ -181,7 +182,7 @@ export function createTask<
           });
       }, params.id);
     },
-    batchTriggerAndWait: async (items) => {
+    batchTriggerAndWait: async (items, options) => {
       const taskMetadata = taskCatalog.getTaskManifest(params.id);
 
       return await batchTriggerAndWait_internal<TIdentifier, TInput, TOutput>(
@@ -191,6 +192,7 @@ export function createTask<
         params.id,
         items,
         undefined,
+        options,
         undefined,
         customQueue
       );
@@ -326,7 +328,7 @@ export function createSchemaTask<
           });
       }, params.id);
     },
-    batchTriggerAndWait: async (items) => {
+    batchTriggerAndWait: async (items, options) => {
       const taskMetadata = taskCatalog.getTaskManifest(params.id);
 
       return await batchTriggerAndWait_internal<TIdentifier, inferSchemaIn<TSchema>, TOutput>(
@@ -336,6 +338,7 @@ export function createSchemaTask<
         params.id,
         items,
         parsePayload,
+        options,
         undefined,
         customQueue
       );
@@ -469,13 +472,14 @@ export function triggerAndWait<TTask extends AnyTask>(
 export async function batchTriggerAndWait<TTask extends AnyTask>(
   id: TaskIdentifier<TTask>,
   items: Array<BatchItem<TaskPayload<TTask>>>,
+  options?: BatchTriggerAndWaitOptions,
   requestOptions?: ApiRequestOptions
 ): Promise<BatchResult<TaskIdentifier<TTask>, TaskOutput<TTask>>> {
   return await batchTriggerAndWait_internal<
     TaskIdentifier<TTask>,
     TaskPayload<TTask>,
     TaskOutput<TTask>
-  >("tasks.batchTriggerAndWait()", id, items, undefined, requestOptions);
+  >("tasks.batchTriggerAndWait()", id, items, undefined, options, requestOptions);
 }
 
 /**
@@ -618,6 +622,7 @@ export async function batchTriggerById<TTask extends AnyTask>(
       spanParentAsLink: true,
       idempotencyKey: await makeIdempotencyKey(options?.idempotencyKey),
       idempotencyKeyTTL: options?.idempotencyKeyTTL,
+      processingStrategy: options?.triggerSequentially ? "sequential" : undefined,
     },
     {
       name: "batch.trigger()",
@@ -740,6 +745,7 @@ export async function batchTriggerById<TTask extends AnyTask>(
  */
 export async function batchTriggerByIdAndWait<TTask extends AnyTask>(
   items: Array<BatchByIdAndWaitItem<InferRunTypes<TTask>>>,
+  options?: BatchTriggerAndWaitOptions,
   requestOptions?: TriggerApiRequestOptions
 ): Promise<BatchByIdResult<TTask>> {
   const ctx = taskContext.ctx;
@@ -786,7 +792,9 @@ export async function batchTriggerByIdAndWait<TTask extends AnyTask>(
           ),
           dependentAttempt: ctx.attempt.id,
         },
-        {},
+        {
+          processingStrategy: options?.triggerSequentially ? "sequential" : undefined,
+        },
         requestOptions
       );
 
@@ -948,6 +956,7 @@ export async function batchTriggerTasks<TTasks extends readonly AnyTask[]>(
       spanParentAsLink: true,
       idempotencyKey: await makeIdempotencyKey(options?.idempotencyKey),
       idempotencyKeyTTL: options?.idempotencyKeyTTL,
+      processingStrategy: options?.triggerSequentially ? "sequential" : undefined,
     },
     {
       name: "batch.triggerByTask()",
@@ -1072,6 +1081,7 @@ export async function batchTriggerAndWaitTasks<TTasks extends readonly AnyTask[]
   items: {
     [K in keyof TTasks]: BatchByTaskAndWaitItem<TTasks[K]>;
   },
+  options?: BatchTriggerAndWaitOptions,
   requestOptions?: TriggerApiRequestOptions
 ): Promise<BatchByTaskResult<TTasks>> {
   const ctx = taskContext.ctx;
@@ -1118,7 +1128,9 @@ export async function batchTriggerAndWaitTasks<TTasks extends readonly AnyTask[]
           ),
           dependentAttempt: ctx.attempt.id,
         },
-        {},
+        {
+          processingStrategy: options?.triggerSequentially ? "sequential" : undefined,
+        },
         requestOptions
       );
 
@@ -1256,6 +1268,7 @@ async function batchTrigger_internal<TRunTypes extends AnyRunTypes>(
       spanParentAsLink: true,
       idempotencyKey: await makeIdempotencyKey(options?.idempotencyKey),
       idempotencyKeyTTL: options?.idempotencyKeyTTL,
+      processingStrategy: options?.triggerSequentially ? "sequential" : undefined,
     },
     {
       name,
@@ -1377,6 +1390,7 @@ async function batchTriggerAndWait_internal<TIdentifier extends string, TPayload
   id: TIdentifier,
   items: Array<BatchTriggerAndWaitItem<TPayload>>,
   parsePayload?: SchemaParseFn<TPayload>,
+  options?: BatchTriggerAndWaitOptions,
   requestOptions?: ApiRequestOptions,
   queue?: QueueOptions
 ): Promise<BatchResult<TIdentifier, TOutput>> {
@@ -1420,7 +1434,9 @@ async function batchTriggerAndWait_internal<TIdentifier extends string, TPayload
           ),
           dependentAttempt: ctx.attempt.id,
         },
-        {},
+        {
+          processingStrategy: options?.triggerSequentially ? "sequential" : undefined,
+        },
         requestOptions
       );
 

pnpm-lock.yaml

@@ -1015,7 +1015,7 @@ importers:
   packages/build:
     dependencies:
       '@trigger.dev/core':
-        specifier: workspace:3.3.4
+        specifier: workspace:3.3.6
         version: link:../core
       pkg-types:
         specifier: ^1.1.3
@@ -1094,10 +1094,10 @@ importers:
         specifier: 1.25.1
         version: 1.25.1
       '@trigger.dev/build':
-        specifier: workspace:3.3.4
+        specifier: workspace:3.3.6
         version: link:../build
       '@trigger.dev/core':
-        specifier: workspace:3.3.4
+        specifier: workspace:3.3.6
         version: link:../core
       c12:
         specifier: ^1.11.1
@@ -1263,8 +1263,8 @@ importers:
   packages/core:
     dependencies:
       '@electric-sql/client':
-        specifier: 0.7.1
-        version: 0.7.1
+        specifier: 0.9.0
+        version: 0.9.0
       '@google-cloud/precise-date':
         specifier: ^4.0.0
         version: 4.0.0
@@ -1390,7 +1390,7 @@ importers:
   packages/react-hooks:
     dependencies:
       '@trigger.dev/core':
-        specifier: workspace:^3.3.4
+        specifier: workspace:^3.3.6
         version: link:../core
       react:
         specifier: '>=18 || >=19.0.0-beta'
@@ -1430,7 +1430,7 @@ importers:
   packages/rsc:
     dependencies:
       '@trigger.dev/core':
-        specifier: workspace:^3.3.4
+        specifier: workspace:^3.3.6
         version: link:../core
       mlly:
         specifier: ^1.7.1
@@ -1446,7 +1446,7 @@ importers:
         specifier: ^0.15.4
         version: 0.15.4
       '@trigger.dev/build':
-        specifier: workspace:^3.3.4
+        specifier: workspace:^3.3.6
         version: link:../build
       '@types/node':
         specifier: ^20.14.14
@@ -1482,7 +1482,7 @@ importers:
         specifier: 1.25.1
         version: 1.25.1
       '@trigger.dev/core':
-        specifier: workspace:3.3.4
+        specifier: workspace:3.3.6
         version: link:../core
       chalk:
         specifier: ^5.2.0
@@ -5112,8 +5112,8 @@ packages:
       '@rollup/rollup-darwin-arm64': 4.21.3
     dev: false
 
-  /@electric-sql/client@0.7.1:
-    resolution: {integrity: sha512-NpKEn5hDSy+NaAdG9Ql8kIGfjrj/XfakJOOHTTutb99db3Dza0uUfnkqycFpyUAarFMQ4hYSKgx8AbOm1PCeFQ==}
+  /@electric-sql/client@0.9.0:
+    resolution: {integrity: sha512-UL2Gep9wPdGMTE0oEWVi0HA8R293R2OzFfHeAsN2LABYYl/boXss7nseNEiIV5+RjHPH7Tm8NsjH9iJW2rZkrQ==}
     optionalDependencies:
       '@rollup/rollup-darwin-arm64': 4.21.3
     dev: false

references/nextjs-realtime/src/app/realtime/page.tsx

@@ -0,0 +1,12 @@
+import RealtimeComparison from "@/components/RealtimeComparison";
+import { auth } from "@trigger.dev/sdk/v3";
+
+export default async function RuntimeComparisonPage() {
+  const accessToken = await auth.createTriggerPublicToken("openai-streaming");
+
+  return (
+    <main className="flex min-h-screen items-center justify-center p-4 bg-gray-900">
+      <RealtimeComparison accessToken={accessToken} />
+    </main>
+  );
+}

references/nextjs-realtime/src/app/runs/[id]/ClientRunDetails.tsx

@@ -27,6 +27,9 @@ function RunDetailsWrapper({
   const { run, error } = useRealtimeRun<typeof exampleTask>(runId, {
     accessToken,
     enabled: accessToken !== undefined,
+    onComplete: (run) => {
+      console.log("Run completed!", run);
+    },
   });
 
   if (error) {

references/nextjs-realtime/src/components/RealtimeComparison.tsx

@@ -0,0 +1,98 @@
+"use client";
+
+import { Button } from "@/components/ui/button";
+import { useRealtimeRunWithStreams, useTaskTrigger } from "@trigger.dev/react-hooks";
+import type { STREAMS, openaiStreaming } from "@/trigger/ai";
+
+export default function RealtimeComparison({ accessToken }: { accessToken: string }) {
+  const trigger = useTaskTrigger<typeof openaiStreaming>("openai-streaming", {
+    accessToken,
+    baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
+  });
+
+  const { streams, stop, run } = useRealtimeRunWithStreams<typeof openaiStreaming, STREAMS>(
+    trigger.handle?.id,
+    {
+      accessToken: trigger.handle?.publicAccessToken,
+      enabled: !!trigger.handle,
+      baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
+      onComplete: (...args) => {
+        console.log("Run completed!", args);
+      },
+    }
+  );
+
+  return (
+    <div className="flex flex-col h-screen bg-gray-900 text-gray-200 text-xs">
+      <div className="p-4">
+        <Button
+          className="bg-gray-100 text-gray-900 hover:bg-gray-200 font-semibold text-xs"
+          onClick={() => {
+            trigger.submit({
+              model: "gpt-4o-mini",
+              prompt:
+                "Based on the temperature, will I need to wear extra clothes today in San Fransico? Please be detailed.",
+            });
+          }}
+        >
+          Debug LLM Streaming
+        </Button>
+
+        {run && (
+          <Button
+            className="bg-gray-100 text-gray-900 hover:bg-gray-200 font-semibold text-xs ml-8"
+            onClick={() => {
+              stop();
+            }}
+          >
+            Stop Streaming
+          </Button>
+        )}
+      </div>
+      <div className="flex-grow flex overflow-hidden">
+        <div className="w-1/2 border-r border-gray-700 overflow-auto">
+          <table className="w-full table-fixed">
+            <thead>
+              <tr className="bg-gray-800">
+                <th className="w-16 p-2 text-left">ID</th>
+                <th className="p-2 text-left">Data</th>
+              </tr>
+            </thead>
+            <tbody>
+              {(streams.openai ?? []).map((part, i) => (
+                <tr key={i} className="border-b border-gray-700">
+                  <td className="w-16 p-2 truncate">{i + 1}</td>
+                  <td className="p-2">
+                    <div className="font-mono whitespace-nowrap overflow-x-auto">
+                      {JSON.stringify(part)}
+                    </div>
+                  </td>
+                </tr>
+              ))}
+            </tbody>
+          </table>
+        </div>
+        <div className="w-1/2 overflow-auto">
+          <table className="w-full table-fixed">
+            <thead>
+              <tr className="bg-gray-800">
+                <th className="w-16 p-2 text-left">ID</th>
+                <th className="p-2 text-left">Data</th>
+              </tr>
+            </thead>
+            <tbody>
+              {(streams.openaiText ?? []).map((text, i) => (
+                <tr key={i} className="border-b border-gray-700">
+                  <td className="w-16 p-2 truncate">{i + 1}</td>
+                  <td className="p-2">
+                    <div className="font-mono whitespace-nowrap overflow-x-auto">{text}</div>
+                  </td>
+                </tr>
+              ))}
+            </tbody>
+          </table>
+        </div>
+      </div>
+    </div>
+  );
+}

references/nextjs-realtime/src/trigger/ai.ts

@@ -9,7 +9,10 @@ const openaiSDK = new OpenAI({
   apiKey: process.env.OPENAI_API_KEY,
 });
 
-export type STREAMS = { openai: TextStreamPart<{ getWeather: typeof weatherTask.tool }> };
+export type STREAMS = {
+  openai: TextStreamPart<{ getWeather: typeof weatherTask.tool }>;
+  openaiText: string;
+};
 
 export const openaiConsumer = schemaTask({
   id: "openai-consumer",
@@ -105,18 +108,7 @@ export const openaiStreaming = schemaTask({
     });
 
     const stream = await metadata.stream("openai", result.fullStream);
-
-    let text = "";
-
-    for await (const chunk of stream) {
-      logger.log("Received chunk", { chunk });
-
-      if (chunk.type === "text-delta") {
-        text += chunk.textDelta;
-      }
-    }
-
-    return { text };
+    await metadata.stream("openaiText", result.textStream);
   },
 });
 

references/v3-catalog/src/trigger/batch.ts

@@ -124,12 +124,17 @@ export const allV2TestTask = task({
   retry: {
     maxAttempts: 1,
   },
-  run: async () => {
-    const response1 = await batch.trigger<typeof allV2ChildTask1 | typeof allV2ChildTask2>([
-      { id: "all-v2-test-child-1", payload: { child1: "foo" } },
-      { id: "all-v2-test-child-2", payload: { child2: "bar" } },
-      { id: "all-v2-test-child-1", payload: { child1: "baz" } },
-    ]);
+  run: async ({ triggerSequentially }: { triggerSequentially?: boolean }) => {
+    const response1 = await batch.trigger<typeof allV2ChildTask1 | typeof allV2ChildTask2>(
+      [
+        { id: "all-v2-test-child-1", payload: { child1: "foo" } },
+        { id: "all-v2-test-child-2", payload: { child2: "bar" } },
+        { id: "all-v2-test-child-1", payload: { child1: "baz" } },
+      ],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.debug("Response 1", { response1 });
 
@@ -156,11 +161,16 @@ export const allV2TestTask = task({
 
     const {
       runs: [batchRun1, batchRun2, batchRun3],
-    } = await batch.triggerByTask([
-      { task: allV2ChildTask1, payload: { child1: "foo" } },
-      { task: allV2ChildTask2, payload: { child2: "bar" } },
-      { task: allV2ChildTask1, payload: { child1: "baz" } },
-    ]);
+    } = await batch.triggerByTask(
+      [
+        { task: allV2ChildTask1, payload: { child1: "foo" } },
+        { task: allV2ChildTask2, payload: { child2: "bar" } },
+        { task: allV2ChildTask1, payload: { child1: "baz" } },
+      ],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.debug("Batch runs", { batchRun1, batchRun2, batchRun3 });
 
@@ -179,11 +189,16 @@ export const allV2TestTask = task({
     type TaskRun3Payload = Expect<Equal<typeof taskRun3.payload, { child1: string } | undefined>>;
     type TaskRun3Output = Expect<Equal<typeof taskRun3.output, { foo: string } | undefined>>;
 
-    const response3 = await batch.triggerAndWait<typeof allV2ChildTask1 | typeof allV2ChildTask2>([
-      { id: "all-v2-test-child-1", payload: { child1: "foo" } },
-      { id: "all-v2-test-child-2", payload: { child2: "bar" } },
-      { id: "all-v2-test-child-1", payload: { child1: "baz" } },
-    ]);
+    const response3 = await batch.triggerAndWait<typeof allV2ChildTask1 | typeof allV2ChildTask2>(
+      [
+        { id: "all-v2-test-child-1", payload: { child1: "foo" } },
+        { id: "all-v2-test-child-2", payload: { child2: "bar" } },
+        { id: "all-v2-test-child-1", payload: { child1: "baz" } },
+      ],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.debug("Response 3", { response3 });
 
@@ -225,11 +240,16 @@ export const allV2TestTask = task({
 
     const {
       runs: [batch2Run1, batch2Run2, batch2Run3],
-    } = await batch.triggerByTaskAndWait([
-      { task: allV2ChildTask1, payload: { child1: "foo" } },
-      { task: allV2ChildTask2, payload: { child2: "bar" } },
-      { task: allV2ChildTask1, payload: { child1: "baz" } },
-    ]);
+    } = await batch.triggerByTaskAndWait(
+      [
+        { task: allV2ChildTask1, payload: { child1: "foo" } },
+        { task: allV2ChildTask2, payload: { child2: "bar" } },
+        { task: allV2ChildTask1, payload: { child1: "baz" } },
+      ],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.debug("Batch 2 runs", { batch2Run1, batch2Run2, batch2Run3 });
 
@@ -276,14 +296,17 @@ export const batchV2TestTask = task({
   retry: {
     maxAttempts: 1,
   },
-  run: async () => {
+  run: async ({ triggerSequentially }: { triggerSequentially?: boolean }) => {
     // First lets try triggering with too many items
     try {
       await tasks.batchTrigger<typeof batchV2TestChild>(
         "batch-v2-test-child",
         Array.from({ length: 501 }, (_, i) => ({
           payload: { foo: `bar${i}` },
-        }))
+        })),
+        {
+          triggerSequentially,
+        }
       );
 
       assert.fail("Batch trigger should have failed");
@@ -299,10 +322,12 @@ export const batchV2TestTask = task({
     // tasks.batchTrigger
     // tasks.batchTriggerAndWait
     // myTask.batchTriggerAndWait
-    const response1 = await batchV2TestChild.batchTrigger([
-      { payload: { foo: "bar" } },
-      { payload: { foo: "baz" } },
-    ]);
+    const response1 = await batchV2TestChild.batchTrigger(
+      [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.info("Response 1", { response1 });
 
@@ -360,7 +385,10 @@ export const batchV2TestTask = task({
     const response2 = await batchV2TestChild.batchTrigger(
       Array.from({ length: 30 }, (_, i) => ({
         payload: { foo: `bar${i}` },
-      }))
+      })),
+      {
+        triggerSequentially,
+      }
     );
 
     logger.info("Response 2", { response2 });
@@ -385,6 +413,7 @@ export const batchV2TestTask = task({
       {
         idempotencyKey: idempotencyKey1,
         idempotencyKeyTTL: "5s",
+        triggerSequentially,
       }
     );
 
@@ -401,6 +430,7 @@ export const batchV2TestTask = task({
       {
         idempotencyKey: idempotencyKey1,
         idempotencyKeyTTL: "5s",
+        triggerSequentially,
       }
     );
 
@@ -429,6 +459,7 @@ export const batchV2TestTask = task({
       {
         idempotencyKey: idempotencyKey1,
         idempotencyKeyTTL: "5s",
+        triggerSequentially,
       }
     );
 
@@ -445,16 +476,21 @@ export const batchV2TestTask = task({
     const idempotencyKeyChild1 = randomUUID();
     const idempotencyKeyChild2 = randomUUID();
 
-    const response6 = await batchV2TestChild.batchTrigger([
+    const response6 = await batchV2TestChild.batchTrigger(
+      [
+        {
+          payload: { foo: "bar" },
+          options: { idempotencyKey: idempotencyKeyChild1, idempotencyKeyTTL: "5s" },
+        },
+        {
+          payload: { foo: "baz" },
+          options: { idempotencyKey: idempotencyKeyChild2, idempotencyKeyTTL: "15s" },
+        },
+      ],
       {
-        payload: { foo: "bar" },
-        options: { idempotencyKey: idempotencyKeyChild1, idempotencyKeyTTL: "5s" },
-      },
-      {
-        payload: { foo: "baz" },
-        options: { idempotencyKey: idempotencyKeyChild2, idempotencyKeyTTL: "15s" },
-      },
-    ]);
+        triggerSequentially,
+      }
+    );
 
     logger.info("Response 6", { response6 });
 
@@ -466,10 +502,15 @@ export const batchV2TestTask = task({
 
     await setTimeout(1000);
 
-    const response7 = await batchV2TestChild.batchTrigger([
-      { payload: { foo: "bar" }, options: { idempotencyKey: idempotencyKeyChild1 } },
-      { payload: { foo: "baz" }, options: { idempotencyKey: idempotencyKeyChild2 } },
-    ]);
+    const response7 = await batchV2TestChild.batchTrigger(
+      [
+        { payload: { foo: "bar" }, options: { idempotencyKey: idempotencyKeyChild1 } },
+        { payload: { foo: "baz" }, options: { idempotencyKey: idempotencyKeyChild2 } },
+      ],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.info("Response 7", { response7 });
 
@@ -490,10 +531,15 @@ export const batchV2TestTask = task({
     await wait.for({ seconds: 6 });
 
     // Now we need to test that the first run is not cached and is a new run, and the second run is cached
-    const response8 = await batchV2TestChild.batchTrigger([
-      { payload: { foo: "bar" }, options: { idempotencyKey: idempotencyKeyChild1 } },
-      { payload: { foo: "baz" }, options: { idempotencyKey: idempotencyKeyChild2 } },
-    ]);
+    const response8 = await batchV2TestChild.batchTrigger(
+      [
+        { payload: { foo: "bar" }, options: { idempotencyKey: idempotencyKeyChild1 } },
+        { payload: { foo: "baz" }, options: { idempotencyKey: idempotencyKeyChild2 } },
+      ],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.info("Response 8", { response8 });
 
@@ -512,10 +558,12 @@ export const batchV2TestTask = task({
     );
 
     // Now we need to test with batchTriggerAndWait
-    const response9 = await batchV2TestChild.batchTriggerAndWait([
-      { payload: { foo: "bar" } },
-      { payload: { foo: "baz" } },
-    ]);
+    const response9 = await batchV2TestChild.batchTriggerAndWait(
+      [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.debug("Response 9", { response9 });
 
@@ -548,7 +596,10 @@ export const batchV2TestTask = task({
     const response10 = await batchV2TestChild.batchTriggerAndWait(
       Array.from({ length: 21 }, (_, i) => ({
         payload: { foo: `bar${i}` },
-      }))
+      })),
+      {
+        triggerSequentially,
+      }
     );
 
     logger.debug("Response 10", { response10 });
@@ -557,10 +608,13 @@ export const batchV2TestTask = task({
     assert.equal(response10.runs.length, 21, "response10: Items length is invalid");
 
     // Now repeat the first few tests using `tasks.batchTrigger`:
-    const response11 = await tasks.batchTrigger<typeof batchV2TestChild>("batch-v2-test-child", [
-      { payload: { foo: "bar" } },
-      { payload: { foo: "baz" } },
-    ]);
+    const response11 = await tasks.batchTrigger<typeof batchV2TestChild>(
+      "batch-v2-test-child",
+      [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }],
+      {
+        triggerSequentially,
+      }
+    );
 
     logger.debug("Response 11", { response11 });
 
@@ -584,7 +638,10 @@ export const batchV2TestTask = task({
       "batch-v2-test-child",
       Array.from({ length: 100 }, (_, i) => ({
         payload: { foo: `bar${i}` },
-      }))
+      })),
+      {
+        triggerSequentially,
+      }
     );
 
     const response12Start = performance.now();