From 6251c96c19e8db14e81c8cbcc98b714240de5491 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 12 Dec 2024 14:31:40 +0000 Subject: [PATCH] Improve the v1 realtime streams (Redis) --- .../v3/ApiRunListPresenter.server.ts | 13 +- .../app/presenters/v3/SpanPresenter.server.ts | 2 +- .../realtime/redisRealtimeStreams.server.ts | 60 ++--- .../realtime/relayRealtimeStreams.server.ts | 226 ++++++++++++++++++ .../app/services/realtime/utils.server.ts | 33 +++ docker/docker-compose.yml | 2 +- internal-packages/testcontainers/src/utils.ts | 4 +- packages/core/package.json | 2 +- packages/core/src/v3/apiClient/runStream.ts | 13 +- packages/core/src/v3/apiClient/stream.ts | 13 +- packages/core/src/v3/runMetadata/manager.ts | 14 +- packages/core/src/v3/utils/ioSerialization.ts | 23 +- pnpm-lock.yaml | 8 +- 13 files changed, 339 insertions(+), 74 deletions(-) create mode 100644 apps/webapp/app/services/realtime/relayRealtimeStreams.server.ts create mode 100644 apps/webapp/app/services/realtime/utils.server.ts diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts index 766ca0367..78f95e324 100644 --- a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts +++ b/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, diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index 539a0c811..d8516ca6a 100644 --- a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts +++ b/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; diff --git a/apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts b/apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts index 808776bd0..6eaee46a3 100644 --- a/apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts +++ b/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); diff --git a/apps/webapp/app/services/realtime/relayRealtimeStreams.server.ts b/apps/webapp/app/services/realtime/relayRealtimeStreams.server.ts new file mode 100644 index 000000000..43e5c81b9 --- /dev/null +++ b/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; + createdAt: number; + lastAccessed: number; + finalized: boolean; +} + +export class RelayRealtimeStreams implements StreamIngestor, StreamResponder { + private _buffers: Map = 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 { + 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, + runId: string, + streamId: string + ): Promise { + 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, + 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 + ): 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 { + const timeout = this.waitForBufferTimeout; + const interval = this.waitForBufferInterval; + const maxAttempts = Math.ceil(timeout / interval); + let attempts = 0; + + return new Promise((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 +); diff --git a/apps/webapp/app/services/realtime/utils.server.ts b/apps/webapp/app/services/realtime/utils.server.ts new file mode 100644 index 000000000..9655878fe --- /dev/null +++ b/apps/webapp/app/services/realtime/utils.server.ts @@ -0,0 +1,33 @@ +export class LineTransformStream extends TransformStream { + 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]); + } + }, + }); + } +} diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 5b9c55333..fd9e0014a 100644 --- a/docker/docker-compose.yml +++ b/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 diff --git a/internal-packages/testcontainers/src/utils.ts b/internal-packages/testcontainers/src/utils.ts index 734a5e562..140628630 100644 --- a/internal-packages/testcontainers/src/utils.ts +++ b/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({ diff --git a/packages/core/package.json b/packages/core/package.json index 6eab0c694..82e2b9460 100644 --- a/packages/core/package.json +++ b/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", diff --git a/packages/core/src/v3/apiClient/runStream.ts b/packages/core/src/v3/apiClient/runStream.ts index 4729f3f34..48494f278 100644 --- a/packages/core/src/v3/apiClient/runStream.ts +++ b/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}`); diff --git a/packages/core/src/v3/apiClient/stream.ts b/packages/core/src/v3/apiClient/stream.ts index 3ce3fdf3a..266b5bc4a 100644 --- a/packages/core/src/v3/apiClient/stream.ts +++ b/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( 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 = Row> { export class LineTransformStream extends TransformStream { 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 { // 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); diff --git a/packages/core/src/v3/runMetadata/manager.ts b/packages/core/src/v3/runMetadata/manager.ts index 83d8e7488..470640fbe 100644 --- a/packages/core/src/v3/runMetadata/manager.ts +++ b/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 diff --git a/packages/core/src/v3/utils/ioSerialization.ts b/packages/core/src/v3/utils/ioSerialization.ts index b369d02ff..8fbfa2e53 100644 --- a/packages/core/src/v3/utils/ioSerialization.ts +++ b/packages/core/src/v3/utils/ioSerialization.ts @@ -14,14 +14,18 @@ export type IOPacket = { dataType: string; }; -export async function parsePacket(value: IOPacket): Promise { +export type ParsePacketOptions = { + filteredKeys?: string[]; +}; + +export async function parsePacket(value: IOPacket, options?: ParsePacketOptions): Promise { 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": diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6369c25fc..d3c5be19f 100644 --- a/pnpm-lock.yaml +++ b/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