perf(webapp): cache deployment logs across deployment switches

Switching between deployments in the dashboard re-read the whole build log
stream from the start every time. Logs are now kept in a small in-memory,
per-tab cache keyed by deployment, seeded instantly on revisit, and the
stream is resumed from the next unread record instead of record zero. A
deployment whose stream has emitted its finalized event and reached a
terminal status is served from the cache without opening a stream at all.

The cache is bounded to 20 deployments and 20k log lines total, evicting
least recently viewed deployments first. Incoming records are also
batched into one state update per tick instead of one per line.

Author: myftija

apps/webapp/app/components/runs/v3/deploymentLogsCache.test.ts

@@ -0,0 +1,97 @@
+import { describe, expect, it } from "vitest";
+import { DeploymentLogsCache, type DeploymentLogEntry } from "./deploymentLogsCache";
+
+function lines(count: number): DeploymentLogEntry[] {
+  return Array.from({ length: count }, (_, i) => ({
+    message: `line ${i}`,
+    timestamp: new Date(0),
+    level: "info" as const,
+  }));
+}
+
+describe("DeploymentLogsCache", () => {
+  it("returns undefined for unknown keys", () => {
+    const cache = new DeploymentLogsCache(2, 100);
+    expect(cache.get("missing")).toBeUndefined();
+  });
+
+  it("stores and returns entries", () => {
+    const cache = new DeploymentLogsCache(2, 100);
+    const value = { logs: lines(3), nextSeqNum: 3, finalized: true, complete: true };
+    cache.set("a", value);
+    expect(cache.get("a")).toBe(value);
+    expect(cache.size).toBe(1);
+    expect(cache.lineCount).toBe(3);
+  });
+
+  it("evicts the least recently used deployment past the entry limit", () => {
+    const cache = new DeploymentLogsCache(2, 100);
+    cache.set("a", { logs: lines(1), nextSeqNum: 1, finalized: false, complete: false });
+    cache.set("b", { logs: lines(1), nextSeqNum: 1, finalized: false, complete: false });
+    cache.get("a");
+    cache.set("c", { logs: lines(1), nextSeqNum: 1, finalized: false, complete: false });
+
+    expect(cache.get("b")).toBeUndefined();
+    expect(cache.get("a")).toBeDefined();
+    expect(cache.get("c")).toBeDefined();
+    expect(cache.size).toBe(2);
+  });
+
+  it("evicts oldest deployments past the total line budget", () => {
+    const cache = new DeploymentLogsCache(10, 10);
+    cache.set("a", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+    cache.set("b", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+    cache.set("c", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+
+    expect(cache.get("a")).toBeUndefined();
+    expect(cache.get("b")).toBeDefined();
+    expect(cache.get("c")).toBeDefined();
+    expect(cache.lineCount).toBe(8);
+  });
+
+  it("always keeps the entry just set, even when it alone exceeds the budget", () => {
+    const cache = new DeploymentLogsCache(10, 10);
+    cache.set("a", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+    cache.set("big", { logs: lines(50), nextSeqNum: 50, finalized: true, complete: true });
+
+    expect(cache.get("a")).toBeUndefined();
+    expect(cache.get("big")?.logs).toHaveLength(50);
+    expect(cache.size).toBe(1);
+    expect(cache.lineCount).toBe(50);
+  });
+
+  it("treats replacing a key as a recent use", () => {
+    const cache = new DeploymentLogsCache(2, 100);
+    cache.set("a", { logs: lines(1), nextSeqNum: 1, finalized: false, complete: false });
+    cache.set("b", { logs: lines(1), nextSeqNum: 1, finalized: false, complete: false });
+    cache.set("a", { logs: lines(2), nextSeqNum: 2, finalized: true, complete: true });
+    cache.set("c", { logs: lines(1), nextSeqNum: 1, finalized: false, complete: false });
+
+    expect(cache.get("b")).toBeUndefined();
+    expect(cache.get("a")?.logs).toHaveLength(2);
+    expect(cache.get("c")).toBeDefined();
+  });
+
+  it("keeps recently read deployments when evicting for the line budget", () => {
+    const cache = new DeploymentLogsCache(10, 10);
+    cache.set("a", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+    cache.set("b", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+    cache.get("a");
+    cache.set("c", { logs: lines(4), nextSeqNum: 4, finalized: true, complete: true });
+
+    expect(cache.get("b")).toBeUndefined();
+    expect(cache.get("a")).toBeDefined();
+    expect(cache.get("c")).toBeDefined();
+    expect(cache.lineCount).toBe(8);
+  });
+
+  it("replaces an existing key without double counting lines", () => {
+    const cache = new DeploymentLogsCache(10, 100);
+    cache.set("a", { logs: lines(4), nextSeqNum: 4, finalized: false, complete: false });
+    cache.set("a", { logs: lines(6), nextSeqNum: 6, finalized: true, complete: true });
+
+    expect(cache.size).toBe(1);
+    expect(cache.lineCount).toBe(6);
+    expect(cache.get("a")?.complete).toBe(true);
+  });
+});

apps/webapp/app/components/runs/v3/deploymentLogsCache.ts

@@ -0,0 +1,57 @@
+export type DeploymentLogEntry = {
+  message: string;
+  timestamp: Date;
+  level: "info" | "error" | "warn" | "debug";
+};
+
+export type CachedDeploymentLogs = {
+  logs: readonly DeploymentLogEntry[];
+  nextSeqNum: number;
+  finalized: boolean;
+  complete: boolean;
+};
+
+export class DeploymentLogsCache {
+  private entries = new Map<string, CachedDeploymentLogs>();
+  private totalLines = 0;
+
+  constructor(
+    private readonly maxDeployments: number,
+    private readonly maxTotalLines: number
+  ) {}
+
+  get(key: string): CachedDeploymentLogs | undefined {
+    const entry = this.entries.get(key);
+    if (!entry) return undefined;
+    this.entries.delete(key);
+    this.entries.set(key, entry);
+    return entry;
+  }
+
+  set(key: string, value: CachedDeploymentLogs) {
+    const existing = this.entries.get(key);
+    if (existing) {
+      this.totalLines -= existing.logs.length;
+      this.entries.delete(key);
+    }
+    this.entries.set(key, value);
+    this.totalLines += value.logs.length;
+
+    for (const [oldestKey, oldest] of this.entries) {
+      if (oldestKey === key) break;
+      if (this.entries.size <= this.maxDeployments && this.totalLines <= this.maxTotalLines) break;
+      this.entries.delete(oldestKey);
+      this.totalLines -= oldest.logs.length;
+    }
+  }
+
+  get size() {
+    return this.entries.size;
+  }
+
+  get lineCount() {
+    return this.totalLines;
+  }
+}
+
+export const deploymentLogsCache = new DeploymentLogsCache(20, 20_000);

apps/webapp/app/hooks/useDeploymentLogs.ts

@@ -0,0 +1,169 @@
+import { S2, S2Error } from "@s2-dev/streamstore";
+import { DeploymentEventFromString } from "@trigger.dev/core/v3/schemas";
+import type { WorkerDeploymentStatus } from "@trigger.dev/database";
+import { useEffect, useState } from "react";
+import {
+  deploymentLogsCache,
+  type DeploymentLogEntry,
+} from "~/components/runs/v3/deploymentLogsCache";
+
+type DeploymentEventStream = {
+  s2: {
+    basin: string;
+    stream: string;
+    accessToken: string;
+  };
+};
+
+const FINISHED_DEPLOYMENT_STATUSES = new Set<WorkerDeploymentStatus>([
+  "DEPLOYED",
+  "FAILED",
+  "CANCELED",
+  "TIMED_OUT",
+]);
+
+type UseDeploymentLogsOptions = {
+  eventStream: DeploymentEventStream | undefined;
+  status: WorkerDeploymentStatus;
+};
+
+export function useDeploymentLogs({ eventStream, status }: UseDeploymentLogsOptions) {
+  const [logs, setLogs] = useState<readonly DeploymentLogEntry[]>([]);
+  const [isStreaming, setIsStreaming] = useState(true);
+  const [streamError, setStreamError] = useState<string | null>(null);
+
+  const basin = eventStream?.s2.basin;
+  const stream = eventStream?.s2.stream;
+  const accessToken = eventStream?.s2.accessToken;
+
+  useEffect(() => {
+    if (!basin || !stream || !accessToken) return;
+
+    const isFinished = FINISHED_DEPLOYMENT_STATUSES.has(status);
+    const cacheKey = `${basin}/${stream}`;
+    const cached = deploymentLogsCache.get(cacheKey);
+
+    let entries = cached?.logs ?? [];
+    let nextSeqNum = cached?.nextSeqNum ?? 0;
+    let pending: DeploymentLogEntry[] = [];
+    let flushTimer: ReturnType<typeof setTimeout> | undefined;
+    let finalized = cached?.finalized ?? false;
+
+    // oxlint-disable-next-line react/set-state-in-effect -- Seed from the cache when the selected deployment changes.
+    setLogs(entries);
+    setStreamError(null);
+
+    if (cached?.complete) {
+      setIsStreaming(false);
+      return;
+    }
+
+    setIsStreaming(true);
+
+    const abortController = new AbortController();
+
+    const flush = () => {
+      clearTimeout(flushTimer);
+      flushTimer = undefined;
+      if (abortController.signal.aborted || pending.length === 0) return;
+      entries = entries.concat(pending);
+      pending = [];
+      setLogs(entries);
+    };
+
+    const push = (entry: DeploymentLogEntry) => {
+      pending.push(entry);
+      flushTimer ??= setTimeout(flush, 0);
+    };
+
+    const store = () => {
+      clearTimeout(flushTimer);
+      flushTimer = undefined;
+      if (pending.length > 0) {
+        entries = entries.concat(pending);
+        pending = [];
+      }
+      if (entries.length === 0 && nextSeqNum === 0 && !finalized) return;
+      deploymentLogsCache.set(cacheKey, {
+        logs: entries,
+        nextSeqNum,
+        finalized,
+        complete: finalized && isFinished,
+      });
+    };
+
+    const streamLogs = async () => {
+      try {
+        const s2 = new S2({ accessToken });
+        const readSession = await s2
+          .basin(basin)
+          .stream(stream)
+          .readSession(
+            {
+              start: { from: { seqNum: nextSeqNum }, clamp: true },
+              stop: { waitSecs: 60 },
+            },
+            { signal: abortController.signal }
+          );
+
+        for await (const record of readSession) {
+          nextSeqNum = record.seqNum + 1;
+
+          const decoded = record.body;
+          const result = DeploymentEventFromString.safeParse(decoded);
+
+          if (!result.success) {
+            // fallback to the previous format in s2 logs for compatibility
+            const headers: Record<string, string> = {};
+            if (record.headers) {
+              for (const [name, value] of record.headers) {
+                headers[name] = value;
+              }
+            }
+            const level =
+              (headers["level"]?.toLowerCase() as DeploymentLogEntry["level"]) ?? "info";
+
+            push({ timestamp: new Date(record.timestamp), message: decoded, level });
+            continue;
+          }
+
+          const event = result.data;
+          if (event.type === "finalized") finalized = true;
+          if (event.type !== "log") continue;
+
+          push({
+            timestamp: new Date(record.timestamp),
+            message: event.data.message,
+            level: event.data.level,
+          });
+        }
+      } catch (error) {
+        if (abortController.signal.aborted) return;
+
+        if (error instanceof S2Error && error.code === "stream_not_found") {
+          finalized = isFinished;
+          return;
+        }
+        if (error instanceof S2Error && error.code === "permission_denied") return;
+
+        console.error("Failed to stream logs:", error);
+        setStreamError("Failed to stream logs");
+      } finally {
+        if (!abortController.signal.aborted) {
+          flush();
+          setIsStreaming(false);
+          store();
+        }
+      }
+    };
+
+    streamLogs();
+
+    return () => {
+      abortController.abort();
+      store();
+    };
+  }, [basin, stream, accessToken, status]);
+
+  return { logs, isStreaming, streamError };
+}