fix bug: incremental_pack repeated count obj num

ceres/src/pack/handler.rs

@@ -62,13 +62,14 @@ pub trait PackHandler: Send + Sync {
         &self,
         tree: Tree,
         exist_objs: &HashSet<String>,
+        counted_obj: &mut HashSet<String>,
         obj_num: &AtomicUsize,
     ) {
         let mut search_tree_ids = vec![];
         let mut search_blob_ids = vec![];
         for item in &tree.tree_items {
             let hash = item.id.to_plain_str();
-            if !exist_objs.contains(&hash) {
+            if !exist_objs.contains(&hash) && counted_obj.insert(hash.clone()) {
                 if item.mode == TreeItemMode::Tree {
                     search_tree_ids.push(hash.clone())
                 } else {
@@ -79,29 +80,47 @@ pub trait PackHandler: Send + Sync {
         obj_num.fetch_add(search_blob_ids.len(), Ordering::SeqCst);
         let trees = self.get_trees_by_hashes(search_tree_ids).await.unwrap();
         for t in trees {
-            self.traverse_for_count(t, exist_objs, obj_num).await;
+            self.traverse_for_count(t, exist_objs, counted_obj, obj_num)
+                .await;
         }
         obj_num.fetch_add(1, Ordering::SeqCst);
     }
 
+    /// Traverse a tree structure asynchronously.
+    ///
+    /// This function traverses a given tree, keeps track of processed objects, and optionally sends
+    /// traversal data to a provided sender. The function will:
+    /// 1. Traverse the tree and calculate the quantities of tree and blob items.
+    /// 2. If a sender is provided, send blob and tree data via the sender.
+    ///
+    /// # Parameters
+    /// - `tree`: The tree structure to traverse.
+    /// - `exist_objs`: A mutable reference to a set containing already processed object IDs.
+    /// - `sender`: An optional sender for sending traversal data.
+    ///
+    /// # Details
+    /// - The function processes tree items, distinguishing between tree and blob items.
+    /// - It collects IDs of items that have not been processed yet.
+    /// - It retrieves and sends blob data if a sender is provided.
+    /// - It recursively traverses sub-trees.
+    /// - It sends the entire tree data if a sender is provided.
     async fn traverse(
         &self,
         tree: Tree,
         exist_objs: &mut HashSet<String>,
         sender: Option<&tokio::sync::mpsc::Sender<Entry>>,
     ) {
-        exist_objs.insert(tree.id.to_plain_str());
         let mut search_tree_ids = vec![];
         let mut search_blob_ids = vec![];
+
         for item in &tree.tree_items {
             let hash = item.id.to_plain_str();
-            if !exist_objs.contains(&hash) {
+            if exist_objs.insert(hash.clone()) {
                 if item.mode == TreeItemMode::Tree {
-                    search_tree_ids.push(hash.clone())
+                    search_tree_ids.push(hash);
                 } else {
-                    search_blob_ids.push(hash.clone());
+                    search_blob_ids.push(hash);
                 }
-                exist_objs.insert(hash);
             }
         }
 
@@ -112,10 +131,12 @@ pub trait PackHandler: Send + Sync {
                 sender.send(blob.into()).await.unwrap();
             }
         }
+
         let trees = self.get_trees_by_hashes(search_tree_ids).await.unwrap();
         for t in trees {
             self.traverse(t, exist_objs, sender).await;
         }
+
         if let Some(sender) = sender {
             sender.send(tree.into()).await.unwrap();
         }

ceres/src/pack/import_repo.rs

@@ -1,3 +1,5 @@
+use std::{collections::{HashMap, HashSet}, str::FromStr, sync::atomic::{AtomicUsize, Ordering}};
+
 use async_trait::async_trait;
 use bytes::Bytes;
 use tokio::sync::mpsc;
@@ -9,7 +11,7 @@ use jupiter::{
     context::Context,
     storage::{batch_query_by_columns, GitStorageProvider},
 };
-use mercury::internal::pack::encode::PackEncoder;
+use mercury::{hash::SHA1, internal::pack::encode::PackEncoder};
 use mercury::{
     errors::GitError,
     internal::{
@@ -119,10 +121,95 @@ impl PackHandler for ImportRepo {
 
     async fn incremental_pack(
         &self,
-        _want: Vec<String>,
-        _have: Vec<String>,
+        mut want: Vec<String>,
+        have: Vec<String>,
     ) -> Result<ReceiverStream<Vec<u8>>, GitError> {
-        unimplemented!()
+        let pack_config = &self.context.config.pack;
+        let storage = self.context.services.git_db_storage.clone();
+        let obj_num = AtomicUsize::new(0);
+
+        let mut exist_objs = HashSet::new();
+
+        let mut want_commits: Vec<Commit> = storage
+            .get_commits_by_hashes(&self.repo, &want)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|x| x.into())
+            .collect();
+        let mut traversal_list: Vec<Commit> = want_commits.clone();
+
+        // traverse commit's all parents to find the commit that client does not have
+        while let Some(temp) = traversal_list.pop() {
+            for p_commit_id in temp.parent_commit_ids {
+                let p_commit_id = p_commit_id.to_plain_str();
+
+                if !have.contains(&p_commit_id) && !want.contains(&p_commit_id) {
+                    let parent: Commit = storage
+                        .get_commit_by_hash(&self.repo, &p_commit_id)
+                        .await
+                        .unwrap()
+                        .unwrap()
+                        .into();
+                    want_commits.push(parent.clone());
+                    want.push(p_commit_id);
+                    traversal_list.push(parent);
+                }
+            }
+        }
+
+        let want_tree_ids = want_commits
+            .iter()
+            .map(|c| c.tree_id.to_plain_str())
+            .collect();
+        let want_trees: HashMap<SHA1, Tree> = storage
+            .get_trees_by_hashes(&self.repo, want_tree_ids)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|m| (SHA1::from_str(&m.tree_id).unwrap(), m.into()))
+            .collect();
+
+        obj_num.fetch_add(want_commits.len(), Ordering::SeqCst);
+
+        let have_commits = storage.get_commits_by_hashes(&self.repo, &have).await.unwrap();
+        let have_trees = storage
+            .get_trees_by_hashes(&self.repo, have_commits.iter().map(|x| x.tree.clone()).collect())
+            .await
+            .unwrap();
+        // traverse to get exist_objs
+        for have_tree in have_trees {
+            self.traverse(have_tree.into(), &mut exist_objs, None).await;
+        }
+
+        let mut counted_obj = HashSet::new();
+        // traverse for get obj nums
+        for c in want_commits.clone() {
+            self.traverse_for_count(
+                want_trees.get(&c.tree_id).unwrap().clone(),
+                &exist_objs,
+                &mut counted_obj,
+                &obj_num,
+            )
+            .await;
+        }
+        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 encoder = PackEncoder::new(obj_num.into_inner(), 0, stream_tx);
+        encoder.encode_async(entry_rx).await.unwrap();
+
+        for c in want_commits {
+            self.traverse(
+                want_trees.get(&c.tree_id).unwrap().clone(),
+                &mut exist_objs,
+                Some(&entry_tx),
+            )
+            .await;
+            entry_tx.send(c.into()).await.unwrap();
+        }
+        drop(entry_tx);
+
+        Ok(ReceiverStream::new(stream_rx))
     }
 
     async fn get_trees_by_hashes(&self, hashes: Vec<String>) -> Result<Vec<Tree>, MegaError> {

ceres/src/pack/monorepo.rs

@@ -188,7 +188,7 @@ impl PackHandler for MonoRepo {
             .unwrap()
             .unwrap()
             .into();
-        self.traverse_for_count(tree.clone(), &HashSet::new(), &obj_num)
+        self.traverse_for_count(tree.clone(), &HashSet::new(), &mut HashSet::new(), &obj_num)
             .await;
 
         obj_num.fetch_add(1, Ordering::SeqCst);
@@ -267,11 +267,13 @@ impl PackHandler for MonoRepo {
             self.traverse(have_tree.into(), &mut exist_objs, None).await;
         }
 
+        let mut counted_obj = HashSet::new();
         // traverse for get obj nums
         for c in want_commits.clone() {
             self.traverse_for_count(
                 want_trees.get(&c.tree_id).unwrap().clone(),
                 &exist_objs,
+                &mut counted_obj,
                 &obj_num,
             )
             .await;

jupiter/src/storage/git_db_storage.rs

@@ -215,6 +215,19 @@ impl GitDbStorage {
             .unwrap())
     }
 
+    pub async fn get_commits_by_hashes(
+        &self,
+        repo: &Repo,
+        hashes: &Vec<String>,
+    ) -> Result<Vec<git_commit::Model>, MegaError> {
+        Ok(git_commit::Entity::find()
+            .filter(git_commit::Column::RepoId.eq(repo.repo_id))
+            .filter(git_commit::Column::CommitId.is_in(hashes))
+            .all(self.get_connection())
+            .await
+            .unwrap())
+    }
+
     pub async fn get_commits_by_repo_id(
         &self,
         repo: &Repo,

mercury/src/internal/pack/encode.rs

@@ -120,7 +120,10 @@ impl PackEncoder {
                 }
                 None => {
                     if self.process_index != self.object_number {
-                        panic!("not all objects are encoded");
+                        panic!(
+                            "not all objects are encoded, process:{}, total:{}",
+                            self.process_index, self.object_number
+                        );
                     }
                     break;
                 }
@@ -296,5 +299,4 @@ mod tests {
         assert_eq!(data[0], 0b_1101_0101);
         assert_eq!(data[1], 0b_0000_0101);
     }
-
 }