perf(webapp): cache deployment logs across navigations

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 s2Stream = new S2({ accessToken }).basin(basin).stream(stream);
+
+        do {
+          const readSession = await s2Stream.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,
+            });
+          }
+        } while (!abortController.signal.aborted && !finalized && !isFinished);
+      } 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 };
+}