fix(realtime): gracefully recover from ECONNRESET errors when sending stream data from tasks to the server

.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/env.server.ts

@@ -198,6 +198,10 @@ 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 * 5), // 5 minutes
 
     REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
       .number()
@@ -1200,6 +1204,14 @@ 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),
   })
   .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/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,20 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ params, request, resource: run, authentication }) => {
-    return relayRealtimeStreams.streamResponse(
+    // Get Last-Event-ID header for resuming from a specific position
+    const lastEventId = request.headers.get("Last-Event-ID") || undefined;
+
+    const realtimeStream = getRealtimeStreamInstance(
+      authentication.environment,
+      run.realtimeStreamsVersion
+    );
+
+    return realtimeStream.streamResponse(
       request,
       run.friendlyId,
       params.streamId,
-      authentication.environment,
-      request.signal
+      request.signal,
+      lastEventId
     );
   }
 );

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

@@ -1,7 +1,11 @@
+import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { $replica } from "~/db.server";
-import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
-import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { $replica, prisma } from "~/db.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -14,10 +18,6 @@ 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,
@@ -54,8 +54,129 @@ const { action } = createActionApiRoute(
       return new Response("Target not found", { status: 404 });
     }
 
-    return relayRealtimeStreams.ingestData(request.body, targetId, params.streamId);
+    if (request.method === "PUT") {
+      // This is the "create" endpoint
+      const updatedRun = await prisma.taskRun.update({
+        where: {
+          friendlyId: targetId,
+          runtimeEnvironmentId: authentication.environment.id,
+        },
+        data: {
+          realtimeStreams: {
+            push: params.streamId,
+          },
+        },
+        select: {
+          realtimeStreamsVersion: true,
+        },
+      });
+
+      const realtimeStream = getRealtimeStreamInstance(
+        authentication.environment,
+        updatedRun.realtimeStreamsVersion
+      );
+
+      const { responseHeaders } = await realtimeStream.initializeStream(targetId, params.streamId);
+
+      return json(
+        {
+          version: updatedRun.realtimeStreamsVersion,
+        },
+        { status: 202, headers: responseHeaders }
+      );
+    } else {
+      // 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");
+      const resumeFromChunkNumber = resumeFromChunk ? parseInt(resumeFromChunk, 10) : undefined;
+
+      const realtimeStream = getRealtimeStreamInstance(authentication.environment, streamVersion);
+
+      return realtimeStream.ingestData(
+        request.body,
+        targetId,
+        params.streamId,
+        clientId,
+        resumeFromChunkNumber
+      );
+    }
+  }
+);
+
+const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    allowJWT: false,
+    corsStrategy: "none",
+    findResource: async (params, authentication) => {
+      return $replica.taskRun.findFirst({
+        where: {
+          friendlyId: params.runId,
+          runtimeEnvironmentId: authentication.environment.id,
+        },
+        select: {
+          id: true,
+          friendlyId: true,
+          parentTaskRun: {
+            select: {
+              friendlyId: true,
+            },
+          },
+          rootTaskRun: {
+            select: {
+              friendlyId: true,
+            },
+          },
+        },
+      });
+    },
+  },
+  async ({ request, params, resource: run, authentication }) => {
+    if (!run) {
+      return new Response("Run not found", { status: 404 });
+    }
+
+    const targetId =
+      params.target === "self"
+        ? run.friendlyId
+        : params.target === "parent"
+        ? run.parentTaskRun?.friendlyId
+        : run.rootTaskRun?.friendlyId;
+
+    if (!targetId) {
+      return new Response("Target not found", { status: 404 });
+    }
+
+    // Handle HEAD request to get last chunk index
+    if (request.method !== "HEAD") {
+      return new Response("Only HEAD requests are allowed for this endpoint", { status: 405 });
+    }
+
+    // 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";
+
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, streamVersion);
+
+    const lastChunkIndex = await realtimeStream.getLastChunkIndex(
+      targetId,
+      params.streamId,
+      clientId
+    );
+
+    return new Response(null, {
+      status: 200,
+      headers: {
+        "X-Last-Chunk-Index": lastChunkIndex.toString(),
+      },
+    });
   }
 );
 
-export { action };
+export { action, loader };

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

@@ -347,6 +347,7 @@ export class RunEngineTriggerTaskService {
                     createdAt: options.overrideCreatedAt,
                     bulkActionId: body.options?.bulkActionId,
                     planType,
+                    realtimeStreamsVersion: options.realtimeStreamsVersion,
                   },
                   this.prisma
                 );