update menorepo mr

ceres/src/pack/handler.rs

@@ -1,10 +1,9 @@
 use std::{
     collections::HashSet,
-    io::Cursor,
     pin::Pin,
     sync::{
         atomic::{AtomicUsize, Ordering},
-        mpsc::{self, Receiver},
+        mpsc::Receiver,
     },
 };
 
@@ -42,27 +41,6 @@ pub trait PackHandler: Send + Sync {
         (head_hash, refs)
     }
 
-    async fn unpack(&self, pack_config: &PackConfig, pack_file: Bytes) -> Result<(), GitError> {
-        // #[cfg(debug_assertions)]
-        // {
-        //     let datetime = chrono::Utc::now().naive_utc();
-        //     let path = format!("{}.pack", datetime);
-        //     let mut output = std::fs::File::create(path).unwrap();
-        //     output.write_all(&pack_file).unwrap();
-        // }
-        let (sender, receiver) = mpsc::channel();
-        let p = Pack::new(
-            None,
-            Some(1024 * 1024 * 1024 * pack_config.pack_decode_mem_size),
-            Some(pack_config.pack_decode_cache_path.clone()),
-            pack_config.clean_cache_after_decode,
-        );
-
-        p.decode_async(Cursor::new(pack_file), sender); //Pack moved here
-
-        self.save_entry(receiver).await
-    }
-
     async fn unpack_stream(
         &self,
         pack_config: &PackConfig,
@@ -81,7 +59,7 @@ pub trait PackHandler: Send + Sync {
         Ok(receiver)
     }
 
-    async fn save_entry(&self, rx: Receiver<Entry>) -> Result<(), GitError>;
+    async fn handle_receiver(&self, rx: Receiver<Entry>) -> Result<(), GitError>;
 
     /// Asynchronously retrieves the full pack data for the specified repository path.
     /// This function collects commits and nodes from the storage and packs them into

ceres/src/pack/import_repo.rs

@@ -49,7 +49,7 @@ impl PackHandler for ImportRepo {
         self.find_head_hash(refs)
     }
 
-    async fn save_entry(&self, receiver: Receiver<Entry>) -> Result<(), GitError> {
+    async fn handle_receiver(&self, receiver: Receiver<Entry>) -> Result<(), GitError> {
         let storage = self.context.services.git_db_storage.clone();
         let mut entry_list = vec![];
         let mut join_tasks = vec![];

ceres/src/pack/monorepo.rs

@@ -10,6 +10,9 @@ use std::{
 };
 
 use async_trait::async_trait;
+use futures::future::join_all;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
 
 use callisto::raw_blob;
 use common::{errors::MegaError, utils::MEGA_BRANCH_NAME};
@@ -23,8 +26,6 @@ use mercury::{
         pack::entry::Entry,
     },
 };
-use tokio::sync::mpsc;
-use tokio_stream::wrappers::ReceiverStream;
 use venus::{
     import_repo::import_refs::{RefCommand, Refs},
     monorepo::mr::MergeRequest,
@@ -114,12 +115,12 @@ impl PackHandler for MonoRepo {
         self.find_head_hash(refs)
     }
 
-    async fn save_entry(&self, receiver: Receiver<Entry>) -> Result<(), GitError> {
+    async fn handle_receiver(&self, receiver: Receiver<Entry>) -> Result<(), GitError> {
         let storage = self.context.services.mega_storage.clone();
 
         let (mut mr, mr_exist) = self.get_mr().await;
 
-        let mut commit_size = 0;
+        let mut unpack_res = Ok(());
         if mr_exist {
             if mr.from_hash == self.from_hash.clone().unwrap() {
                 let to_hash = self.to_hash.clone().unwrap();
@@ -130,7 +131,20 @@ impl PackHandler for MonoRepo {
                         .add_mr_comment(mr.id, 0, Some(comment))
                         .await
                         .unwrap();
-                    commit_size = self.save_entry(receiver).await;
+                    unpack_res = self.save_entry(receiver).await;
+                    if unpack_res.is_err() {
+                        mr.close();
+                        storage
+                            .add_mr_comment(
+                                mr.id,
+                                0,
+                                Some("Mega closed MR due to multi commit detected".to_string()),
+                            )
+                            .await
+                            .unwrap();
+                    }
+                } else {
+                    tracing::info!("repeat commit with mr: {}, do nothing", mr.id);
                 }
             } else {
                 mr.close();
@@ -141,23 +155,12 @@ impl PackHandler for MonoRepo {
             }
             storage.update_mr(mr.clone()).await.unwrap();
         } else {
-            commit_size = self.save_entry(receiver).await;
-
-            storage.save_mr(mr.clone()).await.unwrap();
+            unpack_res = self.save_entry(receiver).await;
+            if unpack_res.is_ok() {
+                storage.save_mr(mr.clone()).await.unwrap();
+            }
         };
-
-        if commit_size > 1 {
-            mr.close();
-            storage
-                .add_mr_comment(
-                    mr.id,
-                    0,
-                    Some("Mega closed MR due to multi commit detected".to_string()),
-                )
-                .await
-                .unwrap();
-        }
-        Ok(())
+        unpack_res
     }
 
     // monorepo full pack should follow the shallow clone command 'git clone --depth=1'
@@ -366,22 +369,36 @@ impl MonoRepo {
         )
     }
 
-    async fn save_entry(&self, receiver: Receiver<Entry>) -> i32 {
+    async fn save_entry(&self, receiver: Receiver<Entry>) -> Result<(), GitError> {
         let storage = self.context.services.mega_storage.clone();
         let mut entry_list = Vec::new();
-
-        let mut commit_size = 0;
+        let mut join_tasks = vec![];
+        let mut commit_id = String::new();
         for entry in receiver {
-            if entry.obj_type == ObjectType::Commit {
-                commit_size += 1;
+            if commit_id.is_empty() {
+                if entry.obj_type == ObjectType::Commit {
+                    commit_id = entry.hash.to_plain_str();
+                }
+            } else {
+                if entry.obj_type == ObjectType::Commit {
+                    return Err(GitError::CustomError(
+                        "only single commit support in each commit".to_string(),
+                    ));
+                }
+                if entry_list.len() >= 1000 {
+                    let stg_clone = storage.clone();
+                    let commit_id = commit_id.clone();
+                    let handle = tokio::spawn(async move {
+                        stg_clone.save_entry(&commit_id, entry_list).await.unwrap();
+                    });
+                    join_tasks.push(handle);
+                    entry_list = vec![];
+                }
             }
             entry_list.push(entry);
-            if entry_list.len() >= 1000 {
-                storage.save_entry(entry_list).await.unwrap();
-                entry_list = Vec::new();
-            }
         }
-        storage.save_entry(entry_list).await.unwrap();
-        commit_size
+        join_all(join_tasks).await;
+        storage.save_entry(&commit_id, entry_list).await.unwrap();
+        Ok(())
     }
 }

ceres/src/protocol/smart.rs

@@ -217,13 +217,13 @@ impl SmartProtocol {
             .await
             .unwrap();
 
+        // can not block main thread here.
         let ph_clone = pack_handler.clone();
-        let unpack_success = tokio::task::spawn_blocking(move || {
+        let unpack_result = tokio::task::spawn_blocking(move || {
             let handle = tokio::runtime::Handle::current();
-            handle.block_on(async { ph_clone.save_entry(receiver).await })
+            handle.block_on(async { ph_clone.handle_receiver(receiver).await })
         })
-        .await
-        .is_ok();
+        .await.unwrap();
 
         // write "unpack ok\n to report"
         add_pkt_line_string(&mut report_status, "unpack ok\n".to_owned());
@@ -240,81 +240,17 @@ impl SmartProtocol {
                 // a.The reference can have changed since the reference discovery phase was originally sent, meaning someone pushed in the meantime.
                 // b.The reference being pushed could be a non-fast-forward reference and the update hooks or configuration could be set to not allow that, etc.
                 // c.Also, some references can be updated while others can be rejected.
-                if unpack_success {
-                    if !default_exist {
-                        command.default_branch = true;
-                        default_exist = true;
+                match unpack_result {
+                    Ok(_) => {
+                        if !default_exist {
+                            command.default_branch = true;
+                            default_exist = true;
+                        }
+                        pack_handler.update_refs(&command).await.unwrap();
                     }
-                    pack_handler.update_refs(&command).await.unwrap();
-                } else {
-                    command.failed(String::from("unpack failed"));
-                }
-            }
-            add_pkt_line_string(&mut report_status, command.get_status());
-        }
-        report_status.put(&PKT_LINE_END_MARKER[..]);
-        let length = report_status.len();
-        let mut buf = self.build_side_band_format(report_status, length);
-        buf.put(&PKT_LINE_END_MARKER[..]);
-        Ok(buf.into())
-    }
-
-    // preserve for ssh server
-    pub async fn git_receive_pack(&mut self, mut body_bytes: Bytes) -> Result<Bytes> {
-        if body_bytes.len() < 1_000 {
-            tracing::debug!("bytes from client: {:?}", body_bytes);
-        } else {
-            tracing::debug!("{} bytes from client", body_bytes.len())
-        }
-        while !body_bytes.starts_with(&[b'P', b'A', b'C', b'K']) && !body_bytes.is_empty() {
-            let (bytes_take, mut pkt_line) = read_pkt_line(&mut body_bytes);
-            if bytes_take != 0 {
-                let command = self.parse_ref_command(&mut pkt_line);
-                self.parse_capabilities(&String::from_utf8(pkt_line.to_vec()).unwrap());
-                tracing::debug!(
-                    "parse ref_command: {:?}, with caps:{:?}",
-                    command,
-                    self.capabilities
-                );
-                self.command_list.push(command);
-            }
-        }
-        // handles situation when client send b"0000"
-        if body_bytes.is_empty() {
-            return Ok(body_bytes);
-        }
-        // After receiving the pack data from the sender, the receiver sends a report
-        let mut report_status = BytesMut::new();
-        let pack_handler = self.pack_handler().await;
-        //1. unpack progress
-        let unpack_success = pack_handler
-            .unpack(&self.context.config.pack, body_bytes)
-            .await
-            .is_ok();
-
-        // write "unpack ok\n to report"
-        add_pkt_line_string(&mut report_status, "unpack ok\n".to_owned());
-
-        let mut default_exist = pack_handler.check_default_branch().await;
-
-        //2. update each refs and build report
-        for mut command in self.command_list.clone() {
-            if command.ref_type == RefType::Tag {
-                // just update if refs type is tag
-                pack_handler.update_refs(&command).await.unwrap();
-            } else {
-                // Updates can be unsuccessful for a number of reasons.
-                // a.The reference can have changed since the reference discovery phase was originally sent, meaning someone pushed in the meantime.
-                // b.The reference being pushed could be a non-fast-forward reference and the update hooks or configuration could be set to not allow that, etc.
-                // c.Also, some references can be updated while others can be rejected.
-                if unpack_success {
-                    if !default_exist {
-                        command.default_branch = true;
-                        default_exist = true;
+                    Err(ref err) => {
+                        command.failed(err.to_string());
                     }
-                    pack_handler.update_refs(&command).await.unwrap();
-                } else {
-                    command.failed(String::from("unpack failed"));
                 }
             }
             add_pkt_line_string(&mut report_status, command.get_status());

gateway/src/api_service/import_service.rs

@@ -21,8 +21,8 @@ pub struct ImportRepoService {
 
 #[async_trait]
 impl ApiHandler for ImportRepoService {
-    async fn get_blob_as_string(&self, _object_id: &str) -> Result<BlobObjects, GitError> {
-        unreachable!()
+    async fn get_blob_as_string(&self, _path: PathBuf, _filename: &str) -> Result<BlobObjects, GitError> {
+        unimplemented!()
     }
 
     async fn get_latest_commit(&self, path: PathBuf) -> Result<LatestCommitInfo, GitError> {

gateway/src/api_service/mod.rs

@@ -16,7 +16,7 @@ const SIGNATURE_END: &str = "-----END PGP SIGNATURE-----";
 
 #[async_trait]
 pub trait ApiHandler: Send + Sync {
-    async fn get_blob_as_string(&self, object_id: &str) -> Result<BlobObjects, GitError>;
+    async fn get_blob_as_string(&self, path: PathBuf, filename: &str) -> Result<BlobObjects, GitError>;
 
     async fn get_latest_commit(&self, path: PathBuf) -> Result<LatestCommitInfo, GitError>;
 
@@ -52,7 +52,7 @@ pub trait ApiHandler: Send + Sync {
             let truncated_text = filtered_text.chars().take(50).collect::<String>();
             truncated_text.to_owned()
         } else {
-            "".to_owned()
+            content
         }
     }
 }