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..fb581aa87ea --- /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 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 = {}; + 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 }; +} 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;