Improve stream dashboard perf with virtualization

Author: ericallam

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx

@@ -1,6 +1,7 @@
 import { BoltIcon, BoltSlashIcon } from "@heroicons/react/20/solid";
 import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { type SSEStreamPart, SSEStreamSubscription } from "@trigger.dev/core/v3";
+import { useVirtualizer } from "@tanstack/react-virtual";
 import { Clipboard, ClipboardCheck } from "lucide-react";
 import { useCallback, useEffect, useRef, useState } from "react";
 import simplur from "simplur";
@@ -117,6 +118,8 @@ export function RealtimeStreamViewer({
   const [mouseOver, setMouseOver] = useState(false);
   const [copied, setCopied] = useState(false);
 
+  console.log("chunks.length", chunks.length);
+
   const getCompactText = useCallback(() => {
     return chunks
       .map((chunk) => {
@@ -208,6 +211,14 @@ export function RealtimeStreamViewer({
   const maxLineNumberWidth = (chunks.length > 0 ? lastLineNumber : firstLineNumber).toString()
     .length;
 
+  // Virtual rendering for list view
+  const rowVirtualizer = useVirtualizer({
+    count: chunks.length,
+    getScrollElement: () => scrollRef.current,
+    estimateSize: () => 28,
+    overscan: 5,
+  });
+
   return (
     <div className="flex h-full flex-col overflow-hidden">
       {/* Header */}
@@ -360,16 +371,33 @@ export function RealtimeStreamViewer({
 
         {chunks.length > 0 && viewMode === "list" && (
           <div className="font-mono text-xs leading-tight">
-            {chunks.map((chunk, index) => (
-              <StreamChunkLine
-                key={index}
-                chunk={chunk}
-                lineNumber={firstLineNumber + index}
-                maxLineNumberWidth={maxLineNumberWidth}
+            <div
+              style={{
+                height: `${rowVirtualizer.getTotalSize()}px`,
+                width: "100%",
+                position: "relative",
+              }}
+            >
+              {rowVirtualizer.getVirtualItems().map((virtualItem) => (
+                <StreamChunkLine
+                  key={virtualItem.key}
+                  chunk={chunks[virtualItem.index]}
+                  lineNumber={firstLineNumber + virtualItem.index}
+                  maxLineNumberWidth={maxLineNumberWidth}
+                  size={virtualItem.size}
+                  start={virtualItem.start}
+                />
+              ))}
+              {/* Sentinel element for IntersectionObserver */}
+              <div
+                ref={bottomRef}
+                className="h-px"
+                style={{
+                  position: "absolute",
+                  top: `${rowVirtualizer.getTotalSize()}px`,
+                }}
               />
-            ))}
-            {/* Sentinel element for IntersectionObserver */}
-            <div ref={bottomRef} className="h-px" />
+            </div>
           </div>
         )}
 
@@ -402,10 +430,14 @@ function StreamChunkLine({
   chunk,
   lineNumber,
   maxLineNumberWidth,
+  size,
+  start,
 }: {
   chunk: StreamChunk;
   lineNumber: number;
   maxLineNumberWidth: number;
+  size: number;
+  start: number;
 }) {
   const formattedData =
     typeof chunk.data === "string" ? chunk.data : JSON.stringify(chunk.data, null, 2);
@@ -421,7 +453,17 @@ function StreamChunkLine({
   const timestamp = `${timeString}.${milliseconds}`;
 
   return (
-    <div className="group flex w-full gap-3 py-1 hover:bg-charcoal-800">
+    <div
+      className="group flex w-full gap-3 py-1 hover:bg-charcoal-800"
+      style={{
+        position: "absolute",
+        top: 0,
+        left: 0,
+        width: "100%",
+        height: `${size}px`,
+        transform: `translateY(${start}px)`,
+      }}
+    >
       {/* Line number */}
       <div
         className="flex-none select-none pl-2 text-right text-charcoal-500"

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

@@ -1,5 +1,5 @@
 import { demoStream } from "@/app/streams";
-import { logger, task } from "@trigger.dev/sdk";
+import { logger, metadata, task } from "@trigger.dev/sdk";
 import { setTimeout } from "timers/promises";
 
 export type STREAMS = {
@@ -46,8 +46,6 @@ export type StreamPayload = {
 export const streamsTask = task({
   id: "streams",
   run: async (payload: StreamPayload = {}, { ctx }) => {
-    await setTimeout(1000);
-
     const scenario = payload.scenario ?? "continuous";
     logger.info("Starting stream scenario", { scenario });
 
@@ -118,7 +116,89 @@ export const streamsTask = task({
 
     await waitUntilComplete();
 
-    await streamsChildTask.triggerAndWait({});
+    logger.info("Stream completed", { scenario });
+
+    return {
+      scenario,
+      scenarioDescription,
+    };
+  },
+});
+
+export const metadataStreamsTask = task({
+  id: "metadata-streams",
+  run: async (payload: StreamPayload = {}, { ctx }) => {
+    const scenario = payload.scenario ?? "continuous";
+    logger.info("Starting stream scenario", { scenario });
+
+    let generator: AsyncGenerator<string>;
+    let scenarioDescription: string;
+
+    switch (scenario) {
+      case "stall": {
+        const stallDurationMs = payload.stallDurationMs ?? 3 * 60 * 1000; // Default 3 minutes
+        const includePing = payload.includePing ?? false;
+        generator = generateLLMTokenStream(includePing, stallDurationMs);
+        scenarioDescription = `Stall scenario: ${stallDurationMs / 1000}s with ${
+          includePing ? "ping tokens" : "no pings"
+        }`;
+        break;
+      }
+      case "continuous": {
+        const durationSec = payload.durationSec ?? 45;
+        const intervalMs = payload.intervalMs ?? 10;
+        generator = generateContinuousTokenStream(durationSec, intervalMs);
+        scenarioDescription = `Continuous scenario: ${durationSec}s with ${intervalMs}ms intervals`;
+        break;
+      }
+      case "burst": {
+        const burstCount = payload.burstCount ?? 10;
+        const tokensPerBurst = payload.tokensPerBurst ?? 20;
+        const burstIntervalMs = payload.burstIntervalMs ?? 5;
+        const pauseBetweenBurstsMs = payload.pauseBetweenBurstsMs ?? 2000;
+        generator = generateBurstTokenStream(
+          burstCount,
+          tokensPerBurst,
+          burstIntervalMs,
+          pauseBetweenBurstsMs
+        );
+        scenarioDescription = `Burst scenario: ${burstCount} bursts of ${tokensPerBurst} tokens`;
+        break;
+      }
+      case "slow-steady": {
+        const durationMin = payload.durationMin ?? 5;
+        const tokenIntervalSec = payload.tokenIntervalSec ?? 5;
+        generator = generateSlowSteadyTokenStream(durationMin, tokenIntervalSec);
+        scenarioDescription = `Slow steady scenario: ${durationMin}min with ${tokenIntervalSec}s intervals`;
+        break;
+      }
+      case "markdown": {
+        const tokenDelayMs = payload.tokenDelayMs ?? 15;
+        generator = generateMarkdownTokenStream(tokenDelayMs);
+        scenarioDescription = `Markdown scenario: generating formatted content with ${tokenDelayMs}ms delays`;
+        break;
+      }
+      case "performance": {
+        const chunkCount = payload.chunkCount ?? 500;
+        const chunkIntervalMs = payload.chunkIntervalMs ?? 10;
+        generator = generatePerformanceStream(chunkCount, chunkIntervalMs);
+        scenarioDescription = `Performance scenario: ${chunkCount} chunks with ${chunkIntervalMs}ms intervals`;
+        break;
+      }
+      default: {
+        throw new Error(`Unknown scenario: ${scenario}`);
+      }
+    }
+
+    logger.info("Starting stream", { scenarioDescription });
+
+    const mockStream = createStreamFromGenerator(generator);
+
+    const stream = await metadata.stream("demo", mockStream);
+
+    for await (const chunk of stream) {
+      logger.info("Received chunk", { chunk });
+    }
 
     logger.info("Stream completed", { scenario });