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

ericallam · ericallam · commit 72724200ed1c · 2025-10-17T10:49:23.000+01:00
diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts
@@ -16,7 +16,15 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return new Response("No body provided", { status: 400 });
   }
 
-  return relayRealtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
+  const resumeFromChunk = request.headers.get("X-Resume-From-Chunk");
+  const resumeFromChunkNumber = resumeFromChunk ? parseInt(resumeFromChunk, 10) : undefined;
+
+  return relayRealtimeStreams.ingestData(
+    request.body,
+    $params.runId,
+    $params.streamId,
+    resumeFromChunkNumber
+  );
 }
 
 export const loader = createLoaderApiRoute(
diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts
@@ -1,7 +1,10 @@
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
-import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import {
+  createActionApiRoute,
+  createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -14,10 +17,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 +53,80 @@ const { action } = createActionApiRoute(
       return new Response("Target not found", { status: 404 });
     }
 
-    return relayRealtimeStreams.ingestData(request.body, targetId, params.streamId);
+    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;
+
+    return relayRealtimeStreams.ingestData(
+      request.body,
+      targetId,
+      params.streamId,
+      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 }) => {
+    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 });
+    }
+
+    const lastChunkIndex = await relayRealtimeStreams.getLastChunkIndex(targetId, params.streamId);
+
+    return new Response(null, {
+      status: 200,
+      headers: {
+        "X-Last-Chunk-Index": lastChunkIndex.toString(),
+      },
+    });
   }
 );
 
