diff --git a/Cargo.lock b/Cargo.lock index f4707dab6..5712ee090 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -541,9 +541,9 @@ dependencies = [ [[package]] name = "async-compression" -version = "0.4.42" +version = "0.4.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac" +checksum = "3976abdc8fe7d1133d43d304afd42abdf5bc3e1319d263d223bde07b5efc4be8" dependencies = [ "compression-codecs", "compression-core", @@ -1548,9 +1548,9 @@ dependencies = [ [[package]] name = "clang-sys" -version = "1.8.1" +version = "1.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a" dependencies = [ "glob", "libc", @@ -2811,9 +2811,9 @@ dependencies = [ [[package]] name = "enum-ordinalize" -version = "4.4.1" +version = "4.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07f808d588c10e464ea6f7d3eaed500049eff30aaac103460f61828c2d65b3eb" +checksum = "89dd01549b09589510cf0647475075d12071456586d70f5c75c98ae2a5537677" dependencies = [ "enum-ordinalize-derive", ] @@ -3712,9 +3712,9 @@ dependencies = [ [[package]] name = "http" -version = "1.4.2" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" dependencies = [ "bytes", "itoa", @@ -6930,9 +6930,9 @@ dependencies = [ [[package]] name = "redis" -version = "1.4.1" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b0b9503711b03773e43b31668c7b5bd279ee7cd9b7d18cff7c23a42cc1d08e5a" +checksum = "3257df217f7eab0044627a268c9cc6cdb60c0c421c88f83ac41c4e31520b6b84" dependencies = [ "arc-swap", "arcstr", @@ -7570,9 +7570,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.42" +version = "0.23.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ "aws-lc-rs", "log", @@ -9596,13 +9596,13 @@ dependencies = [ [[package]] name = "tokio-macros" -version = "2.7.1" +version = "2.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 13ed2a553..7bf4dd644 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -111,7 +111,7 @@ dashmap = "6.2.1" once_cell = "1.21.4" serial_test = "4.0.1" sysinfo = "0.39.6" -http = "1.4.2" +http = "1.5.0" url = "2.5.8" jemallocator = "0.5.4" mimalloc = "0.1.52" @@ -123,9 +123,9 @@ bs58 = "0.5.1" indexmap = "2.14" envsubst = "0.2.1" directories = "6.0.0" -redis = "1.4.1" +redis = "1.5.0" redis-test = "1.0.4" -rustls = "0.23.42" +rustls = "0.23.43" object_store = "0.14.1" parse-display = "0.11.0" toml = "1.1.3" diff --git a/ceres/src/application/api_service/mono/cl_list.rs b/ceres/src/application/api_service/mono/cl_list.rs index 500d54c6b..58c146868 100644 --- a/ceres/src/application/api_service/mono/cl_list.rs +++ b/ceres/src/application/api_service/mono/cl_list.rs @@ -24,6 +24,21 @@ impl ClApplicationService { .bot_names_among(&names) .await?; ItemRes::apply_bot_flags(&mut items, &bot_names); + + let links: Vec = items.iter().map(|i| i.link.clone()).collect(); + match self + .storage() + .cl_service + .cl_store() + .latest_build_status_by_cl_links(&links) + .await + { + Ok(statuses) => ItemRes::apply_build_statuses(&mut items, &statuses), + Err(e) => { + tracing::warn!("Failed to load CL build statuses for list: {e}"); + } + } + Ok(CommonPage { items, total }) } diff --git a/ceres/src/model/issue.rs b/ceres/src/model/issue.rs index 388246873..55b3ade2c 100644 --- a/ceres/src/model/issue.rs +++ b/ceres/src/model/issue.rs @@ -28,6 +28,10 @@ pub struct ItemRes { pub labels: Vec, pub assignees: Vec, pub comment_num: usize, + /// Aggregated Orion build status for this CL (latest task), if any. + /// Issue list leaves this unset/`null`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub build_status: Option, } impl From for ItemRes { @@ -47,6 +51,7 @@ impl From for ItemRes { labels: value.labels.into_iter().map(|m| m.into()).collect(), assignees: value.assignees, comment_num: value.comment_num, + build_status: None, }, ItemKind::Cl(model) => Self { id: model.id, @@ -62,6 +67,7 @@ impl From for ItemRes { labels: value.labels.into_iter().map(|m| m.into()).collect(), assignees: value.assignees, comment_num: value.comment_num, + build_status: None, }, } } @@ -73,6 +79,15 @@ impl ItemRes { item.author_is_bot = bot_names.contains(&item.author) || item.author == "system"; } } + + pub fn apply_build_statuses( + items: &mut [Self], + statuses: &std::collections::HashMap, + ) { + for item in items { + item.build_status = statuses.get(&item.link).cloned(); + } + } } #[derive(Serialize, Deserialize, ToSchema)] diff --git a/jupiter/src/storage/cl_storage.rs b/jupiter/src/storage/cl_storage.rs index 64e682ab3..8769c8410 100644 --- a/jupiter/src/storage/cl_storage.rs +++ b/jupiter/src/storage/cl_storage.rs @@ -447,4 +447,74 @@ impl ClStorage { self.batch_save_model(save_models).await?; Ok(()) } + + /// For each CL link, resolve the latest Orion task and aggregate its + /// `build_targets.latest_state` with Checks-compatible worst-wins priority: + /// Failed > Interrupted > Building > Pending/Uninitialized > Completed. + pub async fn latest_build_status_by_cl_links( + &self, + links: &[String], + ) -> Result, MegaError> { + if links.is_empty() { + return Ok(HashMap::new()); + } + + let tasks = callisto::orion_tasks::Entity::find() + .filter(callisto::orion_tasks::Column::Cl.is_in(links.to_vec())) + .order_by_desc(callisto::orion_tasks::Column::CreatedAt) + .all(self.get_connection()) + .await?; + + // First row per CL wins (already ordered by created_at DESC). + let mut latest_task_by_cl: HashMap = HashMap::new(); + for task in tasks { + latest_task_by_cl.entry(task.cl).or_insert(task.id); + } + + if latest_task_by_cl.is_empty() { + return Ok(HashMap::new()); + } + + let task_ids: Vec = latest_task_by_cl.values().copied().collect(); + let targets = callisto::build_targets::Entity::find() + .filter(callisto::build_targets::Column::TaskId.is_in(task_ids)) + .all(self.get_connection()) + .await?; + + let mut states_by_task: HashMap> = HashMap::new(); + for target in targets { + states_by_task + .entry(target.task_id) + .or_default() + .push(target.latest_state); + } + + let mut result = HashMap::new(); + for (cl, task_id) in latest_task_by_cl { + if let Some(states) = states_by_task.get(&task_id) + && let Some(status) = aggregate_target_states(states) + { + result.insert(cl, status); + } + } + Ok(result) + } +} + +/// Worst-wins aggregation matching Checks UI `getTaskStatus` priority. +fn aggregate_target_states(states: &[String]) -> Option { + if states.is_empty() { + return None; + } + let priority = |s: &str| -> u8 { + match s { + "Failed" => 0, + "Interrupted" => 1, + "Building" => 2, + "Pending" | "Uninitialized" => 3, + "Completed" => 4, + _ => 3, + } + }; + states.iter().min_by_key(|s| priority(s.as_str())).cloned() } diff --git a/mono/Dockerfile b/mono/Dockerfile index 9dcaea0d5..1c74ecc00 100644 --- a/mono/Dockerfile +++ b/mono/Dockerfile @@ -1,6 +1,6 @@ # ────── Stage 0: Chef ────── # Downloads and installs cargo-chef (prebuilt binary, no Rust compilation needed). -FROM rust:1.96.0 AS chef +FROM rust:1.97.1 AS chef ARG CARGO_CHEF_VERSION=0.1.77 diff --git a/moon/apps/web/components/ClView/ClList.tsx b/moon/apps/web/components/ClView/ClList.tsx index 474f01765..8f87a15f4 100644 --- a/moon/apps/web/components/ClView/ClList.tsx +++ b/moon/apps/web/components/ClView/ClList.tsx @@ -1,13 +1,16 @@ import React, { forwardRef, memo, ReactNode } from 'react' import { PostApiIssueListData } from '@gitmono/types' -import { Button, ChatBubbleIcon, Command, ConditionalWrap, LoadingSpinner, useCommand } from '@gitmono/ui' +import { Button, ChatBubbleIcon, Command, ConditionalWrap, LoadingSpinner, Tooltip, useCommand } from '@gitmono/ui' import { MemberHoverAvatarList } from '@/components/Issues/MemberHoverAvatarList' import { SubjectCommand } from '@/components/Subject/SubjectCommand' import { BreadcrumbTitlebarContainer } from '@/components/Titlebar/BreadcrumbTitlebar' import { getFontColor } from '@/utils/getFontColor' +import { identifyStatus } from './components/Checks/cpns/identifyStatus' +import { Status } from './components/Checks/cpns/store' + export function List({ lists, header, @@ -196,9 +199,21 @@ export const ItemLabels = ({ item }: { item: ItemsType[number] }) => { } export const ItemRightIcons = ({ item }: { item: ItemsType[number] }) => { + const buildStatus = item.build_status + const statusEnum = + buildStatus && Object.values(Status).includes(buildStatus as Status) ? (buildStatus as Status) : null + return ( //
+ {statusEnum ? ( + + + {identifyStatus(statusEnum)} + + + ) : null} +
{ - switch (status) { - case Status.Completed: - return - case Status.Failed: - return - case Status.Interrupted: - return - case Status.Building: - return - case Status.Pending: - return - case Status.Uninitialized: - return - case Status.NotFound: - return - - default: - return - } -} +export { identifyStatus } from './identifyStatus' diff --git a/moon/apps/web/components/ClView/components/Checks/cpns/identifyStatus.tsx b/moon/apps/web/components/ClView/components/Checks/cpns/identifyStatus.tsx new file mode 100644 index 000000000..f64fcf95a --- /dev/null +++ b/moon/apps/web/components/ClView/components/Checks/cpns/identifyStatus.tsx @@ -0,0 +1,27 @@ +import { CheckIcon, XIcon } from '@primer/octicons-react' + +import { LoadingSpinner } from '@gitmono/ui/Spinner' + +import { Status } from './store' + +export const identifyStatus = (status: Status[keyof Status]) => { + switch (status) { + case Status.Completed: + return + case Status.Failed: + return + case Status.Interrupted: + return + case Status.Building: + return + case Status.Pending: + return + case Status.Uninitialized: + return + case Status.NotFound: + return + + default: + return + } +} diff --git a/moon/apps/web/components/CodeView/TreeView/CustomTreeItem.tsx b/moon/apps/web/components/CodeView/TreeView/CustomTreeItem.tsx index 68065da2a..29d4ca337 100644 --- a/moon/apps/web/components/CodeView/TreeView/CustomTreeItem.tsx +++ b/moon/apps/web/components/CodeView/TreeView/CustomTreeItem.tsx @@ -20,14 +20,20 @@ import { CustomLabel } from './CustomLabel' import { getIconFromFileType, MuiTreeNode } from './TreeUtils' interface CustomTreeItemProps - extends Omit, Omit, 'onFocus'> { - onLabelClick?: (path: string, isDirectory: boolean) => void - loadingDirectories?: Set -} + extends Omit, Omit, 'onFocus'> {} + +export const LoadingDirectoriesContext = React.createContext>(new Set()) + +// Must live outside the component — defining styled() during render creates a new +// component type every time and remounts the subtree (visible tree "jump"). +const StyledGroupTransition = styled(TreeItemGroupTransition)(({ theme }) => ({ + marginLeft: 15, + borderLeft: `1px dashed ${alpha(theme.palette.text.primary, 0.4)}` +})) // Custom tree structure node component, used to render elements such as icons and labels for each node export const CustomTreeItem = React.forwardRef(function CustomTreeItem( - { loadingDirectories, ...props }: CustomTreeItemProps, + props: CustomTreeItemProps, ref: React.Ref ) { const { id, itemId, label, disabled, children, ...other } = props @@ -43,6 +49,7 @@ export const CustomTreeItem = React.forwardRef(function CustomTreeItem( } = useTreeItem({ id, itemId, children, label, disabled, rootRef: ref }) const item = useTreeItemModel(itemId)! + const loadingDirectories = React.useContext(LoadingDirectoriesContext) // If it is a placeholder node, no content is rendered if (item.isPlaceholder) { @@ -58,12 +65,7 @@ export const CustomTreeItem = React.forwardRef(function CustomTreeItem( } // Check if the current node is loading - const isNodeLoading = loadingDirectories?.has(item.path) - - const StyledGroupTransition = styled(TreeItemGroupTransition)(({ theme }) => ({ - marginLeft: 15, - borderLeft: `1px dashed ${alpha(theme.palette.text.primary, 0.4)}` - })) + const isNodeLoading = loadingDirectories.has(item.path) return ( diff --git a/moon/apps/web/components/CodeView/TreeView/RepoTree.tsx b/moon/apps/web/components/CodeView/TreeView/RepoTree.tsx index 2c284748a..7f09c4be4 100644 --- a/moon/apps/web/components/CodeView/TreeView/RepoTree.tsx +++ b/moon/apps/web/components/CodeView/TreeView/RepoTree.tsx @@ -12,7 +12,7 @@ import { useGetTree } from '@/hooks/useGetTree' import { legacyApiClient } from '@/utils/queryClient' import { expandedNodesAtom, treeAllDataAtom } from './codeTreeAtom' -import { CustomTreeItem } from './CustomTreeItem' +import { CustomTreeItem, LoadingDirectoriesContext } from './CustomTreeItem' import { convertToTreeData, findNode, @@ -186,17 +186,22 @@ const RepoTree = ({ onCommitInfoChange }: { onCommitInfoChange?: Function }) => [router, scope, version, expandedNodes, setExpandedNodes] ) - const handleFocusItem = (_e: React.SyntheticEvent | null, itemId: string) => { - const item = apiRef.current!.getItem(itemId) - - if (item.content_type) { - handleLabelClick(item.path, item.content_type === 'directory') - apiRef.current?.setItemSelection({ - itemId, - keepExistingSelection: false - }) - } - } + // Navigate on click, not focus — focus also fires after remount/selection and + // would re-trigger router.push + scrollIntoView, making the tree jump. + const handleItemClick = useCallback( + (_e: React.SyntheticEvent | null, itemId: string) => { + const item = apiRef.current!.getItem(itemId) + + if (item.content_type) { + handleLabelClick(item.path, item.content_type === 'directory') + apiRef.current?.setItemSelection({ + itemId, + keepExistingSelection: false + }) + } + }, + [handleLabelClick, apiRef] + ) useEffect(() => { if (basePath) { @@ -232,7 +237,7 @@ const RepoTree = ({ onCommitInfoChange }: { onCommitInfoChange?: Function }) => const showInitialSkeleton = (isTreeLoading || loadingDirectories.has(basePath)) && treeAllData?.length === 0 return ( - <> + {showInitialSkeleton ? ( @@ -241,17 +246,16 @@ const RepoTree = ({ onCommitInfoChange }: { onCommitInfoChange?: Function }) => - }} + sx={{ height: 'fit-content', flexGrow: 1, width: '100%', overflow: 'auto', overflowAnchor: 'none' }} + // Stable slot reference — an inline render fn remounts every item on each parent render. + slots={{ item: CustomTreeItem }} /> )} - + ) } diff --git a/moon/apps/web/components/DiffView/TreeView/CustomTreeItem.tsx b/moon/apps/web/components/DiffView/TreeView/CustomTreeItem.tsx index 37beb4642..d2224a7f7 100644 --- a/moon/apps/web/components/DiffView/TreeView/CustomTreeItem.tsx +++ b/moon/apps/web/components/DiffView/TreeView/CustomTreeItem.tsx @@ -23,6 +23,13 @@ interface CustomTreeItemProps onLabelClick?: (path: string, isDirectory: boolean) => void } +// Must live outside the component — defining styled() during render creates a new +// component type every time and remounts the subtree (visible tree "jump"). +const StyledGroupTransition = styled(TreeItemGroupTransition)(({ theme }) => ({ + marginLeft: 15, + borderLeft: `1px dashed ${alpha(theme.palette.text.primary, 0.4)}` +})) + // Custom tree structure node component, used to render elements such as icons and labels for each node export const CustomTreeItem = React.forwardRef(function CustomTreeItem( props: CustomTreeItemProps, @@ -55,11 +62,6 @@ export const CustomTreeItem = React.forwardRef(function CustomTreeItem( icon = getIconFromFileType(item.content_type, false) } - const StyledGroupTransition = styled(TreeItemGroupTransition)(({ theme }) => ({ - marginLeft: 15, - borderLeft: `1px dashed ${alpha(theme.palette.text.primary, 0.4)}` - })) - return ( diff --git a/moon/apps/web/components/DiffView/TreeView/FileTree.tsx b/moon/apps/web/components/DiffView/TreeView/FileTree.tsx index cc9ea2e61..2691390e4 100644 --- a/moon/apps/web/components/DiffView/TreeView/FileTree.tsx +++ b/moon/apps/web/components/DiffView/TreeView/FileTree.tsx @@ -96,9 +96,7 @@ const FileTree = ({ treeData, treeDataLoading, onFileClick }: FileTreeProps) => onExpandedItemsChange={handleNodeToggle} expansionTrigger='iconContainer' sx={{ flexGrow: 1, width: '100%', overflow: 'auto' }} - slots={{ - item: (itemProps) => - }} + slots={{ item: CustomTreeItem }} /> )} diff --git a/moon/apps/web/pages/[org]/code/blob/[version]/[...path]/index.tsx b/moon/apps/web/pages/[org]/code/blob/[version]/[...path]/index.tsx index 6c6740f81..d9b1fcca6 100644 --- a/moon/apps/web/pages/[org]/code/blob/[version]/[...path]/index.tsx +++ b/moon/apps/web/pages/[org]/code/blob/[version]/[...path]/index.tsx @@ -91,6 +91,7 @@ function BlobPage() { flexShrink: 0, borderRadius: 8, overflow: 'auto', + overflowAnchor: 'none', paddingRight: '8px' }} className='bg-primary' diff --git a/moon/apps/web/pages/[org]/code/tree/[version]/[...path]/index.tsx b/moon/apps/web/pages/[org]/code/tree/[version]/[...path]/index.tsx index 8e2313d59..6a1d09bd8 100644 --- a/moon/apps/web/pages/[org]/code/tree/[version]/[...path]/index.tsx +++ b/moon/apps/web/pages/[org]/code/tree/[version]/[...path]/index.tsx @@ -109,6 +109,7 @@ function TreeDetailPage() { flexShrink: 0, height: 'calc(100vh - 96px)', overflow: 'auto', + overflowAnchor: 'none' as const, paddingRight: '8px' } diff --git a/moon/packages/types/generated.ts b/moon/packages/types/generated.ts index 5162fab8f..7aafbdba3 100644 --- a/moon/packages/types/generated.ts +++ b/moon/packages/types/generated.ts @@ -3963,6 +3963,11 @@ export type CommonResultCommonPageItemRes = { author: string /** True when `author` matches a registered bot. */ author_is_bot?: boolean + /** + * Aggregated Orion build status for this CL (latest task), if any. + * Issue list leaves this unset/`null`. + */ + build_status?: string | null /** @format int64 */ closed_at?: number | null /** @min 0 */ @@ -5309,6 +5314,11 @@ export type ItemRes = { author: string /** True when `author` matches a registered bot. */ author_is_bot?: boolean + /** + * Aggregated Orion build status for this CL (latest task), if any. + * Issue list leaves this unset/`null`. + */ + build_status?: string | null /** @format int64 */ closed_at?: number | null /** @min 0 */ diff --git a/orion-server/Dockerfile b/orion-server/Dockerfile index da6289dec..8fed8c200 100644 --- a/orion-server/Dockerfile +++ b/orion-server/Dockerfile @@ -3,7 +3,7 @@ # NOTE: This stage is intentionally identical to `mono/Dockerfile`'s chef stage # (same base image, WORKDIR, cargo-chef version) so BuildKit can share the # resulting layer between the two demo images. -FROM rust:1.96.0 AS chef +FROM rust:1.97.1 AS chef ARG CARGO_CHEF_VERSION=0.1.77 diff --git a/orion-server/src/scheduler.rs b/orion-server/src/scheduler.rs index 38916cbca..18a112f79 100644 --- a/orion-server/src/scheduler.rs +++ b/orion-server/src/scheduler.rs @@ -45,7 +45,7 @@ impl Default for TaskQueueConfig { fn default() -> Self { Self { max_queue_size: 1000, - max_wait_time: Duration::from_secs(30), // Fail builds with no worker after 30s + max_wait_time: Duration::from_secs(3600), // Fail builds with no worker after 60 minutes cleanup_interval: Duration::from_secs(10), // Scan for expired builds every 10 seconds } } @@ -78,6 +78,14 @@ impl TaskQueue { Ok(()) } + /// Put a previously dequeued task back at the front of the queue. + /// + /// Used when dispatch races and no idle worker remains; does not enforce + /// `max_queue_size` so a capacity fill cannot drop an already-accepted build. + pub fn requeue_front_v2(&mut self, task: PendingBuildEventV2) { + self.queue_v2.push_front(task); + } + /// Clean up expired queued builds (v2) pub fn cleanup_expired_v2(&mut self) -> Vec { let now = Instant::now(); @@ -349,25 +357,84 @@ impl TaskScheduler { .await; } - /// Immediately finalize a build that cannot be scheduled because no worker is - /// available, marking it `Interrupted` with an explanatory log. Used instead - /// of queuing the build when there are no idle workers, so the UI shows a - /// clear terminal state (retryable) rather than a build stuck in the queue. - pub async fn fail_unschedulable_build(&self, build_event_id: Uuid, task_id: Uuid, repo: &str) { - self.finalize_interrupted_build( - build_event_id, - task_id, - repo, - "No build worker is currently available; build marked as Interrupted.\n", - ) - .await; + /// Finalize an in-flight build when its worker is gone (WS disconnect, heartbeat + /// timeout, etc.). Removes the build from `active_builds` and marks it Interrupted. + /// + /// No-op when the worker was not `Busy`. Idempotent with other finalize paths via + /// [`BuildEventsRepo::mark_interrupted`]. + pub async fn handle_worker_gone(&self, worker: WorkerInfo, reason: &str) { + let Some((build_id, removed)) = take_busy_active_build(&self.active_builds, &worker.status) + else { + return; + }; + + tracing::warn!( + "Worker gone ({reason}) while busy with build {build_id}; marking as Interrupted" + ); + + let (task_id, repo) = if let Some(info) = removed { + (info.event_payload.task_id, info.event_payload.repo) + } else { + let build_uuid = match build_id.parse::() { + Ok(uuid) => uuid, + Err(_) => { + tracing::warn!("Invalid build id {build_id} when handling gone worker"); + return; + } + }; + match BuildEventsRepo::find_by_id(&self.conn, build_uuid).await { + Ok(Some(event)) => { + let repo = match OrionTasksRepo::find_by_id(&self.conn, event.task_id).await { + Ok(Some(task)) => task.repo_name, + Ok(None) => { + tracing::warn!( + "Task {} not found for build {} when handling gone worker", + event.task_id, + build_id + ); + return; + } + Err(e) => { + tracing::error!( + "Failed to load task for build {build_id} when handling gone worker: {e}" + ); + return; + } + }; + (event.task_id, repo) + } + Ok(None) => { + tracing::warn!("Build event {build_id} not found when handling gone worker"); + return; + } + Err(e) => { + tracing::error!( + "Failed to fetch build event {build_id} when handling gone worker: {e}" + ); + return; + } + } + }; + + let build_uuid = match build_id.parse::() { + Ok(uuid) => uuid, + Err(_) => { + tracing::warn!("Invalid build id {build_id} when finalizing gone-worker build"); + return; + } + }; + + let message = + format!("Worker {reason} while build was in progress; marked as Interrupted.\n"); + self.finalize_interrupted_build(build_uuid, task_id, &repo, &message) + .await; } /// Core, idempotent build finalization: atomically claim the interrupt /// transition, then (only if we won the claim) persist an explanatory log /// line, mark the local log complete, upload it, and move the task's targets /// to `Interrupted`. - async fn finalize_interrupted_build( + pub(crate) async fn finalize_interrupted_build( &self, build_event_id: Uuid, task_id: Uuid, @@ -568,14 +635,29 @@ impl TaskScheduler { /// dead-worker check only covers currently-registered workers. pub async fn reconcile_orphaned_builds_on_startup(&self) { let now = chrono::Utc::now().with_timezone(&FixedOffset::east_opt(0).unwrap()); + self.reconcile_orphaned_builds( + now, + "Build server restarted while this build was in progress; marked as Interrupted.\n", + "Startup reconciliation", + ) + .await; + } - let orphaned = match BuildEventsRepo::list_unfinished_before(&self.conn, now).await { + /// Finalize unfinished builds that this process is not tracking in + /// `active_builds` and that started before `cutoff`. + /// + /// Used at startup (immediate) and periodically (with a grace cutoff) so + /// Building orphans left by WS disconnect / crash are not stuck forever. + pub async fn reconcile_orphaned_builds( + &self, + cutoff: chrono::DateTime, + message: &str, + log_label: &str, + ) { + let orphaned = match BuildEventsRepo::list_unfinished_before(&self.conn, cutoff).await { Ok(builds) => builds, Err(e) => { - tracing::error!( - "Startup reconciliation failed to list unfinished builds: {}", - e - ); + tracing::error!("{log_label} failed to list unfinished builds: {e}"); return; } }; @@ -585,7 +667,7 @@ impl TaskScheduler { } tracing::warn!( - "Startup reconciliation: finalizing {} orphaned unfinished build(s)", + "{log_label}: scanning {} unfinished build(s) older than cutoff", orphaned.len() ); @@ -599,7 +681,7 @@ impl TaskScheduler { Ok(Some(task)) => task.repo_name, Ok(None) => { tracing::warn!( - "Startup reconciliation skipping build {} because task {} no longer exists", + "{log_label} skipping build {} because task {} no longer exists", build.id, build.task_id ); @@ -607,7 +689,7 @@ impl TaskScheduler { } Err(e) => { tracing::error!( - "Startup reconciliation failed to load task {} for build {}: {}", + "{log_label} failed to load task {} for build {}: {}", build.task_id, build.id, e @@ -617,17 +699,12 @@ impl TaskScheduler { }; tracing::warn!( - "Startup reconciliation: marking orphaned build {}/{} as interrupted", + "{log_label}: marking orphaned build {}/{} as interrupted", build.task_id, build.id ); - self.finalize_interrupted_build( - build.id, - build.task_id, - &repo, - "Build server restarted while this build was in progress; marked as Interrupted.\n", - ) - .await; + self.finalize_interrupted_build(build.id, build.task_id, &repo, message) + .await; } } @@ -723,15 +800,26 @@ impl TaskScheduler { } } - // Dispatch tasks concurrently + // Dispatch tasks concurrently; on failure put the build back at the front + // so a race for idle workers cannot silently drop queued work. if !tasks_to_dispatch.is_empty() { let dispatch_futures: Vec<_> = tasks_to_dispatch .into_iter() .map(|task| { let scheduler = self.clone(); tokio::spawn(async move { - if let Err(e) = scheduler.dispatch_task_v2(task).await { - tracing::error!("Failed to dispatch queued task: {}", e); + if let Err(e) = scheduler.dispatch_task_v2(task.clone()).await { + tracing::warn!( + "Failed to dispatch queued task {}/{}; re-queuing: {}", + task.event_payload.task_id, + task.event_payload.build_event_id, + e + ); + { + let mut queue = scheduler.pending_tasks.lock().await; + queue.requeue_front_v2(task); + } + scheduler.notify_task_available(); } }) }) @@ -785,6 +873,21 @@ impl TaskScheduler { started_at: start_at, }; + // Create WebSocket message + let msg = WSMessage::TaskBuild { + build_id: pending_build_event.event_payload.build_event_id.to_string(), + repo: pending_build_event.event_payload.repo.clone(), + cl_link: pending_build_event.event_payload.cl_link.to_string(), + changes: pending_build_event.changes.clone(), + }; + + // Claim + send *before* DB state updates so a failed claim can safely + // re-queue without leaving the target stuck in `Building`. + // Register the build before sending TaskBuild so early TaskBuildOutput + // lines from the worker are not dropped (ws_handler only publishes logs + // when active_builds contains the build_id). + self.claim_worker_for_build(&chosen_id, build_info, msg)?; + // Ensure a per-build target history row exists for this build/target. let _ = callisto::target_state_histories::ActiveModel { id: Set(Uuid::now_v7()), @@ -805,19 +908,6 @@ impl TaskScheduler { .exec(&self.conn) .await; - // Create WebSocket message - let msg = WSMessage::TaskBuild { - build_id: pending_build_event.event_payload.build_event_id.to_string(), - repo: pending_build_event.event_payload.repo, - cl_link: pending_build_event.event_payload.cl_link.to_string(), - changes: pending_build_event.changes.clone(), - }; - - // Register the build *before* sending TaskBuild so early TaskBuildOutput lines - // from the worker are not dropped (ws_handler only publishes logs when - // active_builds contains the build_id). Guard is dropped inside the helper - // before any further awaits in the caller. - self.claim_worker_for_build(&chosen_id, build_info, msg)?; tracing::info!( "Queued task {}/{} dispatched to worker {}", pending_build_event.event_payload.task_id, @@ -886,6 +976,24 @@ impl TaskScheduler { } cleanup_scheduler.cleanup_stale_queued_builds().await; + + // Building orphans not tracked in active_builds (e.g. WS disconnect + // before this fix, or a missed finalize). Grace = max_wait_time. + let max_wait_time = { + let queue = cleanup_scheduler.pending_tasks.lock().await; + queue.config.max_wait_time + }; + let cutoff = (chrono::Utc::now() + - chrono::Duration::from_std(max_wait_time) + .unwrap_or_else(|_| chrono::Duration::seconds(30))) + .with_timezone(&FixedOffset::east_opt(0).unwrap()); + cleanup_scheduler + .reconcile_orphaned_builds( + cutoff, + "Build is no longer tracked by this server; marked as Interrupted.\n", + "Periodic orphan reconciliation", + ) + .await; } }); @@ -901,6 +1009,20 @@ impl TaskScheduler { } } +/// If `status` is [`WorkerStatus::Busy`], remove that build from `active_builds`. +/// +/// Returns `(build_id, removed BuildInfo)` when Busy; `None` for Idle/Error/Lost. +pub(crate) fn take_busy_active_build( + active_builds: &DashMap, + status: &WorkerStatus, +) -> Option<(String, Option)> { + let WorkerStatus::Busy { build_id, .. } = status else { + return None; + }; + let info = active_builds.remove(build_id).map(|(_, v)| v); + Some((build_id.clone(), info)) +} + /// Register `build_info` on `chosen_id` and send `msg`. /// /// Must drop the `workers` DashMap write guard before returning (no `.await` @@ -1009,6 +1131,98 @@ mod tests { assert!(queue.enqueue_v2(task).is_err()); } + #[test] + fn test_default_max_wait_time_is_60_minutes() { + let config = TaskQueueConfig::default(); + assert_eq!(config.max_wait_time, Duration::from_secs(3600)); + } + + #[test] + fn test_requeue_front_preserves_fifo_order_and_bypasses_capacity() { + let config = TaskQueueConfig { + max_queue_size: 1, + max_wait_time: Duration::from_secs(3600), + cleanup_interval: Duration::from_secs(10), + }; + let mut queue = TaskQueue::new(config); + + let first = PendingBuildEventV2 { + event_payload: BuildEventPayload::new( + Uuid::now_v7(), + Uuid::now_v7(), + "first".to_string(), + "repo/a".to_string(), + 0, + ), + targets: vec![], + changes: vec![], + created_at: Instant::now(), + }; + let second = PendingBuildEventV2 { + event_payload: BuildEventPayload::new( + Uuid::now_v7(), + Uuid::now_v7(), + "second".to_string(), + "repo/b".to_string(), + 0, + ), + targets: vec![], + changes: vec![], + created_at: Instant::now(), + }; + let first_id = first.event_payload.build_event_id; + let second_id = second.event_payload.build_event_id; + + assert!(queue.enqueue_v2(second).is_ok()); + // Capacity is full; requeue_front must still accept a previously dequeued task. + queue.requeue_front_v2(first); + + let stats = queue.get_stats(); + assert_eq!(stats.total_queued, 2); + assert_eq!( + queue + .queue_v2 + .front() + .map(|t| t.event_payload.build_event_id), + Some(first_id) + ); + assert_eq!( + queue + .queue_v2 + .back() + .map(|t| t.event_payload.build_event_id), + Some(second_id) + ); + } + + #[test] + fn test_cleanup_expired_respects_max_wait_time() { + let config = TaskQueueConfig { + max_queue_size: 10, + max_wait_time: Duration::from_millis(1), + cleanup_interval: Duration::from_secs(10), + }; + let mut queue = TaskQueue::new(config); + + let task = PendingBuildEventV2 { + event_payload: BuildEventPayload::new( + Uuid::now_v7(), + Uuid::now_v7(), + "stale".to_string(), + "repo".to_string(), + 0, + ), + targets: vec![], + changes: vec![], + created_at: Instant::now() - Duration::from_millis(5), + }; + assert!(queue.enqueue_v2(task).is_ok()); + + let expired = queue.cleanup_expired_v2(); + assert_eq!(expired.len(), 1); + assert_eq!(queue.get_stats().total_queued, 0); + } + /// Regression: after claim+send, the workers DashMap shard must be free so /// concurrent heartbeats / health checks / clients-info cannot block the /// runtime (the old retry path held `get_mut` across DB `.await`). @@ -1083,4 +1297,66 @@ mod tests { Ok(WSMessage::TaskBuild { build_id, .. }) if build_id == build_event_id.to_string() )); } + + #[test] + fn take_busy_active_build_clears_map_when_busy() { + let build_event_id = Uuid::now_v7(); + let build_id = build_event_id.to_string(); + let active_builds = DashMap::new(); + active_builds.insert( + build_id.clone(), + BuildInfo { + event_payload: BuildEventPayload::new( + build_event_id, + Uuid::now_v7(), + "cl".to_string(), + "/repo".to_string(), + 0, + ), + target_id: Uuid::now_v7(), + target_path: "//".to_string(), + changes: vec![], + started_at: chrono::Utc::now(), + worker_id: "w1".to_string(), + }, + ); + + let status = WorkerStatus::Busy { + build_id: build_id.clone(), + phase: None, + }; + let taken = take_busy_active_build(&active_builds, &status); + assert!(taken.is_some()); + let (id, info) = taken.unwrap(); + assert_eq!(id, build_id); + assert!(info.is_some()); + assert!(!active_builds.contains_key(&build_id)); + } + + #[test] + fn take_busy_active_build_idle_is_noop() { + let build_event_id = Uuid::now_v7(); + let build_id = build_event_id.to_string(); + let active_builds = DashMap::new(); + active_builds.insert( + build_id.clone(), + BuildInfo { + event_payload: BuildEventPayload::new( + build_event_id, + Uuid::now_v7(), + "cl".to_string(), + "/repo".to_string(), + 0, + ), + target_id: Uuid::now_v7(), + target_path: "//".to_string(), + changes: vec![], + started_at: chrono::Utc::now(), + worker_id: "w1".to_string(), + }, + ); + + assert!(take_busy_active_build(&active_builds, &WorkerStatus::Idle).is_none()); + assert!(active_builds.contains_key(&build_id)); + } } diff --git a/orion-server/src/server.rs b/orion-server/src/server.rs index e6767bb92..94160eb41 100644 --- a/orion-server/src/server.rs +++ b/orion-server/src/server.rs @@ -1,7 +1,6 @@ use std::{env, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; use axum::{Router, routing::get}; -use chrono::{FixedOffset, Utc}; use common::{ config::{ Config, @@ -26,11 +25,6 @@ use crate::{ log_service::LogService, store::{LogStore, io_orbit_store::IoOrbitLogStore, local_log_store, noop_log_store}, }, - model::target_state::TargetState, - repository::{ - build_events_repo::BuildEventsRepo, build_targets_repo::BuildTargetsRepo, - target_state_histories_repo::TargetStateHistoriesRepo, - }, }; pub async fn init_log_service(config: Config) -> Result { @@ -223,12 +217,10 @@ pub async fn start_server() { } /// Background task that monitors worker health and handles timeouts -/// Removes dead workers and marks their tasks as interrupted +/// Removes dead workers and marks their in-flight builds as interrupted async fn start_health_check_task(state: AppState) { let health_check_interval = Duration::from_secs(30); let worker_timeout = Duration::from_secs(90); - let utc_offset = - FixedOffset::east_opt(0).unwrap_or_else(|| unreachable!("UTC offset must be valid")); tracing::info!( "Health check task started. Interval: {:?}, Worker timeout: {:?}", @@ -258,112 +250,13 @@ async fn start_health_check_task(state: AppState) { tracing::warn!("Found dead workers: {:?}", dead_workers); - // Remove dead workers and handle their tasks for worker_id in dead_workers { if let Some((_, worker_info)) = state.scheduler.workers.remove(&worker_id) { tracing::info!("Removed dead worker: {}", worker_id); - - // If worker was busy, mark task as interrupted - if let crate::scheduler::WorkerStatus::Busy { build_id, .. } = worker_info.status { - tracing::warn!( - "Worker {} was busy with task {}. Marking task as Interrupted.", - worker_id, - build_id - ); - state.scheduler.active_builds.remove(&build_id); - - let build_uuid = match build_id.parse::() { - Ok(uuid) => uuid, - Err(_) => { - tracing::warn!( - "Invalid build id {} when marking interrupted", - build_id - ); - continue; - } - }; - - let end_at = Utc::now().with_timezone(&utc_offset); - if let Err(e) = - BuildEventsRepo::mark_interrupted(build_uuid, end_at, &state.conn).await - { - tracing::error!( - "Failed to mark build event {} interrupted: {}", - build_id, - e - ); - } - - // Keep target-level state consistent with interrupted build state. - let task_id = match BuildEventsRepo::find_by_id(&state.conn, build_uuid).await { - Ok(Some(build_event)) => build_event.task_id, - Ok(None) => { - tracing::warn!( - "Build event {} not found when syncing interrupted target state", - build_id - ); - continue; - } - Err(e) => { - tracing::error!( - "Failed to fetch build event {} when syncing target state: {}", - build_id, - e - ); - continue; - } - }; - - let targets = match BuildTargetsRepo::list_by_task_id(&state.conn, task_id) - .await - { - Ok(targets) => targets, - Err(e) => { - tracing::error!( - "Failed to list targets for task {} when handling dead worker {}: {}", - task_id, - worker_id, - e - ); - continue; - } - }; - - for target in targets { - if let Err(e) = BuildTargetsRepo::update_latest_state( - &state.conn, - target.id, - TargetState::Interrupted, - ) - .await - { - tracing::error!( - "Failed to update target {} to Interrupted for build {}: {}", - target.id, - build_id, - e - ); - continue; - } - - if let Err(e) = TargetStateHistoriesRepo::upsert_state( - &state.conn, - target.id, - build_uuid, - TargetState::Interrupted.to_string(), - end_at, - ) - .await - { - tracing::error!( - "Failed to upsert interrupted history for target {} build {}: {}", - target.id, - build_id, - e - ); - } - } - } + state + .scheduler + .handle_worker_gone(worker_info, "heartbeat timed out") + .await; } } } diff --git a/orion-server/src/service/api_v2_service.rs b/orion-server/src/service/api_v2_service.rs index db4d3d49d..78fc1db02 100644 --- a/orion-server/src/service/api_v2_service.rs +++ b/orion-server/src/service/api_v2_service.rs @@ -712,32 +712,34 @@ pub async fn build_retry( let new_build_id = Uuid::now_v7(); - // If no idle workers, do not queue: create the build and immediately mark it - // Interrupted so it surfaces as a clear, retryable terminal state. + // If no idle workers, enqueue the retry build and wait for a worker. if !state.scheduler.has_idle_workers() { - if let Err(e) = - BuildEventsRepo::insert_build(&state.conn, new_build_id, task.id, repo.clone()).await + match state + .scheduler + .enqueue_task_with_build_id_v2( + new_build_id, + task.id, + &cl_link, + repo, + changes, + retry_count, + ) + .await { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({"message": format!("Failed to create build event: {e}")})), + Ok(()) => ( + StatusCode::OK, + Json(json!({ + "message": "Build queued for later processing", + "build_id": new_build_id.to_string(), + })), ) - .into_response(); + .into_response(), + Err(e) => ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"message": format!("Unable to queue build: {e}")})), + ) + .into_response(), } - - state - .scheduler - .fail_unschedulable_build(new_build_id, task.id, &repo) - .await; - - ( - StatusCode::OK, - Json(json!({ - "message": "No build worker available; build marked as Interrupted", - "build_id": new_build_id.to_string(), - })), - ) - .into_response() } else { if let Err(e) = BuildEventsRepo::insert_build(&state.conn, new_build_id, task.id, repo.clone()).await @@ -927,28 +929,6 @@ async fn handle_immediate_task_dispatch_v2( activate_worker(&build_info, &state.scheduler).await } -/// Create a build event + default target for a task that cannot be scheduled -/// (no idle worker) and immediately mark it `Interrupted`. Returns the new build -/// id so callers can report it to the client. -async fn reject_unschedulable_build( - state: &AppState, - task_id: Uuid, - repo: &str, -) -> Result { - let build_event_id = Uuid::now_v7(); - BuildEventsRepo::insert_build(&state.conn, build_event_id, task_id, repo.to_string()).await?; - - let target_id = Uuid::now_v7(); - BuildTargetsRepo::insert_default_target(target_id, task_id, &state.conn).await?; - - state - .scheduler - .fail_unschedulable_build(build_event_id, task_id, repo) - .await; - - Ok(build_event_id) -} - pub async fn task_handler_v2(state: &AppState, req: TaskBuildRequest) -> Response { let req = TaskBuildRequest { changes: normalize_repo_root_changes(&req.repo, req.changes), @@ -969,19 +949,21 @@ pub async fn task_handler_v2(state: &AppState, req: TaskBuildRequest) -> Respons handle_immediate_task_dispatch_v2(state, task_id, &req.repo, &req.cl_link, req.changes) .await } else { - // No worker is available: instead of queuing (which can leave a build - // stuck), create the build and immediately mark it Interrupted so the UI - // shows a clear, retryable terminal state. - match reject_unschedulable_build(state, task_id, &req.repo).await { + // No idle worker (VM not started or all busy): queue and wait for capacity. + match state + .scheduler + .enqueue_task_v2(task_id, &req.cl_link, req.repo, req.changes, 0) + .await + { Ok(build_id) => OrionBuildResult { build_id: build_id.to_string(), - status: "interrupted".to_string(), - message: "No build worker available; build marked as Interrupted".to_string(), + status: "queued".to_string(), + message: "Task queued for processing when workers become available".to_string(), }, Err(e) => { return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({"message": format!("Unable to record build: {}", e)})), + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"message": format!("Unable to queue task: {}", e)})), ) .into_response(); } diff --git a/orion-server/src/service/ws_service.rs b/orion-server/src/service/ws_service.rs index 4012e010d..580c02ffd 100644 --- a/orion-server/src/service/ws_service.rs +++ b/orion-server/src/service/ws_service.rs @@ -75,9 +75,14 @@ async fn handle_socket(socket: WebSocket, who: SocketAddr, state: AppState) { } } - if let Some(id) = &worker_id { + if let Some(id) = worker_id { tracing::info!("Cleaning up for worker: {id} from {who}."); - state.scheduler.workers.remove(id); + if let Some((_, info)) = state.scheduler.workers.remove(&id) { + state + .scheduler + .handle_worker_gone(info, "disconnected") + .await; + } } else { tracing::info!("Cleaning up unregistered connection from {who}."); } @@ -296,11 +301,11 @@ async fn process_message( } } Message::Close(_) => { - if let Some(id) = worker_id.take() - && let Some(mut worker) = state.scheduler.workers.get_mut(&id) - { - worker.status = WorkerStatus::Lost; - } + // Do not clear worker_id or overwrite Busy → Lost here: the outer + // handle_socket cleanup removes the worker and finalizes any in-flight + // build via handle_worker_gone. Overwriting Busy would lose build_id + // and leave the build stuck as Building forever. + tracing::info!("WebSocket Close from {who}"); return ControlFlow::Break(()); } _ => {}