WIP

Author: ericallam

apps/webapp/app/env.server.ts

@@ -219,10 +219,7 @@ const EnvironmentSchema = z
       .string()
       .default(process.env.REDIS_TLS_DISABLED ?? "false"),
     REALTIME_STREAMS_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
-    REALTIME_STREAMS_INACTIVITY_TIMEOUT_MS: z.coerce
-      .number()
-      .int()
-      .default(60000 * 5), // 5 minutes
+    REALTIME_STREAMS_INACTIVITY_TIMEOUT_MS: z.coerce.number().int().default(60000), // 1 minute
 
     REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
       .number()
@@ -1235,6 +1232,7 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
     REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
     REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
+    WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(60_000),
   })
   .and(GithubAppEnvSchema)
   .and(S2EnvSchema);

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -149,19 +149,19 @@ traceContext.setGlobalManager(standardTraceContextManager);
 
 const durableClock = new DurableClock();
 clock.setGlobalClock(durableClock);
-const runMetadataManager = new StandardMetadataManager(
-  apiClientManager.clientOrThrow(),
-  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
-);
+const runMetadataManager = new StandardMetadataManager(apiClientManager.clientOrThrow());
 runMetadata.setGlobalManager(runMetadataManager);
 
 const standardRealtimeStreamsManager = new StandardRealtimeStreamsManager(
   apiClientManager.clientOrThrow(),
-  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
+  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
+  (getEnvVar("TRIGGER_STREAMS_DEBUG") === "1" || getEnvVar("TRIGGER_STREAMS_DEBUG") === "true") ??
+    false
 );
 realtimeStreams.setGlobalManager(standardRealtimeStreamsManager);
 
-const waitUntilManager = new StandardWaitUntilManager();
+const waitUntilTimeoutInMs = getNumberEnvVar("TRIGGER_WAIT_UNTIL_TIMEOUT_MS", 60_000);
+const waitUntilManager = new StandardWaitUntilManager(waitUntilTimeoutInMs);
 waitUntil.setGlobalManager(waitUntilManager);
 
 const triggerLogLevel = getEnvVar("TRIGGER_LOG_LEVEL");
