feat(realtime): Realtime streams v2

.changeset/quiet-falcons-approve.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+gracefully recover from ECONNRESET errors when sending stream data from tasks to the server

apps/webapp/app/assets/icons/ListBulletIcon.tsx

@@ -0,0 +1,30 @@
+export function ListBulletIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        d="M9 5H20"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M9 12H20"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M9 19H20"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <circle cx="4" cy="5" r="1" fill="currentColor" />
+      <circle cx="4" cy="12" r="1" fill="currentColor" />
+      <circle cx="4" cy="19" r="1" fill="currentColor" />
+    </svg>
+  );
+}

apps/webapp/app/assets/icons/MoveToBottomIcon.tsx

@@ -0,0 +1,27 @@
+export function MoveToBottomIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        d="M12 15L12 3"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M3 21L21 21"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M7.5 12.5L12 17L16.5 12.5"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}

apps/webapp/app/assets/icons/SnakedArrowIcon.tsx

@@ -0,0 +1,20 @@
+export function SnakedArrowIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        d="M5 5H16C17.6569 5 19 6.34315 19 8L19 8.5C19 10.1569 17.6569 11.5 16 11.5H8C6.34314 11.5 5 12.8431 5 14.5L5 15C4.99999 16.6569 6.34314 18 8 18H18.634"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M16 21L19 18L16 15"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}

apps/webapp/app/assets/icons/StreamsIcon.tsx

@@ -0,0 +1,10 @@
+export function StreamsIcon({ className }: { className?: string }) {
+  return (
+<svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+<path d="M3 19C3 19 5.01155 17 8 17C10.9885 17 13 18.9973 16 18.9973C19 18.9973 21 17 21 17" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
+<path d="M3 13.0001C3 13.0001 5.01155 11 8 11C10.9885 11 13 13 16 13C19 13 21 11.0001 21 11.0001" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
+<path d="M3 7C3 7 5.01155 5 8 5C10.9885 5 13 6.9973 16 6.9973C19 6.9973 21 5 21 5" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
+</svg>
+  );
+}
+

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

@@ -20,6 +20,7 @@ import { TriggerIcon } from "~/assets/icons/TriggerIcon";
 import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
 import { TraceIcon } from "~/assets/icons/TraceIcon";
 import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
+import { StreamsIcon } from "~/assets/icons/StreamsIcon";
 
 type TaskIconProps = {
   name: string | undefined;
@@ -107,6 +108,8 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
     case "task-hook-onFailure":
     case "task-hook-catchError":
       return <FunctionIcon className={cn(className, "text-error")} />;
+    case "streams":
+        return <StreamsIcon className={cn(className, "text-text-dimmed")} />;
   }
 
   return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;

apps/webapp/app/env.server.ts

@@ -198,6 +198,7 @@ const EnvironmentSchema = z
       .string()
       .default(process.env.REDIS_TLS_DISABLED ?? "false"),
     REALTIME_STREAMS_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
+    REALTIME_STREAMS_INACTIVITY_TIMEOUT_MS: z.coerce.number().int().default(60000), // 1 minute
 
     REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
       .number()
@@ -1201,6 +1202,16 @@ const EnvironmentSchema = z
     EVENT_LOOP_MONITOR_UTILIZATION_SAMPLE_RATE: z.coerce.number().default(0.05),
 
     VERY_SLOW_QUERY_THRESHOLD_MS: z.coerce.number().int().optional(),
+
+    REALTIME_STREAMS_S2_BASIN: z.string().optional(),
+    REALTIME_STREAMS_S2_ACCESS_TOKEN: z.string().optional(),
+    REALTIME_STREAMS_S2_LOG_LEVEL: z
+      .enum(["log", "error", "warn", "info", "debug"])
+      .default("info"),
+    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),
+    WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
   })
   .and(GithubAppEnvSchema);
 

apps/webapp/app/models/organization.server.ts

@@ -66,7 +66,7 @@ export async function createOrganization(
           role: "ADMIN",
         },
       },
-      v3Enabled: !features.isManagedCloud,
+      v3Enabled: true,
     },
     include: {
       members: true,

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

@@ -19,6 +19,7 @@ import { WaitpointPresenter } from "./WaitpointPresenter.server";
 import { engine } from "~/v3/runEngine.server";
 import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 import { IEventRepository, SpanDetail } from "~/v3/eventRepository/eventRepository.types";
+import { safeJsonParse } from "~/utils/json";
 
 type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
 export type Span = NonNullable<NonNullable<Result>["span"]>;
@@ -551,6 +552,41 @@ export class SpanPresenter extends BasePresenter {
           },
         };
       }