-export { action };
+export { action, loader };
diff --git a/apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts b/apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts
@@ -55,18 +55,36 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
                   lastId = id;
 
                   if (fields && fields.length >= 2) {
-                    if (fields[1] === END_SENTINEL && i === entries.length - 1) {
-                      controller.close();
-                      return;
+                    // Extract the data field from the Redis entry
+                    // Fields format: ["field1", "value1", "field2", "value2", ...]
+                    let data: string | null = null;
+
+                    for (let j = 0; j < fields.length; j += 2) {
+                      if (fields[j] === "data") {
+                        data = fields[j + 1];
+                        break;
+                      }
                     }
 
-                    if (fields[1] !== END_SENTINEL) {
-                      controller.enqueue(fields[1]);
+                    // Handle legacy entries that don't have field names (just data at index 1)
+                    if (data === null && fields.length >= 2) {
+                      data = fields[1];
                     }
 
-                    if (signal.aborted) {
-                      controller.close();
-                      return;
+                    if (data) {
+                      if (data === END_SENTINEL && i === entries.length - 1) {
+                        controller.close();
+                        return;
+                      }
+
+                      if (data !== END_SENTINEL) {
+                        controller.enqueue(data);
+                      }
+
+                      if (signal.aborted) {
+                        controller.close();
+                        return;
+                      }
                     }
                   }
                 }
@@ -127,10 +145,14 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
   async ingestData(
     stream: ReadableStream<Uint8Array>,
     runId: string,
-    streamId: string
+    streamId: string,
+    resumeFromChunk?: number
   ): Promise<Response> {
     const redis = new Redis(this.options.redis ?? {});
     const streamKey = `stream:${runId}:${streamId}`;
+    const startChunk = resumeFromChunk ?? 0;
+    // Start counting from the resume point, not from 0
+    let currentChunkIndex = startChunk;
 
     async function cleanup() {
       try {
@@ -151,9 +173,12 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
           break;
         }
 
-        logger.debug("[RedisRealtimeStreams][ingestData] Reading data", {
+        // Write each chunk with its index
+        logger.debug("[RedisRealtimeStreams][ingestData] Writing chunk", {
           streamKey,
           runId,
+          chunkIndex: currentChunkIndex,
+          resumeFromChunk: startChunk,
           value,
         });
 
@@ -163,9 +188,13 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
           "~",
           String(env.REALTIME_STREAM_MAX_LENGTH),
           "*",
+          "chunkIndex",
+          currentChunkIndex.toString(),
           "data",
           value
         );
+
+        currentChunkIndex++;
       }
 
       // Send the END_SENTINEL and set TTL with a pipeline.
@@ -200,4 +229,50 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
       await cleanup();
     }
   }
+
+  async getLastChunkIndex(runId: string, streamId: string): Promise<number> {
+    const redis = new Redis(this.options.redis ?? {});
+    const streamKey = `stream:${runId}:${streamId}`;
+
+    try {
+      // Get the last entry from the stream using XREVRANGE
+      const entries = await redis.xrevrange(streamKey, "+", "-", "COUNT", 1);
+
+      if (!entries || entries.length === 0) {
+        // No entries in stream, return -1 to indicate no chunks received
+        return -1;
+      }
+
+      const [_id, fields] = entries[0];
+
+      // Find the chunkIndex field
+      for (let i = 0; i < fields.length; i += 2) {
+        if (fields[i] === "chunkIndex") {
+          const chunkIndex = parseInt(fields[i + 1], 10);
+          logger.debug("[RedisRealtimeStreams][getLastChunkIndex] Found last chunk", {
+            streamKey,
+            chunkIndex,
+          });
+          return chunkIndex;
+        }
+      }
+
+      // If no chunkIndex field found (legacy entries), return -1
+      logger.warn("[RedisRealtimeStreams][getLastChunkIndex] No chunkIndex found in entry", {
+        streamKey,
+      });
+      return -1;
+    } catch (error) {
+      logger.error("[RedisRealtimeStreams][getLastChunkIndex] Error getting last chunk:", {
+        error,
+        streamKey,
+      });
+      // Return -1 to indicate we don't know what the server has
+      return -1;
+    } finally {
+      await redis.quit().catch((err) => {
+        logger.error("[RedisRealtimeStreams][getLastChunkIndex] Error in cleanup:", { err });
+      });
+    }
+  }
 }
diff --git a/apps/webapp/app/services/realtime/relayRealtimeStreams.server.ts b/apps/webapp/app/services/realtime/relayRealtimeStreams.server.ts
@@ -134,19 +134,29 @@ export class RelayRealtimeStreams implements StreamIngestor, StreamResponder {
   async ingestData(
     stream: ReadableStream<Uint8Array>,
     runId: string,
-    streamId: string
+    streamId: string,
+    resumeFromChunk?: number
   ): Promise<Response> {
     const [localStream, fallbackStream] = stream.tee();
 
-    logger.debug("[RelayRealtimeStreams][ingestData] Ingesting data", { runId, streamId });
+    logger.debug("[RelayRealtimeStreams][ingestData] Ingesting data", {
+      runId,
+      streamId,
+      resumeFromChunk,
+    });
 
     // Handle local buffering asynchronously and catch errors
     this.handleLocalIngestion(localStream, runId, streamId).catch((err) => {
       logger.error("[RelayRealtimeStreams][ingestData] Error in local ingestion:", { err });
     });
 
     // Forward to the fallback ingestor asynchronously and catch errors
-    return this.options.fallbackIngestor.ingestData(fallbackStream, runId, streamId);
+    return this.options.fallbackIngestor.ingestData(
+      fallbackStream,
+      runId,
+      streamId,
+      resumeFromChunk
+    );
   }
 
   /**
@@ -237,6 +247,11 @@ export class RelayRealtimeStreams implements StreamIngestor, StreamResponder {
     });
   }
 
+  async getLastChunkIndex(runId: string, streamId: string): Promise<number> {
+    // Relay doesn't store chunks, forward to fallback
+    return this.options.fallbackIngestor.getLastChunkIndex(runId, streamId);
+  }
+
   // Don't forget to clear interval on shutdown if needed
   close() {
     clearInterval(this.cleanupInterval);
diff --git a/apps/webapp/app/services/realtime/types.ts b/apps/webapp/app/services/realtime/types.ts
@@ -5,8 +5,11 @@ export interface StreamIngestor {
   ingestData(
     stream: ReadableStream<Uint8Array>,
     runId: string,
-    streamId: string
+    streamId: string,
+    resumeFromChunk?: number
   ): Promise<Response>;
+
+  getLastChunkIndex(runId: string, streamId: string): Promise<number>;
 }
 
 // Interface for stream response
diff --git a/docker/config/toxiproxy.json b/docker/config/toxiproxy.json
@@ -0,0 +1,8 @@
+[
+  {
+    "name": "trigger_webapp_local",
+    "listen": "[::]:30303",
+    "upstream": "host.docker.internal:3030",
+    "enabled": true
+  }
+]
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -141,6 +141,19 @@ services:
     networks:
       - app_network
 
+  toxiproxy:
+    container_name: toxiproxy
+    image: ghcr.io/shopify/toxiproxy:latest
+    restart: always
+    volumes:
+      - ./config/toxiproxy.json:/config/toxiproxy.json
+    ports:
+      - "30303:30303" # Proxied webapp port
+      - "8474:8474" # Toxiproxy API port
+    networks:
+      - app_network
+    command: ["-host", "0.0.0.0", "-config", "/config/toxiproxy.json"]
+
   # otel-collector:
   #   container_name: otel-collector
   #   image: otel/opentelemetry-collector-contrib:latest
diff --git a/packages/core/src/v3/runMetadata/metadataStream.ts b/packages/core/src/v3/runMetadata/metadataStream.ts
diff --git a/references/hello-world/src/trigger/realtime.ts b/references/hello-world/src/trigger/realtime.ts