@@ -531,10 +531,6 @@ const zodIpc = new ZodIpcConnection({
 
         runMetadataManager.runId = execution.run.id;
         runMetadataManager.runIdIsRoot = typeof execution.run.rootTaskRunId === "undefined";
-        runMetadataManager.streamsVersion =
-          typeof execution.run.realtimeStreamsVersion === "undefined"
-            ? "v1"
-            : execution.run.realtimeStreamsVersion;
 
         _executionCount++;
 

packages/cli-v3/src/entryPoints/managed-run-worker.ts

@@ -31,6 +31,7 @@ import {
   WorkerToExecutorMessageCatalog,
   traceContext,
   heartbeats,
+  realtimeStreams,
 } from "@trigger.dev/core/v3";
 import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
 import {
@@ -57,6 +58,7 @@ import {
   UsageTimeoutManager,
   StandardTraceContextManager,
   StandardHeartbeatsManager,
+  StandardRealtimeStreamsManager,
 } from "@trigger.dev/core/v3/workers";
 import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
 import { readFile } from "node:fs/promises";
@@ -127,11 +129,16 @@ clock.setGlobalClock(durableClock);
 const standardTraceContextManager = new StandardTraceContextManager();
 traceContext.setGlobalManager(standardTraceContextManager);
 
-const runMetadataManager = new StandardMetadataManager(
+const runMetadataManager = new StandardMetadataManager(apiClientManager.clientOrThrow());
+runMetadata.setGlobalManager(runMetadataManager);
+
+const standardRealtimeStreamsManager = new StandardRealtimeStreamsManager(
   apiClientManager.clientOrThrow(),
-  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
+  getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev",
+  (getEnvVar("TRIGGER_STREAMS_DEBUG") === "1" || getEnvVar("TRIGGER_STREAMS_DEBUG") === "true") ??
+    false
 );
-runMetadata.setGlobalManager(runMetadataManager);
+realtimeStreams.setGlobalManager(standardRealtimeStreamsManager);
 
 const waitUntilManager = new StandardWaitUntilManager();
 waitUntil.setGlobalManager(waitUntilManager);
@@ -292,6 +299,7 @@ function resetExecutionEnvironment() {
   timeout.reset();
   runMetadataManager.reset();
   waitUntilManager.reset();
+  standardRealtimeStreamsManager.reset();
   _sharedWorkerRuntime?.reset();
   durableClock.reset();
   taskContext.disable();
@@ -300,8 +308,8 @@ function resetExecutionEnvironment() {
 
   // Wait for all streams to finish before completing the run
   waitUntil.register({
-    requiresResolving: () => runMetadataManager.hasActiveStreams(),
-    promise: () => runMetadataManager.waitForAllStreams(),
+    requiresResolving: () => standardRealtimeStreamsManager.hasActiveStreams(),
+    promise: () => standardRealtimeStreamsManager.waitForAllStreams(),
   });
 
   console.log(`[${new Date().toISOString()}] Reset execution environment`);

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

@@ -15,7 +15,8 @@ import { S2MetadataStream } from "../runMetadata/s2MetadataStream.js";
 export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
   constructor(
     private apiClient: ApiClient,
-    private baseUrl: string
+    private baseUrl: string,
+    private debug: boolean = false
   ) {}
   // Add a Map to track active streams
   private activeStreams = new Map<string, { wait: () => Promise<void> }>();
@@ -65,6 +66,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
             source,
             signal: options?.signal,
             limiter: (await import("p-limit")).default,
+            debug: this.debug,
           });
 
     this.activeStreams.set(key, streamInstance);

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

@@ -20,6 +20,7 @@ export type S2MetadataStreamOptions<T = any> = {
   signal?: AbortSignal;
   flushIntervalMs?: number; // How often to flush batched chunks (default 200ms)
   maxRetries?: number; // Max number of retries for failed flushes (default 10)
+  debug?: boolean; // Enable debug logging (default false)
 };
 
 /**
@@ -30,6 +31,7 @@ export type S2MetadataStreamOptions<T = any> = {
  * - Periodic flushing: Flushes buffered chunks every ~200ms (configurable)
  * - Sequential writes: Uses p-limit to ensure writes happen in order
  * - Automatic retries: Retries failed writes with exponential backoff
+ * - Debug logging: Enable with debug: true to see detailed operation logs
  *
  * Example usage:
  * ```typescript
@@ -39,6 +41,7 @@ export type S2MetadataStreamOptions<T = any> = {
  *   accessToken: "s2-token-here",
  *   source: myAsyncIterable,
  *   flushIntervalMs: 200, // Optional: flush every 200ms
+ *   debug: true, // Optional: enable debug logging
  * });
  *
  * // Wait for streaming to complete
@@ -57,6 +60,7 @@ export class S2MetadataStream<T = any> implements StreamInstance {
   private streamPromise: Promise<void>;
   private readonly flushIntervalMs: number;
   private readonly maxRetries: number;
+  private readonly debug: boolean;
 
   // Buffering state
   private streamComplete = false;
@@ -74,11 +78,16 @@ export class S2MetadataStream<T = any> implements StreamInstance {
 
   constructor(private options: S2MetadataStreamOptions<T>) {
     this.limiter = options.limiter(1);
+    this.debug = options.debug ?? false;
 
     this.s2Client = new S2({ accessToken: options.accessToken });
     this.flushIntervalMs = options.flushIntervalMs ?? 200;
     this.maxRetries = options.maxRetries ?? 10;
 
+    this.log(
+      `[S2MetadataStream] Initializing: basin=${options.basin}, stream=${options.stream}, flushIntervalMs=${this.flushIntervalMs}, maxRetries=${this.maxRetries}`
+    );
+
     const [serverStream, consumerStream] = this.createTeeStreams();
     this.serverStream = serverStream;
     this.consumerStream = consumerStream;
@@ -105,7 +114,6 @@ export class S2MetadataStream<T = any> implements StreamInstance {
 
           controller.close();
         } catch (error) {
-          console.error("[S2MetadataStream] Error reading from source", error);
           controller.error(error);
         }
       },
@@ -115,6 +123,7 @@ export class S2MetadataStream<T = any> implements StreamInstance {
   }
 
   private startBuffering(): void {
+    this.log("[S2MetadataStream] Starting buffering task");
     this.streamReader = this.serverStream.getReader();
 
     this.bufferReaderTask = (async () => {
@@ -126,20 +135,29 @@ export class S2MetadataStream<T = any> implements StreamInstance {
 
           if (done) {
             this.streamComplete = true;
+            this.log(`[S2MetadataStream] Stream complete after ${chunkCount} chunks`);
             break;
           }
 
           // Add to pending flushes
           this.pendingFlushes.push(value);
           chunkCount++;
+
+          if (chunkCount % 100 === 0) {
+            this.log(
+              `[S2MetadataStream] Buffered ${chunkCount} chunks, pending flushes: ${this.pendingFlushes.length}`
+            );
+          }
         }
       } catch (error) {
+        this.logError("[S2MetadataStream] Error in buffering task:", error);
         throw error;
       }
     })();
   }
 
   private startPeriodicFlush(): void {
+    this.log(`[S2MetadataStream] Starting periodic flush (every ${this.flushIntervalMs}ms)`);
     this.flushInterval = setInterval(() => {
       this.flush().catch(() => {
         // Errors are already logged in flush()
@@ -154,10 +172,10 @@ export class S2MetadataStream<T = any> implements StreamInstance {
 
     // Take all pending chunks
     const chunksToFlush = this.pendingFlushes.splice(0);
+    this.log(`[S2MetadataStream] Flushing ${chunksToFlush.length} chunks to S2`);
 
     // Add flush to limiter queue to ensure sequential execution
     const flushPromise = this.limiter(async () => {
-      const startTime = Date.now();
       try {
         // Convert chunks to S2 record format (body as JSON string)
         const records = chunksToFlush.map((data) => ({
@@ -170,32 +188,31 @@ export class S2MetadataStream<T = any> implements StreamInstance {
           appendInput: { records },
         });
 
-        const duration = Date.now() - startTime;
+        this.log(`[S2MetadataStream] Successfully flushed ${chunksToFlush.length} chunks`);
 
         // Reset retry count on success
         this.retryCount = 0;
       } catch (error) {
-        console.error("[S2MetadataStream] Flush error", {
-          error,
-          count: chunksToFlush.length,
-          retryCount: this.retryCount,
-        });
-
         // Handle retryable errors
         if (this.isRetryableError(error) && this.retryCount < this.maxRetries) {
           this.retryCount++;
           const delayMs = this.calculateBackoffDelay();
 
+          this.logError(
+            `[S2MetadataStream] Flush failed (attempt ${this.retryCount}/${this.maxRetries}), retrying in ${delayMs}ms:`,
+            error
+          );
+
           await this.delay(delayMs);
 
           // Re-add chunks to pending flushes and retry
           this.pendingFlushes.unshift(...chunksToFlush);
           await this.flush();
         } else {
-          console.error("[S2MetadataStream] Max retries exceeded or non-retryable error", {
-            retryCount: this.retryCount,
-            maxRetries: this.maxRetries,
-          });
+          this.logError(
+            `[S2MetadataStream] Flush failed permanently after ${this.retryCount} retries:`,
+            error
+          );
           throw error;
         }
       }
@@ -205,20 +222,28 @@ export class S2MetadataStream<T = any> implements StreamInstance {
   }
 
   private async initializeServerStream(): Promise<void> {
+    this.log("[S2MetadataStream] Waiting for buffer task to complete");
     // Wait for buffer task and all flushes to complete
     await this.bufferReaderTask;
 
+    this.log(
+      `[S2MetadataStream] Buffer task complete, performing final flush (${this.pendingFlushes.length} pending chunks)`
+    );
     // Final flush
     await this.flush();
 
+    this.log(`[S2MetadataStream] Waiting for ${this.flushPromises.length} flush promises`);
     // Wait for all pending flushes
     await Promise.all(this.flushPromises);
 
+    this.log("[S2MetadataStream] All flushes complete, cleaning up");
     // Clean up
     if (this.flushInterval) {
       clearInterval(this.flushInterval);
       this.flushInterval = null;
     }
+
+    this.log("[S2MetadataStream] Stream completed successfully");
   }
 
   public async wait(): Promise<void> {
@@ -231,6 +256,18 @@ export class S2MetadataStream<T = any> implements StreamInstance {
 
   // Helper methods
 
+  private log(message: string): void {
+    if (this.debug) {
+      console.log(message);
+    }
+  }
+
+  private logError(message: string, error?: any): void {
+    if (this.debug) {
+      console.error(message, error);
+    }
+  }
+
   private isRetryableError(error: any): boolean {
     if (!error) return false;
 

packages/core/src/v3/waitUntil/index.ts

@@ -8,7 +8,7 @@ class NoopManager implements WaitUntilManager {
     // noop
   }
 
-  blockUntilSettled(timeout: number): Promise<void> {
+  blockUntilSettled(): Promise<void> {
     return Promise.resolve();
   }
 
@@ -44,8 +44,8 @@ export class WaitUntilAPI implements WaitUntilManager {
     return this.#getManager().register(promise);
   }
 
-  blockUntilSettled(timeout: number): Promise<void> {
-    return this.#getManager().blockUntilSettled(timeout);
+  blockUntilSettled(): Promise<void> {
+    return this.#getManager().blockUntilSettled();
   }
 
   requiresResolving(): boolean {

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

@@ -3,6 +3,8 @@ import { MaybeDeferredPromise, WaitUntilManager } from "./types.js";
 export class StandardWaitUntilManager implements WaitUntilManager {
   private maybeDeferredPromises: Set<MaybeDeferredPromise> = new Set();
 
+  constructor(private timeoutInMs: number = 60_000) {}
+
   reset(): void {
     this.maybeDeferredPromises.clear();
   }
@@ -11,7 +13,7 @@ export class StandardWaitUntilManager implements WaitUntilManager {
     this.maybeDeferredPromises.add(promise);
   }
 
-  async blockUntilSettled(timeout: number): Promise<void> {
+  async blockUntilSettled(): Promise<void> {
     if (this.promisesRequringResolving.length === 0) {
       return;
     }
@@ -22,7 +24,7 @@ export class StandardWaitUntilManager implements WaitUntilManager {
 
     await Promise.race([
       Promise.allSettled(promises),
-      new Promise<void>((resolve, _) => setTimeout(() => resolve(), timeout)),
+      new Promise<void>((resolve, _) => setTimeout(() => resolve(), this.timeoutInMs)),
     ]);
 
     this.maybeDeferredPromises.clear();

packages/core/src/v3/waitUntil/types.ts

@@ -5,6 +5,6 @@ export type MaybeDeferredPromise = {
 
 export interface WaitUntilManager {
   register(promise: MaybeDeferredPromise): void;
-  blockUntilSettled(timeout: number): Promise<void>;
+  blockUntilSettled(): Promise<void>;
   requiresResolving(): boolean;
 }

packages/core/src/v3/workers/taskExecutor.ts

@@ -1079,7 +1079,7 @@ export class TaskExecutor {
     return this._tracer.startActiveSpan(
       "waitUntil",
       async (span) => {
-        return await waitUntil.blockUntilSettled(60_000);
+        return await waitUntil.blockUntilSettled();
       },
       {
         attributes: {