+      case "realtime-stream": {
+        if (!span.entity.id) {
+          logger.error(`SpanPresenter: No realtime stream id`, {
+            spanId,
+            realtimeStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const [runId, streamKey] = span.entity.id.split(":");
+
+        if (!runId || !streamKey) {
+          logger.error(`SpanPresenter: Invalid realtime stream id`, {
+            spanId,
+            realtimeStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const metadata = span.entity.metadata
+          ? (safeJsonParse(span.entity.metadata) as Record<string, unknown> | undefined)
+          : undefined;
+
+        return {
+          ...data,
+          entity: {
+            type: "realtime-stream" as const,
+            object: {
+              runId,
+              streamKey,
+              metadata,
+            },
+          },
+        };
+      }
       default:
         return { ...data, entity: null };
     }

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

@@ -33,6 +33,7 @@ export const HeadersSchema = z.object({
   "x-trigger-client": z.string().nullish(),
   "x-trigger-engine-version": RunEngineVersionSchema.nullish(),
   "x-trigger-request-idempotency-key": z.string().nullish(),
+  "x-trigger-realtime-streams-version": z.string().nullish(),
   traceparent: z.string().optional(),
   tracestate: z.string().optional(),
 });
@@ -63,6 +64,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-client": triggerClient,
       "x-trigger-engine-version": engineVersion,
       "x-trigger-request-idempotency-key": requestIdempotencyKey,
+      "x-trigger-realtime-streams-version": realtimeStreamsVersion,
     } = headers;
 
     const cachedResponse = await handleRequestIdempotency(requestIdempotencyKey, {
@@ -108,14 +110,7 @@ const { action, loader } = createActionApiRoute(
         options: body.options,
         isFromWorker,
         traceContext,
-      });
-
-      logger.debug("[otelContext]", {
-        taskId: params.taskId,
-        headers,
-        options: body.options,
-        isFromWorker,
-        traceContext,
+        realtimeStreamsVersion,
       });
 
       const idempotencyKeyExpiresAt = resolveIdempotencyKeyTTL(idempotencyKeyTTL);
@@ -131,6 +126,7 @@ const { action, loader } = createActionApiRoute(
           traceContext,
           spanParentAsLink: spanParentAsLink === 1,
           oneTimeUseToken,
+          realtimeStreamsVersion: realtimeStreamsVersion ?? undefined,
         },
         engineVersion ?? undefined
       );

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

@@ -1,24 +1,13 @@
-import { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
-import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
   streamId: z.string(),
 });
 
-export async function action({ request, params }: ActionFunctionArgs) {
-  const $params = ParamsSchema.parse(params);
-
-  if (!request.body) {
-    return new Response("No body provided", { status: 400 });
-  }
-
-  return relayRealtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
-}
-
 export const loader = createLoaderApiRoute(
   {
     params: ParamsSchema,
@@ -51,12 +40,32 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ params, request, resource: run, authentication }) => {
-    return relayRealtimeStreams.streamResponse(
-      request,
-      run.friendlyId,
-      params.streamId,
+    // Get Last-Event-ID header for resuming from a specific position
+    const lastEventId = request.headers.get("Last-Event-ID") || undefined;
+
+    const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
+    const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
+
+    if (timeoutInSeconds && isNaN(timeoutInSeconds)) {
+      return new Response("Invalid timeout seconds", { status: 400 });
+    }
+
+    if (timeoutInSeconds && timeoutInSeconds < 1) {
+      return new Response("Timeout seconds must be greater than 0", { status: 400 });
+    }
+
+    if (timeoutInSeconds && timeoutInSeconds > 600) {
+      return new Response("Timeout seconds must be less than 600", { status: 400 });
+    }
+
+    const realtimeStream = getRealtimeStreamInstance(
       authentication.environment,
-      request.signal
+      run.realtimeStreamsVersion
     );
+
+    return realtimeStream.streamResponse(request, run.friendlyId, params.streamId, request.signal, {
+      lastEventId,
+      timeoutInSeconds,
+    });
   }
 );