Configure the waitUntil timeout via an env var

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1232,7 +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),
+    WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
   })
   .and(GithubAppEnvSchema)
   .and(S2EnvSchema);

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

@@ -1185,6 +1185,14 @@ async function resolveCommonBuiltInVariables(
         String(env.TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT)
       ),
     },
+    {
+      key: "TRIGGER_WAIT_UNTIL_TIMEOUT_MS",
+      value: resolveBuiltInEnvironmentVariableOverrides(
+        "TRIGGER_WAIT_UNTIL_TIMEOUT_MS",
+        runtimeEnvironment,
+        String(env.WAIT_UNTIL_TIMEOUT_MS)
+      ),
+    },
   ];
 }
 

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

@@ -140,7 +140,8 @@ const standardRealtimeStreamsManager = new StandardRealtimeStreamsManager(
 );
 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 standardHeartbeatsManager = new StandardHeartbeatsManager(

references/realtime-streams/src/trigger/streams.ts

@@ -113,7 +113,7 @@ export const streamsTask = task({
 
     const mockStream = createStreamFromGenerator(generator);
 
-    await streams.append("stream", mockStream);
+    const { wait } = await streams.append("stream", mockStream);
 
     await setTimeout(1000);
 
@@ -128,6 +128,8 @@ export const streamsTask = task({
       tokenCount++;
     }
 
+    await wait();
+
     logger.info("Stream completed", { scenario, tokenCount });
 
     return {