Various coderabbit fixes

Author: ericallam

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -97,7 +97,7 @@ const { action, loader } = createActionApiRoute(
       } else if (error instanceof OutOfEntitlementError) {
         return json({ error: error.message }, { status: 422 });
       } else if (error instanceof Error) {
-        return json({ error: error.message }, { status: 400 });
+        return json({ error: error.message }, { status: 500 });
       }
 
       return json({ error: "Something went wrong" }, { status: 500 });

apps/webapp/app/routes/realtime.v1.streams.test.ts

@@ -59,14 +59,18 @@ export class RealtimeStreams {
             } catch (error) {
               if (signal.aborted) break;
 
-              console.error("Error reading from Redis stream:", error);
+              logger.error("[RealtimeStreams][streamResponse] Error reading from Redis stream:", {
+                error,
+              });
               retryCount++;
               if (retryCount >= maxRetries) throw error;
               await new Promise((resolve) => setTimeout(resolve, 1000 * retryCount));
             }
           }
         } catch (error) {
-          console.error("Fatal error in stream processing:", error);
+          logger.error("[RealtimeStreams][streamResponse] Fatal error in stream processing:", {
+            error,
+          });
           controller.error(error);
         } finally {
           await cleanup();
@@ -163,7 +167,8 @@ export class RealtimeStreams {
 
       return new Response(null, { status: 200 });
     } catch (error) {
-      console.error("Error in ingestData:", error);
+      logger.error("[RealtimeStreams][ingestData] Error in ingestData:", { error });
+
       return new Response(null, { status: 500 });
     } finally {
       await cleanup();

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

@@ -14,9 +14,9 @@ import { RegistryProxy } from "~/v3/registryProxy.server";
 
 const app = express();
 
-// if (process.env.DISABLE_COMPRESSION !== "1") {
-//   app.use(compression());
-// }
+if (process.env.DISABLE_COMPRESSION !== "1") {
+  app.use(compression());
+}
 
 // http://expressjs.com/en/advanced/best-practice-security.html#at-a-minimum-disable-x-powered-by-header
 app.disable("x-powered-by");
@@ -73,63 +73,15 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
     next();
   });
 
-  app.post("/realtime/v1/streams/express/test", async (req, res) => {
-    // Ensure the request is a readable stream
-    const { method, headers } = req;
-    console.log("Inside /realtime/v1/streams/express/test");
-
-    if (method !== "POST") {
-      return res.status(405).send("Method Not Allowed");
-    }
+  app.use((req, res, next) => {
+    // Generate a unique request ID for each request
+    const requestId = nanoid();
 
-    // Set encoding to UTF-8 to read string data
-    req.setEncoding("utf8");
-
-    let buffer = "";
-
-    try {
-      req.on("data", (chunk) => {
-        buffer += chunk;
-        const lines = buffer.split("\n");
-        buffer = lines.pop() || "";
-
-        for (const line of lines) {
-          if (line.trim()) {
-            const data = JSON.parse(line);
-            console.log(`${new Date().toISOString()} Received data:`, data);
-            // You can process each data chunk as needed
-          }
-        }
-      });
-
-      req.on("end", () => {
-        if (buffer) {
-          const data = JSON.parse(buffer);
-          console.log(`${new Date().toISOString()} Received data at end:`, data);
-          // You can process the remaining data as needed
-        }
-        res.status(200).send(); // Send a success response
-      });
-
-      req.on("error", (error) => {
-        console.error("Error processing stream:", error);
-        res.status(500).send("Internal Server Error");
-      });
-    } catch (error) {
-      console.error("Error processing stream:", error);
-      res.status(500).send("Internal Server Error");
-    }
+    runWithHttpContext({ requestId, path: req.url, host: req.hostname, method: req.method }, next);
   });
 
-  // app.use((req, res, next) => {
-  //   // Generate a unique request ID for each request
-  //   const requestId = nanoid();
-
-  //   runWithHttpContext({ requestId, path: req.url, host: req.hostname, method: req.method }, next);
-  // });
-
   if (process.env.DASHBOARD_AND_API_DISABLED !== "true") {
-    // app.use(apiRateLimiter);
+    app.use(apiRateLimiter);
 
     app.all(
       "*",

apps/webapp/server.ts

@@ -147,6 +147,10 @@ export class SSEStreamSubscriptionFactory implements StreamSubscriptionFactory {
   ) {}
 
   createSubscription(runId: string, streamKey: string, baseUrl?: string): StreamSubscription {
+    if (!runId || !streamKey) {
+      throw new Error("runId and streamKey are required");
+    }
+
     const url = `${baseUrl ?? this.baseUrl}/realtime/v1/streams/${runId}/${streamKey}`;
     return new SSEStreamSubscription(url, this.options);
   }

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

@@ -260,7 +260,7 @@ export class StandardMetadataManager implements RunMetadataManager {
   }
 
   // Waits for all the streams to finish
-  public async waitForAllStreams(timeout: number = 30_000): Promise<void> {
+  public async waitForAllStreams(timeout: number = 60_000): Promise<void> {
     if (this.activeStreams.size === 0) {
       return;
     }

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

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