Skip to content

fix[mono]: fix files-changed invalid after mr merged #1097

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
genedna merged 1 commit into from
Jun 5, 2025
Merged

fix[mono]: fix files-changed invalid after mr merged #1097

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 64 additions & 59 deletions ceres/src/api_service/mono_api_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,8 +224,8 @@ impl MonoApiService {
self.update_parent_tree(path, tree_vec, commit)
.await
.unwrap();
// remove refs start with path
storage.remove_refs(&mr.path).await.unwrap();
// remove refs start with path exceprt mr type

Copilot AI Jun 5, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Typo in comment: exceprt should be except.

Suggested change
// remove refs start with path exceprt mr type
// remove refs start with path except mr type

Copilot uses AI. Check for mistakes.
storage.remove_none_mr_refs(&mr.path).await.unwrap();
// TODO: self.clean_dangling_commits().await;
}
// update mr
Expand Down Expand Up @@ -315,79 +315,84 @@ impl MonoApiService {
env::set_current_dir(&base_path).unwrap();
let clone_path = base_path.join(mr_link);
if !fs::exists(&clone_path).unwrap() {
// fs::remove_dir_all(&clone_path).unwrap();
Command::new("mkdir")
.arg(mr_link)
.output()
.await
.expect("Failed to mkdir");
// cd mr
env::set_current_dir(&clone_path).unwrap();
// libra init
Command::new("libra")
.arg("init")
.output()
.await
.expect("Failed to execute libra init");
// libra remote add origin http://localhost:8000/project
let git_remote = if mr.path.starts_with("/") {
format!("{}{}", listen_addr, mr.path)
} else {
format!("{}/{}", listen_addr, mr.path)
};
Command::new("libra")
.arg("remote")
.arg("add")
.arg("origin")
.arg(git_remote)
.output()
.await
.expect("Failed to execute libra remote add");
// libra fetch origin refs/mr/PC0EPG1L
Command::new("libra")
.arg("fetch")
.arg("origin")
.arg(format!("refs/mr/{}", mr_link))
.output()
.await
.expect("Failed to execute libra fetch");
// libra branch PC0EPG1L origin/mr/PC0EPG1L
Command::new("libra")
.arg("branch")
.arg(mr_link)
.arg(format!("origin/mr/{}", mr_link))
.output()
.await
.expect("Failed to execute libra branch");
// libra switch PC0EPG1L
Command::new("libra")
.arg("switch")
.arg(mr_link)
.output()
.await
.expect("Failed to execute libra switch");
let result = self.run_libra_diff(&mr, listen_addr, &clone_path).await;
if result.is_err() && fs::exists(&clone_path).unwrap() {
fs::remove_dir_all(&clone_path).unwrap();
}
} else {
env::set_current_dir(&clone_path).unwrap();
}
// libra diff --old hash
tracing::debug!("Run libra Command: libra diff --old {}", mr.from_hash);
let output = Command::new("libra")
.arg("diff")
.arg("--old")
.arg(mr.from_hash)
.output()
.await
.expect("Failed to execute libra diff");
if output.status.success() {
return Ok(String::from_utf8_lossy(&output.stdout).to_string());
} else {
.await?;

let stderr_str = String::from_utf8_lossy(&output.stderr);

if !stderr_str.trim().is_empty() || !output.status.success() {
tracing::error!(
"Command failed: {}",
String::from_utf8_lossy(&output.stderr)
);
fs::remove_dir_all(&clone_path).unwrap();
} else {
return Ok(String::from_utf8_lossy(&output.stdout).to_string());
}
}
Ok(String::new())
}

async fn run_libra_diff(
&self,
mr: &callisto::mega_mr::Model,
listen_addr: &str,
clone_path: &PathBuf,
) -> Result<(), anyhow::Error> {
Command::new("mkdir").arg(&mr.link).output().await?;
env::set_current_dir(clone_path).unwrap();
Command::new("libra").arg("init").output().await?;
let git_remote = if mr.path.starts_with("/") {
format!("{}{}", listen_addr, mr.path)
} else {
format!("{}/{}", listen_addr, mr.path)
};
tracing::debug!("Run libra Command: libra remote add origin {}", &git_remote);
Command::new("libra")
.arg("remote")
.arg("add")
.arg("origin")
.arg(git_remote)
.output()
.await?;
tracing::debug!("Run libra Command: libra fetch origin refs/mr/{}", &mr.link);
Command::new("libra")
.arg("fetch")
.arg("origin")
.arg(format!("refs/mr/{}", &mr.link))
.output()
.await?;
tracing::debug!(
"Run libra Command: libra branch {} origin/mr/{}",
&mr.link,
&mr.link
);
Command::new("libra")
.arg("branch")
.arg(&mr.link)
.arg(format!("origin/mr/{}", &mr.link))
.output()
.await?;
tracing::debug!("Run libra Command: libra switch {}", &mr.link);
Command::new("libra")
.arg("switch")
.arg(&mr.link)
.output()
.await?;
Ok(())
}
}

