From 424339db498f3c10d6a2265c0ca6e3dc80e03be9 Mon Sep 17 00:00:00 2001 From: Saadi Myftija Date: Tue, 25 Aug 2026 13:18:38 +0200 Subject: [PATCH 1/2] 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. --- .../runs/v3/deploymentLogsCache.test.ts | 97 ++++++++++ .../components/runs/v3/deploymentLogsCache.ts | 57 ++++++ apps/webapp/app/hooks/useDeploymentLogs.ts | 169 ++++++++++++++++++ .../route.tsx | 120 +------------ 4 files changed, 330 insertions(+), 113 deletions(-) create mode 100644 apps/webapp/app/components/runs/v3/deploymentLogsCache.test.ts create mode 100644 apps/webapp/app/components/runs/v3/deploymentLogsCache.ts create mode 100644 apps/webapp/app/hooks/useDeploymentLogs.ts diff --git a/apps/webapp/app/components/runs/v3/deploymentLogsCache.test.ts b/apps/webapp/app/components/runs/v3/deploymentLogsCache.test.ts new file mode 100644 index 00000000000..3253b11411b --- /dev/null +++ b/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); + }); +}); diff --git a/apps/webapp/app/components/runs/v3/deploymentLogsCache.ts b/apps/webapp/app/components/runs/v3/deploymentLogsCache.ts new file mode 100644 index 00000000000..0cb3977437d --- /dev/null +++ b/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(); + 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); diff --git a/apps/webapp/app/hooks/useDeploymentLogs.ts b/apps/webapp/app/hooks/useDeploymentLogs.ts new file mode 100644 index 00000000000..b4cc6f0079f --- /dev/null +++ b/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([ + "DEPLOYED", + "FAILED", + "CANCELED", + "TIMED_OUT", +]); + +type UseDeploymentLogsOptions = { + eventStream: DeploymentEventStream | undefined; + status: WorkerDeploymentStatus; +}; + +export function useDeploymentLogs({ eventStream, status }: UseDeploymentLogsOptions) { + const [logs, setLogs] = useState([]); + const [isStreaming, setIsStreaming] = useState(true); + const [streamError, setStreamError] = useState(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 | 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 = {}; + 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 }; +} diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments.$deploymentParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments.$deploymentParam/route.tsx index 343df5bff82..bfa7e5918fb 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments.$deploymentParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments.$deploymentParam/route.tsx @@ -2,7 +2,6 @@ import { useLocation } from "@remix-run/react"; import { type LoaderFunctionArgs } from "@remix-run/server-runtime"; import { typedjson, useTypedLoaderData } from "remix-typedjson"; import { useEffect, useState, useRef, useCallback } from "react"; -import { S2, S2Error } from "@s2-dev/streamstore"; import { Clipboard, ClipboardCheck, @@ -51,7 +50,8 @@ import { cn } from "~/utils/cn"; import { v3DeploymentParams, v3DeploymentsPath, v3RunsPath } from "~/utils/pathBuilder"; import { capitalizeWord } from "~/utils/string"; import { UserTag } from "../_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments/route"; -import { DeploymentEventFromString } from "@trigger.dev/core/v3/schemas"; +import { useDeploymentLogs } from "~/hooks/useDeploymentLogs"; +import { type DeploymentLogEntry } from "~/components/runs/v3/deploymentLogsCache"; import { deploymentAgentPageContext } from "~/components/dashboard-agent/suggested-prompts"; import type { Handle } from "~/utils/handle"; @@ -91,12 +91,6 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { } }; -type LogEntry = { - message: string; - timestamp: Date; - level: "info" | "error" | "warn" | "debug"; -}; - function getTriggeredViaDisplay(triggeredVia: string | null | undefined): { icon: React.ReactNode; label: string; @@ -205,110 +199,10 @@ export default function Page() { const page = new URLSearchParams(location.search).get("page"); const logsDisabled = eventStream === undefined; - const [logs, setLogs] = useState([]); - const [isStreaming, setIsStreaming] = useState(true); - const [streamError, setStreamError] = useState(null); - const isPending = deployment.status === "PENDING"; - - useEffect(() => { - if (logsDisabled) return; - - const abortController = new AbortController(); - - // oxlint-disable-next-line react/set-state-in-effect -- This effect intentionally synchronizes route state after an external or lifecycle change. - setLogs([]); - setStreamError(null); - setIsStreaming(true); - - const streamLogs = async () => { - try { - const s2 = new S2({ accessToken: eventStream.s2.accessToken }); - const basin = s2.basin(eventStream.s2.basin); - const stream = basin.stream(eventStream.s2.stream); - - const readSession = await stream.readSession( - { - start: { from: { seqNum: 0 }, clamp: true }, - stop: { waitSecs: 60 }, - }, - { signal: abortController.signal } - ); - - for await (const record of readSession) { - const decoded = record.body; - const result = DeploymentEventFromString.safeParse(decoded); - - if (!result.success) { - // fallback to the previous format in s2 logs for compatibility - try { - const headers: Record = {}; - - if (record.headers) { - for (const [name, value] of record.headers) { - headers[name] = value; - } - } - const level = (headers["level"]?.toLowerCase() as LogEntry["level"]) ?? "info"; - - setLogs((prevLogs) => [ - ...prevLogs, - { - timestamp: new Date(record.timestamp), - message: decoded, - level, - }, - ]); - } catch (err) { - console.error("Failed to parse log record:", err); - } - - continue; - } - - const event = result.data; - if (event.type !== "log") { - continue; - } - - setLogs((prevLogs) => [ - ...prevLogs, - { - timestamp: new Date(record.timestamp), - message: event.data.message, - level: event.data.level, - }, - ]); - } - } catch (error) { - if (abortController.signal.aborted) return; - - const isNotFoundError = - error instanceof S2Error && - error.code && - ["permission_denied", "stream_not_found"].includes(error.code); - if (isNotFoundError) return; - - console.error("Failed to stream logs:", error); - setStreamError("Failed to stream logs"); - } finally { - if (!abortController.signal.aborted) { - setIsStreaming(false); - } - } - }; - - streamLogs(); - - return () => { - abortController.abort(); - }; - }, [ - eventStream?.s2?.basin, - eventStream?.s2?.stream, - eventStream?.s2?.accessToken, - isPending, - logsDisabled, - ]); + const { logs, isStreaming, streamError } = useDeploymentLogs({ + eventStream, + status: deployment.status, + }); return (
@@ -622,7 +516,7 @@ function LogsDisplay({ streamError, initialCollapsed = false, }: { - logs: LogEntry[]; + logs: readonly DeploymentLogEntry[]; isStreaming: boolean; streamError: string | null; initialCollapsed?: boolean; From c29e7bbe594d6afda6ff96d5d99fa4d28d53bf15 Mon Sep 17 00:00:00 2001 From: Saadi Myftija Date: Tue, 25 Aug 2026 13:34:13 +0200 Subject: [PATCH 2/2] fix(webapp): keep following deployment logs across idle read-session timeouts An in-progress deployment that produced no log output for 60 seconds ended the read session and stopped streaming until the next status change. The session is now reopened from the next unread record until the deployment is finalized. --- apps/webapp/app/hooks/useDeploymentLogs.ts | 60 +++++++++++----------- 1 file changed, 30 insertions(+), 30 deletions(-) diff --git a/apps/webapp/app/hooks/useDeploymentLogs.ts b/apps/webapp/app/hooks/useDeploymentLogs.ts index b4cc6f0079f..fb581aa87ea 100644 --- a/apps/webapp/app/hooks/useDeploymentLogs.ts +++ b/apps/webapp/app/hooks/useDeploymentLogs.ts @@ -94,11 +94,10 @@ export function useDeploymentLogs({ eventStream, status }: UseDeploymentLogsOpti const streamLogs = async () => { try { - const s2 = new S2({ accessToken }); - const readSession = await s2 - .basin(basin) - .stream(stream) - .readSession( + const s2Stream = new S2({ accessToken }).basin(basin).stream(stream); + + do { + const readSession = await s2Stream.readSession( { start: { from: { seqNum: nextSeqNum }, clamp: true }, stop: { waitSecs: 60 }, @@ -106,37 +105,38 @@ export function useDeploymentLogs({ eventStream, status }: UseDeploymentLogsOpti { signal: abortController.signal } ); - for await (const record of readSession) { - nextSeqNum = record.seqNum + 1; + for await (const record of readSession) { + nextSeqNum = record.seqNum + 1; - const decoded = record.body; - const result = DeploymentEventFromString.safeParse(decoded); + 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 = {}; - if (record.headers) { - for (const [name, value] of record.headers) { - headers[name] = value; + if (!result.success) { + // fallback to the previous format in s2 logs for compatibility + const headers: Record = {}; + if (record.headers) { + for (const [name, value] of record.headers) { + headers[name] = value; + } } - } - const level = - (headers["level"]?.toLowerCase() as DeploymentLogEntry["level"]) ?? "info"; + const level = + (headers["level"]?.toLowerCase() as DeploymentLogEntry["level"]) ?? "info"; - push({ timestamp: new Date(record.timestamp), message: decoded, level }); - continue; - } + 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; + 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, - }); - } + 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;