Improve the v1 realtime streams (Redis)

apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts

@@ -224,10 +224,15 @@ export class ApiRunListPresenter extends BasePresenter {
 
       const data: ListRunResponseItem[] = await Promise.all(
         results.runs.map(async (run) => {
-          const metadata = await parsePacket({
-            data: run.metadata ?? undefined,
-            dataType: run.metadataType,
-          });
+          const metadata = await parsePacket(
+            {
+              data: run.metadata ?? undefined,
+              dataType: run.metadataType,
+            },
+            {
+              filteredKeys: ["$$streams", "$$streamsVersion", "$$streamsBaseUrl"],
+            }
+          );
 
           return {
             id: run.friendlyId,

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -216,7 +216,7 @@ export class SpanPresenter extends BasePresenter {
 
     const metadata = run.metadata
       ? await prettyPrintPacket(run.metadata, run.metadataType, {
-          filteredKeys: ["$$streams", "$$streamsVersion"],
+          filteredKeys: ["$$streams", "$$streamsVersion", "$$streamsBaseUrl"],
         })
       : undefined;
 

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

@@ -1,7 +1,8 @@
-import Redis, { RedisKey, RedisOptions, RedisValue } from "ioredis";
+import Redis, { RedisOptions } from "ioredis";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
 import { logger } from "../logger.server";
 import { StreamIngestor, StreamResponder } from "./types";
-import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { LineTransformStream } from "./utils.server";
 
 export type RealtimeStreamsOptions = {
   redis: RedisOptions | undefined;
@@ -56,7 +57,7 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
                       controller.close();
                       return;
                     }
-                    controller.enqueue(`data: ${fields[1]}\n\n`);
+                    controller.enqueue(fields[1]);
 
                     if (signal.aborted) {
                       controller.close();
@@ -88,7 +89,18 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
       cancel: async () => {
         await cleanup();
       },
-    });
+    })
+      .pipeThrough(new LineTransformStream())
+      .pipeThrough(
+        new TransformStream({
+          transform(chunk, controller) {
+            for (const line of chunk) {
+              controller.enqueue(`data: ${line}\n\n`);
+            }
+          },
+        })
+      )
+      .pipeThrough(new TextEncoderStream());
 
     async function cleanup() {
       if (isCleanedUp) return;
@@ -98,7 +110,7 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
 
     signal.addEventListener("abort", cleanup);
 
-    return new Response(stream.pipeThrough(new TextEncoderStream()), {
+    return new Response(stream, {
       headers: {
         "Content-Type": "text/event-stream",
         "Cache-Control": "no-cache",
@@ -119,7 +131,7 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
       try {
         await redis.quit();
       } catch (error) {
-        logger.error("[RealtimeStreams][ingestData] Error in cleanup:", { error });
+        logger.error("[RedisRealtimeStreams][ingestData] Error in cleanup:", { error });
       }
     }
 
@@ -127,42 +139,20 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
       const textStream = stream.pipeThrough(new TextDecoderStream());
       const reader = textStream.getReader();
 
-      const batchSize = 10;
-      let batchCommands: Array<[key: RedisKey, ...args: RedisValue[]]> = [];
-
       while (true) {
         const { done, value } = await reader.read();
 
-        if (done) {
+        if (done || !value) {
           break;
         }
 
-        logger.debug("[RealtimeStreams][ingestData] Reading data", { streamKey, value });
+        logger.debug("[RedisRealtimeStreams][ingestData] Reading data", {
+          streamKey,
+          runId,
+          value,
+        });
 
-        const lines = value.split("\n");
-
-        for (const line of lines) {
-          if (line.trim()) {
-            batchCommands.push([streamKey, "MAXLEN", "~", "2500", "*", "data", line]);
-
-            if (batchCommands.length >= batchSize) {
-              const pipeline = redis.pipeline();
-              for (const args of batchCommands) {
-                pipeline.xadd(...args);
-              }
-              await pipeline.exec();
-              batchCommands = [];
-            }
-          }
-        }
-      }
-
-      if (batchCommands.length > 0) {
-        const pipeline = redis.pipeline();
-        for (const args of batchCommands) {
-          pipeline.xadd(...args);
-        }
-        await pipeline.exec();
+        await redis.xadd(streamKey, "MAXLEN", "~", "1000", "*", "data", value);
       }
 
       await redis.xadd(streamKey, "MAXLEN", "~", "1000", "*", "data", END_SENTINEL);

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

@@ -0,0 +1,226 @@
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { logger } from "../logger.server";
+import { StreamIngestor, StreamResponder } from "./types";
+import { LineTransformStream } from "./utils.server";
+import { v1RealtimeStreams } from "./v1StreamsGlobal.server";
+import { singleton } from "~/utils/singleton";
+
+export type RelayRealtimeStreamsOptions = {
+  ttl: number;
+  fallbackIngestor: StreamIngestor;
+  fallbackResponder: StreamResponder;
+  waitForBufferTimeout?: number; // Time to wait for buffer in ms (default: 500ms)
+  waitForBufferInterval?: number; // Polling interval in ms (default: 50ms)
+};
+
+interface RelayedStreamRecord {
+  stream: ReadableStream<Uint8Array>;
+  createdAt: number;
+  lastAccessed: number;
+  finalized: boolean;
+}
+
+export class RelayRealtimeStreams implements StreamIngestor, StreamResponder {
+  private _buffers: Map<string, RelayedStreamRecord> = new Map();
+  private cleanupInterval: NodeJS.Timeout;
+  private waitForBufferTimeout: number;
+  private waitForBufferInterval: number;
+
+  constructor(private options: RelayRealtimeStreamsOptions) {
+    this.waitForBufferTimeout = options.waitForBufferTimeout ?? 5000;
+    this.waitForBufferInterval = options.waitForBufferInterval ?? 50;
+
+    // Periodic cleanup
+    this.cleanupInterval = setInterval(() => {
+      this.cleanup();
+    }, this.options.ttl).unref();
+  }
+
+  async streamResponse(
+    request: Request,
+    runId: string,
+    streamId: string,
+    environment: AuthenticatedEnvironment,
+    signal: AbortSignal
+  ): Promise<Response> {
+    let record = this._buffers.get(`${runId}:${streamId}`);
+
+    if (!record) {
+      logger.debug(
+        "[RelayRealtimeStreams][streamResponse] No ephemeral record found, waiting to see if one becomes available",
+        {
+          streamId,
+          runId,
+        }
+      );
+
+      record = await this.waitForBuffer(`${runId}:${streamId}`);
+
+      if (!record) {
+        logger.debug(
+          "[RelayRealtimeStreams][streamResponse] No ephemeral record found, using fallback",
+          {
+            streamId,
+            runId,
+          }
+        );
+
+        // No ephemeral record, use fallback
+        return this.options.fallbackResponder.streamResponse(
+          request,
+          runId,
+          streamId,
+          environment,
+          signal
+        );
+      }
+    }
+
+    record.lastAccessed = Date.now();
+
+    logger.debug("[RelayRealtimeStreams][streamResponse] Streaming from ephemeral record", {
+      streamId,
+      runId,
+    });
+
+    // Create a streaming response from the buffered data
+    const stream = record.stream
+      .pipeThrough(new TextDecoderStream())
+      .pipeThrough(new LineTransformStream())
+      .pipeThrough(
+        new TransformStream({
+          transform(chunk, controller) {
+            for (const line of chunk) {
+              controller.enqueue(`data: ${line}\n\n`);
+            }
+          },
+        })
+      )
+      .pipeThrough(new TextEncoderStream());
+
+    // Once we start streaming, consider deleting the buffer when done.
+    // For a simple approach, we can rely on finalized and no more reads.
+    // Or we can let TTL cleanup handle it if multiple readers might come in.
+    return new Response(stream, {
+      headers: {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        Connection: "keep-alive",
+      },
+    });
+  }
+
+  async ingestData(
+    stream: ReadableStream<Uint8Array>,
+    runId: string,
+    streamId: string
+  ): Promise<Response> {
+    const [localStream, fallbackStream] = stream.tee();
+
+    logger.debug("[RelayRealtimeStreams][ingestData] Ingesting data", { runId, streamId });
+
+    // 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);
+  }
+
+  /**
+   * Handles local buffering of the stream data.
+   * @param stream The readable stream to buffer.
+   * @param streamId The unique identifier for the stream.
+   */
+  private async handleLocalIngestion(
+    stream: ReadableStream<Uint8Array>,
+    runId: string,
+    streamId: string
+  ) {
+    this.createOrUpdateRelayedStream(`${runId}:${streamId}`, stream);
+  }
+
+  /**
+   * Retrieves an existing buffer or creates a new one for the given streamId.
+   * @param streamId The unique identifier for the stream.
+   */
+  private createOrUpdateRelayedStream(
+    bufferKey: string,
+    stream: ReadableStream<Uint8Array>
+  ): RelayedStreamRecord {
+    let record = this._buffers.get(bufferKey);
+    if (!record) {
+      record = {
+        stream,
+        createdAt: Date.now(),
+        lastAccessed: Date.now(),
+        finalized: false,
+      };
+      this._buffers.set(bufferKey, record);
+    } else {
+      record.lastAccessed = Date.now();
+    }
+    return record;
+  }
+
+  private cleanup() {
+    const now = Date.now();
+    for (const [key, record] of this._buffers.entries()) {
+      // If last accessed is older than ttl, clean up
+      if (now - record.lastAccessed > this.options.ttl) {
+        this.deleteBuffer(key);
+      }
+    }
+  }
+
+  private deleteBuffer(bufferKey: string) {
+    this._buffers.delete(bufferKey);
+  }
+
+  /**
+   * Waits for a buffer to be created within a specified timeout.
+   * @param streamId The unique identifier for the stream.
+   * @returns A promise that resolves to true if the buffer was created, false otherwise.
+   */
+  private async waitForBuffer(bufferKey: string): Promise<RelayedStreamRecord | undefined> {
+    const timeout = this.waitForBufferTimeout;
+    const interval = this.waitForBufferInterval;
+    const maxAttempts = Math.ceil(timeout / interval);
+    let attempts = 0;
+
+    return new Promise<RelayedStreamRecord | undefined>((resolve) => {
+      const checkBuffer = () => {
+        attempts++;
+        if (this._buffers.has(bufferKey)) {
+          resolve(this._buffers.get(bufferKey));
+          return;
+        }
+        if (attempts >= maxAttempts) {
+          resolve(undefined);
+          return;
+        }
+        setTimeout(checkBuffer, interval);
+      };
+      checkBuffer();
+    });
+  }
+
+  // Don't forget to clear interval on shutdown if needed
+  close() {
+    clearInterval(this.cleanupInterval);
+  }
+}
+
+function initializeRelayRealtimeStreams() {
+  return new RelayRealtimeStreams({
+    ttl: 1000 * 60 * 5, // 5 minutes
+    fallbackIngestor: v1RealtimeStreams,
+    fallbackResponder: v1RealtimeStreams,
+  });
+}
+
+export const relayRealtimeStreams = singleton(
+  "relayRealtimeStreams",
+  initializeRelayRealtimeStreams
+);

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

@@ -0,0 +1,33 @@
+export class LineTransformStream extends TransformStream<string, string[]> {
+  private buffer = "";
+
+  constructor() {
+    super({
+      transform: (chunk, controller) => {
+        // Append the chunk to the buffer
+        this.buffer += chunk;
+
+        // Split on newlines
+        const lines = this.buffer.split("\n");
+
+        // The last element might be incomplete, hold it back in buffer
+        this.buffer = lines.pop() || "";
+
+        // Filter out empty or whitespace-only lines
+        const fullLines = lines.filter((line) => line.trim().length > 0);
+
+        // If we got any complete lines, emit them as an array
+        if (fullLines.length > 0) {
+          controller.enqueue(fullLines);
+        }
+      },
+      flush: (controller) => {
+        // On stream end, if there's leftover text, emit it as a single-element array
+        const trimmed = this.buffer.trim();
+        if (trimmed.length > 0) {
+          controller.enqueue([trimmed]);
+        }
+      },
+    });
+  }
+}

docker/docker-compose.yml

@@ -65,7 +65,7 @@ services:
       - 6379:6379
 
   electric:
-    image: electricsql/electric:0.9.4
+    image: electricsql/electric:1.0.0-beta.1@sha256:2262f6f09caf5fa45f233731af97b84999128170a9529e5f9b9b53642308493f
     restart: always
     environment:
       DATABASE_URL: postgresql://postgres:postgres@database:5432/postgres?sslmode=disable

internal-packages/testcontainers/src/utils.ts

@@ -55,7 +55,9 @@ export async function createElectricContainer(
     network.getName()
   )}:5432/${postgresContainer.getDatabase()}?sslmode=disable`;
 
-  const container = await new GenericContainer("electricsql/electric:0.9.4")
+  const container = await new GenericContainer(
+    "electricsql/electric:1.0.0-beta.1@sha256:2262f6f09caf5fa45f233731af97b84999128170a9529e5f9b9b53642308493f"
+  )
     .withExposedPorts(3000)
     .withNetwork(network)
     .withEnvironment({

packages/core/package.json

@@ -182,7 +182,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@electric-sql/client": "0.9.0",
+    "@electric-sql/client": "1.0.0-beta.1",
     "@google-cloud/precise-date": "^4.0.0",
     "@jsonhero/path": "^1.0.21",
     "@opentelemetry/api": "1.9.0",

packages/core/src/v3/apiClient/runStream.ts

@@ -218,19 +218,15 @@ export class ElectricStreamSubscription implements StreamSubscription {
       .pipeThrough(
         new TransformStream({
           transform(chunk, controller) {
-            console.log("ElectricStreamSubscription chunk.value", chunk.value);
-
             controller.enqueue(chunk.value);
           },
         })
       )
-      .pipeThrough(new LineTransformStream(this.url))
+      .pipeThrough(new LineTransformStream())
       .pipeThrough(
         new TransformStream({
           transform(chunk, controller) {
             for (const line of chunk) {
-              console.log("ElectricStreamSubscription line", line);
-
               controller.enqueue(safeParseJSON(line));
             }
           },
@@ -281,12 +277,15 @@ export class VersionedStreamSubscriptionFactory implements StreamSubscriptionFac
     const version =
       typeof metadata.$$streamsVersion === "string" ? metadata.$$streamsVersion : "v1";
 
+    const $baseUrl =
+      typeof metadata.$$streamsBaseUrl === "string" ? metadata.$$streamsBaseUrl : baseUrl;
+
     if (version === "v1") {
-      return this.version1.createSubscription(metadata, runId, streamKey, baseUrl);
+      return this.version1.createSubscription(metadata, runId, streamKey, $baseUrl);
     }
 
     if (version === "v2") {
-      return this.version2.createSubscription(metadata, runId, streamKey, baseUrl);
+      return this.version2.createSubscription(metadata, runId, streamKey, $baseUrl);
     }
 
     throw new Error(`Unknown stream version: ${version}`);

packages/core/src/v3/apiClient/stream.ts

@@ -8,7 +8,6 @@ import {
   type Message,
   type Row,
   type ShapeStreamInterface,
-  // @ts-ignore it's safe to import types from the client
 } from "@electric-sql/client";
 
 export type ZodShapeStreamOptions = {
@@ -26,7 +25,7 @@ export function zodShapeStream<TShapeSchema extends z.ZodTypeAny>(
     url,
     headers: {
       ...options?.headers,
-      "x-trigger-electric-version": "0.8.1",
+      "x-trigger-electric-version": "1.0.0-beta.1",
     },
     fetchClient: options?.fetchClient,
     signal: options?.signal,
@@ -207,7 +206,7 @@ class ReadableShapeStream<T extends Row<unknown> = Row> {
 export class LineTransformStream extends TransformStream<string, string[]> {
   private buffer = "";
 
-  constructor(streamId: string) {
+  constructor() {
     super({
       transform: (chunk, controller) => {
         // Append the chunk to the buffer
@@ -222,14 +221,6 @@ export class LineTransformStream extends TransformStream<string, string[]> {
         // Filter out empty or whitespace-only lines
         const fullLines = lines.filter((line) => line.trim().length > 0);
 
-        console.log("LineTransformStream", {
-          chunk,
-          lines,
-          fullLines,
-          buffer: this.buffer,
-          streamId,
-        });
-
         // If we got any complete lines, emit them as an array
         if (fullLines.length > 0) {
           controller.enqueue(fullLines);

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

@@ -230,13 +230,6 @@ export class StandardMetadataManager implements RunMetadataManager {
     }
 
     try {
-      // Add the key to the special stream metadata object
-      this.appendKey(`$$streams`, key);
-      this.setKey("$$streamsVersion", this.streamsVersion);
-      this.setKey("$$streamsBaseUrl", this.streamsBaseUrl);
-
-      await this.flush();
-
       const streamInstance = new MetadataStream({
         key,
         runId: this.runId,
@@ -252,6 +245,13 @@ export class StandardMetadataManager implements RunMetadataManager {
       // Clean up when stream completes
       streamInstance.wait().finally(() => this.activeStreams.delete(key));
 
+      // Add the key to the special stream metadata object
+      this.appendKey(`$$streams`, key);
+      this.setKey("$$streamsVersion", this.streamsVersion);
+      this.setKey("$$streamsBaseUrl", this.streamsBaseUrl);
+
+      await this.flush();
+
       return streamInstance;
     } catch (error) {
       // Clean up metadata key if stream creation fails

packages/core/src/v3/utils/ioSerialization.ts

@@ -14,14 +14,18 @@ export type IOPacket = {
   dataType: string;
 };
 
-export async function parsePacket(value: IOPacket): Promise<any> {
+export type ParsePacketOptions = {
+  filteredKeys?: string[];
+};
+
+export async function parsePacket(value: IOPacket, options?: ParsePacketOptions): Promise<any> {
   if (!value.data) {
     return undefined;
   }
 
   switch (value.dataType) {
     case "application/json":
-      return JSON.parse(value.data);
+      return JSON.parse(value.data, makeSafeReviver(options));
     case "application/super+json":
       const { parse } = await loadSuperJSON();
 
@@ -400,6 +404,21 @@ function makeSafeReplacer(options?: ReplacerOptions) {
   };
 }
 
+function makeSafeReviver(options?: ReplacerOptions) {
+  if (!options) {
+    return undefined;
+  }
+
+  return function reviver(key: string, value: any) {
+    // Check if the key should be filtered out
+    if (options?.filteredKeys?.includes(key)) {
+      return undefined;
+    }
+
+    return value;
+  };
+}
+
 function getPacketExtension(outputType: string): string {
   switch (outputType) {
     case "application/json":

pnpm-lock.yaml

@@ -1263,8 +1263,8 @@ importers:
   packages/core:
     dependencies:
       '@electric-sql/client':
-        specifier: 0.9.0
-        version: 0.9.0
+        specifier: 1.0.0-beta.1
+        version: 1.0.0-beta.1
       '@google-cloud/precise-date':
         specifier: ^4.0.0
         version: 4.0.0
@@ -5112,8 +5112,8 @@ packages:
       '@rollup/rollup-darwin-arm64': 4.21.3
     dev: false
 
-  /@electric-sql/client@0.9.0:
-    resolution: {integrity: sha512-UL2Gep9wPdGMTE0oEWVi0HA8R293R2OzFfHeAsN2LABYYl/boXss7nseNEiIV5+RjHPH7Tm8NsjH9iJW2rZkrQ==}
+  /@electric-sql/client@1.0.0-beta.1:
+    resolution: {integrity: sha512-Ei9jN3pDoGzc+a/bGqnB5ajb52IvSv7/n2btuyzUlcOHIR2kM9fqtYTJXPwZYKLkGZlHWlpHgWyRtrinkP2nHg==}
     optionalDependencies:
       '@rollup/rollup-darwin-arm64': 4.21.3
     dev: false