realtime: fix streams missing chunks when streams are longer than 5 minutes and receive 408 Request Timeout errors. Also now support multiple client streams being sent to a single stream key (#1993)

* realtime: fix streams missing chunks when streams are longer than 5 minutes and receive 408 Request Timeout errors. Also now support multiple client streams being sent to a single stream key

* Safely release the reader lock

Author: ericallam

.changeset/gentle-waves-suffer.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fixed an issue with realtime streams that timeout and resume streaming dropping chunks

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

@@ -50,15 +50,19 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
               if (messages && messages.length > 0) {
                 const [_key, entries] = messages[0];
 
-                for (const [id, fields] of entries) {
+                for (let i = 0; i < entries.length; i++) {
+                  const [id, fields] = entries[i];
                   lastId = id;
 
                   if (fields && fields.length >= 2) {
-                    if (fields[1] === END_SENTINEL) {
+                    if (fields[1] === END_SENTINEL && i === entries.length - 1) {
                       controller.close();
                       return;
                     }
-                    controller.enqueue(fields[1]);
+
+                    if (fields[1] !== END_SENTINEL) {
+                      controller.enqueue(fields[1]);
+                    }
 
                     if (signal.aborted) {
                       controller.close();

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

@@ -22,14 +22,12 @@ export class MetadataStream<T> {
   private retryCount = 0;
   private readonly maxRetries: number;
   private currentChunkIndex = 0;
-  private reader: ReadableStreamDefaultReader<T>;
 
   constructor(private options: MetadataOptions<T>) {
     const [serverStream, consumerStream] = this.createTeeStreams();
     this.serverStream = serverStream;
     this.consumerStream = consumerStream;
     this.maxRetries = options.maxRetries ?? 10;
-    this.reader = this.serverStream.getReader();
 
     this.streamPromise = this.initializeServerStream();
   }
@@ -52,6 +50,8 @@ export class MetadataStream<T> {
   }
 
   private async makeRequest(startFromChunk: number = 0): Promise<void> {
+    const reader = this.serverStream.getReader();
+
     return new Promise((resolve, reject) => {
       const url = new URL(this.buildUrl());
       const timeout = 15 * 60 * 1000; // 15 minutes
@@ -71,15 +71,20 @@ export class MetadataStream<T> {
       });
 
       req.on("error", (error) => {
+        safeReleaseLock(reader);
         reject(error);
       });
 
       req.on("timeout", () => {
+        safeReleaseLock(reader);
+
         req.destroy(new Error("Request timed out"));
       });
 
       req.on("response", (res) => {
         if (res.statusCode === 408) {
+          safeReleaseLock(reader);
+
           if (this.retryCount < this.maxRetries) {
             this.retryCount++;
 
@@ -112,7 +117,7 @@ export class MetadataStream<T> {
       const processStream = async () => {
         try {
           while (true) {
-            const { done, value } = await this.reader.read();
+            const { done, value } = await reader.read();
 
             if (done) {
               req.end();
@@ -124,7 +129,7 @@ export class MetadataStream<T> {
             this.currentChunkIndex++;
           }
         } catch (error) {
-          req.destroy(error as Error);
+          reject(error);
         }
       };
 
@@ -135,12 +140,7 @@ export class MetadataStream<T> {
   }
 
   private async initializeServerStream(): Promise<void> {
-    try {
-      await this.makeRequest(0);
-    } catch (error) {
-      this.reader.releaseLock();
-      throw error;
-    }
+    await this.makeRequest(0);
   }
 
   public async wait(): Promise<void> {
@@ -174,6 +174,12 @@ async function* streamToAsyncIterator<T>(stream: ReadableStream<T>): AsyncIterab
       yield value;
     }
   } finally {
-    reader.releaseLock();
+    safeReleaseLock(reader);
   }
 }
+
+function safeReleaseLock(reader: ReadableStreamDefaultReader<any>) {
+  try {
+    reader.releaseLock();
+  } catch (error) {}
+}