triggerdotdev · ericallam · Apr 29, 2025 · Apr 29, 2025 · Apr 29, 2025
diff --git a/.changeset/gentle-waves-suffer.md b/.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
diff --git a/apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts b/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();

diff --git a/packages/core/src/v3/runMetadata/metadataStream.ts b/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) {}
+}