feat: optimize Redis stream handling with batching

Add STREAM_ORIGIN to environment schema. Improve performance in
RealtimeStreams by using TextDecoderStream for simpler text
decoding and implementing batching of XADD commands for Redis
streams. Limit stream size using MAXLEN option. Update
environment variable repository with new variable type. Adjust
import statements for Redis key and value types.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -32,6 +32,7 @@ const EnvironmentSchema = z.object({
   LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ORIGIN: z.string().default("http://localhost:3030"),
   API_ORIGIN: z.string().optional(),
+  STREAM_ORIGIN: z.string().optional(),
   ELECTRIC_ORIGIN: z.string().default("http://localhost:3060"),
   APP_ENV: z.string().default(process.env.NODE_ENV),
   SERVICE_NAME: z.string().default("trigger.dev webapp"),

apps/webapp/app/services/realtimeStreams.server.ts

@@ -1,4 +1,4 @@
-import Redis, { RedisOptions } from "ioredis";
+import Redis, { RedisKey, RedisOptions, RedisValue } from "ioredis";
 import { logger } from "./logger.server";
 
 export type RealtimeStreamsOptions = {
@@ -107,55 +107,68 @@ export class RealtimeStreams {
     streamId: string
   ): Promise<Response> {
     const redis = new Redis(this.options.redis ?? {});
-
     const streamKey = `stream:${runId}:${streamId}`;
 
-    async function cleanup(stream?: TransformStream) {
+    async function cleanup() {
       try {
         await redis.quit();
-        if (stream) {
-          const writer = stream.writable.getWriter();
-          await writer.close(); // Catch in case the stream is already closed
-        }
       } catch (error) {
         logger.error("[RealtimeStreams][ingestData] Error in cleanup:", { error });
       }
     }
 
     try {
-      const reader = stream.getReader();
-      const decoder = new TextDecoder();
-      let buffer = "";
+      // Use TextDecoderStream to simplify text decoding
+      const textStream = stream.pipeThrough(new TextDecoderStream());
+      const reader = textStream.getReader();
+
+      const batchSize = 10; // Adjust this value based on performance testing
+      let batchCommands: Array<[key: RedisKey, ...args: RedisValue[]]> = [];
 
       while (true) {
         const { done, value } = await reader.read();
 
-        logger.debug("[RealtimeStreams][ingestData] Reading data", { streamKey, done });
-
         if (done) {
-          if (buffer) {
-            const data = JSON.parse(buffer);
-            await redis.xadd(streamKey, "*", "data", JSON.stringify(data));
-          }
           break;
         }
 
-        buffer += decoder.decode(value, { stream: true });
-        const lines = buffer.split("\n");
-        buffer = lines.pop() || "";
+        logger.debug("[RealtimeStreams][ingestData] Reading data", { streamKey, value });
+
+        // 'value' is a string containing the decoded text
+        const lines = value.split("\n");
 
         for (const line of lines) {
           if (line.trim()) {
-            const data = JSON.parse(line);
-
-            logger.debug("[RealtimeStreams][ingestData] Ingesting data", { streamKey });
-
-            await redis.xadd(streamKey, "*", "data", JSON.stringify(data));
+            // Avoid unnecessary parsing; assume 'line' is already a JSON string
+            // Add XADD command with MAXLEN option to limit stream size
+            batchCommands.push([streamKey, "MAXLEN", "~", "1000", "*", "data", line]);
+
+            if (batchCommands.length >= batchSize) {
+              // Send batch using a pipeline
+              const pipeline = redis.pipeline();
+              for (const args of batchCommands) {
+                pipeline.xadd(...args);
+              }
+              await pipeline.exec();
+              batchCommands = [];
+            }
           }
         }
       }
 
-      await redis.xadd(streamKey, "*", "data", JSON.stringify({ __end: true }));
+      // Send any remaining commands
+      if (batchCommands.length > 0) {
+        const pipeline = redis.pipeline();
+        for (const args of batchCommands) {
+          pipeline.xadd(...args);
+        }
+        await pipeline.exec();
+      }
+
+      // Send the __end message to indicate the end of the stream
+      const endData = JSON.stringify({ __end: true });
+      await redis.xadd(streamKey, "MAXLEN", "~", "1000", "*", "data", endData);
+
       return new Response(null, { status: 200 });
     } catch (error) {
       console.error("Error in ingestData:", error);

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -734,6 +734,10 @@ async function resolveBuiltInProdVariables(runtimeEnvironment: RuntimeEnvironmen
       key: "TRIGGER_API_URL",
       value: env.API_ORIGIN ?? env.APP_ORIGIN,
     },
+    {
+      key: "TRIGGER_STREAM_URL",
+      value: env.STREAM_ORIGIN ?? env.API_ORIGIN ?? env.APP_ORIGIN,
+    },
     {
       key: "TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS",
       value: String(env.CHECKPOINT_THRESHOLD_IN_MS),

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

@@ -1,3 +1,5 @@
+import { run } from "node:test";
+
 export type MetadataOptions<T> = {
   baseUrl: string;
   runId: string;