From 8eb9771a4ed8efa790f33f49628ac14a3b0152f0 Mon Sep 17 00:00:00 2001 From: mbeaulne Date: Tue, 21 Jul 2026 11:54:00 -0400 Subject: [PATCH] Load and normalize run timing data --- .../components/RunTiming/RunTimingView.tsx | 33 +- .../components/RunTiming/runTiming.test.ts | 278 ++++++++++ .../RunView/components/RunTiming/runTiming.ts | 522 ++++++++++++++++++ .../components/RunTiming/runTiming.types.ts | 71 +++ .../RunTiming/runTimingService.test.ts | 158 ++++++ .../components/RunTiming/runTimingService.ts | 199 +++++++ .../RunTiming/useRunTimingData.test.tsx | 82 +++ .../components/RunTiming/useRunTimingData.ts | 96 ++++ src/services/executionService.ts | 6 +- 9 files changed, 1442 insertions(+), 3 deletions(-) create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/runTiming.test.ts create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/runTiming.ts create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/runTiming.types.ts create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/runTimingService.test.ts create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/runTimingService.ts create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.test.tsx create mode 100644 src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.ts diff --git a/src/routes/v2/pages/RunView/components/RunTiming/RunTimingView.tsx b/src/routes/v2/pages/RunView/components/RunTiming/RunTimingView.tsx index 0d7d4bb88b..a59bd8178a 100644 --- a/src/routes/v2/pages/RunView/components/RunTiming/RunTimingView.tsx +++ b/src/routes/v2/pages/RunView/components/RunTiming/RunTimingView.tsx @@ -1,8 +1,39 @@ +import { InfoBox } from "@/components/shared/InfoBox"; +import { LoadingScreen } from "@/components/shared/LoadingScreen"; import { Icon } from "@/components/ui/icon"; import { BlockStack } from "@/components/ui/layout"; import { Heading, Paragraph } from "@/components/ui/typography"; +import { useExecutionData } from "@/providers/ExecutionDataProvider"; +import { + flattenExecutionStatusStats, + isExecutionComplete, +} from "@/utils/executionStatus"; + +import { useRunTimingData } from "./useRunTimingData"; export function RunTimingView() { + const { rootDetails, rootState, metadata } = useExecutionData(); + const runComplete = isExecutionComplete( + flattenExecutionStatusStats(rootState?.child_execution_status_stats), + ); + const { data, error, isLoading } = useRunTimingData({ + rootDetails, + runCreatedAt: metadata?.created_at, + runComplete, + }); + + if (isLoading) return ; + + if (error) { + return ( + + + {error.message} + + + ); + } + return ( Run timing - Task timing and run performance will appear here. + {data?.tasks.length ?? 0} tasks ready for timing analysis. ); diff --git a/src/routes/v2/pages/RunView/components/RunTiming/runTiming.test.ts b/src/routes/v2/pages/RunView/components/RunTiming/runTiming.test.ts new file mode 100644 index 0000000000..cea0074405 --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/runTiming.test.ts @@ -0,0 +1,278 @@ +import { describe, expect, it } from "vitest"; + +import type { + GetContainerExecutionStateResponse, + GetExecutionInfoResponse, +} from "@/api/types.gen"; + +import { formatTimingDuration, normalizeRunTimingData } from "./runTiming"; +import type { RunTimingTaskSource } from "./runTiming.types"; + +const MINUTE = 60_000; +const RUN_START = Date.parse("2026-07-14T10:00:00.000Z"); + +function details({ + id, + inputArtifactIds = [], + outputArtifactIds = [], +}: { + id: string; + inputArtifactIds?: string[]; + outputArtifactIds?: string[]; +}): GetExecutionInfoResponse { + return { + id, + child_task_execution_ids: {}, + input_artifacts: Object.fromEntries( + inputArtifactIds.map((artifactId, index) => [ + `input-${index}`, + { id: artifactId }, + ]), + ), + output_artifacts: Object.fromEntries( + outputArtifactIds.map((artifactId, index) => [ + `output-${index}`, + { id: artifactId }, + ]), + ), + task_spec: { + componentRef: { + name: "Python component", + digest: "sha256:abc", + spec: { + name: "Python component", + implementation: { container: { image: "python:3.12" } }, + }, + }, + }, + }; +} + +function containerState( + startMinutes: number, + endMinutes: number | undefined, + status: GetContainerExecutionStateResponse["status"] = "SUCCEEDED", +): GetContainerExecutionStateResponse { + return { + status, + started_at: new Date(RUN_START + startMinutes * MINUTE).toISOString(), + ended_at: + endMinutes === undefined + ? undefined + : new Date(RUN_START + endMinutes * MINUTE).toISOString(), + }; +} + +function source({ + executionId, + parentExecutionId = "root", + depth = 0, + isSubgraph = false, + inputArtifactIds = [], + outputArtifactIds = [], + state, +}: { + executionId: string; + parentExecutionId?: string; + depth?: number; + isSubgraph?: boolean; + inputArtifactIds?: string[]; + outputArtifactIds?: string[]; + state?: GetContainerExecutionStateResponse; +}): RunTimingTaskSource { + return { + executionId, + parentExecutionId, + taskId: executionId, + navigationPath: depth === 0 ? ["Pipeline"] : ["Pipeline", "subgraph"], + depth, + inputArtifactIds, + outputArtifactIds, + details: details({ id: executionId, inputArtifactIds, outputArtifactIds }), + containerState: state, + isSubgraph, + }; +} + +describe("normalizeRunTimingData", () => { + it("reconstructs dependencies and startup gaps from artifact readiness", () => { + const data = normalizeRunTimingData( + [ + source({ + executionId: "prepare", + outputArtifactIds: ["artifact-a"], + state: containerState(1, 3), + }), + source({ + executionId: "train", + inputArtifactIds: ["artifact-a"], + state: containerState(4, 6), + }), + ], + new Date(RUN_START).toISOString(), + RUN_START + 10 * MINUTE, + ); + + expect(data.tasks[0].phases).toEqual([ + { + name: "startup", + startAt: RUN_START, + endAt: RUN_START + MINUTE, + durationMs: MINUTE, + }, + { + name: "runtime", + startAt: RUN_START + MINUTE, + endAt: RUN_START + 3 * MINUTE, + durationMs: 2 * MINUTE, + }, + ]); + expect(data.tasks[1]).toMatchObject({ + dependencyExecutionIds: ["prepare"], + readyAt: RUN_START + 3 * MINUTE, + phases: [ + { + name: "startup", + startAt: RUN_START + 3 * MINUTE, + endAt: RUN_START + 4 * MINUTE, + durationMs: MINUTE, + }, + { + name: "runtime", + startAt: RUN_START + 4 * MINUTE, + endAt: RUN_START + 6 * MINUTE, + durationMs: 2 * MINUTE, + }, + ], + }); + expect(data.rangeEnd).toBe(RUN_START + 6 * MINUTE); + expect(data.metrics).toMatchObject({ + wallClockDurationMs: 6 * MINUTE, + totalTaskCount: 2, + cachedTaskCount: 0, + averageStartupMs: MINUTE, + busyRuntimeMs: 4 * MINUTE, + busyPercent: 67, + criticalPathDurationMs: 6 * MINUTE, + }); + expect(data.criticalPathExecutionIds).toEqual( + new Set(["prepare", "train"]), + ); + }); + + it("uses the union of overlapping runtime intervals for compute time", () => { + const data = normalizeRunTimingData( + [ + source({ executionId: "a", state: containerState(1, 5) }), + source({ executionId: "b", state: containerState(3, 7) }), + ], + new Date(RUN_START).toISOString(), + RUN_START + 10 * MINUTE, + ); + + expect(data.metrics.busyRuntimeMs).toBe(6 * MINUTE); + expect(data.metrics.busyPercent).toBe(86); + }); + + it("propagates completion through cached artifact dependencies", () => { + const data = normalizeRunTimingData( + [ + source({ + executionId: "fresh-upstream", + outputArtifactIds: ["fresh-output"], + state: containerState(1, 2), + }), + source({ + executionId: "cached-middle", + inputArtifactIds: ["fresh-output"], + outputArtifactIds: ["cached-output"], + state: containerState(-10, -9), + }), + source({ + executionId: "fresh-downstream", + inputArtifactIds: ["cached-output"], + state: containerState(3, 4), + }), + ], + new Date(RUN_START).toISOString(), + RUN_START + 10 * MINUTE, + ); + + expect(data.tasks[1]).toMatchObject({ + cacheState: "hit", + durationMs: 0, + historicalRuntimeMs: MINUTE, + phases: [], + }); + expect(data.tasks[2]).toMatchObject({ + dependencyExecutionIds: ["cached-middle"], + readyAt: RUN_START + 2 * MINUTE, + }); + expect(data.metrics.cachedTaskCount).toBe(1); + expect(data.criticalPathExecutionIds).toEqual( + new Set(["fresh-upstream", "cached-middle", "fresh-downstream"]), + ); + }); + + it("resolves a subgraph output dependency to its latest-ending leaf", () => { + const data = normalizeRunTimingData( + [ + source({ + executionId: "subgraph", + isSubgraph: true, + outputArtifactIds: ["subgraph-output"], + }), + source({ + executionId: "child-a", + parentExecutionId: "subgraph", + depth: 1, + state: containerState(1, 2), + }), + source({ + executionId: "child-b", + parentExecutionId: "subgraph", + depth: 1, + state: containerState(1, 4), + }), + source({ + executionId: "downstream", + inputArtifactIds: ["subgraph-output"], + state: containerState(5, 6), + }), + ], + new Date(RUN_START).toISOString(), + RUN_START + 10 * MINUTE, + ); + + expect(data.tasks[0]).toMatchObject({ + isSubgraph: true, + startAt: RUN_START, + endAt: RUN_START + 4 * MINUTE, + }); + expect(data.tasks[3]).toMatchObject({ + dependencyExecutionIds: ["child-b"], + readyAt: RUN_START + 4 * MINUTE, + }); + }); + + it("keeps never-started leaves visible with unavailable timing", () => { + const data = normalizeRunTimingData( + [source({ executionId: "pending" })], + new Date(RUN_START).toISOString(), + RUN_START + MINUTE, + ); + + expect(data.tasks[0]).toMatchObject({ + phases: [], + timingQuality: "unavailable", + cacheState: "unknown", + }); + }); +}); + +describe("formatTimingDuration", () => { + it("formats millisecond durations for timing labels", () => { + expect(formatTimingDuration(3_725_000)).toBe("1h 2m 5s"); + expect(formatTimingDuration(undefined)).toBe("—"); + }); +}); diff --git a/src/routes/v2/pages/RunView/components/RunTiming/runTiming.ts b/src/routes/v2/pages/RunView/components/RunTiming/runTiming.ts new file mode 100644 index 0000000000..f76455a837 --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/runTiming.ts @@ -0,0 +1,522 @@ +import { getOverallExecutionStatusFromStats } from "@/utils/executionStatus"; + +import type { + RunTimingData, + RunTimingMetrics, + RunTimingPhase, + RunTimingTask, + RunTimingTaskSource, +} from "./runTiming.types"; + +function parseTimestamp(value: string | null | undefined): number | undefined { + if (!value) return undefined; + const timestamp = new Date(value).getTime(); + return Number.isFinite(timestamp) ? timestamp : undefined; +} + +function makePhase( + name: RunTimingPhase["name"], + startAt: number | undefined, + endAt: number | undefined, +): RunTimingPhase | undefined { + if (startAt === undefined || endAt === undefined || endAt < startAt) { + return undefined; + } + + return { name, startAt, endAt, durationMs: endAt - startAt }; +} + +interface BaseTaskTiming { + status?: string; + runtimeStart?: number; + runtimeEnd?: number; + historicalRuntimeMs?: number; + cacheState: RunTimingTask["cacheState"]; +} + +function normalizeBaseTiming( + source: RunTimingTaskSource, + runCreatedAt: number | undefined, + now: number, +): BaseTaskTiming { + if (source.isSubgraph) return { cacheState: "unknown" }; + + const status = source.containerState?.status; + const runtimeStart = parseTimestamp(source.containerState?.started_at); + const reportedEnd = parseTimestamp(source.containerState?.ended_at); + const runtimeEnd = + runtimeStart !== undefined && reportedEnd === undefined ? now : reportedEnd; + const reportedRuntimeMs = + runtimeStart !== undefined && runtimeEnd !== undefined + ? Math.max(0, runtimeEnd - runtimeStart) + : undefined; + const cacheState = + runtimeStart === undefined + ? "unknown" + : reportedEnd !== undefined && + runCreatedAt !== undefined && + runtimeStart < runCreatedAt + ? "hit" + : "miss"; + + return { + status, + runtimeStart, + runtimeEnd, + historicalRuntimeMs: cacheState === "hit" ? reportedRuntimeMs : undefined, + cacheState, + }; +} + +function descendantExecutionIds( + executionId: string, + childrenByParent: Map, +): string[] { + const descendants: string[] = []; + const queue = [...(childrenByParent.get(executionId) ?? [])]; + for (let index = 0; index < queue.length; index += 1) { + const child = queue[index]; + descendants.push(child.executionId); + queue.push(...(childrenByParent.get(child.executionId) ?? [])); + } + return descendants; +} + +function resolveArtifactDependencies( + sources: RunTimingTaskSource[], + baseTimingById: Map, +): Map { + const sourceById = new Map( + sources.map((source) => [source.executionId, source]), + ); + const childrenByParent = new Map(); + const producerByArtifactId = new Map(); + + for (const source of sources) { + const siblings = childrenByParent.get(source.parentExecutionId) ?? []; + siblings.push(source); + childrenByParent.set(source.parentExecutionId, siblings); + for (const artifactId of source.outputArtifactIds) { + if (!producerByArtifactId.has(artifactId)) { + producerByArtifactId.set(artifactId, source.executionId); + } + } + } + + const resolveProducerToLeaf = (executionId: string): string | undefined => { + const source = sourceById.get(executionId); + if (!source) return undefined; + if (!source.isSubgraph) return executionId; + + return descendantExecutionIds(executionId, childrenByParent) + .filter((descendantId) => { + const descendant = sourceById.get(descendantId); + return descendant && !descendant.isSubgraph; + }) + .reduce((latestId, candidateId) => { + const candidateEnd = baseTimingById.get(candidateId)?.runtimeEnd; + if (candidateEnd === undefined) return latestId; + if (latestId === undefined) return candidateId; + const latestEnd = baseTimingById.get(latestId)?.runtimeEnd; + return latestEnd === undefined || candidateEnd > latestEnd + ? candidateId + : latestId; + }, undefined); + }; + + return new Map( + sources.map((source) => { + const dependencies = new Set(); + for (const artifactId of source.inputArtifactIds) { + const producerId = producerByArtifactId.get(artifactId); + if (!producerId || producerId === source.executionId) continue; + const leafProducerId = resolveProducerToLeaf(producerId); + if (leafProducerId && leafProducerId !== source.executionId) { + dependencies.add(leafProducerId); + } + } + return [source.executionId, [...dependencies]]; + }), + ); +} + +function deriveLeafTasks( + sources: RunTimingTaskSource[], + baseTimingById: Map, + dependencies: Map, + rangeStart: number, +): RunTimingTask[] { + const sourceById = new Map( + sources.map((source) => [source.executionId, source]), + ); + const completionById = new Map(); + const visiting = new Set(); + + const effectiveCompletion = (executionId: string): number => { + const knownCompletion = completionById.get(executionId); + if (knownCompletion !== undefined) return knownCompletion; + if (visiting.has(executionId)) return rangeStart; + + visiting.add(executionId); + const timing = baseTimingById.get(executionId); + let completion = rangeStart; + if (timing?.cacheState !== "hit" && timing?.runtimeEnd !== undefined) { + completion = timing.runtimeEnd; + } else { + for (const dependencyId of dependencies.get(executionId) ?? []) { + completion = Math.max(completion, effectiveCompletion(dependencyId)); + } + } + visiting.delete(executionId); + completionById.set(executionId, completion); + return completion; + }; + + return sources + .filter((source) => !source.isSubgraph) + .map((source): RunTimingTask => { + const timing = baseTimingById.get(source.executionId) ?? { + cacheState: "unknown", + }; + const dependencyExecutionIds = ( + dependencies.get(source.executionId) ?? [] + ).filter((executionId) => sourceById.has(executionId)); + const readyAt = dependencyExecutionIds.reduce( + (latest, executionId) => + Math.max(latest, effectiveCompletion(executionId)), + rangeStart, + ); + const isFresh = timing.cacheState !== "hit"; + const startupPhase = isFresh + ? makePhase("startup", readyAt, timing.runtimeStart) + : undefined; + const runtimePhase = isFresh + ? makePhase("runtime", timing.runtimeStart, timing.runtimeEnd) + : undefined; + const phases = [startupPhase, runtimePhase].filter( + (phase): phase is RunTimingPhase => phase !== undefined, + ); + const startAt = timing.cacheState === "hit" ? rangeStart : readyAt; + const endAt = + timing.cacheState === "hit" ? rangeStart : timing.runtimeEnd; + const durationMs = + startAt !== undefined && endAt !== undefined + ? Math.max(0, endAt - startAt) + : undefined; + const componentRef = source.details?.task_spec.componentRef; + + return { + executionId: source.executionId, + parentExecutionId: source.parentExecutionId, + taskId: source.taskId, + taskName: source.taskId, + navigationPath: source.navigationPath, + depth: source.depth, + dependencyExecutionIds, + componentName: + componentRef?.spec?.name ?? componentRef?.name ?? undefined, + componentDigest: componentRef?.digest ?? undefined, + isSubgraph: false, + status: timing.status, + phases, + readyAt, + startAt, + endAt, + durationMs, + historicalRuntimeMs: timing.historicalRuntimeMs, + cacheState: timing.cacheState, + timingQuality: + timing.runtimeStart === undefined + ? "unavailable" + : timing.status === "RUNNING" + ? "partial" + : "complete", + }; + }); +} + +function deriveSubgraphTasks( + sources: RunTimingTaskSource[], + leafTasks: RunTimingTask[], + dependencies: Map, +): RunTimingTask[] { + const tasksByParent = new Map(); + for (const task of leafTasks) { + const siblings = tasksByParent.get(task.parentExecutionId) ?? []; + siblings.push(task); + tasksByParent.set(task.parentExecutionId, siblings); + } + + const subgraphSources = sources + .filter((source) => source.isSubgraph) + .sort((left, right) => right.depth - left.depth); + const subgraphTasks: RunTimingTask[] = []; + + for (const source of subgraphSources) { + const children = tasksByParent.get(source.executionId) ?? []; + const freshDescendants = children.filter( + (task) => task.cacheState !== "hit" && task.endAt !== undefined, + ); + const starts = freshDescendants.flatMap((task) => + task.readyAt === undefined ? [] : [task.readyAt], + ); + const ends = freshDescendants.flatMap((task) => + task.endAt === undefined ? [] : [task.endAt], + ); + const statusCounts = children.reduce>( + (counts, task) => { + if (task.status) counts[task.status] = (counts[task.status] ?? 0) + 1; + return counts; + }, + {}, + ); + const startAt = starts.length > 0 ? Math.min(...starts) : undefined; + const endAt = ends.length > 0 ? Math.max(...ends) : undefined; + const componentRef = source.details?.task_spec.componentRef; + const task: RunTimingTask = { + executionId: source.executionId, + parentExecutionId: source.parentExecutionId, + taskId: source.taskId, + taskName: source.taskId, + navigationPath: source.navigationPath, + depth: source.depth, + dependencyExecutionIds: dependencies.get(source.executionId) ?? [], + componentName: + componentRef?.spec?.name ?? componentRef?.name ?? undefined, + componentDigest: componentRef?.digest ?? undefined, + isSubgraph: true, + status: getOverallExecutionStatusFromStats(statusCounts), + phases: [], + readyAt: startAt, + startAt, + endAt, + durationMs: + startAt !== undefined && endAt !== undefined + ? endAt - startAt + : undefined, + cacheState: "unknown", + timingQuality: + freshDescendants.length === 0 + ? "unavailable" + : freshDescendants.every( + (descendant) => descendant.timingQuality === "complete", + ) + ? "complete" + : "partial", + }; + subgraphTasks.push(task); + + const siblings = tasksByParent.get(source.parentExecutionId) ?? []; + siblings.push(task); + tasksByParent.set(source.parentExecutionId, siblings); + } + + return subgraphTasks; +} + +interface CriticalPathResult { + executionIds: string[]; + durationMs?: number; +} + +export function calculateCriticalPath( + tasks: RunTimingTask[], + rangeStart?: number, +): CriticalPathResult { + const leafById = new Map( + tasks + .filter((task) => !task.isSubgraph) + .map((task) => [task.executionId, task]), + ); + const freshLeaves = [...leafById.values()].filter( + (task) => task.cacheState !== "hit" && task.endAt !== undefined, + ); + if (freshLeaves.length === 0) return { executionIds: [] }; + + const effectiveCompletionById = new Map(); + const visiting = new Set(); + const effectiveCompletion = (executionId: string): number => { + const known = effectiveCompletionById.get(executionId); + if (known !== undefined) return known; + if (visiting.has(executionId)) return rangeStart ?? 0; + + visiting.add(executionId); + const task = leafById.get(executionId); + let completion = rangeStart ?? 0; + if (task?.cacheState !== "hit" && task?.endAt !== undefined) { + completion = task.endAt; + } else { + for (const dependencyId of task?.dependencyExecutionIds ?? []) { + completion = Math.max(completion, effectiveCompletion(dependencyId)); + } + } + visiting.delete(executionId); + effectiveCompletionById.set(executionId, completion); + return completion; + }; + + const executionIds: string[] = []; + const seen = new Set(); + let current = freshLeaves.reduce((latest, task) => + (task.endAt ?? 0) > (latest.endAt ?? 0) ? task : latest, + ); + + while (!seen.has(current.executionId)) { + seen.add(current.executionId); + executionIds.push(current.executionId); + const dependencies = current.dependencyExecutionIds + .map((executionId) => leafById.get(executionId)) + .filter((task): task is RunTimingTask => task !== undefined); + if (dependencies.length === 0) break; + current = dependencies.reduce((latest, task) => + effectiveCompletion(task.executionId) > + effectiveCompletion(latest.executionId) + ? task + : latest, + ); + } + + executionIds.reverse(); + const durationMs = executionIds.reduce((total, executionId) => { + const task = leafById.get(executionId); + if (!task || task.cacheState === "hit") return total; + return ( + total + task.phases.reduce((sum, phase) => sum + phase.durationMs, 0) + ); + }, 0); + + return { executionIds, durationMs }; +} + +function intervalUnionDuration( + intervals: Array<{ startAt: number; endAt: number }>, +): number { + const sorted = [...intervals].sort( + (left, right) => left.startAt - right.startAt, + ); + if (sorted.length === 0) return 0; + + let total = 0; + let currentStart = sorted[0].startAt; + let currentEnd = sorted[0].endAt; + for (const interval of sorted.slice(1)) { + if (interval.startAt <= currentEnd) { + currentEnd = Math.max(currentEnd, interval.endAt); + } else { + total += currentEnd - currentStart; + currentStart = interval.startAt; + currentEnd = interval.endAt; + } + } + return total + currentEnd - currentStart; +} + +function calculateMetrics( + tasks: RunTimingTask[], + rangeStart: number, + rangeEnd: number, + criticalPathDurationMs: number | undefined, +): RunTimingMetrics { + const leafTasks = tasks.filter((task) => !task.isSubgraph); + const freshLeafTasks = leafTasks.filter((task) => task.cacheState !== "hit"); + const startupPhases = freshLeafTasks.flatMap((task) => + task.phases.filter((phase) => phase.name === "startup"), + ); + const runtimePhases = freshLeafTasks.flatMap((task) => + task.phases.filter((phase) => phase.name === "runtime"), + ); + const busyRuntimeMs = intervalUnionDuration(runtimePhases); + + return { + wallClockDurationMs: rangeEnd - rangeStart, + totalTaskCount: leafTasks.length, + cachedTaskCount: leafTasks.filter((task) => task.cacheState === "hit") + .length, + averageStartupMs: + startupPhases.length > 0 + ? startupPhases.reduce((sum, phase) => sum + phase.durationMs, 0) / + startupPhases.length + : undefined, + startupCoverage: startupPhases.length, + busyRuntimeMs, + busyPercent: + rangeEnd > rangeStart + ? Math.round((busyRuntimeMs / (rangeEnd - rangeStart)) * 100) + : 0, + criticalPathDurationMs, + }; +} + +export function normalizeRunTimingData( + sources: RunTimingTaskSource[], + runCreatedAt: string | null | undefined, + now: number, + truncated = false, +): RunTimingData { + const createdAt = parseTimestamp(runCreatedAt); + const baseTimingById = new Map( + sources.map((source) => [ + source.executionId, + normalizeBaseTiming(source, createdAt, now), + ]), + ); + const freshStarts = [...baseTimingById.values()].flatMap((timing) => + timing.cacheState !== "hit" && timing.runtimeStart !== undefined + ? [timing.runtimeStart] + : [], + ); + const rangeStart = createdAt ?? Math.min(...freshStarts, now); + const dependencies = resolveArtifactDependencies(sources, baseTimingById); + const leafTasks = deriveLeafTasks( + sources, + baseTimingById, + dependencies, + rangeStart, + ); + const subgraphTasks = deriveSubgraphTasks(sources, leafTasks, dependencies); + const tasks = sources.flatMap((source) => { + const candidates = source.isSubgraph ? subgraphTasks : leafTasks; + const task = candidates.find( + (candidate) => candidate.executionId === source.executionId, + ); + return task ? [task] : []; + }); + const freshEnds = leafTasks.flatMap((task) => + task.cacheState !== "hit" && task.endAt !== undefined ? [task.endAt] : [], + ); + const rangeEnd = + freshEnds.length > 0 + ? Math.max(...freshEnds) + : now > rangeStart + ? now + : rangeStart + 1_000; + const criticalPath = calculateCriticalPath(tasks, rangeStart); + + return { + tasks, + truncated, + rangeStart, + rangeEnd, + criticalPathExecutionIds: new Set(criticalPath.executionIds), + metrics: calculateMetrics( + tasks, + rangeStart, + rangeEnd, + criticalPath.durationMs, + ), + }; +} + +export function formatTimingDuration(durationMs: number | undefined): string { + if (durationMs === undefined || !Number.isFinite(durationMs)) return "—"; + + const totalSeconds = Math.max(0, Math.round(durationMs / 1000)); + const seconds = totalSeconds % 60; + const totalMinutes = Math.floor(totalSeconds / 60); + const minutes = totalMinutes % 60; + const hours = Math.floor(totalMinutes / 60); + + if (hours > 0) return `${hours}h ${minutes}m ${seconds}s`; + if (minutes > 0) return `${minutes}m ${seconds}s`; + return `${seconds}s`; +} diff --git a/src/routes/v2/pages/RunView/components/RunTiming/runTiming.types.ts b/src/routes/v2/pages/RunView/components/RunTiming/runTiming.types.ts new file mode 100644 index 0000000000..922cab6743 --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/runTiming.types.ts @@ -0,0 +1,71 @@ +import type { + GetContainerExecutionStateResponse, + GetExecutionInfoResponse, +} from "@/api/types.gen"; + +export type RunTimingPhaseName = "startup" | "runtime"; +export type RunTimingQuality = "complete" | "partial" | "unavailable"; +export type RunTimingCacheState = "hit" | "miss" | "unknown"; + +export interface RunTimingPhase { + name: RunTimingPhaseName; + startAt: number; + endAt: number; + durationMs: number; +} + +export interface RunTimingTaskSource { + executionId: string; + parentExecutionId: string; + taskId: string; + navigationPath: string[]; + depth: number; + inputArtifactIds: string[]; + outputArtifactIds: string[]; + details?: GetExecutionInfoResponse; + containerState?: GetContainerExecutionStateResponse; + isSubgraph: boolean; + loadFailed?: boolean; +} + +export interface RunTimingTask { + executionId: string; + parentExecutionId: string; + taskId: string; + taskName: string; + navigationPath: string[]; + depth: number; + dependencyExecutionIds: string[]; + componentName?: string; + componentDigest?: string; + isSubgraph: boolean; + status?: string; + phases: RunTimingPhase[]; + readyAt?: number; + startAt?: number; + endAt?: number; + durationMs?: number; + historicalRuntimeMs?: number; + cacheState: RunTimingCacheState; + timingQuality: RunTimingQuality; +} + +export interface RunTimingMetrics { + wallClockDurationMs?: number; + totalTaskCount: number; + cachedTaskCount: number; + averageStartupMs?: number; + startupCoverage: number; + busyRuntimeMs: number; + busyPercent: number; + criticalPathDurationMs?: number; +} + +export interface RunTimingData { + tasks: RunTimingTask[]; + truncated: boolean; + rangeStart?: number; + rangeEnd?: number; + criticalPathExecutionIds: Set; + metrics: RunTimingMetrics; +} diff --git a/src/routes/v2/pages/RunView/components/RunTiming/runTimingService.test.ts b/src/routes/v2/pages/RunView/components/RunTiming/runTimingService.test.ts new file mode 100644 index 0000000000..2adca149c7 --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/runTimingService.test.ts @@ -0,0 +1,158 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import type { GetExecutionInfoResponse } from "@/api/types.gen"; +import { + fetchContainerExecutionState, + fetchExecutionDetails, +} from "@/services/executionService"; + +import { fetchRunTimingData } from "./runTimingService"; + +vi.mock("@/services/executionService", () => ({ + fetchContainerExecutionState: vi.fn(), + fetchExecutionDetails: vi.fn(), +})); + +const leafComponent = { + name: "Leaf component", + implementation: { container: { image: "python:3.12" } }, +}; + +const subgraphComponent = { + name: "Nested pipeline", + implementation: { + graph: { + tasks: { + nested: { componentRef: { spec: leafComponent } }, + }, + }, + }, +}; + +const rootDetails: GetExecutionInfoResponse = { + id: "root-exec", + child_task_execution_ids: { + first: "exec-first", + second: "exec-second", + subgraph: "exec-subgraph", + }, + task_spec: { + componentRef: { + spec: { + name: "Example pipeline", + implementation: { + graph: { + tasks: { + first: { componentRef: { spec: leafComponent } }, + second: { + componentRef: { spec: leafComponent }, + arguments: { + input: { + taskOutput: { taskId: "first", outputName: "result" }, + }, + }, + }, + subgraph: { componentRef: { spec: subgraphComponent } }, + }, + }, + }, + }, + }, + }, +}; + +function leafDetails(id: string): GetExecutionInfoResponse { + return { + id, + child_task_execution_ids: {}, + input_artifacts: + id === "exec-second" ? { input: { id: "artifact-first" } } : {}, + output_artifacts: + id === "exec-first" ? { output: { id: "artifact-first" } } : {}, + task_spec: { componentRef: { spec: leafComponent } }, + status_history: [ + { status: "RUNNING", first_observed_at: "2026-07-14T10:00:00Z" }, + { status: "SUCCEEDED", first_observed_at: "2026-07-14T10:01:00Z" }, + ], + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(fetchExecutionDetails).mockImplementation(async (executionId) => { + if (executionId === "root-exec") return rootDetails; + if (executionId === "exec-subgraph") { + return { + id: executionId, + child_task_execution_ids: { nested: "exec-nested" }, + task_spec: { componentRef: { spec: subgraphComponent } }, + }; + } + return leafDetails(executionId); + }); + vi.mocked(fetchContainerExecutionState).mockResolvedValue({ + status: "SUCCEEDED", + started_at: "2026-07-14T10:00:00Z", + ended_at: "2026-07-14T10:01:00Z", + }); +}); + +describe("fetchRunTimingData", () => { + it("loads nested executions and preserves dependency and navigation context", async () => { + const data = await fetchRunTimingData({ + rootDetails, + backendUrl: "https://example.test", + runCreatedAt: "2026-07-14T09:59:00Z", + now: Date.parse("2026-07-14T10:02:00Z"), + }); + + expect(data.tasks.map((task) => task.executionId)).toEqual([ + "exec-first", + "exec-second", + "exec-subgraph", + "exec-nested", + ]); + expect( + data.tasks.find((task) => task.executionId === "exec-second"), + ).toMatchObject({ dependencyExecutionIds: ["exec-first"] }); + expect( + data.tasks.find((task) => task.executionId === "exec-nested"), + ).toMatchObject({ + depth: 1, + navigationPath: ["Example pipeline", "subgraph"], + parentExecutionId: "exec-subgraph", + }); + expect(fetchExecutionDetails).toHaveBeenCalledTimes(5); + expect(fetchExecutionDetails).toHaveBeenCalledWith( + "root-exec", + "https://example.test", + { signal: undefined }, + ); + expect(fetchContainerExecutionState).toHaveBeenCalledTimes(3); + expect(data.truncated).toBe(false); + }); + + it("caps recursive requests for large runs", async () => { + const child_task_execution_ids = Object.fromEntries( + Array.from({ length: 251 }, (_, index) => [ + `task-${index}`, + `exec-${index}`, + ]), + ); + + vi.mocked(fetchExecutionDetails).mockResolvedValueOnce({ + ...rootDetails, + child_task_execution_ids, + }); + + const data = await fetchRunTimingData({ + rootDetails: { ...rootDetails, child_task_execution_ids }, + backendUrl: "https://example.test", + now: Date.parse("2026-07-14T10:02:00Z"), + }); + + expect(data.tasks).toHaveLength(250); + expect(data.truncated).toBe(true); + expect(fetchExecutionDetails).toHaveBeenCalledTimes(251); + }); +}); diff --git a/src/routes/v2/pages/RunView/components/RunTiming/runTimingService.ts b/src/routes/v2/pages/RunView/components/RunTiming/runTimingService.ts new file mode 100644 index 0000000000..c0064a7d5f --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/runTimingService.ts @@ -0,0 +1,199 @@ +import type { GetExecutionInfoResponse, TaskSpecOutput } from "@/api/types.gen"; +import { + fetchContainerExecutionState, + fetchExecutionDetails, +} from "@/services/executionService"; +import { CONTAINER_STATUSES_PRE_LAUNCH } from "@/utils/executionStatus"; +import { RemoteAuthError } from "@/utils/fetchWithErrorHandling"; + +import { normalizeRunTimingData } from "./runTiming"; +import type { RunTimingData, RunTimingTaskSource } from "./runTiming.types"; + +const MAX_CONCURRENT_REQUESTS = 6; +const MAX_TASK_EXECUTIONS = 250; + +interface ExecutionQueueItem { + executionId: string; + parentExecutionId: string; + taskId: string; + navigationPath: string[]; + depth: number; + taskSpec?: TaskSpecOutput; +} + +interface FetchRunTimingDataOptions { + rootDetails: GetExecutionInfoResponse; + backendUrl: string; + runCreatedAt?: string | null; + signal?: AbortSignal; + now?: number; +} + +function graphTasksForExecution( + details: GetExecutionInfoResponse, +): Record { + const implementation = details.task_spec.componentRef.spec?.implementation; + return implementation && "graph" in implementation + ? implementation.graph.tasks + : {}; +} + +function isSubgraphTask( + details: GetExecutionInfoResponse | undefined, + fallbackTaskSpec: TaskSpecOutput | undefined, +): boolean { + if ( + details && + Object.keys(details.child_task_execution_ids ?? {}).length > 0 + ) { + return true; + } + + const implementation = + details?.task_spec.componentRef.spec?.implementation ?? + fallbackTaskSpec?.componentRef.spec?.implementation; + return ( + implementation !== undefined && + implementation !== null && + "graph" in implementation + ); +} + +function childQueueItems( + details: GetExecutionInfoResponse, + navigationPath: string[], + depth: number, +): ExecutionQueueItem[] { + const graphTasks = graphTasksForExecution(details); + + return Object.entries(details.child_task_execution_ids ?? {}).map( + ([taskId, executionId]) => { + const taskSpec = graphTasks[taskId]; + return { + executionId, + parentExecutionId: details.id, + taskId, + navigationPath, + depth, + taskSpec, + }; + }, + ); +} + +function shouldFetchContainerState( + details: GetExecutionInfoResponse, + isSubgraph: boolean, +): boolean { + if (isSubgraph) return false; + const status = details.status_history?.at(-1)?.status; + return !status || !CONTAINER_STATUSES_PRE_LAUNCH.has(status); +} + +export async function fetchRunTimingData({ + rootDetails, + backendUrl, + runCreatedAt, + signal, + now = Date.now(), +}: FetchRunTimingDataOptions): Promise { + let latestRootDetails = rootDetails; + try { + latestRootDetails = await fetchExecutionDetails( + rootDetails.id, + backendUrl, + { + signal, + }, + ); + } catch (error) { + if (signal?.aborted || error instanceof RemoteAuthError) throw error; + } + + const rootName = + latestRootDetails.task_spec.componentRef.spec?.name ?? "Pipeline Run"; + const rootChildren = childQueueItems(latestRootDetails, [rootName], 0); + const queue = rootChildren.slice(0, MAX_TASK_EXECUTIONS); + let truncated = rootChildren.length > MAX_TASK_EXECUTIONS; + const sourceByExecutionId = new Map(); + let cursor = 0; + + const loadExecution = async (item: ExecutionQueueItem) => { + let details: GetExecutionInfoResponse | undefined; + try { + details = await fetchExecutionDetails(item.executionId, backendUrl, { + signal, + }); + } catch (error) { + if (signal?.aborted || error instanceof RemoteAuthError) throw error; + sourceByExecutionId.set(item.executionId, { + ...item, + inputArtifactIds: [], + outputArtifactIds: [], + isSubgraph: isSubgraphTask(undefined, item.taskSpec), + loadFailed: true, + }); + return; + } + + const isSubgraph = isSubgraphTask(details, item.taskSpec); + let containerState; + if (shouldFetchContainerState(details, isSubgraph)) { + try { + containerState = await fetchContainerExecutionState( + item.executionId, + backendUrl, + { signal }, + ); + } catch (error) { + if (signal?.aborted || error instanceof RemoteAuthError) throw error; + } + } + + sourceByExecutionId.set(item.executionId, { + ...item, + details, + containerState, + inputArtifactIds: Object.values(details.input_artifacts ?? {}).map( + (artifact) => artifact.id, + ), + outputArtifactIds: Object.values(details.output_artifacts ?? {}).map( + (artifact) => artifact.id, + ), + isSubgraph, + }); + + if (details.child_task_execution_ids) { + const children = childQueueItems( + details, + [...item.navigationPath, item.taskId], + item.depth + 1, + ); + const remainingCapacity = MAX_TASK_EXECUTIONS - queue.length; + if (children.length > remainingCapacity) truncated = true; + queue.push(...children.slice(0, Math.max(0, remainingCapacity))); + } + }; + + const worker = async () => { + while (cursor < queue.length) { + const item = queue[cursor]; + cursor += 1; + await loadExecution(item); + } + }; + + await Promise.all( + Array.from( + { length: Math.min(MAX_CONCURRENT_REQUESTS, queue.length) }, + worker, + ), + ); + + const sources = queue.flatMap((item) => { + const source = sourceByExecutionId.get(item.executionId); + return source ? [source] : []; + }); + + return normalizeRunTimingData(sources, runCreatedAt, now, truncated); +} diff --git a/src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.test.tsx b/src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.test.tsx new file mode 100644 index 0000000000..d584f2ab53 --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.test.tsx @@ -0,0 +1,82 @@ +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { cleanup, renderHook, waitFor } from "@testing-library/react"; +import type { PropsWithChildren } from "react"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { GetExecutionInfoResponse } from "@/api/types.gen"; + +import type { RunTimingData } from "./runTiming.types"; +import { fetchRunTimingData } from "./runTimingService"; +import { useRunTimingData } from "./useRunTimingData"; + +vi.mock("@/providers/BackendProvider", () => ({ + useBackend: () => ({ backendUrl: "https://example.test" }), +})); + +vi.mock("./runTimingService", () => ({ + fetchRunTimingData: vi.fn(), +})); + +const rootDetails: GetExecutionInfoResponse = { + id: "root-exec", + child_task_execution_ids: {}, + task_spec: { componentRef: { spec: { name: "Example pipeline" } } }, +}; + +const timingData: RunTimingData = { + tasks: [], + truncated: false, + criticalPathExecutionIds: new Set(), + metrics: { + totalTaskCount: 0, + cachedTaskCount: 0, + startupCoverage: 0, + busyRuntimeMs: 0, + busyPercent: 0, + }, +}; + +const queryClients: QueryClient[] = []; + +function createWrapper(queryClient: QueryClient) { + return function Wrapper({ children }: PropsWithChildren) { + return ( + {children} + ); + }; +} + +afterEach(() => { + cleanup(); + vi.clearAllMocks(); + for (const queryClient of queryClients) queryClient.clear(); + queryClients.length = 0; +}); + +describe("useRunTimingData", () => { + it("refreshes immediately on mount when cached timing data is still fresh", async () => { + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + queryClients.push(queryClient); + queryClient.setQueryData( + ["run-timing", "https://example.test", "root-exec", undefined], + timingData, + ); + vi.mocked(fetchRunTimingData).mockResolvedValue(timingData); + + renderHook( + () => + useRunTimingData({ + rootDetails, + runCreatedAt: undefined, + runComplete: true, + }), + { wrapper: createWrapper(queryClient) }, + ); + + await waitFor(() => { + expect(fetchRunTimingData).toHaveBeenCalledTimes(1); + }); + }); +}); diff --git a/src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.ts b/src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.ts new file mode 100644 index 0000000000..2ef8cbc69d --- /dev/null +++ b/src/routes/v2/pages/RunView/components/RunTiming/useRunTimingData.ts @@ -0,0 +1,96 @@ +import { Query, useQuery } from "@tanstack/react-query"; +import { useRef } from "react"; + +import type { GetExecutionInfoResponse } from "@/api/types.gen"; +import { useBackend } from "@/providers/BackendProvider"; +import { RemoteAuthError } from "@/utils/fetchWithErrorHandling"; + +import { fetchRunTimingData } from "./runTimingService"; + +const REFRESH_INTERVAL_MS = 5_000; +const COMPLETED_RUN_STABLE_REFRESHES = 2; + +interface CompletedRunRefreshState { + dataUpdatedAt: number; + taskSignature: string; + stableRefreshes: number; +} + +interface UseRunTimingDataOptions { + rootDetails: GetExecutionInfoResponse | undefined; + runCreatedAt: string | null | undefined; + runComplete: boolean; +} + +export function useRunTimingData({ + rootDetails, + runCreatedAt, + runComplete, +}: UseRunTimingDataOptions) { + const { backendUrl } = useBackend(); + const completedRunRefreshState = useRef(undefined); + + return useQuery({ + queryKey: ["run-timing", backendUrl, rootDetails?.id, runCreatedAt], + queryFn: ({ signal }) => { + if (!rootDetails) + throw new Error("Run execution details are unavailable"); + return fetchRunTimingData({ + rootDetails, + backendUrl, + runCreatedAt, + signal, + }); + }, + enabled: rootDetails !== undefined, + staleTime: REFRESH_INTERVAL_MS, + refetchInterval: (query) => { + if (!runComplete) return REFRESH_INTERVAL_MS; + if (!(query instanceof Query) || !query.state.data) { + return REFRESH_INTERVAL_MS; + } + + const previous = completedRunRefreshState.current; + if (previous?.dataUpdatedAt === query.state.dataUpdatedAt) { + return previous.stableRefreshes >= COMPLETED_RUN_STABLE_REFRESHES + ? false + : REFRESH_INTERVAL_MS; + } + + const taskSignature = JSON.stringify( + query.state.data.tasks + .map((task) => ({ + executionId: task.executionId, + status: task.status, + readyAt: task.readyAt, + startAt: task.startAt, + endAt: task.endAt, + durationMs: task.durationMs, + cacheState: task.cacheState, + timingQuality: task.timingQuality, + phases: task.phases, + })) + .sort((left, right) => + left.executionId.localeCompare(right.executionId), + ), + ); + completedRunRefreshState.current = { + dataUpdatedAt: query.state.dataUpdatedAt, + taskSignature, + stableRefreshes: + previous?.taskSignature === taskSignature + ? previous.stableRefreshes + 1 + : 0, + }; + + return completedRunRefreshState.current.stableRefreshes >= + COMPLETED_RUN_STABLE_REFRESHES + ? false + : REFRESH_INTERVAL_MS; + }, + refetchOnMount: "always", + refetchOnWindowFocus: false, + retry: (failureCount, error) => + !(error instanceof RemoteAuthError) && failureCount < 3, + }); +} diff --git a/src/services/executionService.ts b/src/services/executionService.ts index 81a80c8cb5..d56c6cfa2b 100644 --- a/src/services/executionService.ts +++ b/src/services/executionService.ts @@ -40,9 +40,10 @@ export const fetchExecutionState = async ( export const fetchExecutionDetails = async ( executionId: string, backendUrl: string, + options?: RequestInit, ): Promise => { const url = `${backendUrl}/api/executions/${executionId}/details`; - return fetchWithErrorHandling(url); + return fetchWithErrorHandling(url, options); }; export const useFetchExecutionDetails = ( @@ -85,9 +86,10 @@ export const useFetchPipelineRunMetadata = (runId: string | undefined) => { export const fetchContainerExecutionState = async ( executionId: string, backendUrl: string, + options?: RequestInit, ): Promise => { const url = `${backendUrl}/api/executions/${executionId}/container_state`; - return fetchWithErrorHandling(url); + return fetchWithErrorHandling(url, options); }; export const fetchContainerLog = async (