From d7c8137c2dad383eb243b9f36d9868ae279e2c6e Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Tue, 12 Aug 2025 18:54:03 +0800 Subject: [PATCH 1/4] [orion-server] feat: add buck2 log segment function Signed-off-by: Xiaoyang Han --- orion-server/Cargo.toml | 5 +- orion-server/src/api.rs | 44 +++++++ orion-server/src/lib.rs | 5 + orion-server/src/scheduler.rs | 158 +++++++++++++++++++++++++- orion-server/src/server.rs | 9 +- orion-server/tests/log_segment_api.rs | 48 ++++++++ 6 files changed, 263 insertions(+), 6 deletions(-) create mode 100644 orion-server/src/lib.rs create mode 100644 orion-server/tests/log_segment_api.rs diff --git a/orion-server/Cargo.toml b/orion-server/Cargo.toml index 654915fec..965830b1b 100644 --- a/orion-server/Cargo.toml +++ b/orion-server/Cargo.toml @@ -32,4 +32,7 @@ utoipa.workspace = true utoipa-swagger-ui = { workspace = true, features = ["axum"] } chrono = { version = "0.4", features = ["serde"] } reqwest.workspace = true -thiserror = "2.0" \ No newline at end of file +thiserror = "2.0" + +[dev-dependencies] +tempfile = "3" \ No newline at end of file diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index 0301efc19..415dd8a71 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -3,6 +3,7 @@ use crate::scheduler::{ BuildInfo, BuildRequest, TaskQueueStats, TaskScheduler, WorkerInfo, WorkerStatus, create_log_file, get_build_log_dir, }; +use crate::scheduler::{LogSegment, LogReadError}; use axum::{ Json, Router, extract::{ @@ -84,6 +85,7 @@ pub fn routers() -> Router { .route("/task", axum::routing::post(task_handler)) .route("/task-status/{id}", get(task_status_handler)) .route("/task-output/{id}", get(task_output_handler)) + .route("/task-output-segment/{id}", get(task_output_segment_handler)) .route("/mr-task/{mr}", get(task_query_by_mr)) .route("/queue-stats", get(queue_stats_handler)) } @@ -243,6 +245,48 @@ pub async fn task_output_handler( Ok(Sse::new(stream.boxed()).keep_alive(KeepAlive::new())) } +#[utoipa::path( + get, + path = "/task-output-segment/{id}", + params( + ("id" = String, Path, description = "Task ID whose log to read"), + ("offset" = u64, Query, description = "Start byte offset", example = 0), + ("len" = usize, Query, description = "Max bytes to read", example = 4096) + ), + responses( + (status = 200, description = "Log segment", body = LogSegment), + (status = 404, description = "Log file not found"), + (status = 416, description = "Offset out of range"), + (status = 400, description = "Invalid parameters") + ) +)] +pub async fn task_output_segment_handler( + State(state): State, + Path(id): Path, + axum::extract::Query(params): axum::extract::Query>, +) -> impl IntoResponse { + let offset = params.get("offset").and_then(|v| v.parse::().ok()).unwrap_or(0); + let len = params.get("len").and_then(|v| v.parse::().ok()).unwrap_or(4096); + // Cap len to reasonable maximum (e.g., 1MB) + let len = len.min(1024 * 1024); + + match state.scheduler.read_log_segment(&id, offset, len).await { + Ok(seg) => (StatusCode::OK, Json(seg)).into_response(), + Err(LogReadError::NotFound) => StatusCode::NOT_FOUND.into_response(), + Err(LogReadError::OffsetOutOfRange { size }) => ( + StatusCode::RANGE_NOT_SATISFIABLE, + Json(serde_json::json!({ + "message": "Offset out of range", + "file_size": size + })) + ).into_response(), + Err(LogReadError::Io(e)) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({"message": format!("IO error: {e}")})) + ).into_response(), + } +} + #[utoipa::path( post, path = "/task", diff --git a/orion-server/src/lib.rs b/orion-server/src/lib.rs new file mode 100644 index 000000000..045891432 --- /dev/null +++ b/orion-server/src/lib.rs @@ -0,0 +1,5 @@ +pub mod api; +pub mod buck2; +pub mod model; +pub mod scheduler; +pub mod server; diff --git a/orion-server/src/scheduler.rs b/orion-server/src/scheduler.rs index 864c2b94c..c79fa61f1 100644 --- a/orion-server/src/scheduler.rs +++ b/orion-server/src/scheduler.rs @@ -8,6 +8,8 @@ use std::collections::VecDeque; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::{Mutex, Notify, mpsc::UnboundedSender}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; +use tokio::io::SeekFrom; use utoipa::ToSchema; use uuid::Uuid; @@ -164,6 +166,37 @@ pub struct TaskScheduler { pub conn: DatabaseConnection, } +/// Log segment read result +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct LogSegment { + /// Task id / log file name + pub task_id: String, + /// Requested starting offset + pub offset: u64, + /// Bytes actually read + pub len: usize, + /// UTF-8 (lossy) decoded data slice + pub data: String, + /// Next offset (offset + len) + pub next_offset: u64, + /// Total file size in bytes + pub file_size: u64, + /// Whether we reached end of file + pub eof: bool, +} + +/// Errors when reading a log segment +#[derive(Debug)] +pub enum LogReadError { + NotFound, + OffsetOutOfRange { size: u64 }, + Io(std::io::Error), +} + +impl From for LogReadError { + fn from(e: std::io::Error) -> Self { Self::Io(e) } +} + impl TaskScheduler { /// Create new task scheduler instance pub fn new( @@ -182,6 +215,16 @@ impl TaskScheduler { } } + /// Read a segment of a log file by task id at given offset, limited to max_len bytes. + pub async fn read_log_segment( + &self, + task_id: &str, + offset: u64, + max_len: usize, + ) -> Result { + read_log_segment_raw(task_id, offset, max_len).await + } + /// Add task to queue pub async fn enqueue_task( &self, @@ -431,12 +474,87 @@ impl TaskScheduler { } } +/// Read a segment of a task log file. +/// Returns metadata and data slice (UTF-8 lossy converted). +pub async fn read_log_segment_raw( + task_id: &str, + offset: u64, + max_len: usize, +) -> Result { + let log_path = format!("{}/{}", get_build_log_dir(), task_id); + let path = std::path::Path::new(&log_path); + if !path.exists() { return Err(LogReadError::NotFound); } + + let meta = tokio::fs::metadata(path).await.map_err(LogReadError::Io)?; + let size = meta.len(); + if offset > size { return Err(LogReadError::OffsetOutOfRange { size }); } + + // Fast path: only metadata + if max_len == 0 || offset == size { + return Ok(LogSegment { + task_id: task_id.to_string(), + offset, + len: 0, + data: String::new(), + next_offset: offset, + file_size: size, + eof: offset >= size, + }); + } + + let mut file = tokio::fs::File::open(path).await.map_err(LogReadError::Io)?; + file.seek(SeekFrom::Start(offset)).await.map_err(LogReadError::Io)?; + + let remaining = (size - offset) as usize; + let to_read = remaining.min(max_len); + let mut buf = vec![0u8; to_read]; + let read_bytes = file.read(&mut buf).await.map_err(LogReadError::Io)?; + buf.truncate(read_bytes); + let data = String::from_utf8_lossy(&buf).to_string(); + let next_offset = offset + read_bytes as u64; + let eof = next_offset >= size; + + Ok(LogSegment { + task_id: task_id.to_string(), + offset, + len: read_bytes, + data, + next_offset, + file_size: size, + eof, + }) +} + /// Get build log directory pub fn get_build_log_dir() -> &'static str { - use once_cell::sync::Lazy; - static BUILD_LOG_DIR: Lazy = - Lazy::new(|| std::env::var("BUILD_LOG_DIR").expect("BUILD_LOG_DIR must be set")); - &BUILD_LOG_DIR + #[cfg(not(test))] + { + use once_cell::sync::Lazy; + static BUILD_LOG_DIR: Lazy = Lazy::new(|| { + std::env::var("BUILD_LOG_DIR").expect("BUILD_LOG_DIR must be set") + }); + &BUILD_LOG_DIR + } + #[cfg(test)] + { + // In tests allow changing BUILD_LOG_DIR per test case. + thread_local! { + static BUILD_LOG_DIR_TLS: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; + } + // Read from env each call and cache in TLS for &'static str simulation. + // We leak the String to extend lifetime intentionally (test scope only). + BUILD_LOG_DIR_TLS.with(|cell| { + if cell.borrow().is_none() { + let val = std::env::var("BUILD_LOG_DIR").expect("BUILD_LOG_DIR must be set"); + let leaked: &'static str = Box::leak(val.into_boxed_str()); + *cell.borrow_mut() = Some(leaked.to_string()); + } + let current = cell.borrow(); + let s = current.as_ref().unwrap(); + // Leak clone to satisfy 'static return each call. + Box::leak(s.clone().into_boxed_str()) + }) + } } /// Create log file @@ -459,6 +577,7 @@ pub fn create_log_file(task_id: &str) -> Result { #[cfg(test)] mod tests { use super::*; + use std::io::Write; /// Test task queue basic functionality #[test] @@ -536,4 +655,35 @@ mod tests { // Should fail when full assert!(queue.enqueue(task).is_err()); } + + #[tokio::test] + async fn test_read_log_segment_basic() { + // Prepare temp dir + let tmp = tempfile::tempdir().unwrap(); + unsafe { std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); } + let task_id = "segment-test"; + let mut file = create_log_file(task_id).unwrap(); + write!(file, "Hello World! This is a test log.").unwrap(); + + // Read first 5 bytes + let seg = read_log_segment_raw(task_id, 0, 5).await.unwrap(); + assert_eq!(seg.offset, 0); + assert_eq!(seg.len, 5); + assert_eq!(seg.data, "Hello"); + assert!(!seg.eof); + + // Read next bytes + let seg2 = read_log_segment_raw(task_id, seg.next_offset, 100).await.unwrap(); + assert!(seg2.data.starts_with(" World")); + } + + #[tokio::test] + async fn test_read_log_segment_offset_out_of_range() { + let tmp = tempfile::tempdir().unwrap(); + unsafe { std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); } + let task_id = "segment-oob"; + let _ = create_log_file(task_id).unwrap(); + let res = read_log_segment_raw(task_id, 10, 10).await; + assert!(matches!(res, Err(LogReadError::OffsetOutOfRange { .. }))); + } } diff --git a/orion-server/src/server.rs b/orion-server/src/server.rs index 4bf1f124a..a72648dc9 100644 --- a/orion-server/src/server.rs +++ b/orion-server/src/server.rs @@ -19,10 +19,17 @@ use utoipa_swagger_ui::SwaggerUi; api::task_handler, api::task_status_handler, api::task_output_handler, + api::task_output_segment_handler, api::task_query_by_mr, ), components( - schemas(crate::scheduler::BuildRequest, api::TaskStatus, api::TaskStatusEnum, api::BuildDTO) + schemas( + crate::scheduler::BuildRequest, + crate::scheduler::LogSegment, + api::TaskStatus, + api::TaskStatusEnum, + api::BuildDTO + ) ), tags( (name = "Build", description = "Build related endpoints") diff --git a/orion-server/tests/log_segment_api.rs b/orion-server/tests/log_segment_api.rs new file mode 100644 index 000000000..e0100fb9f --- /dev/null +++ b/orion-server/tests/log_segment_api.rs @@ -0,0 +1,48 @@ +use orion_server::scheduler::{create_log_file, read_log_segment_raw, LogReadError}; +use std::io::Write; +use once_cell::sync::Lazy; +use std::sync::Mutex; + +static TEST_LOG_DIR: Lazy>> = Lazy::new(|| Mutex::new(None)); + +fn init_log_dir() { + let mut guard = TEST_LOG_DIR.lock().unwrap(); + if guard.is_none() { + let tmp = tempfile::tempdir().expect("temp dir"); + unsafe { std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); } + *guard = Some(tmp); + } +} + +fn write_log(task_id: &str, content: &str) { + init_log_dir(); + let mut f = create_log_file(task_id).expect("create log file"); + write!(f, "{content}").unwrap(); +} + +fn create_empty_log(task_id: &str) { init_log_dir(); let _ = create_log_file(task_id).unwrap(); } + +#[tokio::test] +async fn test_log_segment_read_basic() { + write_log("segment-basic", "Hello Log Segment Test!"); + let seg = read_log_segment_raw("segment-basic", 0, 5).await.expect("segment"); + assert_eq!(seg.data, "Hello"); + let seg2 = read_log_segment_raw("segment-basic", seg.next_offset, 1024).await.expect("segment2"); + assert!(seg2.data.starts_with(" Log Segment")); +} + +#[tokio::test] +async fn test_log_segment_offset_out_of_range() { + create_empty_log("segment-oob"); + let res = read_log_segment_raw("segment-oob", 10, 10).await; + assert!(matches!(res, Err(LogReadError::OffsetOutOfRange { .. }))); +} + +#[tokio::test] +async fn test_log_segment_zero_len_metadata() { + write_log("segment-zero", "1234567890"); + let seg = read_log_segment_raw("segment-zero", 0, 0).await.expect("segment"); + assert_eq!(seg.len, 0); + assert_eq!(seg.file_size, 10); + assert!(!seg.eof); +} From 0714d8e55a198976603b5b2d98864bba1157a330 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 13 Aug 2025 09:56:23 +0800 Subject: [PATCH 2/4] [orion-server]: fix box leak Signed-off-by: Xiaoyang Han --- orion-server/src/scheduler.rs | 52 +++++++++++++++++++++++++++-------- 1 file changed, 41 insertions(+), 11 deletions(-) diff --git a/orion-server/src/scheduler.rs b/orion-server/src/scheduler.rs index c79fa61f1..dd0078567 100644 --- a/orion-server/src/scheduler.rs +++ b/orion-server/src/scheduler.rs @@ -525,8 +525,42 @@ pub async fn read_log_segment_raw( }) } -/// Get build log directory +/// Unified accessor for the build log directory (BUILD_LOG_DIR). +/// +/// Behavior differs between test and non-test builds: +/// +/// Non-test (`cfg(not(test))`): +/// * Uses `once_cell::sync::Lazy` to read the env var exactly once at first access. +/// * Panics early if the variable is missing (surfacing deployment misconfiguration). +/// * Cannot be changed at runtime (subsequent env var edits are ignored). +/// +/// Test (`cfg(test)`): +/// * Allows setting `BUILD_LOG_DIR` before the first call in each test thread. +/// * Uses `thread_local!` + `Cell>`; on first access leaks the string via `Box::leak` only once per thread (bounded leak acceptable in tests). +/// * Motivation: +/// - Avoid a global `Lazy` capturing a temporary directory too early for all tests. +/// - Keep memory growth bounded (one leaked string per thread at most). +/// * Changing the environment variable in the same thread after first access has no effect. +/// +/// Usage in tests: +/// ```ignore +/// let tmp = tempfile::tempdir().unwrap(); +/// std::env::set_var("BUILD_LOG_DIR", tmp.path()); +/// let dir = get_build_log_dir(); +/// ``` +/// +/// # Panics +/// Panics if `BUILD_LOG_DIR` is not set at first access. +/// +/// # Thread Safety +/// Returns an immutable `&'static str`. Non-test mode uses a `Lazy` (thread-safe once init); +/// test mode uses per-thread initialization to avoid cross-thread contention / early capture. +/// +/// # Possible Future Improvement +/// If hot-swapping the directory is ever required, this could return `Arc` and expose +/// an atomic update mechanism. Current requirements favor simplicity and immutability. pub fn get_build_log_dir() -> &'static str { + // Body only distinguishes cfg paths; see doc comment above for detailed rationale. #[cfg(not(test))] { use once_cell::sync::Lazy; @@ -537,22 +571,18 @@ pub fn get_build_log_dir() -> &'static str { } #[cfg(test)] { - // In tests allow changing BUILD_LOG_DIR per test case. + // Test mode: allow setting BUILD_LOG_DIR before first use; only leak once per thread. + use std::cell::Cell; thread_local! { - static BUILD_LOG_DIR_TLS: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; + static BUILD_LOG_DIR_TLS: Cell> = Cell::new(None); } - // Read from env each call and cache in TLS for &'static str simulation. - // We leak the String to extend lifetime intentionally (test scope only). BUILD_LOG_DIR_TLS.with(|cell| { - if cell.borrow().is_none() { + if cell.get().is_none() { let val = std::env::var("BUILD_LOG_DIR").expect("BUILD_LOG_DIR must be set"); let leaked: &'static str = Box::leak(val.into_boxed_str()); - *cell.borrow_mut() = Some(leaked.to_string()); + cell.set(Some(leaked)); } - let current = cell.borrow(); - let s = current.as_ref().unwrap(); - // Leak clone to satisfy 'static return each call. - Box::leak(s.clone().into_boxed_str()) + cell.get().unwrap() }) } } From be887860b55b44482772a7f581b1e2f0906d23cd Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 13 Aug 2025 10:57:41 +0800 Subject: [PATCH 3/4] [orion-server]: fix typo Signed-off-by: Xiaoyang Han --- orion-server/src/api.rs | 27 ++++++++++++----- orion-server/src/scheduler.rs | 43 ++++++++++++++++++--------- orion-server/tests/log_segment_api.rs | 25 +++++++++++----- 3 files changed, 66 insertions(+), 29 deletions(-) diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index 415dd8a71..4b1e4ee37 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -3,7 +3,7 @@ use crate::scheduler::{ BuildInfo, BuildRequest, TaskQueueStats, TaskScheduler, WorkerInfo, WorkerStatus, create_log_file, get_build_log_dir, }; -use crate::scheduler::{LogSegment, LogReadError}; +use crate::scheduler::{LogReadError, LogSegment}; use axum::{ Json, Router, extract::{ @@ -85,7 +85,10 @@ pub fn routers() -> Router { .route("/task", axum::routing::post(task_handler)) .route("/task-status/{id}", get(task_status_handler)) .route("/task-output/{id}", get(task_output_handler)) - .route("/task-output-segment/{id}", get(task_output_segment_handler)) + .route( + "/task-output-segment/{id}", + get(task_output_segment_handler), + ) .route("/mr-task/{mr}", get(task_query_by_mr)) .route("/queue-stats", get(queue_stats_handler)) } @@ -265,8 +268,14 @@ pub async fn task_output_segment_handler( Path(id): Path, axum::extract::Query(params): axum::extract::Query>, ) -> impl IntoResponse { - let offset = params.get("offset").and_then(|v| v.parse::().ok()).unwrap_or(0); - let len = params.get("len").and_then(|v| v.parse::().ok()).unwrap_or(4096); + let offset = params + .get("offset") + .and_then(|v| v.parse::().ok()) + .unwrap_or(0); + let len = params + .get("len") + .and_then(|v| v.parse::().ok()) + .unwrap_or(4096); // Cap len to reasonable maximum (e.g., 1MB) let len = len.min(1024 * 1024); @@ -278,12 +287,14 @@ pub async fn task_output_segment_handler( Json(serde_json::json!({ "message": "Offset out of range", "file_size": size - })) - ).into_response(), + })), + ) + .into_response(), Err(LogReadError::Io(e)) => ( StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"message": format!("IO error: {e}")})) - ).into_response(), + Json(serde_json::json!({"message": format!("IO error: {e}")})), + ) + .into_response(), } } diff --git a/orion-server/src/scheduler.rs b/orion-server/src/scheduler.rs index dd0078567..d322b329d 100644 --- a/orion-server/src/scheduler.rs +++ b/orion-server/src/scheduler.rs @@ -7,9 +7,9 @@ use serde::{Deserialize, Serialize}; use std::collections::VecDeque; use std::sync::Arc; use std::time::{Duration, Instant}; -use tokio::sync::{Mutex, Notify, mpsc::UnboundedSender}; -use tokio::io::{AsyncReadExt, AsyncSeekExt}; use tokio::io::SeekFrom; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; +use tokio::sync::{Mutex, Notify, mpsc::UnboundedSender}; use utoipa::ToSchema; use uuid::Uuid; @@ -194,7 +194,9 @@ pub enum LogReadError { } impl From for LogReadError { - fn from(e: std::io::Error) -> Self { Self::Io(e) } + fn from(e: std::io::Error) -> Self { + Self::Io(e) + } } impl TaskScheduler { @@ -483,11 +485,15 @@ pub async fn read_log_segment_raw( ) -> Result { let log_path = format!("{}/{}", get_build_log_dir(), task_id); let path = std::path::Path::new(&log_path); - if !path.exists() { return Err(LogReadError::NotFound); } + if !path.exists() { + return Err(LogReadError::NotFound); + } let meta = tokio::fs::metadata(path).await.map_err(LogReadError::Io)?; let size = meta.len(); - if offset > size { return Err(LogReadError::OffsetOutOfRange { size }); } + if offset > size { + return Err(LogReadError::OffsetOutOfRange { size }); + } // Fast path: only metadata if max_len == 0 || offset == size { @@ -502,8 +508,12 @@ pub async fn read_log_segment_raw( }); } - let mut file = tokio::fs::File::open(path).await.map_err(LogReadError::Io)?; - file.seek(SeekFrom::Start(offset)).await.map_err(LogReadError::Io)?; + let mut file = tokio::fs::File::open(path) + .await + .map_err(LogReadError::Io)?; + file.seek(SeekFrom::Start(offset)) + .await + .map_err(LogReadError::Io)?; let remaining = (size - offset) as usize; let to_read = remaining.min(max_len); @@ -564,9 +574,8 @@ pub fn get_build_log_dir() -> &'static str { #[cfg(not(test))] { use once_cell::sync::Lazy; - static BUILD_LOG_DIR: Lazy = Lazy::new(|| { - std::env::var("BUILD_LOG_DIR").expect("BUILD_LOG_DIR must be set") - }); + static BUILD_LOG_DIR: Lazy = + Lazy::new(|| std::env::var("BUILD_LOG_DIR").expect("BUILD_LOG_DIR must be set")); &BUILD_LOG_DIR } #[cfg(test)] @@ -574,7 +583,7 @@ pub fn get_build_log_dir() -> &'static str { // Test mode: allow setting BUILD_LOG_DIR before first use; only leak once per thread. use std::cell::Cell; thread_local! { - static BUILD_LOG_DIR_TLS: Cell> = Cell::new(None); + static BUILD_LOG_DIR_TLS: Cell> = const { Cell::new(None) }; } BUILD_LOG_DIR_TLS.with(|cell| { if cell.get().is_none() { @@ -690,7 +699,9 @@ mod tests { async fn test_read_log_segment_basic() { // Prepare temp dir let tmp = tempfile::tempdir().unwrap(); - unsafe { std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); } + unsafe { + std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); + } let task_id = "segment-test"; let mut file = create_log_file(task_id).unwrap(); write!(file, "Hello World! This is a test log.").unwrap(); @@ -703,14 +714,18 @@ mod tests { assert!(!seg.eof); // Read next bytes - let seg2 = read_log_segment_raw(task_id, seg.next_offset, 100).await.unwrap(); + let seg2 = read_log_segment_raw(task_id, seg.next_offset, 100) + .await + .unwrap(); assert!(seg2.data.starts_with(" World")); } #[tokio::test] async fn test_read_log_segment_offset_out_of_range() { let tmp = tempfile::tempdir().unwrap(); - unsafe { std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); } + unsafe { + std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); + } let task_id = "segment-oob"; let _ = create_log_file(task_id).unwrap(); let res = read_log_segment_raw(task_id, 10, 10).await; diff --git a/orion-server/tests/log_segment_api.rs b/orion-server/tests/log_segment_api.rs index e0100fb9f..b5f0d48d0 100644 --- a/orion-server/tests/log_segment_api.rs +++ b/orion-server/tests/log_segment_api.rs @@ -1,6 +1,6 @@ -use orion_server::scheduler::{create_log_file, read_log_segment_raw, LogReadError}; -use std::io::Write; use once_cell::sync::Lazy; +use orion_server::scheduler::{LogReadError, create_log_file, read_log_segment_raw}; +use std::io::Write; use std::sync::Mutex; static TEST_LOG_DIR: Lazy>> = Lazy::new(|| Mutex::new(None)); @@ -9,7 +9,9 @@ fn init_log_dir() { let mut guard = TEST_LOG_DIR.lock().unwrap(); if guard.is_none() { let tmp = tempfile::tempdir().expect("temp dir"); - unsafe { std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); } + unsafe { + std::env::set_var("BUILD_LOG_DIR", tmp.path().to_str().unwrap()); + } *guard = Some(tmp); } } @@ -20,14 +22,21 @@ fn write_log(task_id: &str, content: &str) { write!(f, "{content}").unwrap(); } -fn create_empty_log(task_id: &str) { init_log_dir(); let _ = create_log_file(task_id).unwrap(); } +fn create_empty_log(task_id: &str) { + init_log_dir(); + let _ = create_log_file(task_id).unwrap(); +} #[tokio::test] async fn test_log_segment_read_basic() { write_log("segment-basic", "Hello Log Segment Test!"); - let seg = read_log_segment_raw("segment-basic", 0, 5).await.expect("segment"); + let seg = read_log_segment_raw("segment-basic", 0, 5) + .await + .expect("segment"); assert_eq!(seg.data, "Hello"); - let seg2 = read_log_segment_raw("segment-basic", seg.next_offset, 1024).await.expect("segment2"); + let seg2 = read_log_segment_raw("segment-basic", seg.next_offset, 1024) + .await + .expect("segment2"); assert!(seg2.data.starts_with(" Log Segment")); } @@ -41,7 +50,9 @@ async fn test_log_segment_offset_out_of_range() { #[tokio::test] async fn test_log_segment_zero_len_metadata() { write_log("segment-zero", "1234567890"); - let seg = read_log_segment_raw("segment-zero", 0, 0).await.expect("segment"); + let seg = read_log_segment_raw("segment-zero", 0, 0) + .await + .expect("segment"); assert_eq!(seg.len, 0); assert_eq!(seg.file_size, 10); assert!(!seg.eof); From db24f0c3877dda8ca274ee2b1d7482c52f4284b9 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 13 Aug 2025 11:10:50 +0800 Subject: [PATCH 4/4] [orion-server] tasks' status interface Signed-off-by: Xiaoyang Han --- ceres/src/api_service/mono_api_service.rs | 163 +++++++++++++++------- orion-server/src/api.rs | 85 ++++++++++- 2 files changed, 193 insertions(+), 55 deletions(-) diff --git a/ceres/src/api_service/mono_api_service.rs b/ceres/src/api_service/mono_api_service.rs index e4c8731f3..0328ec998 100644 --- a/ceres/src/api_service/mono_api_service.rs +++ b/ceres/src/api_service/mono_api_service.rs @@ -44,7 +44,6 @@ use async_trait::async_trait; use callisto::sea_orm_active_enums::ConvTypeEnum; use callisto::{mega_blob, mega_mr, mega_tree, raw_blob}; use common::errors::MegaError; -use neptune::neptune_engine::Diff; use jupiter::storage::base_storage::StorageConnector; use jupiter::storage::Storage; use jupiter::utils::converter::generate_git_keep_with_timestamp; @@ -53,11 +52,11 @@ use mercury::hash::SHA1; use mercury::internal::object::blob::Blob; use mercury::internal::object::commit::Commit; use mercury::internal::object::tree::{Tree, TreeItem, TreeItemMode}; +use neptune::neptune_engine::Diff; use crate::api_service::ApiHandler; use crate::model::git::CreateFileInfo; -use crate::model::mr::{MrDiffFile, MrDiff, MrPageInfo}; - +use crate::model::mr::{MrDiff, MrDiffFile, MrPageInfo}; #[derive(Clone)] pub struct MonoApiService { @@ -387,18 +386,23 @@ impl MonoApiService { ) -> Result { // old and new blobs for comparison let stg = self.storage.mr_storage(); - let mr = stg.get_mr(mr_link).await.unwrap().ok_or_else(|| { - GitError::CustomError(format!("Merge request not found: {mr_link}")) - })?; - let old_blobs = self.get_commit_blobs(&mr.from_hash).await.map_err(|e| { - GitError::CustomError(format!("Failed to get old commit blobs: {e}")) - })?; - let new_blobs = self.get_commit_blobs(&mr.to_hash).await.map_err(|e| { - GitError::CustomError(format!("Failed to get new commit blobs: {e}")) - })?; + let mr = + stg.get_mr(mr_link).await.unwrap().ok_or_else(|| { + GitError::CustomError(format!("Merge request not found: {mr_link}")) + })?; + let old_blobs = self + .get_commit_blobs(&mr.from_hash) + .await + .map_err(|e| GitError::CustomError(format!("Failed to get old commit blobs: {e}")))?; + let new_blobs = self + .get_commit_blobs(&mr.to_hash) + .await + .map_err(|e| GitError::CustomError(format!("Failed to get new commit blobs: {e}")))?; // calculate pages - let sorted_changed_files = self.mr_files_list(old_blobs.clone(), new_blobs.clone()).await?; + let sorted_changed_files = self + .mr_files_list(old_blobs.clone(), new_blobs.clone()) + .await?; // ensure page_id is within bounds let start = (page_id.saturating_sub(1)) * page_size; @@ -441,7 +445,7 @@ impl MonoApiService { } // Simple synchronous closure that uses the pre-fetched cache - let read_content = |_file: &PathBuf, hash: &SHA1| -> Vec{ + let read_content = |_file: &PathBuf, hash: &SHA1| -> Vec { blob_cache.get(hash).cloned().unwrap_or_default() }; @@ -452,15 +456,16 @@ impl MonoApiService { "histogram".to_string(), Vec::new(), read_content, - ).await; + ) + .await; Ok(MrDiff { data: diff_output, page_info: Some(MrPageInfo { - total_pages: (sorted_changed_files.len()-1).div_ceil(page_size), + total_pages: (sorted_changed_files.len() - 1).div_ceil(page_size), current_page: page_id, page_size, - }) + }), }) } @@ -517,7 +522,9 @@ impl MonoApiService { // Sort the results res.sort_by(|a, b| { - a.path().cmp(b.path()).then_with(|| a.kind_weight().cmp(&b.kind_weight())) + a.path() + .cmp(b.path()) + .then_with(|| a.kind_weight().cmp(&b.kind_weight())) }); Ok(res) } @@ -566,10 +573,10 @@ impl MonoApiService { #[cfg(test)] mod test { use super::*; - use std::path::PathBuf; + use crate::model::mr::{MrDiffFile, MrPageInfo}; use mercury::hash::SHA1; + use std::path::PathBuf; use std::str::FromStr; - use crate::model::mr::{MrDiffFile, MrPageInfo}; #[test] pub fn test_path() { @@ -585,9 +592,19 @@ mod test { #[test] fn test_paging_calculation_basic() { let files: Vec = vec![ - MrDiffFile::New(PathBuf::from("file1.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - MrDiffFile::Modified(PathBuf::from("file2.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap()), - MrDiffFile::Deleted(PathBuf::from("file3.txt"), SHA1::from_str("1111111111111111111111111111111111111111").unwrap()), + MrDiffFile::New( + PathBuf::from("file1.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + ), + MrDiffFile::Modified( + PathBuf::from("file2.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap(), + ), + MrDiffFile::Deleted( + PathBuf::from("file3.txt"), + SHA1::from_str("1111111111111111111111111111111111111111").unwrap(), + ), ]; let page_size = 2u32; @@ -613,10 +630,23 @@ mod test { #[test] fn test_paging_calculation_second_page() { let files: Vec = vec![ - MrDiffFile::New(PathBuf::from("file1.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - MrDiffFile::Modified(PathBuf::from("file2.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap()), - MrDiffFile::Deleted(PathBuf::from("file3.txt"), SHA1::from_str("1111111111111111111111111111111111111111").unwrap()), - MrDiffFile::New(PathBuf::from("file4.txt"), SHA1::from_str("2222222222222222222222222222222222222222").unwrap()), + MrDiffFile::New( + PathBuf::from("file1.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + ), + MrDiffFile::Modified( + PathBuf::from("file2.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap(), + ), + MrDiffFile::Deleted( + PathBuf::from("file3.txt"), + SHA1::from_str("1111111111111111111111111111111111111111").unwrap(), + ), + MrDiffFile::New( + PathBuf::from("file4.txt"), + SHA1::from_str("2222222222222222222222222222222222222222").unwrap(), + ), ]; let page_size = 2u32; @@ -644,9 +674,19 @@ mod test { #[test] fn test_paging_calculation_partial_page() { let files: Vec = vec![ - MrDiffFile::New(PathBuf::from("file1.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - MrDiffFile::Modified(PathBuf::from("file2.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap()), - MrDiffFile::Deleted(PathBuf::from("file3.txt"), SHA1::from_str("1111111111111111111111111111111111111111").unwrap()), + MrDiffFile::New( + PathBuf::from("file1.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + ), + MrDiffFile::Modified( + PathBuf::from("file2.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap(), + ), + MrDiffFile::Deleted( + PathBuf::from("file3.txt"), + SHA1::from_str("1111111111111111111111111111111111111111").unwrap(), + ), ]; let page_size = 5u32; @@ -671,9 +711,10 @@ mod test { #[test] fn test_paging_calculation_out_of_bounds() { - let files: Vec = vec![ - MrDiffFile::New(PathBuf::from("file1.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - ]; + let files: Vec = vec![MrDiffFile::New( + PathBuf::from("file1.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + )]; let page_size = 2u32; let page_id = 3u32; // Page that doesn't exist @@ -697,9 +738,10 @@ mod test { #[test] fn test_paging_calculation_edge_case_zero_page_size() { - let files: Vec = vec![ - MrDiffFile::New(PathBuf::from("file1.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - ]; + let files: Vec = vec![MrDiffFile::New( + PathBuf::from("file1.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + )]; let page_size = 0u32; let page_id = 1u32; @@ -724,8 +766,15 @@ mod test { #[test] fn test_paging_calculation_zero_page_id() { let files: Vec = vec![ - MrDiffFile::New(PathBuf::from("file1.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - MrDiffFile::Modified(PathBuf::from("file2.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap()), + MrDiffFile::New( + PathBuf::from("file1.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + ), + MrDiffFile::Modified( + PathBuf::from("file2.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap(), + ), ]; let page_size = 2u32; @@ -771,9 +820,10 @@ mod test { storage: Storage::mock(), }; - let files = vec![ - MrDiffFile::New(PathBuf::from("new_file.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - ]; + let files = vec![MrDiffFile::New( + PathBuf::from("new_file.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + )]; let mut old_blobs = Vec::new(); let mut new_blobs = Vec::new(); @@ -791,9 +841,10 @@ mod test { storage: Storage::mock(), }; - let files = vec![ - MrDiffFile::Deleted(PathBuf::from("deleted_file.txt"), SHA1::from_str("1234567890123456789012345678901234567890").unwrap()), - ]; + let files = vec![MrDiffFile::Deleted( + PathBuf::from("deleted_file.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + )]; let mut old_blobs = Vec::new(); let mut new_blobs = Vec::new(); @@ -811,13 +862,11 @@ mod test { storage: Storage::mock(), }; - let files = vec![ - MrDiffFile::Modified( - PathBuf::from("modified_file.txt"), - SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), - SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap() - ), - ]; + let files = vec![MrDiffFile::Modified( + PathBuf::from("modified_file.txt"), + SHA1::from_str("1234567890123456789012345678901234567890").unwrap(), + SHA1::from_str("abcdefabcdefabcdefabcdefabcdefabcdefabcd").unwrap(), + )]; let mut old_blobs = Vec::new(); let mut new_blobs = Vec::new(); @@ -837,12 +886,18 @@ mod test { }; let files = vec![ - MrDiffFile::New(PathBuf::from("new.txt"), SHA1::from_str("1111111111111111111111111111111111111111").unwrap()), - MrDiffFile::Deleted(PathBuf::from("deleted.txt"), SHA1::from_str("2222222222222222222222222222222222222222").unwrap()), + MrDiffFile::New( + PathBuf::from("new.txt"), + SHA1::from_str("1111111111111111111111111111111111111111").unwrap(), + ), + MrDiffFile::Deleted( + PathBuf::from("deleted.txt"), + SHA1::from_str("2222222222222222222222222222222222222222").unwrap(), + ), MrDiffFile::Modified( PathBuf::from("modified.txt"), SHA1::from_str("3333333333333333333333333333333333333333").unwrap(), - SHA1::from_str("4444444444444444444444444444444444444444").unwrap() + SHA1::from_str("4444444444444444444444444444444444444444").unwrap(), ), ]; diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index 4b1e4ee37..b2f6a42c4 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -39,6 +39,8 @@ use uuid::Uuid; /// Enumeration of possible task statuses #[derive(Debug, Serialize, Default, ToSchema)] pub enum TaskStatusEnum { + /// Task is queued and waiting to be assigned to a worker + Pending, Building, Interrupted, // Task was interrupted, exit code is None Failed, @@ -90,6 +92,7 @@ pub fn routers() -> Router { get(task_output_segment_handler), ) .route("/mr-task/{mr}", get(task_query_by_mr)) + .route("/tasks", get(tasks_handler)) .route("/queue-stats", get(queue_stats_handler)) } @@ -146,7 +149,8 @@ pub async fn task_status_handler( Some(model) => { // Determine task status based on database fields let status = if model.end_at.is_none() { - TaskStatusEnum::Building + // Not in active_builds and end_at is None => still queued (pending) + TaskStatusEnum::Pending } else if model.exit_code.is_none() { TaskStatusEnum::Interrupted } else if model.exit_code.unwrap() == 0 { @@ -773,6 +777,85 @@ pub async fn task_query_by_mr( } } +/// Task information including current status +#[derive(Debug, Serialize, ToSchema)] +pub struct TaskInfoDTO { + pub build_id: String, + pub output_file: String, + pub exit_code: Option, + pub start_at: String, + pub end_at: Option, + pub repo_name: String, + pub target: String, + pub arguments: String, + pub mr: String, + pub status: TaskStatusEnum, +} + +impl TaskInfoDTO { + fn from_model_with_status(model: builds::Model, status: TaskStatusEnum) -> Self { + Self { + build_id: model.build_id.to_string(), + output_file: model.output_file, + exit_code: model.exit_code, + start_at: model.start_at.to_rfc3339(), + end_at: model.end_at.map(|dt| dt.to_rfc3339()), + repo_name: model.repo_name, + target: model.target, + arguments: model.arguments, + mr: model.mr, + status, + } + } +} + +#[utoipa::path( + get, + path = "/tasks", + responses( + (status = 200, description = "All tasks with their current status", body = [TaskInfoDTO]), + (status = 500, description = "Internal error", body = serde_json::Value) + ) +)] +/// Return all tasks with their current status (combining /mr-task and /task-status logic) +pub async fn tasks_handler( + State(state): State, +) -> Result>, (StatusCode, Json)> { + let db = &state.conn; + let active_builds = state.scheduler.active_builds.clone(); + match builds::Entity::find().all(db).await { + Ok(models) => { + let tasks: Vec = models + .into_iter() + .map(|m| { + let id_str = m.build_id.to_string(); + let status = if active_builds.contains_key(&id_str) { + TaskStatusEnum::Building + } else if m.end_at.is_none() { + // In queue waiting for a worker assignment + TaskStatusEnum::Pending + } else if m.exit_code.is_none() { + TaskStatusEnum::Interrupted + } else if m.exit_code == Some(0) { + TaskStatusEnum::Completed + } else { + TaskStatusEnum::Failed + }; + TaskInfoDTO::from_model_with_status(m, status) + }) + .collect(); + Ok(Json(tasks)) + } + Err(e) => { + tracing::error!("Failed to fetch tasks: {e}"); + Err(( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({"message": "Failed to fetch tasks"})), + )) + } + } +} + #[cfg(test)] mod tests { /// Test random number generation for worker selection