diff --git a/extensions/rag/index/src/qdrant.rs b/extensions/rag/index/src/qdrant.rs index 0a51f88f3..0fbca747c 100644 --- a/extensions/rag/index/src/qdrant.rs +++ b/extensions/rag/index/src/qdrant.rs @@ -30,7 +30,10 @@ impl QdrantNode { log::info!("Collection '{}' already exists", self.collection_name); } Ok(false) => { - log::info!("Collection '{}' does not exist, creating...", self.collection_name); + log::info!( + "Collection '{}' does not exist, creating...", + self.collection_name + ); if let Err(e) = self .client .create_collection( @@ -45,7 +48,10 @@ impl QdrantNode { ) .await { - log::error!("Failed to create collection '{}': {e}", self.collection_name); + log::error!( + "Failed to create collection '{}': {e}", + self.collection_name + ); } } Err(e) => { @@ -53,7 +59,6 @@ impl QdrantNode { } } } - // async fn ensure_collection(&self) { // if self diff --git a/jupiter/callisto/src/builds.rs b/jupiter/callisto/src/builds.rs index 66edddc79..3f42effec 100644 --- a/jupiter/callisto/src/builds.rs +++ b/jupiter/callisto/src/builds.rs @@ -1,4 +1,4 @@ -//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.14 +//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.15 use sea_orm::entity::prelude::*; use serde::{Deserialize, Serialize}; @@ -7,21 +7,34 @@ use serde::{Deserialize, Serialize}; #[sea_orm(table_name = "builds")] pub struct Model { #[sea_orm(primary_key, auto_increment = false)] - pub build_id: Uuid, + pub id: Uuid, + pub task_id: Uuid, pub exit_code: Option, - pub start_at: DateTime, - pub end_at: Option, - pub repo_name: String, + pub start_at: DateTimeWithTimeZone, + pub end_at: Option, + pub repo: String, pub target: String, - #[sea_orm(column_type = "Text")] + pub args: Option>, pub output_file: String, - #[sea_orm(column_type = "Text")] - pub arguments: String, - #[sea_orm(column_type = "Text")] - pub mr: String, + pub created_at: DateTimeWithTimeZone, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] -pub enum Relation {} +pub enum Relation { + #[sea_orm( + belongs_to = "super::tasks::Entity", + from = "Column::TaskId", + to = "super::tasks::Column::Id", + on_update = "NoAction", + on_delete = "Cascade" + )] + Tasks, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Tasks.def() + } +} impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/callisto/src/mod.rs b/jupiter/callisto/src/mod.rs index 879c28351..d8065dc66 100644 --- a/jupiter/callisto/src/mod.rs +++ b/jupiter/callisto/src/mod.rs @@ -3,6 +3,7 @@ pub mod prelude; pub mod access_token; +pub mod builds; pub mod check_result; pub mod entity_ext; pub mod git_blob; diff --git a/jupiter/callisto/src/prelude.rs b/jupiter/callisto/src/prelude.rs index 54b7d5911..9a4ee3e2a 100644 --- a/jupiter/callisto/src/prelude.rs +++ b/jupiter/callisto/src/prelude.rs @@ -1,6 +1,7 @@ //! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.15 pub use super::access_token::Entity as AccessToken; +pub use super::builds::Entity as Builds; pub use super::check_result::Entity as CheckResult; pub use super::git_blob::Entity as GitBlob; pub use super::git_commit::Entity as GitCommit; diff --git a/jupiter/callisto/src/tasks.rs b/jupiter/callisto/src/tasks.rs index 238b195e0..54f3b20ef 100644 --- a/jupiter/callisto/src/tasks.rs +++ b/jupiter/callisto/src/tasks.rs @@ -7,21 +7,24 @@ use serde::{Deserialize, Serialize}; #[sea_orm(table_name = "tasks")] pub struct Model { #[sea_orm(primary_key, auto_increment = false)] - pub task_id: Uuid, - #[sea_orm(column_type = "JsonBinary")] - pub build_ids: Json, - #[sea_orm(column_type = "JsonBinary")] - pub output_files: Json, - pub exit_code: Option, - pub start_at: DateTimeWithTimeZone, - pub end_at: Option, - pub repo_name: String, - pub target: String, - pub arguments: String, - pub mr: String, + pub id: Uuid, + pub mr_id: i64, + pub task_name: Option, + #[sea_orm(column_type = "JsonBinary", nullable)] + pub template: Option, + pub created_at: DateTimeWithTimeZone, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] -pub enum Relation {} +pub enum Relation { + #[sea_orm(has_many = "super::builds::Entity")] + Builds, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Builds.def() + } +} impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/src/migration/m20250904_074945_modify_tasks_and_builds.rs b/jupiter/src/migration/m20250904_074945_modify_tasks_and_builds.rs new file mode 100644 index 000000000..24047797f --- /dev/null +++ b/jupiter/src/migration/m20250904_074945_modify_tasks_and_builds.rs @@ -0,0 +1,128 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(Tasks::Table).to_owned()) + .await?; + manager + .create_table( + Table::create() + .table(Tasks::Table) + .if_not_exists() + .col(ColumnDef::new(Tasks::Id).uuid().not_null().primary_key()) + .col(ColumnDef::new(Tasks::MrId).big_integer().not_null()) + .col(ColumnDef::new(Tasks::TaskName).string()) + .col(ColumnDef::new(Tasks::Template).json_binary()) + .col( + ColumnDef::new(Tasks::CreatedAt) + .timestamp_with_time_zone() + .not_null() + .default(Expr::current_timestamp()), + ) + .to_owned(), + ) + .await?; + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_tasks_created_at") + .table(Tasks::Table) + .col(Tasks::CreatedAt) + .to_owned(), + ) + .await?; + + manager + .create_table( + Table::create() + .table(Builds::Table) + .if_not_exists() + .col(ColumnDef::new(Builds::Id).uuid().not_null().primary_key()) + .col(ColumnDef::new(Builds::TaskId).uuid().not_null()) + .col(ColumnDef::new(Builds::ExitCode).integer()) + .col( + ColumnDef::new(Builds::StartAt) + .timestamp_with_time_zone() + .not_null(), + ) + .col(ColumnDef::new(Builds::EndAt).timestamp_with_time_zone()) + .col(ColumnDef::new(Builds::Repo).string().not_null()) + .col(ColumnDef::new(Builds::Target).string().not_null()) + .col(ColumnDef::new(Builds::Args).json_binary()) + .col(ColumnDef::new(Builds::OutputFile).string().not_null()) + .col( + ColumnDef::new(Builds::CreatedAt) + .timestamp_with_time_zone() + .not_null() + .default(Expr::current_timestamp()), + ) + .foreign_key( + ForeignKey::create() + .name("fk_builds_task_id") + .from(Builds::Table, Builds::TaskId) + .to(Tasks::Table, Tasks::Id) + .on_delete(ForeignKeyAction::Cascade), + ) + .to_owned(), + ) + .await?; + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_builds_task_id") + .table(Builds::Table) + .col(Builds::TaskId) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_builds_start_at") + .table(Builds::Table) + .col(Builds::StartAt) + .to_owned(), + ) + .await?; + + Ok(()) + } + + async fn down(&self, _: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } +} + +#[derive(DeriveIden)] +enum Tasks { + Table, + Id, + MrId, + TaskName, + Template, + CreatedAt, +} + +#[derive(DeriveIden)] +enum Builds { + Table, + Id, + TaskId, + ExitCode, + StartAt, + EndAt, + Repo, + Target, + Args, + OutputFile, + CreatedAt, +} diff --git a/jupiter/src/migration/mod.rs b/jupiter/src/migration/mod.rs index 2262c570d..c2833d5a1 100644 --- a/jupiter/src/migration/mod.rs +++ b/jupiter/src/migration/mod.rs @@ -55,6 +55,7 @@ mod m20250828_092459_remove_gpg_table; mod m20250828_092729_create_standalone_table; mod m20250903_013904_create_task_table; mod m20250903_071928_add_issue_refs; +mod m20250904_074945_modify_tasks_and_builds; mod m20250905_163011_add_mr_reviewer; /// Creates a primary key column definition with big integer type. /// @@ -98,11 +99,11 @@ impl MigratorTrait for Migrator { Box::new(m20250828_092729_create_standalone_table::Migration), Box::new(m20250903_013904_create_task_table::Migration), Box::new(m20250903_071928_add_issue_refs::Migration), + Box::new(m20250904_074945_modify_tasks_and_builds::Migration), Box::new(m20250905_163011_add_mr_reviewer::Migration), ] } } - /// Applies database migrations to the given database connection. /// /// # Arguments diff --git a/libra/examples/test_dry_run.rs b/libra/examples/test_dry_run.rs index 34554f69d..5fd194e40 100644 --- a/libra/examples/test_dry_run.rs +++ b/libra/examples/test_dry_run.rs @@ -6,38 +6,41 @@ use tempfile::TempDir; fn main() { println!("Testing libra rm --dry-run functionality..."); - + // Create a temporary directory for test files let temp_dir = TempDir::new().expect("Failed to create temporary directory"); let temp_path = temp_dir.path(); - + // Create test files in the temporary directory let file1_path = temp_path.join("file1.txt"); let file2_path = temp_path.join("file2.txt"); - + let mut file1 = fs::File::create(&file1_path).unwrap(); file1.write_all(b"Test content 1").unwrap(); - + let mut file2 = fs::File::create(&file2_path).unwrap(); file2.write_all(b"Test content 2").unwrap(); - + println!("Created test files in temporary directory: {:?}", temp_path); - + // Test dry-run functionality let args = RemoveArgs { - pathspec: vec![file1_path.to_string_lossy().to_string(), file2_path.to_string_lossy().to_string()], + pathspec: vec![ + file1_path.to_string_lossy().to_string(), + file2_path.to_string_lossy().to_string(), + ], cached: false, recursive: false, force: true, // Use force mode to avoid requiring git repository dry_run: true, }; - + println!("\nExecuting: libra rm --dry-run --force on test files"); - + match execute(args) { Ok(_) => { println!("✓ dry-run executed successfully!"); - + // Verify files still exist if file1_path.exists() && file2_path.exists() { println!(" Files still exist after dry-run (correct behavior)"); @@ -49,7 +52,7 @@ fn main() { println!("✗ dry-run execution failed: {:?}", e); } } - + println!("\nTest completed, temporary directory will be automatically cleaned up"); // The temporary directory will be automatically cleaned up when temp_dir goes out of scope } diff --git a/libra/src/command/mod.rs b/libra/src/command/mod.rs index 8ad8407fe..6c5732464 100644 --- a/libra/src/command/mod.rs +++ b/libra/src/command/mod.rs @@ -38,8 +38,6 @@ use std::io; use std::io::Write; use std::path::Path; - - // impl load for all objects pub fn load_object(hash: &SHA1) -> Result where diff --git a/libra/src/command/status.rs b/libra/src/command/status.rs index 6e8152c26..0ab57a249 100644 --- a/libra/src/command/status.rs +++ b/libra/src/command/status.rs @@ -10,9 +10,9 @@ use crate::command::calc_file_blob_hash; use crate::internal::head::Head; use crate::utils::object_ext::{CommitExt, TreeExt}; use crate::utils::{path, util}; +use clap::Parser; use mercury::internal::index::Index; use std::io::Write; -use clap::Parser; #[derive(Parser, Debug, Default)] pub struct StatusArgs { @@ -50,11 +50,11 @@ impl Changes { * 1. unstaged * 2. staged to be committed */ - pub async fn execute_to(args: StatusArgs,writer: &mut impl Write) { +pub async fn execute_to(args: StatusArgs, writer: &mut impl Write) { if !util::check_repo_exist() { return; } - + // Do not output branch info in porcelain mode if !args.porcelain { match Head::current().await { @@ -69,8 +69,8 @@ impl Changes { if Head::current_commit().await.is_none() { writeln!(writer, "\nNo commits yet\n").unwrap(); } - } - + } + // to cur_dir relative path let staged = changes_to_be_committed().await.to_relative(); let unstaged = changes_to_be_staged().to_relative(); @@ -81,9 +81,8 @@ impl Changes { return; } - if staged.is_empty() && unstaged.is_empty() { - writeln!(writer,"nothing to commit, working tree clean").unwrap(); + writeln!(writer, "nothing to commit, working tree clean").unwrap(); return; } @@ -92,15 +91,15 @@ impl Changes { println!(" use \"libra restore --staged ...\" to unstage"); staged.deleted.iter().for_each(|f| { let str = format!("\tdeleted: {}", f.display()); - writeln!(writer,"{}", str.bright_green()).unwrap(); + writeln!(writer, "{}", str.bright_green()).unwrap(); }); staged.modified.iter().for_each(|f| { let str = format!("\tmodified: {}", f.display()); - writeln!(writer,"{}", str.bright_green()).unwrap(); + writeln!(writer, "{}", str.bright_green()).unwrap(); }); staged.new.iter().for_each(|f| { let str = format!("\tnew file: {}", f.display()); - writeln!(writer,"{}", str.bright_green()).unwrap(); + writeln!(writer, "{}", str.bright_green()).unwrap(); }); } @@ -110,11 +109,11 @@ impl Changes { println!(" use \"libra restore ...\" to discard changes in working directory"); unstaged.deleted.iter().for_each(|f| { let str = format!("\tdeleted: {}", f.display()); - writeln!(writer,"{}", str.bright_red()).unwrap(); + writeln!(writer, "{}", str.bright_red()).unwrap(); }); unstaged.modified.iter().for_each(|f| { let str = format!("\tmodified: {}", f.display()); - writeln!(writer,"{}", str.bright_red()).unwrap(); + writeln!(writer, "{}", str.bright_red()).unwrap(); }); } if !unstaged.new.is_empty() { @@ -122,7 +121,7 @@ impl Changes { println!(" use \"libra add ...\" to include in what will be committed"); unstaged.new.iter().for_each(|f| { let str = format!("\t{}", f.display()); - writeln!(writer,"{}", str.bright_red()).unwrap(); + writeln!(writer, "{}", str.bright_red()).unwrap(); }); } } @@ -138,7 +137,7 @@ pub fn output_porcelain(staged: &Changes, unstaged: &Changes, writer: &mut impl for file in &staged.deleted { writeln!(writer, "D {}", file.display()).unwrap(); } - + // Output unstaged changes for file in &unstaged.modified { writeln!(writer, " M {}", file.display()).unwrap(); @@ -146,12 +145,11 @@ pub fn output_porcelain(staged: &Changes, unstaged: &Changes, writer: &mut impl for file in &unstaged.deleted { writeln!(writer, " D {}", file.display()).unwrap(); } - + // Output untracked files for file in &unstaged.new { writeln!(writer, "?? {}", file.display()).unwrap(); } - } pub async fn execute(args: StatusArgs) { diff --git a/libra/src/internal/db.rs b/libra/src/internal/db.rs index 8ef0e2931..44c28e529 100644 --- a/libra/src/internal/db.rs +++ b/libra/src/internal/db.rs @@ -210,7 +210,7 @@ mod tests { } let conn = create_database(db_path).await.unwrap(); assert!(Path::new(db_path).exists()); - + let result = create_database(db_path).await; assert!(result.is_err()); diff --git a/libra/src/utils/client_storage.rs b/libra/src/utils/client_storage.rs index 6cb75e072..6296821b6 100644 --- a/libra/src/utils/client_storage.rs +++ b/libra/src/utils/client_storage.rs @@ -491,7 +491,7 @@ impl ClientStorage { } Some(buf) => buf, }; - + let pack_cursor = Cursor::new(pack_file_buf); let mut pack_reader = BufReader::new(pack_cursor); pack_reader.seek(io::SeekFrom::Start(offset))?; diff --git a/libra/tests/command/remove_test.rs b/libra/tests/command/remove_test.rs index 76d0bf57a..91ab87647 100644 --- a/libra/tests/command/remove_test.rs +++ b/libra/tests/command/remove_test.rs @@ -497,7 +497,7 @@ async fn test_remove_dry_run_cached() { // Verify the file still exists in both filesystem and index assert!(file_path.exists(), "File should still exist in filesystem"); - + // Verify file doesn't appear as deleted in changes let changes = changes_to_be_staged(); assert!( @@ -548,8 +548,14 @@ async fn test_remove_dry_run_recursive() { assert!(file1.exists(), "File 1 should still exist after dry-run"); assert!(file2.exists(), "File 2 should still exist after dry-run"); assert!(file3.exists(), "File 3 should still exist after dry-run"); - assert!(PathBuf::from("test_dir").exists(), "Directory should still exist"); - assert!(PathBuf::from("test_dir/subdir").exists(), "Subdirectory should still exist"); + assert!( + PathBuf::from("test_dir").exists(), + "Directory should still exist" + ); + assert!( + PathBuf::from("test_dir/subdir").exists(), + "Subdirectory should still exist" + ); // Verify files are still tracked by checking they don't appear as deleted let changes = changes_to_be_staged(); diff --git a/libra/tests/command/status_test.rs b/libra/tests/command/status_test.rs index e97237f83..5a08f0885 100644 --- a/libra/tests/command/status_test.rs +++ b/libra/tests/command/status_test.rs @@ -1,9 +1,9 @@ use super::*; -use std::fs; -use std::io::Write; -use libra::command::status::StatusArgs; use libra::command::status::execute_to as status_execute; use libra::command::status::output_porcelain; +use libra::command::status::StatusArgs; +use std::fs; +use std::io::Write; #[tokio::test] #[serial] /// Tests the file status detection functionality with respect to ignore patterns. @@ -107,32 +107,32 @@ async fn test_changes_to_be_staged() { fn test_output_porcelain_format() { use libra::command::status::Changes; use std::path::PathBuf; - + // Create test data let staged = Changes { new: vec![PathBuf::from("new_file.txt")], modified: vec![PathBuf::from("modified_file.txt")], deleted: vec![PathBuf::from("deleted_file.txt")], }; - + let unstaged = Changes { new: vec![PathBuf::from("untracked_file.txt")], modified: vec![PathBuf::from("unstaged_modified.txt")], deleted: vec![PathBuf::from("unstaged_deleted.txt")], }; - + // Create a buffer to capture the output let mut output = Vec::new(); - + // Call the output_porcelain function output_porcelain(&staged, &unstaged, &mut output); - + // Get the output as a string let output_str = String::from_utf8(output).unwrap(); - + // Verify the output format let lines: Vec<&str> = output_str.trim().split('\n').collect(); - + assert!(lines.contains(&"A new_file.txt")); assert!(lines.contains(&"M modified_file.txt")); assert!(lines.contains(&"D deleted_file.txt")); @@ -153,7 +153,7 @@ async fn test_status_porcelain() { // Create test data let mut file1 = fs::File::create("file1.txt").unwrap(); file1.write_all(b"content").unwrap(); - + let mut file2 = fs::File::create("file2.txt").unwrap(); file2.write_all(b"content").unwrap(); @@ -166,7 +166,8 @@ async fn test_status_porcelain() { dry_run: false, ignore_errors: false, refresh: false, - }).await; + }) + .await; // Add another file to the staging area and modify it add::execute(AddArgs { @@ -177,7 +178,8 @@ async fn test_status_porcelain() { dry_run: false, ignore_errors: false, refresh: false, - }).await; + }) + .await; file2.write_all(b"modified content").unwrap(); // Create a new file (untracked) @@ -186,13 +188,13 @@ async fn test_status_porcelain() { // Create a buffer to capture the output let mut output = Vec::new(); - + // Execute the status command with the --porcelain flag status_execute(StatusArgs { porcelain: true }, &mut output).await; - + // Get the output as a string let output_str = String::from_utf8(output).unwrap(); - + // Verify the porcelain output format let lines: Vec<&str> = output_str.trim().split('\n').collect(); @@ -201,10 +203,10 @@ async fn test_status_porcelain() { assert!(lines.iter().any(|line| line.starts_with("A file2.txt"))); // Should contain modified but unstaged files assert!(lines.iter().any(|line| line.starts_with(" M file2.txt"))); - + // Should contain untracked files assert!(lines.iter().any(|line| line.starts_with("?? file3.txt"))); - + // Should not contain human-readable text assert!(!output_str.contains("Changes to be committed")); assert!(!output_str.contains("Untracked files")); diff --git a/mercury/delta/src/lib.rs b/mercury/delta/src/lib.rs index 9593b44e2..93af2614d 100644 --- a/mercury/delta/src/lib.rs +++ b/mercury/delta/src/lib.rs @@ -1,6 +1,6 @@ -use std::hash::{DefaultHasher, Hash, Hasher}; use encode::DeltaDiff; use rayon::prelude::*; +use std::hash::{DefaultHasher, Hash, Hasher}; mod decode; mod encode; @@ -9,8 +9,8 @@ mod utils; pub use decode::delta_decode as decode; -const SAMPLE_STEP: usize = 64; -const MIN_DELTA_RATE: f64 = 0.5; +const SAMPLE_STEP: usize = 64; +const MIN_DELTA_RATE: f64 = 0.5; /// approximate calculation delta rate to enhance efficiency #[allow(clippy::manual_div_ceil)] @@ -19,21 +19,19 @@ pub fn heuristic_encode_rate(old_data: &[u8], new_data: &[u8]) -> f64 { let new_len = new_data.len(); if old_len == 0 && new_len == 0 { - return 1.0; + return 1.0; } if old_len == 0 || new_len == 0 { - return 0.0; + return 0.0; } let step = SAMPLE_STEP; let mut match_count = 0; let mut sample_count = 0; - let min_len = old_len.min(new_len); - - let total_samples = (min_len + step - 1) / step; + let total_samples = (min_len + step - 1) / step; let mut i = 0; while i < min_len { let old_chunk = &old_data[i..(i + step).min(old_len)]; @@ -55,7 +53,6 @@ pub fn heuristic_encode_rate(old_data: &[u8], new_data: &[u8]) -> f64 { i += step; } - match_count as f64 / sample_count as f64 } @@ -69,13 +66,17 @@ pub fn heuristic_encode_rate_parallel(old_data: &[u8], new_data: &[u8]) -> f64 { let old_len = old_data.len(); let new_len = new_data.len(); - if old_len == 0 && new_len == 0 { return 1.0; } - if old_len == 0 || new_len == 0 { return 0.0; } + if old_len == 0 && new_len == 0 { + return 1.0; + } + if old_len == 0 || new_len == 0 { + return 0.0; + } let min_len = old_len.min(new_len); - let step = if min_len > 10_000_000{ - 1024 + let step = if min_len > 10_000_000 { + 1024 } else if min_len > 1_000_000 { 512 } else if min_len > 100_000 { @@ -84,16 +85,19 @@ pub fn heuristic_encode_rate_parallel(old_data: &[u8], new_data: &[u8]) -> f64 { 16 }; - let chunks: Vec<_> = old_data[..min_len].chunks(step) + let chunks: Vec<_> = old_data[..min_len] + .chunks(step) .zip(new_data[..min_len].chunks(step)) .collect(); - let match_count: usize = chunks.par_iter() - .filter(|(a, b)| a == b) - .count(); + let match_count: usize = chunks.par_iter().filter(|(a, b)| a == b).count(); let rate = match_count as f64 / chunks.len() as f64; - if rate < MIN_DELTA_RATE { 0.0 } else { rate } + if rate < MIN_DELTA_RATE { + 0.0 + } else { + rate + } } pub fn encode_rate(old_data: &[u8], new_data: &[u8]) -> f64 { @@ -108,63 +112,68 @@ pub fn encode(old_data: &[u8], new_data: &[u8]) -> Vec { #[cfg(test)] mod tests { - use crate::{ encode_rate, heuristic_encode_rate, heuristic_encode_rate_parallel}; + use crate::{encode_rate, heuristic_encode_rate, heuristic_encode_rate_parallel}; #[test] fn test_heuristic_encode_rate() { - let data1 = b"hello world, this is a test for delta rate"; let data2 = b"hello world, this is a test for delta rate"; let rate = heuristic_encode_rate(data1, data2); println!("rate = {}", rate); assert!((rate - 1.0).abs() < 1e-6, "Expected 1.0 for identical data"); - let data3 = b"hello world, this is a test for delta rate"; let data4 = b"hello worll, this is a test for delta rate"; let rate = encode_rate(data3, data4); println!("rate = {}", rate); - assert!(rate > 0.5 && rate < 1.0, "Expected partial match for similar data"); + assert!( + rate > 0.5 && rate < 1.0, + "Expected partial match for similar data" + ); - let data5 = b"abcdefghijklmno"; let data6 = b"1234567890!@#"; let rate = heuristic_encode_rate(data5, data6); println!("rate = {}", rate); assert!(rate < 0.2, "Expected low match rate for different data"); - let data7 = b""; let data8 = b""; let rate = heuristic_encode_rate(data7, data8); println!("rate = {}", rate); assert_eq!(rate, 1.0, "Empty slices should be fully matching"); - let rate = heuristic_encode_rate(data7, data1); assert_eq!(rate, 0.0, "Empty vs non-empty should give 0 rate"); } #[test] fn test_heuristic_encode_rate_large_files() { - let data1 = vec![0u8; 100_000]; let data2 = vec![1u8; 100_000]; let rate = heuristic_encode_rate(&data1, &data2); println!("Large non-matching data rate = {}", rate); - assert_eq!(rate, 0.0, "Large completely different data should early stop with 0 rate"); + assert_eq!( + rate, 0.0, + "Large completely different data should early stop with 0 rate" + ); - - let data3 = vec![0u8; 100]; + let data3 = vec![0u8; 100]; let mut data4 = vec![0u8; 100]; - + for i in 0..2 { data4[i] = 1; } let rate1 = heuristic_encode_rate_parallel(&data3, &data4); let rate2 = encode_rate(&data3, &data4); - println!("Large partially matching data rate = {}, accurate rate = {}", rate1,rate2); - - assert!((rate2-rate1).abs() < 0.2, "Large partially matching data should preserve partial rate"); + println!( + "Large partially matching data rate = {}, accurate rate = {}", + rate1, rate2 + ); + + assert!( + (rate2 - rate1).abs() < 0.2, + "Large partially matching data should preserve partial rate" + ); } } diff --git a/mercury/src/internal/pack/encode.rs b/mercury/src/internal/pack/encode.rs index 18ea699ea..3e3902d79 100644 --- a/mercury/src/internal/pack/encode.rs +++ b/mercury/src/internal/pack/encode.rs @@ -129,7 +129,6 @@ fn magic_sort(a: &Entry, b: &Entry) -> Ordering { (a as *const Entry).cmp(&(b as *const Entry)) } - fn calc_hash(data: &[u8]) -> u64 { let mut hasher = AHasher::default(); data.hash(&mut hasher); @@ -261,7 +260,6 @@ impl PackEncoder { ) .map_err(|e| GitError::PackEncodeError(format!("Task join error: {e}")))?; - let all_encoded_data = [ commit_results .map_err(|e| GitError::PackEncodeError(format!("Commit encoding error: {e}")))?, @@ -670,7 +668,6 @@ mod tests { let entries = get_entries_for_test().await; let entries_number = entries.lock().await.len(); - let total_original_size: usize = entries .lock() .await diff --git a/mercury/src/utils.rs b/mercury/src/utils.rs index a36ee8297..2ed62f41e 100644 --- a/mercury/src/utils.rs +++ b/mercury/src/utils.rs @@ -12,9 +12,6 @@ pub fn read_sha1(file: &mut impl Read) -> io::Result { SHA1::from_stream(file) } - - - /// A lightweight wrapper that counts bytes read from the underlying reader. /// replace deflate.intotal() in decompress_data pub struct CountingReader { @@ -46,7 +43,7 @@ impl BufRead for CountingReader { } fn consume(&mut self, amt: usize) { - self.bytes_read += amt as u64; + self.bytes_read += amt as u64; self.inner.consume(amt); } } diff --git a/orion-server/db/schema.sql b/orion-server/db/schema.sql index 534875e01..aabfafd68 100644 --- a/orion-server/db/schema.sql +++ b/orion-server/db/schema.sql @@ -1,16 +1,31 @@ -- Orion schema (no CREATE DATABASE here) +-- Tasks table - represents a user's build request CREATE TABLE IF NOT EXISTS public.tasks ( - task_id UUID PRIMARY KEY, - build_ids JSONB NOT NULL, - output_files JSONB NOT NULL, - exit_code INTEGER, - start_at TIMESTAMPTZ NOT NULL, - end_at TIMESTAMPTZ, - repo_name TEXT NOT NULL, - target TEXT NOT NULL, - arguments TEXT NOT NULL, - mr TEXT NOT NULL + id UUID PRIMARY KEY, + mr_id BIGINT NOT NULL, + task_name TEXT, + template JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); -CREATE INDEX IF NOT EXISTS idx_tasks_mr ON public.tasks (mr); -CREATE INDEX IF NOT EXISTS idx_tasks_start_at ON public.tasks (start_at); \ No newline at end of file +CREATE INDEX IF NOT EXISTS idx_tasks_mr ON public.tasks (mr_id); + +CREATE INDEX IF NOT EXISTS idx_tasks_start_at ON public.tasks (created_at); + +-- Builds table - represents individual builds belonging to a task +CREATE TABLE IF NOT EXISTS public.builds ( + id UUID PRIMARY KEY, + task_id UUID NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + exit_code INT, + start_at TIMESTAMPTZ NOT NULL, + end_at TIMESTAMPTZ, + repo TEXT NOT NULL, + target TEXT NOT NULL, + args TEXT [] DEFAULT NULL, + output_file TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_builds_task_id ON public.builds (task_id); + +CREATE INDEX IF NOT EXISTS idx_builds_start_at ON public.builds (start_at); \ No newline at end of file diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index a8de222b4..dcf1c01ff 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -1,6 +1,6 @@ -use crate::model::tasks; +use crate::model::{builds, tasks}; use crate::scheduler::{ - BuildInfo, BuildRequest, TaskQueueStats, TaskScheduler, WorkerInfo, WorkerStatus, + self, BuildInfo, BuildRequest, TaskQueueStats, TaskScheduler, WorkerInfo, WorkerStatus, create_log_file, get_build_log_dir, }; use axum::{ @@ -18,11 +18,9 @@ use dashmap::DashMap; use futures_util::{SinkExt, Stream, StreamExt}; use orion::ws::WSMessage; use rand::Rng; -use sea_orm::{ - ActiveModelTrait, ActiveValue::Set, ColumnTrait, DatabaseConnection, EntityTrait, - QueryFilter as _, -}; +use sea_orm::{ActiveValue::Set, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter as _}; use serde::{Deserialize, Serialize}; +use serde_json::Value; use std::convert::Infallible; use std::io::Write; use std::net::SocketAddr; @@ -37,6 +35,19 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use utoipa::ToSchema; use uuid::Uuid; +#[derive(Debug, Serialize, Default, ToSchema)] +struct BuildResult { + build_id: String, + status: String, + message: String, +} + +#[derive(Debug, Serialize, Default, ToSchema)] +struct TaskResponse { + task_id: String, + results: Vec, +} + /// Enumeration of possible task statuses #[derive(Debug, Serialize, Default, ToSchema)] pub enum TaskStatusEnum { @@ -49,27 +60,19 @@ pub enum TaskStatusEnum { #[default] NotFound, } - /// Default log limit for segmented log retrieval const DEFAULT_LOG_LIMIT: usize = 4096; /// Default log offset for segmented log retrieval const DEFAULT_LOG_OFFSET: u64 = 1; -/// Response structure for task status queries +/// Request structure for creating a task #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct TaskRequest { pub repo: String, - pub mr: Option, -} - -/// Response structure for task status queries -#[derive(Debug, Serialize, Default, ToSchema)] -pub struct TaskStatus { - status: TaskStatusEnum, - #[serde(skip_serializing_if = "Option::is_none")] - exit_code: Option, - #[serde(skip_serializing_if = "Option::is_none")] - message: Option, + pub mr: i64, + pub task_name: Option, + pub template: Option, + pub builds: Vec, } /// Shared application state containing worker connections, database, and active builds @@ -98,15 +101,12 @@ pub fn routers() -> Router { Router::new() .route("/ws", any(ws_handler)) .route("/task", axum::routing::post(task_handler)) - .route("/task-build/{id}", axum::routing::post(task_build_handler)) - .route("/task-status/{id}", get(task_status_handler)) .route("/task-build-list/{id}", get(task_build_list_handler)) .route("/task-output/{id}", get(task_output_handler)) .route( "/task-history-output/{id}", get(task_history_output_handler), ) - .route("/mr-task/{mr}", get(task_query_by_mr)) .route("/tasks/{mr}", get(tasks_handler)) .route("/queue-stats", get(queue_stats_handler)) } @@ -130,81 +130,6 @@ pub async fn queue_stats_handler(State(state): State) -> impl IntoResp (StatusCode::OK, Json(stats)) } -#[utoipa::path( - get, - path = "/task-status/{id}", - params( - ("id" = String, Path, description = "Task id") - ), - responses( - (status = 200, description = "Task status", body = TaskStatus) - ) -)] -/// Retrieves the current status of a build task by its ID -/// Returns status information including exit code and current state -pub async fn task_status_handler( - Path(id): Path, - State(state): State, -) -> impl IntoResponse { - let (code, status) = if state.scheduler.active_builds.contains_key(&id) { - // Task is currently active/building - ( - StatusCode::OK, - TaskStatus { - status: TaskStatusEnum::Building, - ..Default::default() - }, - ) - } else { - // Check database for completed/historical tasks - match Uuid::parse_str(&id) { - Ok(id_uuid) => { - let output = tasks::Model::get_by_task_id(id_uuid, &state.conn).await; - match output { - Some(model) => { - // Determine task status based on database fields - let status = if model.end_at.is_none() { - // 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 { - TaskStatusEnum::Completed - } else { - TaskStatusEnum::Failed - }; - ( - StatusCode::OK, - TaskStatus { - status, - exit_code: model.exit_code, - ..Default::default() - }, - ) - } - None => ( - StatusCode::NOT_FOUND, - TaskStatus { - status: TaskStatusEnum::NotFound, - message: Some("Build task not found".to_string()), - ..Default::default() - }, - ), - } - } - Err(_) => ( - StatusCode::BAD_REQUEST, - TaskStatus { - status: TaskStatusEnum::NotFound, - message: Some("Invalid build id".to_string()), - ..Default::default() - }, - ), - } - }; - (code, Json(status)) -} - /// Streams build output logs in real-time using Server-Sent Events (SSE) /// Continuously monitors the log file and streams new content as it becomes available #[utoipa::path( @@ -394,195 +319,127 @@ pub async fn task_history_output_handler( #[utoipa::path( post, - path = "/task-build/{id}", - params( - ("id" = String, Path, description = "Task ID to get build IDs for") - ), - request_body = BuildRequest, + path = "/task", + request_body = TaskRequest, responses( - (status = 200, description = "Task start build", body = serde_json::Value), - (status = 401, description = "Not found task id", body = serde_json::Value), + (status = 200, description = "Task created", body = serde_json::Value), (status = 503, description = "Queue is full", body = serde_json::Value) - ), + ) )] -pub async fn task_build_handler( +/// Creates new build tasks based on the builds count in TaskRequest and either assigns them immediately or queues them for later processing +/// Returns task ID and status information upon successful creation +pub async fn task_handler( State(state): State, - Path(id): Path, - Json(req): Json, + Json(req): Json, ) -> impl IntoResponse { - let db = &state.conn; + // for now we do not extract from file, just use the fixed build target. + let target = "//...".to_string(); - // 解析task_id - let task_id = match id.parse::() { - Ok(uuid) => uuid, - Err(_) => { - return ( - StatusCode::BAD_REQUEST, - Json(serde_json::json!({"message": "Invalid task ID format"})), - ) - .into_response(); - } - }; + // create task id + let task_id = Uuid::now_v7(); - // 检查task_id是否存在 - if !tasks::Model::exists_by_task_id(task_id, db).await { + // Process each build + let mut results = Vec::with_capacity(req.builds.len()); + // Insert task into the database using the model's insert method + if let Err(err) = tasks::Model::insert_task( + task_id, + req.mr, + req.task_name.clone(), + req.template.clone(), + chrono::Utc::now().into(), + &state.conn, + ) + .await + { + tracing::error!("Failed to insert task into DB: {}", err); return ( - StatusCode::UNAUTHORIZED, - Json(serde_json::json!({"message": "Task ID does not exist"})), + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ + "message": format!("Failed to insert task into database: {}", err) + })), ) .into_response(); } - // Download and get buck2 targets first - // let target = match download_and_get_buck2_targets(&req.buck_hash, &req.buckconfig_hash).await { - // Ok(target) => target, - // Err(e) => { - // tracing::error!("Failed to download buck2 targets: {}", e); - // return ( - // StatusCode::INTERNAL_SERVER_ERROR, - // Json(serde_json::json!({ "message": format!("Failed to download buck2 targets: {}", e) })), - // ).into_response(); - // } - // }; - // for now we do not extract from file, just use the fixed build target. - let target = "//...".to_string(); - // Check if there are idle workers available - if state.scheduler.has_idle_workers() { - // Have idle workers, directly dispatch task (keep original logic) - handle_immediate_task_dispatch(state, req, target, task_id).await - } else { - // No idle workers, add task to queue - match state - .scheduler - .enqueue_task(req.clone(), target.clone(), task_id) - .await - { - Ok(build_id) => { - tracing::info!("Build {}/{} queued for later processing", task_id, build_id); - - // Save to database (mark as Pending status) - let model = tasks::ActiveModel { - task_id: Set(task_id), - build_ids: Set(serde_json::json!([build_id])), - output_files: Set(serde_json::json!([format!( - "{}/{}", - get_build_log_dir(), - build_id - )])), - exit_code: Set(None), - start_at: Set(Utc::now().with_timezone(&FixedOffset::east_opt(0).unwrap())), - end_at: Set(None), - repo_name: Set(req.repo.clone()), - target: Set(target.clone()), - arguments: Set(req.args.clone().unwrap_or_default().join(" ")), - mr: Set(req.mr.clone().unwrap_or_default()), - }; - - if let Err(e) = model.insert(&state.conn).await { - tracing::error!("Failed to insert queued task into DB: {}", e); - } - - ( - StatusCode::OK, - Json(serde_json::json!({ - "task_id": task_id.to_string(), - "build_id":build_id.to_string(), - "status": "queued", - "message": "Task queued for processing when workers become available" - })), - ) - .into_response() - } - Err(e) => { - tracing::warn!("Failed to queue task: {}", e); - ( - StatusCode::SERVICE_UNAVAILABLE, - Json(serde_json::json!({ - "message": format!("Unable to queue task: {}", e) - })), + for build in &req.builds { + // Check if there are idle workers available + if state.scheduler.has_idle_workers() { + // Have idle workers, directly dispatch task (keep original logic) + let result: BuildResult = handle_immediate_task_dispatch( + state.clone(), + task_id, + &req.repo, + req.mr, + build.clone(), + target.clone(), + ) + .await; + results.push(result); + } else { + // No idle workers, add task to queue + match state + .scheduler + .enqueue_task( + task_id, + build.clone(), + target.clone(), + req.repo.clone(), + req.mr, ) - .into_response() + .await + { + Ok(build_id) => { + tracing::info!("Build {}/{} queued for later processing", task_id, build_id); + let result: BuildResult = BuildResult { + build_id: build_id.to_string(), + status: "queued".to_string(), + message: "Task queued for processing when workers become available" + .to_string(), + }; + results.push(result); + } + Err(e) => { + tracing::warn!("Failed to queue task: {}", e); + let result: BuildResult = BuildResult { + build_id: "".to_string(), + status: "error".to_string(), + message: format!("Unable to queue task: {}", e), + }; + results.push(result); + } } } } -} -#[utoipa::path( - post, - path = "/task", - request_body = TaskRequest, - responses( - (status = 200, description = "Task created successfully", body = serde_json::Value), - (status = 500, description = "Failed to create task", body = serde_json::Value) + ( + StatusCode::OK, + Json(TaskResponse { + task_id: task_id.to_string(), + results, + }), ) -)] -/// Creates a new build task and either assigns it immediately or queues it for later processing -/// Returns task ID and status information upon successful creation -pub async fn task_handler( - State(state): State, - Json(req): Json, -) -> impl IntoResponse { - let task_id = Uuid::now_v7(); - let db = &state.conn; - - // Create empty JSON arrays for build_ids and output_files - let empty_json_array = serde_json::Value::Array(vec![]); - - // Insert new task into database - let task_model = tasks::ActiveModel { - task_id: Set(task_id), - build_ids: Set(empty_json_array.clone()), - output_files: Set(empty_json_array), - exit_code: Set(None), - start_at: Set(Utc::now().with_timezone(&FixedOffset::east_opt(0).unwrap())), - end_at: Set(None), - repo_name: Set(req.repo), - target: Set(String::new()), // Empty for now, may be populated from request in the future - arguments: Set(String::new()), // Empty for now, may be populated from request in the future - mr: Set(req.mr.unwrap_or_default()), - }; - - match task_model.insert(db).await { - Ok(_) => { - // Return task_id - ( - StatusCode::OK, - Json(serde_json::json!({ - "task_id": task_id.to_string() - })), - ) - .into_response() - } - Err(e) => { - tracing::error!("Failed to insert task into database: {}", e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({ - "message": "Failed to create task" - })), - ) - .into_response() - } - } + .into_response() } /// Handle immediate task dispatch logic (original task_handler logic) async fn handle_immediate_task_dispatch( state: AppState, + task_id: Uuid, + repo: &str, + mr: i64, req: BuildRequest, target: String, - task_id: Uuid, -) -> axum::response::Response { +) -> BuildResult { // Find all idle workers let idle_workers = state.scheduler.get_idle_workers(); // Return error if no workers are available (this shouldn't happen theoretically since we already checked) if idle_workers.is_empty() { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(serde_json::json!({"message": "No available workers at the moment"})), - ) - .into_response(); + return BuildResult { + build_id: "".to_string(), + status: "error".to_string(), + message: "No available workers at the moment".to_string(), + }; } // Randomly select an idle worker @@ -603,135 +460,79 @@ async fn handle_immediate_task_dispatch( build_id, e ); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"message": "Failed to create log file"})), - ) - .into_response(); + return BuildResult { + build_id: "".to_string(), + status: "error".to_string(), + message: "Failed to create log file".to_string(), + }; } }; // Create build information structure let build_info = BuildInfo { - repo: req.repo.clone(), + repo: repo.to_string(), target: target.clone(), args: req.args.clone(), start_at: chrono::Utc::now(), - mr: req.mr.clone(), + mr: mr.to_string(), _worker_id: chosen_id.clone(), log_file, }; - // Retrieve existing task to get current build_ids and output_files - let existing_task = match tasks::Model::get_by_task_id(task_id, &state.conn).await { - Some(task) => task, - None => { - tracing::error!("Task not found: {}", task_id); - return ( - StatusCode::BAD_REQUEST, - Json(serde_json::json!({"message": "Task not found"})), - ) - .into_response(); - } - }; - - // Update build_ids and output_files arrays - let mut build_ids: Vec = existing_task - .build_ids - .as_array() - .cloned() - .unwrap_or_default(); - build_ids.push(serde_json::Value::String(build_id.to_string())); - - let mut output_files: Vec = existing_task - .output_files - .as_array() - .cloned() - .unwrap_or_default(); - output_files.push(serde_json::Value::String(format!( - "{}/{}", - get_build_log_dir(), - build_id - ))); - - let model = tasks::ActiveModel { - task_id: Set(task_id), - build_ids: Set(serde_json::Value::Array(build_ids)), - output_files: Set(serde_json::Value::Array(output_files)), - exit_code: Set(None), - start_at: Set(build_info - .start_at - .with_timezone(&FixedOffset::east_opt(0).unwrap())), - end_at: Set(None), - repo_name: Set(build_info.repo.clone()), - target: Set(build_info.target.clone()), - arguments: Set(build_info.args.clone().unwrap_or_default().join(" ")), - mr: Set(build_info.mr.clone().unwrap_or_default()), - }; - if let Err(e) = model.update(&state.conn).await { - tracing::error!("Failed to insert new build task into DB: {}", e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"message": "Failed to create task in database"})), - ) - .into_response(); + // Use the model's insert_build method for direct insertion + if let Err(err) = builds::Model::insert_build( + task_id, + repo.to_string(), + target.clone(), + req.clone(), + &state.conn, + ) + .await + { + tracing::error!("Failed to insert builds into DB: {}", err); + return BuildResult { + build_id: "".to_string(), + status: "error".to_string(), + message: format!("Failed to insert builds into database: {}", err), + }; } - // Create WebSocket message for the worker + // Create WebSocket message for the worker (use first build's args) let msg = WSMessage::Task { id: build_id.to_string(), - repo: req.repo, + repo: repo.to_string(), target, - args: req.args, - mr: req.mr.unwrap_or_default(), + args: req.args.clone(), + mr: mr.to_string(), }; // Send task to the selected worker - if let Some(mut worker) = state.scheduler.workers.get_mut(&chosen_id) { - if worker.sender.send(msg).is_ok() { - worker.status = WorkerStatus::Busy(build_id.to_string()); - state - .scheduler - .active_builds - .insert(build_id.to_string(), build_info); - tracing::info!( - "Build {}/{} dispatched immediately to worker {}", - task_id, - build_id, - chosen_id - ); - ( - StatusCode::OK, - Json(serde_json::json!({ - "task_id": task_id.to_string(), - "build_id": build_id.to_string(), - "client_id": chosen_id, - "status": "dispatched" - })), - ) - .into_response() - } else { - tracing::error!( - "Failed to send task to supposedly idle worker {}. It might have just disconnected.", - chosen_id - ); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json( - serde_json::json!({"message": "Failed to dispatch task to worker. Please try again."}), - ), - ).into_response() - } - } else { - tracing::error!( - "Chosen idle worker {} not found in map. This should not happen.", + if let Some(mut worker) = state.scheduler.workers.get_mut(&chosen_id) + && worker.sender.send(msg).is_ok() + { + worker.status = WorkerStatus::Busy(build_id.to_string()); + state + .scheduler + .active_builds + .insert(build_id.to_string(), build_info); + tracing::info!( + "Build {}/{} dispatched immediately to worker {}", + task_id, + build_id, chosen_id ); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"message": "Internal scheduler error."})), - ) - .into_response() + return BuildResult { + build_id: build_id.to_string(), + status: "dispatched".to_string(), + message: format!("Build dispatched to worker {}", chosen_id), + }; + } + + // If we reach here, sending failed + BuildResult { + build_id: "".to_string(), + status: "error".to_string(), + message: "Failed to dispatch task to worker".to_string(), } } @@ -754,7 +555,6 @@ pub async fn task_build_list_handler( ) -> impl IntoResponse { let db = &state.conn; - // 解析task_id let task_id = match id.parse::() { Ok(uuid) => uuid, Err(_) => { @@ -766,7 +566,6 @@ pub async fn task_build_list_handler( } }; - // 使用get_builds_by_task_id方法获取build_ids match tasks::Model::get_builds_by_task_id(task_id, db).await { Some(build_ids) => { let build_ids_str: Vec = @@ -942,15 +741,15 @@ async fn process_message( // Remove from active builds and update database state.scheduler.active_builds.remove(&id); - let _ = tasks::Entity::update_many() - .set(tasks::ActiveModel { + let _ = builds::Entity::update_many() + .set(builds::ActiveModel { exit_code: Set(exit_code), end_at: Set(Some( Utc::now().with_timezone(&FixedOffset::east_opt(0).unwrap()), )), ..Default::default() }) - .filter(tasks::Column::TaskId.eq(id.parse::().unwrap())) + .filter(builds::Column::Id.eq(id.parse::().unwrap())) .exec(&state.conn) .await; @@ -976,75 +775,51 @@ async fn process_message( /// Data transfer object for build information in API responses #[derive(Debug, Serialize, ToSchema)] -pub struct TaskDTO { +pub struct BuildDTO { + pub id: String, pub task_id: String, - pub build_ids: serde_json::Value, - pub output_files: serde_json::Value, pub exit_code: Option, pub start_at: String, pub end_at: Option, - pub repo_name: String, + pub repo: String, pub target: String, - pub arguments: String, - pub mr: String, + pub args: Option>, + pub output_file: String, + pub created_at: String, + pub status: TaskStatusEnum, } -impl TaskDTO { +impl BuildDTO { /// Converts a database model to a DTO for API responses - pub fn from_model(model: tasks::Model) -> Self { + pub fn from_model(model: builds::Model, status: TaskStatusEnum) -> Self { Self { + id: model.id.to_string(), task_id: model.task_id.to_string(), - build_ids: model.build_ids, - output_files: model.output_files, exit_code: model.exit_code, start_at: model.start_at.with_timezone(&Utc).to_rfc3339(), end_at: model.end_at.map(|dt| dt.with_timezone(&Utc).to_rfc3339()), - repo_name: model.repo_name, + repo: model.repo, target: model.target, - arguments: model.arguments, - mr: model.mr, + args: model.args, + output_file: model.output_file, + created_at: model.created_at.with_timezone(&Utc).to_rfc3339(), + status, } } -} -#[utoipa::path( - get, - path = "/mr-task/{mr}", - params( - ("mr" = String, Path, description = "MR number") - ), - responses( - (status = 200, description = "Task for MR", body = [TaskDTO]), - (status = 404, description = "No builds found for the given MR", body = serde_json::Value), - (status = 500, description = "Internal server error", body = serde_json::Value) - ) -)] -/// Retrieves all build tasks associated with a specific merge request -/// Returns a list of task filtered by MR number -pub async fn task_query_by_mr( - State(state): State, - Path(mr): Path, -) -> Result>, (StatusCode, Json)> { - let db = &state.conn; - match tasks::Entity::find() - .filter(tasks::Column::Mr.eq(mr)) - .all(db) - .await - { - Ok(models) if !models.is_empty() => { - let dtos = models.into_iter().map(TaskDTO::from_model).collect(); - Ok(Json(dtos)) - } - Ok(_) => Err(( - StatusCode::NOT_FOUND, - Json(serde_json::json!({ "message": "No builds found for the given MR" })), - )), - Err(e) => { - tracing::error!("Database error: {}", e); - Err(( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({ "message": "Internal server error" })), - )) + /// Determine build status based on database fields and active builds + pub fn determine_status(model: &builds::Model, is_active: bool) -> TaskStatusEnum { + if is_active { + TaskStatusEnum::Building + } else if model.end_at.is_none() { + // 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 { + TaskStatusEnum::Completed + } else { + TaskStatusEnum::Failed } } } @@ -1052,31 +827,23 @@ pub async fn task_query_by_mr( /// Task information including current status #[derive(Debug, Serialize, ToSchema)] pub struct TaskInfoDTO { - pub build_id: serde_json::Value, - pub output_files: serde_json::Value, - 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, + pub task_id: String, + pub mr_id: i64, + pub task_name: Option, + pub template: Option, + pub created_at: String, + pub build_list: Vec, } impl TaskInfoDTO { - fn from_model_with_status(model: tasks::Model, status: TaskStatusEnum) -> Self { + fn from_model(model: tasks::Model, build_list: Vec) -> Self { Self { - build_id: model.build_ids, - output_files: model.output_files, - exit_code: model.exit_code, - start_at: model.start_at.with_timezone(&Utc).to_rfc3339(), - end_at: model.end_at.map(|dt| dt.with_timezone(&Utc).to_rfc3339()), - repo_name: model.repo_name, - target: model.target, - arguments: model.arguments, - mr: model.mr, - status, + task_id: model.id.to_string(), + mr_id: model.mr_id, + task_name: model.task_name, + template: model.template, + created_at: model.created_at.with_timezone(&Utc).to_rfc3339(), + build_list, } } } @@ -1085,7 +852,7 @@ impl TaskInfoDTO { get, path = "/tasks/{mr}", params( - ("mr" = String, Path, description = "MR number to filter tasks by") + ("mr" = i64, Path, description = "MR number to filter tasks by") ), responses( (status = 200, description = "All tasks with their current status", body = [TaskInfoDTO]), @@ -1095,35 +862,40 @@ impl TaskInfoDTO { /// Return all tasks with their current status (combining /mr-task and /task-status logic) pub async fn tasks_handler( State(state): State, - Path(mr): Path, + Path(mr): Path, ) -> Result>, (StatusCode, Json)> { let db = &state.conn; let active_builds = state.scheduler.active_builds.clone(); match tasks::Entity::find() - .filter(tasks::Column::Mr.eq(mr)) + .filter(tasks::Column::MrId.eq(mr)) .all(db) .await { - Ok(models) => { - let tasks: Vec = models - .into_iter() - .map(|m| { - let id_str = m.task_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(task_models) => { + let mut tasks: Vec = Vec::new(); + + for m in task_models { + // Query builds associated with this task + let build_models = builds::Entity::find() + .filter(builds::Column::TaskId.eq(m.id)) + .all(db) + .await + .unwrap_or_else(|_| vec![]); + + // Convert build models to DTOs with individual status + let build_list: Vec = build_models + .into_iter() + .map(|build_model| { + let build_id_str = build_model.id.to_string(); + let is_active = active_builds.contains_key(&build_id_str); + let status = BuildDTO::determine_status(&build_model, is_active); + BuildDTO::from_model(build_model, status) + }) + .collect(); + + tasks.push(TaskInfoDTO::from_model(m, build_list)); + } + Ok(Json(tasks)) } Err(e) => { diff --git a/orion-server/src/model/builds.rs b/orion-server/src/model/builds.rs new file mode 100644 index 000000000..2f0412561 --- /dev/null +++ b/orion-server/src/model/builds.rs @@ -0,0 +1,82 @@ +//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.15 + +use chrono::Utc; +use sea_orm::entity::prelude::*; +use sea_orm::{ActiveValue::Set, ConnectionTrait}; +use serde::{Deserialize, Serialize}; + +use crate::scheduler::BuildRequest; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "builds")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: Uuid, + pub task_id: Uuid, + pub exit_code: Option, + pub start_at: DateTimeWithTimeZone, + pub end_at: Option, + pub repo: String, + pub target: String, + #[sea_orm(column_type = "JsonBinary")] + pub args: Option>, + pub output_file: String, + pub created_at: DateTimeWithTimeZone, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::tasks::Entity", + from = "Column::TaskId", + to = "super::tasks::Column::Id", + on_update = "NoAction", + on_delete = "Cascade" + )] + Tasks, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Tasks.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} + +impl Model { + /// Create a new build ActiveModel for database insertion + pub fn create_build( + task_id: Uuid, + repo: String, + target: String, + args: Option>, + ) -> ActiveModel { + let now = Utc::now().into(); + let build_id = Uuid::now_v7(); + ActiveModel { + id: Set(build_id), + task_id: Set(task_id), + exit_code: Set(None), + start_at: Set(now), + end_at: Set(None), + repo: Set(repo), + target: Set(target), + args: Set(args), + output_file: Set(format!("./logs/{}", build_id)), + created_at: Set(now), + } + } + + /// Insert a single build directly into the database + pub async fn insert_build( + task_id: Uuid, + repo: String, + target: String, + build: BuildRequest, + db: &impl ConnectionTrait, + ) -> Result { + let build_model = Self::create_build(task_id, repo, target, build.args); + build_model.insert(db).await + } +} diff --git a/orion-server/src/model/mod.rs b/orion-server/src/model/mod.rs index f053a00b2..5941e361a 100644 --- a/orion-server/src/model/mod.rs +++ b/orion-server/src/model/mod.rs @@ -1 +1,2 @@ +pub mod builds; pub mod tasks; diff --git a/orion-server/src/model/tasks.rs b/orion-server/src/model/tasks.rs index 5b54cbb0c..eb1af0d1b 100644 --- a/orion-server/src/model/tasks.rs +++ b/orion-server/src/model/tasks.rs @@ -1,67 +1,87 @@ -use sea_orm::QuerySelect; +//! `SeaORM` Entity, @generated by sea-orm-codegen 1.1.15 + use sea_orm::entity::prelude::*; +use sea_orm::{ActiveValue::Set, ConnectionTrait, DbErr}; use serde::{Deserialize, Serialize}; +use uuid::Uuid; -/// Database model for build tasks -/// Stores information about build jobs including their status, timing, and metadata #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] #[sea_orm(table_name = "tasks")] pub struct Model { #[sea_orm(primary_key, auto_increment = false)] - pub task_id: Uuid, - #[sea_orm(column_type = "JsonBinary")] - pub build_ids: Json, - #[sea_orm(column_type = "JsonBinary")] - pub output_files: Json, - pub exit_code: Option, - pub start_at: DateTimeWithTimeZone, - pub end_at: Option, - pub repo_name: String, - pub target: String, - pub arguments: String, - pub mr: String, + pub id: Uuid, + pub mr_id: i64, + pub task_name: Option, + #[sea_orm(column_type = "JsonBinary", nullable)] + pub template: Option, + pub created_at: DateTimeWithTimeZone, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] -pub enum Relation {} +pub enum Relation { + #[sea_orm(has_many = "super::builds::Entity")] + Builds, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Builds.def() + } +} impl ActiveModelBehavior for ActiveModel {} impl Model { - /// Retrieves a task record by its UUID from the database - pub async fn get_by_task_id(task_id: Uuid, conn: &DatabaseConnection) -> Option { - Entity::find() - .filter(Column::TaskId.eq(task_id)) - .one(conn) - .await - .expect("Failed to get by `task_id`") - } - - /// Retrieves build_ids list by task_id + /// Retrieves build IDs associated with a task ID pub async fn get_builds_by_task_id( task_id: Uuid, - conn: &DatabaseConnection, + db: &DatabaseConnection, ) -> Option> { - let build_ids: Option = Entity::find() - .filter(Column::TaskId.eq(task_id)) - .select_only() - .column(Column::BuildIds) - .into_tuple::() - .one(conn) + // Query the builds table for all builds with the given task_id + match super::builds::Entity::find() + .filter(super::builds::Column::TaskId.eq(task_id)) + .all(db) .await - .expect("Failed to get `build_ids` by `task_id`"); + { + Ok(builds) => { + // Extract the IDs from the builds + let build_ids: Vec = builds.into_iter().map(|build| build.id).collect(); + Some(build_ids) + } + Err(e) => { + tracing::error!("Failed to fetch builds for task_id {}: {}", task_id, e); + None + } + } + } - build_ids - .map(|json| serde_json::from_value::>(json).unwrap_or_else(|_| Vec::new())) + /// Create a new task ActiveModel for database insertion + pub fn create_task( + task_id: Uuid, + mr_id: i64, + task_name: Option, + template: Option, + created_at: DateTimeWithTimeZone, + ) -> ActiveModel { + ActiveModel { + id: Set(task_id), + mr_id: Set(mr_id), + task_name: Set(task_name), + template: Set(template), + created_at: Set(created_at), + } } - /// Checks if a task with the given task_id exists in the database - pub async fn exists_by_task_id(task_id: Uuid, conn: &DatabaseConnection) -> bool { - Entity::find() - .filter(Column::TaskId.eq(task_id)) - .count(conn) - .await - .expect("Failed to check if task exists") - > 0 + /// Insert a task directly into the database + pub async fn insert_task( + task_id: Uuid, + mr_id: i64, + task_name: Option, + template: Option, + created_at: DateTimeWithTimeZone, + db: &impl ConnectionTrait, + ) -> Result { + let task_model = Self::create_task(task_id, mr_id, task_name, template, created_at); + task_model.insert(db).await } } diff --git a/orion-server/src/scheduler.rs b/orion-server/src/scheduler.rs index 029485877..25adb7ca9 100644 --- a/orion-server/src/scheduler.rs +++ b/orion-server/src/scheduler.rs @@ -1,9 +1,10 @@ -use crate::model::tasks; +use crate::model::builds; use chrono::FixedOffset; use dashmap::DashMap; use orion::ws::WSMessage; use rand::Rng; -use sea_orm::{ActiveModelTrait, ActiveValue::Set, DatabaseConnection, prelude::DateTimeUtc}; +use sea_orm::ActiveModelTrait; +use sea_orm::{ActiveValue::Set, DatabaseConnection, prelude::DateTimeUtc}; use serde::{Deserialize, Serialize}; use std::collections::VecDeque; use std::sync::Arc; @@ -18,11 +19,9 @@ use uuid::Uuid; #[allow(dead_code)] #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct BuildRequest { - pub repo: String, pub buck_hash: String, pub buckconfig_hash: String, pub args: Option>, - pub mr: Option, } /// Pending task waiting for dispatch @@ -30,6 +29,8 @@ pub struct BuildRequest { pub struct PendingTask { pub task_id: Uuid, pub build_id: Uuid, + pub repo: String, + pub mr: i64, pub request: BuildRequest, pub target: String, pub created_at: Instant, @@ -128,12 +129,13 @@ pub struct TaskQueueStats { /// Information about an active build task #[derive(Clone)] +#[allow(dead_code)] pub struct BuildInfo { pub repo: String, pub target: String, pub args: Option>, pub start_at: DateTimeUtc, - pub mr: Option, + pub mr: String, pub _worker_id: String, pub log_file: Arc>, } @@ -234,9 +236,11 @@ impl TaskScheduler { /// Add task-bound build to queue pub async fn enqueue_task( &self, + task_id: Uuid, request: BuildRequest, target: String, - task_id: Uuid, + repo: String, + mr: i64, ) -> Result { let build_id = Uuid::now_v7(); @@ -246,6 +250,8 @@ impl TaskScheduler { request, target, created_at: Instant::now(), + repo, + mr, }; { @@ -361,47 +367,42 @@ impl TaskScheduler { // Create build information let build_info = BuildInfo { - repo: pending_task.request.repo.clone(), + repo: pending_task.repo.clone(), target: pending_task.target.clone(), args: pending_task.request.args.clone(), start_at: chrono::Utc::now(), - mr: pending_task.request.mr.clone(), + mr: pending_task.mr.to_string(), _worker_id: chosen_id.clone(), log_file, }; - // Save to database - let model = tasks::ActiveModel { + // Insert build record + let _ = builds::ActiveModel { + id: Set(pending_task.build_id), task_id: Set(pending_task.task_id), - build_ids: Set(serde_json::json!([pending_task.build_id])), - output_files: Set(serde_json::json!([format!( - "{}/{}", - get_build_log_dir(), - pending_task.build_id - )])), exit_code: Set(None), start_at: Set(build_info .start_at .with_timezone(&FixedOffset::east_opt(0).unwrap())), end_at: Set(None), - repo_name: Set(build_info.repo.clone()), - target: Set(build_info.target.clone()), - arguments: Set(build_info.args.clone().unwrap_or_default().join(" ")), - mr: Set(build_info.mr.clone().unwrap_or_default()), - }; - - if let Err(e) = model.insert(&self.conn).await { - tracing::error!("Failed to insert queued task into DB: {}", e); - return Err(format!("Failed to create task in database: {e}")); + repo: Set(build_info.repo.clone()), + target: Set("//...".to_string()), + args: Set(build_info.args.clone()), + output_file: Set(format!("{}/{}", get_build_log_dir(), pending_task.build_id)), + created_at: Set(build_info + .start_at + .with_timezone(&FixedOffset::east_opt(0).unwrap())), } + .insert(&self.conn) + .await; // Create WebSocket message let msg = WSMessage::Task { id: pending_task.build_id.to_string(), - repo: pending_task.request.repo, + repo: pending_task.repo, target: pending_task.target, args: pending_task.request.args, - mr: pending_task.request.mr.unwrap_or_default(), + mr: pending_task.mr.to_string(), }; // Send task to worker @@ -477,7 +478,7 @@ impl TaskScheduler { "Expired build: {}/{} ({})", task.task_id, task.build_id, - task.request.repo + task.repo ); } } @@ -649,28 +650,28 @@ mod tests { task_id: Uuid::now_v7(), build_id: Uuid::now_v7(), request: BuildRequest { - repo: "test1".to_string(), buck_hash: "hash1".to_string(), buckconfig_hash: "config1".to_string(), args: None, - mr: None, }, target: "target1".to_string(), created_at: Instant::now(), + repo: "/test/repo".to_string(), + mr: 123456, }; let task2 = PendingTask { task_id: Uuid::now_v7(), build_id: Uuid::now_v7(), request: BuildRequest { - repo: "test2".to_string(), buck_hash: "hash2".to_string(), buckconfig_hash: "config2".to_string(), args: None, - mr: None, }, target: "target2".to_string(), created_at: Instant::now(), + repo: "/test2/repo".to_string(), + mr: 123457, }; // Test FIFO behavior @@ -679,11 +680,11 @@ mod tests { let dequeued1 = queue.dequeue().unwrap(); assert_eq!(dequeued1.build_id, task1.build_id); - assert_eq!(dequeued1.request.repo, "test1"); + assert_eq!(dequeued1.repo, "/test/repo"); let dequeued2 = queue.dequeue().unwrap(); assert_eq!(dequeued2.build_id, task2.build_id); - assert_eq!(dequeued2.request.repo, "test2"); + assert_eq!(dequeued2.repo, "/test2/repo"); } /// Test queue capacity limit @@ -700,14 +701,14 @@ mod tests { task_id: Uuid::now_v7(), build_id: Uuid::now_v7(), request: BuildRequest { - repo: "test".to_string(), buck_hash: "hash".to_string(), buckconfig_hash: "config".to_string(), args: None, - mr: None, }, target: "target".to_string(), created_at: Instant::now(), + repo: "/test/repo".to_string(), + mr: 123456, }; // Fill queue to capacity diff --git a/orion-server/src/server.rs b/orion-server/src/server.rs index f519a00c2..90c5a9624 100644 --- a/orion-server/src/server.rs +++ b/orion-server/src/server.rs @@ -16,27 +16,23 @@ use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; use crate::api::{self, AppState}; -use crate::model::tasks; +use crate::model::{builds, tasks}; /// OpenAPI documentation configuration #[derive(OpenApi)] #[openapi( paths( api::task_handler, - api::task_build_handler, - api::task_status_handler, api::task_build_list_handler, api::task_output_handler, api::task_history_output_handler, - api::task_query_by_mr, api::tasks_handler, ), components( schemas( crate::scheduler::BuildRequest, crate::scheduler::LogSegment, - api::TaskStatus, api::TaskStatusEnum, - api::TaskDTO, + api::BuildDTO, api::TaskInfoDTO ) @@ -169,14 +165,14 @@ async fn start_health_check_task(state: AppState) { ); state.scheduler.active_builds.remove(&task_id); - let update_res = tasks::Entity::update_many() - .set(tasks::ActiveModel { + let update_res = builds::Entity::update_many() + .set(builds::ActiveModel { end_at: Set(Some( Utc::now().with_timezone(&FixedOffset::east_opt(0).unwrap()), )), ..Default::default() }) - .filter(tasks::Column::TaskId.eq(task_id.parse::().unwrap())) + .filter(builds::Column::TaskId.eq(task_id.parse::().unwrap())) .exec(&state.conn) .await;