#[cfg(test)]
Expand Down
66 changes: 44 additions & 22 deletions ceres/src/pack/import_repo.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,18 @@ use std::{
collections::{HashMap, HashSet},
path::PathBuf,
str::FromStr,
sync::atomic::{AtomicUsize, Ordering},
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
};

use async_trait::async_trait;
use futures::{future::join_all, StreamExt};
use tokio::sync::mpsc::{self, UnboundedReceiver};
use tokio::sync::{
mpsc::{self, UnboundedReceiver},
Semaphore,
};
use tokio_stream::wrappers::ReceiverStream;

use callisto::{mega_tree, raw_blob, sea_orm_active_enums::RefTypeEnum};
Expand Down Expand Up @@ -58,20 +64,37 @@ impl PackHandler for ImportRepo {
) -> Result<Option<Commit>, GitError> {
let storage = self.context.services.git_db_storage.clone();
let mut entry_list = vec![];
let semaphore = Arc::new(Semaphore::new(8));

Copilot AI Jun 5, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[nitpick] The semaphore limit 8 is a magic number. Extract it into a named constant or configuration to clarify its purpose and make it configurable.

Suggested change
let semaphore = Arc::new(Semaphore::new(8));
let semaphore = Arc::new(Semaphore::new(SEMAPHORE_LIMIT));

Copilot uses AI. Check for mistakes.
let mut join_tasks = vec![];
let repo_id = self.repo.repo_id;

while let Some(entry) = receiver.recv().await {
entry_list.push(entry);
if entry_list.len() >= 10000 {
if entry_list.len() >= 1000 {
let stg_clone = storage.clone();
let acquired = semaphore.clone().acquire_owned().await.unwrap();

let handle = tokio::spawn(async move {
stg_clone.save_entry(repo_id, entry_list).await.unwrap();
let _acquired = acquired;
stg_clone.save_entry(repo_id, entry_list).await
});
join_tasks.push(handle);
entry_list = vec![];
}
}
join_all(join_tasks).await;
let results = join_all(join_tasks).await;
for (i, res) in results.into_iter().enumerate() {
match res {
Ok(Ok(())) => {}
Ok(Err(e)) => {
tracing::error!("Task {} save_entry Err: {:?}", i, e);
}
Err(join_err) => {
tracing::error!("Task {} panic or cancle: {:?}", i, join_err);
}
}
}

storage.save_entry(repo_id, entry_list).await.unwrap();
self.attach_to_monorepo_parent().await.unwrap();
Ok(None)
Expand Down Expand Up @@ -126,29 +149,28 @@ impl PackHandler for ImportRepo {
}
}

let mut blob_handler = vec![];
for chunk in bids.chunks(10000) {
// let mut blob_handler = vec![];
for chunk in bids.chunks(1000) {
let raw_storage = raw_storage.clone();
let sender_clone = entry_tx.clone();
let chunk_clone = chunk.to_vec();
let handler = tokio::spawn(async move {
let mut blob_stream =
raw_storage.get_raw_blobs_stream(chunk_clone).await.unwrap();
while let Some(model) = blob_stream.next().await {
match model {
Ok(m) => {
// todo handle storage type
let b: Blob = m.into();
let entry: Entry = b.into();
sender_clone.send(entry).await.unwrap();
}
Err(err) => eprintln!("Error: {:?}", err),
// let handler = tokio::spawn(async move {
Comment on lines +152 to +157

Copilot AI Jun 5, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[nitpick] There are several commented-out blocks related to blob handling. Remove or restore this code to reduce clutter and improve readability.

Copilot uses AI. Check for mistakes.
let mut blob_stream = raw_storage.get_raw_blobs_stream(chunk_clone).await.unwrap();
while let Some(model) = blob_stream.next().await {
match model {
Ok(m) => {
// TODO handle storage type
let b: Blob = m.into();
let entry: Entry = b.into();
sender_clone.send(entry).await.unwrap();
}
Err(err) => eprintln!("Error: {:?}", err),
}
});
blob_handler.push(handler);
}
// });
// blob_handler.push(handler);
}
join_all(blob_handler).await;
// join_all(blob_handler).await;
tracing::info!("send blobs end");

let tags = storage.get_tags_by_repo_id(repo_id).await.unwrap();
Expand Down
77 changes: 7 additions & 70 deletions ceres/src/pack/monorepo.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,12 @@ impl PackHandler for MonoRepo {
let storage = self.context.services.mono_storage.clone();

let result = storage.get_refs(self.path.to_str().unwrap()).await.unwrap();
let refs = if !result.is_empty() {

let heads_exist = result
.iter()
.any(|x| x.ref_name == common::utils::MEGA_BRANCH_NAME);

let refs = if heads_exist {
let refs: Vec<Refs> = result.into_iter().map(|x| x.into()).collect();
refs
} else {
Expand Down Expand Up @@ -164,75 +169,7 @@ impl PackHandler for MonoRepo {

// monorepo full pack should follow the shallow clone command 'git clone --depth=1'
async fn full_pack(&self, want: Vec<String>) -> Result<ReceiverStream<Vec<u8>>, GitError> {
let pack_config = &self.context.config.pack;
let storage = self.context.services.mono_storage.clone();
let obj_num = AtomicUsize::new(0);
let mut trees = Vec::new();

let refs = storage
.get_ref(self.path.to_str().unwrap())
.await
.unwrap()
.unwrap();
let commit: Commit = storage
.get_commit_by_hash(&refs.ref_commit_hash)
.await
.unwrap()
.unwrap()
.into();
let tree: Tree = storage
.get_tree_by_hash(&refs.ref_tree_hash)
.await
.unwrap()
.unwrap()
.into();
trees.push(tree.clone());
let mut exist_objs = HashSet::new();
let mut counted_obj = HashSet::new();
self.traverse_for_count(tree.clone(), &exist_objs, &mut counted_obj, &obj_num)
.await;
obj_num.fetch_add(1, Ordering::SeqCst);

exist_objs.extend(counted_obj.clone());

let (entry_tx, entry_rx) = mpsc::channel(pack_config.channel_message_size);
let (stream_tx, stream_rx) = mpsc::channel(pack_config.channel_message_size);

let want_c = want.first().unwrap();
if refs.ref_commit_hash != *want_c {
let refs = storage
.get_ref_by_commit(self.path.to_str().unwrap(), want_c)
.await
.unwrap()
.unwrap();
let commit: Commit = storage
.get_commit_by_hash(&refs.ref_commit_hash)
.await
.unwrap()
.unwrap()
.into();
let tree: Tree = storage
.get_tree_by_hash(&refs.ref_tree_hash)
.await
.unwrap()
.unwrap()
.into();
trees.push(tree.clone());
self.traverse_for_count(tree, &exist_objs, &mut counted_obj, &obj_num)
.await;
obj_num.fetch_add(1, Ordering::SeqCst);
entry_tx.send(commit.into()).await.unwrap();
}

let encoder = PackEncoder::new(obj_num.into_inner(), 0, stream_tx);
encoder.encode_async(entry_rx).await.unwrap();
let mut send_exist = HashSet::new();
for tree in trees {
self.traverse(tree, &mut send_exist, Some(&entry_tx)).await;
}
entry_tx.send(commit.into()).await.unwrap();
drop(entry_tx);
Ok(ReceiverStream::new(stream_rx))
self.incremental_pack(want, Vec::new()).await
}

async fn incremental_pack(
Expand Down
2 changes: 2 additions & 0 deletions jupiter/migration/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ use sea_orm_migration::schema::big_integer;

mod m20250314_025943_init;
mod m20250427_031332_add_mr_refs_tag;
mod m20250605_013340_alter_mega_mr_index;

pub struct Migrator;

Expand All @@ -12,6 +13,7 @@ impl MigratorTrait for Migrator {
vec![
Box::new(m20250314_025943_init::Migration),
Box::new(m20250427_031332_add_mr_refs_tag::Migration),
Box::new(m20250605_013340_alter_mega_mr_index::Migration),
]
}
}
Expand Down
Loading
Loading