Merge branch 'main' into docs/improve-env-var-guide

Author: lindseystead

.changeset/tiny-carrots-rest.md

@@ -0,0 +1,41 @@
+---
+"@trigger.dev/sdk": minor
+---
+
+Prevent uncaught errors in the `onSuccess`, `onComplete`, and `onFailure` lifecycle hooks from failing attempts/runs.
+
+Deprecated the `onStart` lifecycle hook (which only fires before the `run` function on the first attempt). Replaced with `onStartAttempt` that fires before the run function on every attempt:
+
+```ts
+export const taskWithOnStartAttempt = task({
+  id: "task-with-on-start-attempt",
+  onStartAttempt: async ({ payload, ctx }) => {
+    //...
+  },
+  run: async (payload: any, { ctx }) => {
+    //...
+  },
+});
+
+// Default a global lifecycle hook using tasks
+tasks.onStartAttempt(({ ctx, payload, task }) => {
+  console.log(
+    `Run ${ctx.run.id} started on task ${task} attempt ${ctx.run.attempt.number}`,
+    ctx.run
+  );
+});
+```
+
+If you want to execute code before just the first attempt, you can use the `onStartAttempt` function and check `ctx.run.attempt.number === 1`:
+
+```ts /trigger/on-start-attempt.ts
+export const taskWithOnStartAttempt = task({
+  id: "task-with-on-start-attempt",
+  onStartAttempt: async ({ payload, ctx }) => {
+    if (ctx.run.attempt.number === 1) {
+      console.log("Run started on attempt 1", ctx.run);
+    }
+  },
+});
+```
+

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

@@ -98,6 +98,7 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
       return <RunFunctionIcon className={cn(className, "text-text-dimmed")} />;
     case "task-hook-init":
     case "task-hook-onStart":
+    case "task-hook-onStartAttempt":
     case "task-hook-onSuccess":
     case "task-hook-onWait":
     case "task-hook-onResume":

apps/webapp/app/env.server.ts

@@ -1236,6 +1236,7 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
     REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
     REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
+    REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
     WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
   })
   .and(GithubAppEnvSchema)

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

