From 8b31254f2a78572750445c8ef232f0760530316e Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Wed, 8 Apr 2026 18:33:04 +0800 Subject: [PATCH] fix: improve orion/orion-server dev builds - Make orion compile on macOS by gating scorpiofs/Linux-only Antares code --- .dockerignore | 4 + ceres/src/pack/import_repo.rs | 28 +- orion-server/Dockerfile | 5 +- orion-server/src/main.rs | 19 +- orion-server/src/service/ws_service.rs | 9 + orion/Cargo.toml | 4 +- orion/src/antares.rs | 689 +++++++++++++------------ orion/src/buck_controller.rs | 42 +- orion/src/ws.rs | 93 ++-- orion/systemd/orion-runner.service | 12 +- 10 files changed, 526 insertions(+), 379 deletions(-) diff --git a/.dockerignore b/.dockerignore index accee5252..e31713c71 100644 --- a/.dockerignore +++ b/.dockerignore @@ -37,6 +37,10 @@ ztm_agent_db* # buck2 out buck-out +# local docker build cache (huge, never needed in image) +.buildx-cache +.buildx-cache/** + # Dockderfile docker diff --git a/ceres/src/pack/import_repo.rs b/ceres/src/pack/import_repo.rs index d850cbf12..8a2b0a865 100644 --- a/ceres/src/pack/import_repo.rs +++ b/ceres/src/pack/import_repo.rs @@ -459,14 +459,29 @@ impl ImportRepo { let mono_api_service: MonoApiService = self.into(); let storage = self.storage.mono_storage(); + // Import tip commit + message do not depend on monorepo root; load once so retries + // under the root lock do not repeat git_db reads. + let latest_commit: Commit = Commit::from_git_model( + self.storage + .git_db_storage() + .get_commit_by_hash(self.repo.repo_id, &commit_id) + .await? + .ok_or_else(|| MegaError::Other(format!("commit {commit_id} not found")))?, + ); + let commit_msg = latest_commit.format_message(); + // Concurrent attaches need CAS on root mega_refs; retry when head moved. + // Redis lock reduces retry storms; DB still enforces correctness via StaleMonorepoRootRef. + // + // `search_and_create_tree` walks the live root tree via the API (`get_root_tree`); it must + // run against the same root snapshot as `get_main_ref` for this attempt, so it stays + // inside the locked section. Further reduction would require threading an explicit root + // tree/commit into `tree_ops` so tree building can run without holding the global lock. const MAX_ATTACH_ATTEMPTS: u32 = 64; let mut root_lock_wait_max_ms: u128 = 0; let mut root_lock_wait_sum_ms: u128 = 0; for attempt in 0..MAX_ATTACH_ATTEMPTS { - // Only the root mega_refs update needs cross-repo serialization. - // Keep the lock scope as small as possible: just the root ref read + attach transaction. let t_lock = Instant::now(); let guard = self.unpack_redlock.clone().lock().await?; let lock_wait_ms = t_lock.elapsed().as_millis(); @@ -483,15 +498,6 @@ impl ImportRepo { let save_trees = tree_ops::search_and_create_tree(&mono_api_service, &path).await?; - let latest_commit: Commit = Commit::from_git_model( - self.storage - .git_db_storage() - .get_commit_by_hash(self.repo.repo_id, &commit_id) - .await? - .ok_or_else(|| MegaError::Other(format!("commit {} not found", commit_id)))?, - ); - - let commit_msg = latest_commit.format_message(); let new_commit = Commit::from_tree_id( save_trees .back() diff --git a/orion-server/Dockerfile b/orion-server/Dockerfile index 03d0248a8..ce0d58c07 100644 --- a/orion-server/Dockerfile +++ b/orion-server/Dockerfile @@ -45,7 +45,10 @@ COPY . . # -- Step 2: build binary into normal target directory ENV CARGO_TARGET_DIR=/app/target -RUN cargo build -p orion-server --release +RUN --mount=type=cache,target=/usr/local/cargo/registry \ + --mount=type=cache,target=/usr/local/cargo/git \ + --mount=type=cache,target=/app/target \ + cargo build -p orion-server --release # ────── Stage 3: Runtime ────── FROM debian:bookworm-slim diff --git a/orion-server/src/main.rs b/orion-server/src/main.rs index 7de7c770f..5d4e36a08 100644 --- a/orion-server/src/main.rs +++ b/orion-server/src/main.rs @@ -1,3 +1,4 @@ +use std::io::IsTerminal; /// Orion Build Server /// A distributed build system that manages build tasks and worker nodes use std::path::PathBuf; @@ -23,27 +24,33 @@ fn init_tracing_from_config() { cli_path: None, env_path: std::env::var_os("MEGA_CONFIG").map(PathBuf::from), }; - let level = match ConfigLoader::new(input).load() { + let (level, with_ansi) = match ConfigLoader::new(input).load() { Ok(loaded) => match loaded.path.to_str() { Some(p) => match Config::new(p) { - Ok(cfg) => tracing_level_from_config(&cfg.log.level), + Ok(cfg) => ( + tracing_level_from_config(&cfg.log.level), + cfg.log.with_ansi && std::io::stdout().is_terminal(), + ), Err(e) => { eprintln!("Failed to parse config for log level: {e}; using info"); - tracing::Level::INFO + (tracing::Level::INFO, std::io::stdout().is_terminal()) } }, None => { eprintln!("Config path is not valid UTF-8; using info"); - tracing::Level::INFO + (tracing::Level::INFO, std::io::stdout().is_terminal()) } }, Err(e) => { eprintln!("Failed to locate config for log level: {e}; using info"); - tracing::Level::INFO + (tracing::Level::INFO, std::io::stdout().is_terminal()) } }; - tracing_subscriber::fmt().with_max_level(level).init(); + tracing_subscriber::fmt() + .with_max_level(level) + .with_ansi(with_ansi) + .init(); } #[tokio::main] diff --git a/orion-server/src/service/ws_service.rs b/orion-server/src/service/ws_service.rs index 002f34057..f4bd70c66 100644 --- a/orion-server/src/service/ws_service.rs +++ b/orion-server/src/service/ws_service.rs @@ -148,6 +148,15 @@ async fn process_message( } } } + WSMessage::TaskAck { + build_id, + success, + message, + } => { + tracing::info!( + "TaskAck from worker {current_worker_id}: build_id={build_id} success={success} message={message}" + ); + } WSMessage::TaskBuildOutput { build_id, output } => { if let Some(build_info) = state.scheduler.active_builds.get(&build_id) { let log_event = LogEvent { diff --git a/orion/Cargo.toml b/orion/Cargo.toml index b1b097ca5..f33c79e3c 100644 --- a/orion/Cargo.toml +++ b/orion/Cargo.toml @@ -30,9 +30,11 @@ td_util_buck = { path = "./buck" } thiserror = { workspace = true } utoipa.workspace = true common = { path = "../common" } -scorpiofs = "0.2.1" tokio-util = { workspace = true } [dev-dependencies] serial_test = { workspace = true } tempfile = { workspace = true } + +[target.'cfg(target_os = "linux")'.dependencies] +scorpiofs = "0.2.1" diff --git a/orion/src/antares.rs b/orion/src/antares.rs index a63c95c8d..4b90b1cc4 100644 --- a/orion/src/antares.rs +++ b/orion/src/antares.rs @@ -2,381 +2,434 @@ //! //! This module provides a singleton wrapper around `scorpiofs::AntaresManager` //! for managing overlay filesystem mounts used during build operations. +//! +//! Platform support: +//! +//! Orion integrates with the FUSE filesystem via `scorpiofs`, which is only used +//! on Linux in this repository. To keep local development tooling usable on +//! macOS/Windows, this module provides: +//! +//! - **Linux**: real implementation backed by `scorpiofs` +//! - **Non-Linux**: stub implementation (compiles; mount/unmount return errors) + +#[cfg(target_os = "linux")] +mod imp { + //! Linux implementation backed by `scorpiofs`. + + use std::{ + any::Any, + error::Error, + io, + panic::AssertUnwindSafe, + path::{Path, PathBuf}, + sync::Arc, + time::Duration, + }; -use std::{ - any::Any, - error::Error, - io, - panic::AssertUnwindSafe, - path::{Path, PathBuf}, - sync::Arc, - time::Duration, -}; - -use futures_util::FutureExt; -use scorpiofs::{AntaresConfig, AntaresManager, AntaresPaths}; -use tokio::sync::OnceCell; - -static MANAGER: OnceCell> = OnceCell::const_new(); -const TEST_BROWSE_JOB_ID: &str = "antares_test"; - -type DynError = Box; - -/// Get the global AntaresManager instance. -/// -/// Initializes the manager on first call by loading the scorpio configuration -/// from the path specified by `SCORPIO_CONFIG` environment variable. -/// -/// If `SCORPIO_CONFIG` is not set, Orion will look for `scorpio.toml` in: -/// 1. Current working directory -/// 2. Next to the executable -/// 3. `/etc/scorpio/scorpio.toml` (system default) -/// -/// Returns an error if no config file is found. -async fn get_manager() -> Result<&'static Arc, DynError> { - MANAGER - .get_or_try_init(|| async { - let config_path = resolve_config_path()?; - let config_path_str = config_path.to_str().ok_or_else(|| -> DynError { - Box::new(io_other("Invalid SCORPIO_CONFIG path (non-UTF8)")) - })?; - - tracing::info!("Initializing Antares with config: {}", config_path_str); - - scorpiofs::util::config::init_config(config_path_str).map_err(|e| { - io_other(format!( - "Failed to load scorpio config from {config_path_str}: {e}. \ + use futures_util::FutureExt; + use scorpiofs::{AntaresConfig, AntaresManager, AntaresPaths}; + use tokio::sync::OnceCell; + + static MANAGER: OnceCell> = OnceCell::const_new(); + const TEST_BROWSE_JOB_ID: &str = "antares_test"; + + type DynError = Box; + + /// Get the global AntaresManager instance. + /// + /// Initializes the manager on first call by loading the scorpio configuration + /// from the path specified by `SCORPIO_CONFIG` environment variable. + /// + /// If `SCORPIO_CONFIG` is not set, Orion will look for `scorpio.toml` in: + /// 1. Current working directory + /// 2. Next to the executable + /// 3. `/etc/scorpio/scorpio.toml` (system default) + /// + /// Returns an error if no config file is found. + async fn get_manager() -> Result<&'static Arc, DynError> { + MANAGER + .get_or_try_init(|| async { + let config_path = resolve_config_path()?; + let config_path_str = config_path.to_str().ok_or_else(|| -> DynError { + Box::new(io_other("Invalid SCORPIO_CONFIG path (non-UTF8)")) + })?; + + tracing::info!("Initializing Antares with config: {}", config_path_str); + + scorpiofs::util::config::init_config(config_path_str).map_err(|e| { + io_other(format!( + "Failed to load scorpio config from {config_path_str}: {e}. \ Hint: set SCORPIO_CONFIG=/path/to/scorpio.toml or create /etc/scorpio/scorpio.toml" - )) - })?; + )) + })?; - let paths = AntaresPaths::from_global_config(); - let manager = AntaresManager::new(paths).await; - Ok(Arc::new(manager)) - }) - .await -} + let paths = AntaresPaths::from_global_config(); + let manager = AntaresManager::new(paths).await; + Ok(Arc::new(manager)) + }) + .await + } -fn io_other(message: impl Into) -> io::Error { - io::Error::other(message.into()) -} + fn io_other(message: impl Into) -> io::Error { + io::Error::other(message.into()) + } -fn resolve_config_path() -> Result { - // 1. Check SCORPIO_CONFIG environment variable - if let Ok(path) = std::env::var("SCORPIO_CONFIG") { - let config_path = PathBuf::from(&path); - if config_path.exists() { - return Ok(config_path); + /// Resolve scorpio configuration path. + fn resolve_config_path() -> Result { + // 1. Check SCORPIO_CONFIG environment variable + if let Ok(path) = std::env::var("SCORPIO_CONFIG") { + let config_path = PathBuf::from(&path); + if config_path.exists() { + return Ok(config_path); + } + return Err(Box::new(io_other(format!( + "SCORPIO_CONFIG is set but file does not exist: {}", + config_path.display() + )))); } - return Err(Box::new(io_other(format!( - "SCORPIO_CONFIG is set but file does not exist: {}", - config_path.display() - )))); - } - // 2. Check current working directory - let cwd = std::env::current_dir().map_err(|e| { - Box::new(io_other(format!( - "Failed to get current working directory: {e}" - ))) as DynError - })?; - let cwd_candidate = cwd.join("scorpio.toml"); - if cwd_candidate.exists() { - return Ok(cwd_candidate); - } + // 2. Check current working directory + let cwd = std::env::current_dir().map_err(|e| { + Box::new(io_other(format!( + "Failed to get current working directory: {e}" + ))) as DynError + })?; + let cwd_candidate = cwd.join("scorpio.toml"); + if cwd_candidate.exists() { + return Ok(cwd_candidate); + } - // 3. Check next to executable - if let Ok(exe) = std::env::current_exe() - && let Some(exe_dir) = exe.parent() - { - let exe_candidate = exe_dir.join("scorpio.toml"); - if exe_candidate.exists() { - return Ok(exe_candidate); + // 3. Check next to executable + if let Ok(exe) = std::env::current_exe() + && let Some(exe_dir) = exe.parent() + { + let exe_candidate = exe_dir.join("scorpio.toml"); + if exe_candidate.exists() { + return Ok(exe_candidate); + } } - } - // 4. Check system default path - let system_candidate = PathBuf::from("/etc/scorpio/scorpio.toml"); - if system_candidate.exists() { - return Ok(system_candidate); - } + // 4. Check system default path + let system_candidate = PathBuf::from("/etc/scorpio/scorpio.toml"); + if system_candidate.exists() { + return Ok(system_candidate); + } - Err(Box::new(io_other(format!( - "Scorpio config not found. Set SCORPIO_CONFIG=/path/to/scorpio.toml, \ + Err(Box::new(io_other(format!( + "Scorpio config not found. Set SCORPIO_CONFIG=/path/to/scorpio.toml, \ place scorpio.toml in the working directory ({}), \ or create /etc/scorpio/scorpio.toml", - cwd.display() - )))) -} - -/// Mount a job overlay filesystem. -/// -/// Creates a new Antares overlay mount for the specified job. The underlying -/// Dicfuse layer provides read-only access to the repository, while the overlay -/// allows copy-on-write modifications. -/// -/// # Arguments -/// * `job_id` - Unique identifier for this build job -/// * `cl` - Optional changelist layer name -/// -/// # Returns -/// The `AntaresConfig` containing mountpoint and job metadata on success. -pub async fn mount_job(job_id: &str, cl: Option<&str>) -> Result { - tracing::debug!("Mounting Antares job: job_id={}, cl={:?}", job_id, cl); - let manager = get_manager().await?; - run_with_panic_guard( - format!("Antares mount panicked for job_id={job_id}, cl={cl:?}"), - manager.mount_job(job_id, cl), - ) - .await -} - -/// Initialize Antares during Orion startup and eagerly trigger Dicfuse import. -/// -/// This keeps the first build request from paying the full Dicfuse cold-start -/// cost. Readiness waiting runs in the background so Orion can continue booting. -#[allow(dead_code)] // Called from the bin target (main.rs), not visible to lib check. -pub(crate) async fn warmup_dicfuse() -> Result<(), DynError> { - tracing::info!("Initializing Antares Dicfuse during Orion startup"); - let manager = get_manager().await?; - let manager_for_test_mount = Arc::clone(manager); - let dicfuse = manager.dicfuse(); - - // Idempotent: safe even if the manager already started import internally. - dicfuse.start_import(); - - tokio::spawn(async move { - let warmup_timeout_secs: u64 = std::env::var("ORION_DICFUSE_WARMUP_TIMEOUT_SECS") - .ok() - .and_then(|v| v.parse().ok()) - .unwrap_or(1200); - tracing::info!( - "Waiting for Antares Dicfuse warmup to finish (timeout: {}s)", - warmup_timeout_secs - ); + cwd.display() + )))) + } - match tokio::time::timeout( - Duration::from_secs(warmup_timeout_secs), - dicfuse.store.wait_for_ready(), + /// Mount a job overlay filesystem. + /// + /// Creates a new Antares overlay mount for the specified job. The underlying + /// Dicfuse layer provides read-only access to the repository, while the overlay + /// allows copy-on-write modifications. + /// + /// # Arguments + /// * `job_id` - Unique identifier for this build job + /// * `cl` - Optional changelist layer name + /// + /// # Returns + /// The `AntaresConfig` containing mountpoint and job metadata on success. + pub async fn mount_job(job_id: &str, cl: Option<&str>) -> Result { + tracing::debug!("Mounting Antares job: job_id={}, cl={:?}", job_id, cl); + let manager = get_manager().await?; + run_with_panic_guard( + format!("Antares mount panicked for job_id={job_id}, cl={cl:?}"), + manager.mount_job(job_id, cl), ) .await - { - Ok(_) => { - tracing::info!("Antares Dicfuse warmup completed"); - log_dicfuse_root_tree(); - if is_test_mount_enabled() { - ensure_test_mount(manager_for_test_mount.as_ref()).await; - } else { - tracing::info!( - "Antares test mount disabled by ORION_ENABLE_ANTARES_TEST_MOUNT" - ); - } - } - Err(_) => tracing::warn!( - "Antares Dicfuse warmup timed out after {}s", + } + + /// Initialize Antares during Orion startup and eagerly trigger Dicfuse import. + /// + /// This keeps the first build request from paying the full Dicfuse cold-start + /// cost. Readiness waiting runs in the background so Orion can continue booting. + #[allow(dead_code)] + pub(crate) async fn warmup_dicfuse() -> Result<(), DynError> { + tracing::info!("Initializing Antares Dicfuse during Orion startup"); + let manager = get_manager().await?; + let manager_for_test_mount = Arc::clone(manager); + let dicfuse = manager.dicfuse(); + + dicfuse.start_import(); + + tokio::spawn(async move { + let warmup_timeout_secs: u64 = std::env::var("ORION_DICFUSE_WARMUP_TIMEOUT_SECS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(1200); + tracing::info!( + "Waiting for Antares Dicfuse warmup to finish (timeout: {}s)", warmup_timeout_secs - ), - } - }); + ); - Ok(()) -} + match tokio::time::timeout( + Duration::from_secs(warmup_timeout_secs), + dicfuse.store.wait_for_ready(), + ) + .await + { + Ok(_) => { + tracing::info!("Antares Dicfuse warmup completed"); + log_dicfuse_root_tree(); + if is_test_mount_enabled() { + ensure_test_mount(manager_for_test_mount.as_ref()).await; + } else { + tracing::info!( + "Antares test mount disabled by ORION_ENABLE_ANTARES_TEST_MOUNT" + ); + } + } + Err(_) => tracing::warn!( + "Antares Dicfuse warmup timed out after {}s", + warmup_timeout_secs + ), + } + }); -fn is_test_mount_enabled() -> bool { - match std::env::var("ORION_ENABLE_ANTARES_TEST_MOUNT") { - Ok(v) => { - let v = v.trim().to_ascii_lowercase(); - !(v == "0" || v == "false" || v == "no" || v == "off") - } - Err(_) => true, + Ok(()) } -} -async fn ensure_test_mount(manager: &AntaresManager) { - match run_with_panic_guard( - format!("Antares test mount panicked for job_id={TEST_BROWSE_JOB_ID}"), - manager.mount_job(TEST_BROWSE_JOB_ID, None), - ) - .await - { - Ok(config) => { - tracing::info!( - "Antares test mount ready: job_id={}, mountpoint={}", - TEST_BROWSE_JOB_ID, - config.mountpoint.display() - ); - } - Err(err) => { - tracing::warn!( - "Failed to create Antares test mount job_id={}: {}", - TEST_BROWSE_JOB_ID, - err - ); + fn is_test_mount_enabled() -> bool { + match std::env::var("ORION_ENABLE_ANTARES_TEST_MOUNT") { + Ok(v) => { + let v = v.trim().to_ascii_lowercase(); + !(v == "0" || v == "false" || v == "no" || v == "off") + } + Err(_) => true, } } -} -fn log_dicfuse_root_tree() { - let root = PathBuf::from(scorpiofs::util::config::workspace()); - let max_depth = std::env::var("ORION_DICFUSE_ROOT_TREE_DEPTH") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(2); - let max_entries = std::env::var("ORION_DICFUSE_ROOT_TREE_MAX_ENTRIES") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(200); - - tracing::info!( - root = %root.display(), - max_depth, - max_entries, - "Dicfuse init: printing workspace root tree" - ); - - if !root.exists() { - tracing::warn!("Dicfuse workspace path does not exist: {}", root.display()); - return; + async fn ensure_test_mount(manager: &AntaresManager) { + match run_with_panic_guard( + format!("Antares test mount panicked for job_id={TEST_BROWSE_JOB_ID}"), + manager.mount_job(TEST_BROWSE_JOB_ID, None), + ) + .await + { + Ok(config) => { + tracing::info!( + "Antares test mount ready: job_id={}, mountpoint={}", + TEST_BROWSE_JOB_ID, + config.mountpoint.display() + ); + } + Err(err) => { + tracing::warn!( + "Failed to create Antares test mount job_id={}: {}", + TEST_BROWSE_JOB_ID, + err + ); + } + } } - let mut printed = 0usize; - tracing::info!("[dicfuse-root] /"); - log_tree_recursive(&root, &root, 0, max_depth, max_entries, &mut printed); + fn log_dicfuse_root_tree() { + let root = PathBuf::from(scorpiofs::util::config::workspace()); + let max_depth = std::env::var("ORION_DICFUSE_ROOT_TREE_DEPTH") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(2); + let max_entries = std::env::var("ORION_DICFUSE_ROOT_TREE_MAX_ENTRIES") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(200); - if printed >= max_entries { tracing::info!( - "Dicfuse root tree output truncated at {} entries (set ORION_DICFUSE_ROOT_TREE_MAX_ENTRIES to increase)", - max_entries + root = %root.display(), + max_depth, + max_entries, + "Dicfuse init: printing workspace root tree" ); - } -} -fn log_tree_recursive( - root: &Path, - current: &Path, - depth: usize, - max_depth: usize, - max_entries: usize, - printed: &mut usize, -) { - if depth >= max_depth || *printed >= max_entries { - return; + if !root.exists() { + tracing::warn!("Dicfuse workspace path does not exist: {}", root.display()); + return; + } + + let mut printed = 0usize; + tracing::info!("[dicfuse-root] /"); + log_tree_recursive(&root, &root, 0, max_depth, max_entries, &mut printed); + + if printed >= max_entries { + tracing::info!( + "Dicfuse root tree output truncated at {} entries (set ORION_DICFUSE_ROOT_TREE_MAX_ENTRIES to increase)", + max_entries + ); + } } - let entries = match std::fs::read_dir(current) { - Ok(entries) => entries, - Err(err) => { - tracing::warn!("Failed to read {}: {}", current.display(), err); + fn log_tree_recursive( + root: &Path, + current: &Path, + depth: usize, + max_depth: usize, + max_entries: usize, + printed: &mut usize, + ) { + if depth >= max_depth || *printed >= max_entries { return; } - }; - let mut children: Vec<(String, PathBuf, bool)> = Vec::new(); - for entry in entries { - let entry = match entry { - Ok(entry) => entry, + let entries = match std::fs::read_dir(current) { + Ok(entries) => entries, Err(err) => { - tracing::warn!("read_dir entry error under {}: {}", current.display(), err); - continue; + tracing::warn!("Failed to read {}: {}", current.display(), err); + return; } }; - let path = entry.path(); - let name = entry.file_name().to_string_lossy().to_string(); - let is_dir = entry.file_type().map(|t| t.is_dir()).unwrap_or(false); - children.push((name, path, is_dir)); - } - - children.sort_by(|a, b| a.0.cmp(&b.0)); + let mut children: Vec<(String, PathBuf, bool)> = Vec::new(); + for entry in entries { + let entry = match entry { + Ok(entry) => entry, + Err(err) => { + tracing::warn!("read_dir entry error under {}: {}", current.display(), err); + continue; + } + }; - for (_name, path, is_dir) in children { - if *printed >= max_entries { - return; + let path = entry.path(); + let name = entry.file_name().to_string_lossy().to_string(); + let is_dir = entry.file_type().map(|t| t.is_dir()).unwrap_or(false); + children.push((name, path, is_dir)); } - let rel = path - .strip_prefix(root) - .map(|p| p.display().to_string()) - .unwrap_or_else(|_| path.display().to_string()); - let indent = " ".repeat(depth + 1); - if is_dir { - tracing::info!("[dicfuse-root] {}{}/", indent, rel); - } else { - tracing::info!("[dicfuse-root] {}{}", indent, rel); - } - *printed += 1; + children.sort_by(|a, b| a.0.cmp(&b.0)); - if is_dir { - log_tree_recursive(root, &path, depth + 1, max_depth, max_entries, printed); + for (_name, path, is_dir) in children { + if *printed >= max_entries { + return; + } + + let rel = path + .strip_prefix(root) + .map(|p| p.display().to_string()) + .unwrap_or_else(|_| path.display().to_string()); + let indent = " ".repeat(depth + 1); + if is_dir { + tracing::info!("[dicfuse-root] {}{}/", indent, rel); + } else { + tracing::info!("[dicfuse-root] {}{}", indent, rel); + } + *printed += 1; + + if is_dir { + log_tree_recursive(root, &path, depth + 1, max_depth, max_entries, printed); + } } } -} -/// Unmount and cleanup a job overlay filesystem. -/// -/// # Arguments -/// * `job_id` - The job identifier to unmount -/// -/// # Returns -/// The `AntaresConfig` of the unmounted job if it existed. -#[allow(dead_code)] -pub async fn unmount_job(job_id: &str) -> Result, DynError> { - tracing::debug!("Unmounting Antares job: job_id={}", job_id); - get_manager() - .await? - .umount_job(job_id) - .await - .map_err(Into::into) -} + /// Unmount and cleanup a job overlay filesystem. + /// + /// # Arguments + /// * `job_id` - The job identifier to unmount + /// + /// # Returns + /// The `AntaresConfig` of the unmounted job if it existed. + #[allow(dead_code)] + pub async fn unmount_job(job_id: &str) -> Result, DynError> { + tracing::debug!("Unmounting Antares job: job_id={}", job_id); + get_manager() + .await? + .umount_job(job_id) + .await + .map_err(Into::into) + } -async fn run_with_panic_guard(context: String, future: F) -> Result -where - F: std::future::Future>, - E: Into, -{ - match AssertUnwindSafe(future).catch_unwind().await { - Ok(result) => result.map_err(Into::into), - Err(payload) => Err(Box::new(io_other(format!( - "{context}: {}", - panic_payload_to_string(payload.as_ref()) - )))), + /// Convert panics within scorpiofs calls into regular errors. + async fn run_with_panic_guard(context: String, future: F) -> Result + where + F: std::future::Future>, + E: Into, + { + match AssertUnwindSafe(future).catch_unwind().await { + Ok(result) => result.map_err(Into::into), + Err(payload) => Err(Box::new(io_other(format!( + "{context}: {}", + panic_payload_to_string(payload.as_ref()) + )))), + } } -} -fn panic_payload_to_string(payload: &(dyn Any + Send)) -> String { - if let Some(message) = payload.downcast_ref::<&str>() { - return (*message).to_string(); + /// Best-effort stringify for panic payloads. + fn panic_payload_to_string(payload: &(dyn Any + Send)) -> String { + if let Some(message) = payload.downcast_ref::<&str>() { + return (*message).to_string(); + } + if let Some(message) = payload.downcast_ref::() { + return message.clone(); + } + "non-string panic payload".to_string() } - if let Some(message) = payload.downcast_ref::() { - return message.clone(); + + #[cfg(test)] + mod tests { + use std::io; + + use super::{panic_payload_to_string, run_with_panic_guard}; + + #[tokio::test] + async fn test_run_with_panic_guard_converts_panic_to_error() { + let result: Result<(), Box> = + run_with_panic_guard("panic guard".to_string(), async move { + panic!("fuse mount failed"); + #[allow(unreachable_code)] + Ok::<(), io::Error>(()) + }) + .await; + + let err = result.expect_err("panic should be converted into error"); + assert!(err.to_string().contains("panic guard")); + assert!(err.to_string().contains("fuse mount failed")); + } + + #[test] + fn test_panic_payload_to_string_handles_common_payloads() { + assert_eq!(panic_payload_to_string(&"oops"), "oops"); + assert_eq!(panic_payload_to_string(&"boom".to_string()), "boom"); + } } - "non-string panic payload".to_string() } -#[cfg(test)] -mod tests { - use std::io; +#[cfg(not(target_os = "linux"))] +mod imp { + //! Stub implementation for non-Linux platforms. - use super::{panic_payload_to_string, run_with_panic_guard}; + use std::{error::Error, path::PathBuf}; - #[tokio::test] - async fn test_run_with_panic_guard_converts_panic_to_error() { - let result: Result<(), Box> = - run_with_panic_guard("panic guard".to_string(), async move { - panic!("fuse mount failed"); - #[allow(unreachable_code)] - Ok::<(), io::Error>(()) - }) - .await; + type DynError = Box; + + #[derive(Debug, Clone)] + pub struct AntaresConfig { + pub mountpoint: PathBuf, + pub job_id: String, + } - let err = result.expect_err("panic should be converted into error"); - assert!(err.to_string().contains("panic guard")); - assert!(err.to_string().contains("fuse mount failed")); + /// Mounting Antares requires `scorpiofs` (Linux-only in this repository). + pub async fn mount_job(_job_id: &str, _cl: Option<&str>) -> Result { + Err(Box::new(std::io::Error::other( + "Antares/scorpiofs is only supported on Linux", + ))) } - #[test] - fn test_panic_payload_to_string_handles_common_payloads() { - assert_eq!(panic_payload_to_string(&"oops"), "oops"); - assert_eq!(panic_payload_to_string(&"boom".to_string()), "boom"); + /// Unmounting Antares requires `scorpiofs` (Linux-only in this repository). + #[allow(dead_code)] + pub async fn unmount_job(_job_id: &str) -> Result, DynError> { + Err(Box::new(std::io::Error::other( + "Antares/scorpiofs is only supported on Linux", + ))) + } + + #[allow(dead_code)] + pub(crate) async fn warmup_dicfuse() -> Result<(), DynError> { + Ok(()) } } + +pub use imp::*; diff --git a/orion/src/buck_controller.rs b/orion/src/buck_controller.rs index 028145e60..fc25eb1c5 100644 --- a/orion/src/buck_controller.rs +++ b/orion/src/buck_controller.rs @@ -598,6 +598,19 @@ pub async fn build( ) -> Result> { tracing::info!("[Task {}] Building in repo {}", id, repo); + // Until `buck2 build` starts, the worker only had Busy with phase=None. The UI then + // looks like the client is not reporting progress during FUSE mount + target discovery. + if let Err(e) = sender.send(WSMessage::TaskPhaseUpdate { + build_id: id.clone(), + phase: TaskPhase::DownloadingSource, + }) { + tracing::warn!( + "[Task {}] Failed to send DownloadingSource phase (server may not show prep progress): {}", + id, + e + ); + } + let task_id = id.trim(); // Handle empty cl string as None to mount the base repo without a CL layer. let cl_trimmed = cl.trim(); @@ -677,6 +690,11 @@ pub async fn build( Ok(found_targets) => { mount_point = Some(repo_mount_point); old_repo_mount_point_saved = Some(old_repo_mount_point.clone()); + tracing::info!( + "[Task {}] Target discovery succeeded: {} targets", + id, + found_targets.len() + ); targets = found_targets; break; } @@ -733,6 +751,12 @@ pub async fn build( // Run buck2 build from the sub-project directory, not the monorepo root. // This ensures buck2 uses the sub-project's .buckconfig and PACKAGE files. let project_root = PathBuf::from(&mount_point).join(repo_prefix); + tracing::info!( + "[Task {}] Starting buck2 build. project_root={}, targets={}", + id, + project_root.display(), + targets.len() + ); let mut cmd = Command::new("buck2"); // --event-log and --build-report are used to collect build execution status // at target level (e.g. pending / running / succeeded / failed). @@ -756,7 +780,15 @@ pub async fn build( tracing::debug!("[Task {}] Executing command: {:?}", id, cmd); - let mut child = cmd.spawn()?; + let mut child = cmd.spawn().map_err(|e| { + tracing::error!( + "[Task {}] Failed to spawn buck2 (cwd={}): {}", + id, + project_root.display(), + e + ); + e + })?; if let Err(e) = sender.send(WSMessage::TaskPhaseUpdate { build_id: id.clone(), @@ -765,7 +797,13 @@ pub async fn build( tracing::error!("Failed to send RunningBuild phase update: {}", e); } - let target_build_track = start_build_status_tracker(&project_root, sender.clone(), cl_trimmed, task_id); + tracing::info!( + "[Task {}] Starting buck2 event-log tracker: file={}", + id, + project_root.join(EVENT_LOG_FILE).display() + ); + let target_build_track = + start_build_status_tracker(&project_root, sender.clone(), cl_trimmed, task_id); let stdout = child.stdout.take().unwrap(); let stderr = child.stderr.take().unwrap(); diff --git a/orion/src/ws.rs b/orion/src/ws.rs index 0ba4aa88b..16137a845 100644 --- a/orion/src/ws.rs +++ b/orion/src/ws.rs @@ -59,8 +59,8 @@ pub async fn run_client(server_addr: String, worker_id: String) { /// Processes an established WebSocket connection. /// /// Coordinates three concurrent tasks: -/// - Heartbeat transmission to maintain connection -/// - Message sending from internal channels +/// - Heartbeat on a timer with priority over build output +/// - Message sending from internal channels /// - Message receiving and processing from server /// /// # Arguments @@ -77,48 +77,66 @@ async fn handle_connection( let worker_id_clone = worker_id.clone(); let hostname_clone = server_addr.clone(); - let orion_version = env!("CARGO_PKG_VERSION").to_string(); // Get from Cargo.toml + let orion_version = env!("CARGO_PKG_VERSION").to_string(); + + let send_task = tokio::spawn(async move { + // Heartbeats must not sit behind unbounded build output/status batches: the server drops + // workers after ~90s without a Heartbeat (see orion-server health check). + let mut heartbeat_interval = tokio::time::interval(Duration::from_secs(30)); - let internal_tx_clone = internal_tx.clone(); - let heartbeat_task = tokio::spawn(async move { tracing::info!("Registering with worker ID: {}", worker_id_clone); - if internal_tx_clone - .send(WSMessage::Register { - id: worker_id_clone, - hostname: hostname_clone, - orion_version, - }) - .is_err() - { - tracing::error!("Failed to queue register message. Internal channel closed."); - return; - } - let heartbeat_interval = Duration::from_secs(30); - loop { - tokio::time::sleep(heartbeat_interval).await; - tracing::debug!("Sending heartbeat..."); - if internal_tx_clone.send(WSMessage::Heartbeat).is_err() { - tracing::warn!("Failed to queue heartbeat message. Internal channel closed."); - break; + let register = WSMessage::Register { + id: worker_id_clone, + hostname: hostname_clone, + orion_version, + }; + let register_str = match serde_json::to_string(®ister) { + Ok(s) => s, + Err(e) => { + tracing::error!("Failed to serialize Register: {}", e); + return; } + }; + let mut ws_sender = ws_sender; + if let Err(e) = ws_sender.send(Message::Text(register_str.into())).await { + tracing::error!( + "Failed to send Register to server: {}. Terminating send task.", + e + ); + return; } - }); - let mut ws_sender = ws_sender; - let send_task = tokio::spawn(async move { - while let Some(msg) = internal_rx.recv().await { - match serde_json::to_string(&msg) { - Ok(msg_str) => { - if let Err(e) = ws_sender.send(Message::Text(msg_str.into())).await { - tracing::error!( - "Failed to send message to server: {}. Terminating send task.", - e - ); - break; + loop { + tokio::select! { + biased; + _ = heartbeat_interval.tick() => { + tracing::info!("Sending heartbeat..."); + match serde_json::to_string(&WSMessage::Heartbeat) { + Ok(payload) => { + if let Err(e) = ws_sender.send(Message::Text(payload.into())).await { + tracing::warn!("Failed to send heartbeat on WebSocket: {}", e); + break; + } + } + Err(e) => { + tracing::error!("Failed to serialize Heartbeat: {}", e); + } } } - Err(e) => { - tracing::error!("Failed to serialize WSMessage: {}", e); + maybe_msg = internal_rx.recv() => { + let Some(msg) = maybe_msg else { break; }; + match serde_json::to_string(&msg) { + Ok(msg_str) => { + if let Err(e) = ws_sender.send(Message::Text(msg_str.into())).await { + tracing::error!( + "Failed to send message to server: {}. Terminating send task.", + e + ); + break; + } + } + Err(e) => tracing::error!("Failed to serialize WSMessage: {}", e), + } } } } @@ -138,7 +156,6 @@ async fn handle_connection( // Wait for any task to complete tokio::select! { - _ = heartbeat_task => tracing::info!("Heartbeat task finished."), _ = send_task => tracing::info!("Send task finished."), _ = recv_task => tracing::info!("Receive task finished."), } diff --git a/orion/systemd/orion-runner.service b/orion/systemd/orion-runner.service index a8384ca90..c9f76ca33 100644 --- a/orion/systemd/orion-runner.service +++ b/orion/systemd/orion-runner.service @@ -5,8 +5,8 @@ Wants=network-online.target [Service] Type=simple -User=orion -Group=orion +User=root +Group=root WorkingDirectory=/home/orion/orion-runner # Optional runtime config. Create this file on the VM to set SERVER_WS, @@ -27,6 +27,14 @@ ExecStart=/home/orion/orion-runner/run.sh Restart=always RestartSec=2 +# When FUSE/kernel I/O gets stuck, `orion` can become unkillable (D-state). Keep +# stop bounded so deploy/ops doesn't hang indefinitely. +TimeoutStopSec=20 +KillMode=control-group +KillSignal=SIGTERM +SendSIGKILL=yes +FinalKillSignal=SIGKILL + # Grant capabilities for FUSE mounting and read-search bypass AmbientCapabilities=CAP_SYS_ADMIN CapabilityBoundingSet=CAP_SYS_ADMIN CAP_DAC_OVERRIDE CAP_DAC_READ_SEARCH