@@ -9,6 +9,7 @@ import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 const ParamsSchema = z.object({
   taskId: z.string(),
@@ -40,6 +41,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
     "trigger-version": triggerVersion,
     "x-trigger-span-parent-as-link": spanParentAsLink,
     "x-trigger-worker": isFromWorker,
+    "x-trigger-realtime-streams-version": realtimeStreamsVersion,
     traceparent,
     tracestate,
   } = headers.data;
@@ -100,6 +102,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
       triggerVersion: triggerVersion ?? undefined,
       traceContext,
       spanParentAsLink: spanParentAsLink === 1,
+      realtimeStreamsVersion: determineRealtimeStreamsVersion(realtimeStreamsVersion ?? undefined),
     });
 
     if (!result) {

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -11,6 +11,7 @@ import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { ApiAuthenticationResultSuccess, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import {
@@ -126,7 +127,9 @@ const { action, loader } = createActionApiRoute(
           traceContext,
           spanParentAsLink: spanParentAsLink === 1,
           oneTimeUseToken,
-          realtimeStreamsVersion: realtimeStreamsVersion ?? undefined,
+          realtimeStreamsVersion: determineRealtimeStreamsVersion(
+            realtimeStreamsVersion ?? undefined
+          ),
         },
         engineVersion ?? undefined
       );

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

@@ -16,6 +16,7 @@ import {
 } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 const { action, loader } = createActionApiRoute(
   {
@@ -69,6 +70,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-client": triggerClient,
       "x-trigger-engine-version": engineVersion,
       "batch-processing-strategy": batchProcessingStrategy,
+      "x-trigger-realtime-streams-version": realtimeStreamsVersion,
       traceparent,
       tracestate,
     } = headers;
@@ -107,6 +109,9 @@ const { action, loader } = createActionApiRoute(
         traceContext,
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
+        realtimeStreamsVersion: determineRealtimeStreamsVersion(
+          realtimeStreamsVersion ?? undefined
+        ),
       });
 
       const $responseHeaders = await responseHeaders(

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

@@ -18,6 +18,7 @@ import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { BatchProcessingStrategy } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 
 const { action, loader } = createActionApiRoute(
   {
@@ -59,6 +60,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-engine-version": engineVersion,
       "batch-processing-strategy": batchProcessingStrategy,
       "x-trigger-request-idempotency-key": requestIdempotencyKey,
+      "x-trigger-realtime-streams-version": realtimeStreamsVersion,
       traceparent,
       tracestate,
     } = headers;
@@ -119,6 +121,9 @@ const { action, loader } = createActionApiRoute(
         traceContext,
         spanParentAsLink: spanParentAsLink === 1,
         oneTimeUseToken,
+        realtimeStreamsVersion: determineRealtimeStreamsVersion(
+          realtimeStreamsVersion ?? undefined
+        ),
       });
 
       const $responseHeaders = await responseHeaders(

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

@@ -1,13 +1,79 @@
+import { type ActionFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
   streamId: z.string(),
 });
 
+// Plain action for backwards compatibility with older clients that don't send auth headers
+export async function action({ request, params }: ActionFunctionArgs) {
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return new Response("Invalid parameters", { status: 400 });
+  }
+
+  const { runId, streamId } = parsedParams.data;
+
+  // Look up the run without environment scoping for backwards compatibility
+  const run = await $replica.taskRun.findFirst({
+    where: {
+      friendlyId: runId,
+    },
+    select: {
+      id: true,
+      friendlyId: true,
+      runtimeEnvironment: {
+        include: {
+          project: true,
+          organization: true,
+          orgMember: true,
+        },
+      },
+    },
+  });
+
+  if (!run) {
+    return new Response("Run not found", { status: 404 });
+  }
+
+  // Extract client ID from header, default to "default" if not provided
+  const clientId = request.headers.get("X-Client-Id") || "default";
+  const streamVersion = request.headers.get("X-Stream-Version") || "v1";
+
+  if (!request.body) {
+    return new Response("No body provided", { status: 400 });
+  }
+
+  const resumeFromChunk = request.headers.get("X-Resume-From-Chunk");
+  let resumeFromChunkNumber: number | undefined = undefined;
+  if (resumeFromChunk) {
+    const parsed = parseInt(resumeFromChunk, 10);
+    if (isNaN(parsed) || parsed < 0) {
+      return new Response(`Invalid X-Resume-From-Chunk header value: ${resumeFromChunk}`, {
+        status: 400,
+      });
+    }
+    resumeFromChunkNumber = parsed;
+  }
+
+  // The runtimeEnvironment from the run is already in the correct shape for AuthenticatedEnvironment
+  const realtimeStream = getRealtimeStreamInstance(run.runtimeEnvironment, streamVersion);
+
+  return realtimeStream.ingestData(
+    request.body,
+    run.friendlyId,
+    streamId,
+    clientId,
+    resumeFromChunkNumber
+  );
+}
+
 export const loader = createLoaderApiRoute(
   {
     params: ParamsSchema,

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

@@ -47,6 +47,7 @@ export type BatchTriggerTaskServiceOptions = {
   traceContext?: Record<string, string | undefined | Record<string, string | undefined>>;
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
+  realtimeStreamsVersion?: "v1" | "v2";
 };
 
 /**
@@ -708,6 +709,7 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         batchIndex: currentIndex,
         skipChecks: true, // Skip entitlement and queue checks since we already validated at batch/chunk level
         planType, // Pass planType from batch-level entitlement check
+        realtimeStreamsVersion: options?.realtimeStreamsVersion,
       },
       "V2"
     );

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

@@ -60,6 +60,22 @@ export function getRealtimeStreamInstance(
   }
 }
 
+export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
+  if (!streamVersion) {
+    return env.REALTIME_STREAMS_DEFAULT_VERSION;
+  }
+
+  if (
+    streamVersion === "v2" &&
+    env.REALTIME_STREAMS_S2_BASIN &&
+    env.REALTIME_STREAMS_S2_ACCESS_TOKEN
+  ) {
+    return "v2";
+  }
+
+  return "v1";
+}
+
 const s2RealtimeStreamsCache = singleton(
   "s2RealtimeStreamsCache",
   initializeS2RealtimeStreamsCache