fix: batch save concurrent limit (#2044)

Author: benjamin-747

ceres/src/pack/import_repo.rs

@@ -6,6 +6,7 @@ use std::{
         Arc, Mutex,
         atomic::{AtomicUsize, Ordering},
     },
+    time::Instant,
 };
 
 use async_recursion::async_recursion;
@@ -47,6 +48,7 @@ pub struct ImportRepo {
     pub command_list: Mutex<Vec<RefCommand>>,
     pub unpack_redlock: Arc<RedLock>,
     pub git_object_cache: Arc<GitObjectCache>,
+    pub receive_pack_extra_timings_ms: Mutex<Vec<(String, u128)>>,
 }
 
 #[async_trait]
@@ -55,6 +57,19 @@ impl RepoHandler for ImportRepo {
         false
     }
 
+    fn save_entry_concurrency(&self) -> usize {
+        self.storage.config().pack.save_entry_concurrency
+    }
+
+    fn receive_pack_extra_timings_ms(&self) -> Vec<(String, u128)> {
+        std::mem::take(
+            &mut self
+                .receive_pack_extra_timings_ms
+                .lock()
+                .expect("receive_pack_extra_timings_ms lock poisoned"),
+        )
+    }
+
     fn sync_commands_after_unpack(&self, commands: &[RefCommand]) {
         *self
             .command_list
@@ -75,10 +90,32 @@ impl RepoHandler for ImportRepo {
     }
 
     async fn finalize_receive_pack(&self) -> Result<(), MegaError> {
-        let guard = self.unpack_redlock.clone().lock().await?;
+        let t0 = Instant::now();
+        let t_fp = Instant::now();
         self.traverses_tree_and_update_filepath().await?;
+        self.receive_pack_extra_timings_ms
+            .lock()
+            .expect("receive_pack_extra_timings_ms lock poisoned")
+            .push((
+                "import_filepath_update_ms".to_string(),
+                t_fp.elapsed().as_millis(),
+            ));
+
+        let t_attach = Instant::now();
         self.attach_to_monorepo_parent().await?;
-        guard.unlock().await?;
+        self.receive_pack_extra_timings_ms
+            .lock()
+            .expect("receive_pack_extra_timings_ms lock poisoned")
+            .extend([
+                (
+                    "import_attach_to_monorepo_parent_ms".to_string(),
+                    t_attach.elapsed().as_millis(),
+                ),
+                (
+                    "import_finalize_total_ms".to_string(),
+                    t0.elapsed().as_millis(),
+                ),
+            ]);
         Ok(())
     }
 
@@ -424,8 +461,18 @@ impl ImportRepo {
 
         // Concurrent attaches need CAS on root mega_refs; retry when head moved.
         const MAX_ATTACH_ATTEMPTS: u32 = 64;
+        let mut root_lock_wait_max_ms: u128 = 0;
+        let mut root_lock_wait_sum_ms: u128 = 0;
 
         for attempt in 0..MAX_ATTACH_ATTEMPTS {
+            // Only the root mega_refs update needs cross-repo serialization.
+            // Keep the lock scope as small as possible: just the root ref read + attach transaction.
+            let t_lock = Instant::now();
+            let guard = self.unpack_redlock.clone().lock().await?;
+            let lock_wait_ms = t_lock.elapsed().as_millis();
+            root_lock_wait_max_ms = root_lock_wait_max_ms.max(lock_wait_ms);
+            root_lock_wait_sum_ms += lock_wait_ms;
+
             let root_ref = storage
                 .get_main_ref("/")
                 .await?
@@ -479,6 +526,7 @@ impl ImportRepo {
                 }
             }
 
+            let t_attach_txn = Instant::now();
             match storage
                 .attach_to_monorepo_parent_in_txn(
                     &txn,
@@ -492,10 +540,38 @@ impl ImportRepo {
             {
                 Ok(()) => {
                     txn.commit().await.map_err(MegaError::Db)?;
+                    let t_unlock = Instant::now();
+                    guard.unlock().await?;
+                    self.receive_pack_extra_timings_ms
+                        .lock()
+                        .expect("receive_pack_extra_timings_ms lock poisoned")
+                        .extend([
+                            (
+                                "import_attach_attempts_count".to_string(),
+                                (attempt + 1) as u128,
+                            ),
+                            (
+                                "import_root_lock_wait_sum_ms".to_string(),
+                                root_lock_wait_sum_ms,
+                            ),
+                            (
+                                "import_root_lock_wait_max_ms".to_string(),
+                                root_lock_wait_max_ms,
+                            ),
+                            (
+                                "import_attach_txn_ms".to_string(),
+                                t_attach_txn.elapsed().as_millis(),
+                            ),
+                            (
+                                "import_root_lock_unlock_ms".to_string(),
+                                t_unlock.elapsed().as_millis(),
+                            ),
+                        ]);
                     return Ok(());
                 }
                 Err(MegaError::StaleMonorepoRootRef) if attempt + 1 < MAX_ATTACH_ATTEMPTS => {
                     let _ = txn.rollback().await;
+                    let _ = guard.unlock().await;
                     tracing::warn!(
                         attempt = attempt,
                         repo_path = %self.repo.repo_path,
@@ -505,6 +581,28 @@ impl ImportRepo {
                 }
                 Err(e) => {
                     let _ = txn.rollback().await;
+                    let _ = guard.unlock().await;
+                    self.receive_pack_extra_timings_ms
+                        .lock()
+                        .expect("receive_pack_extra_timings_ms lock poisoned")
+                        .extend([
+                            (
+                                "import_attach_attempts_count".to_string(),
+                                (attempt + 1) as u128,
+                            ),
+                            (
+                                "import_root_lock_wait_sum_ms".to_string(),
+                                root_lock_wait_sum_ms,
+                            ),
+                            (
+                                "import_root_lock_wait_max_ms".to_string(),
+                                root_lock_wait_max_ms,
+                            ),
+                            (
+                                "import_attach_txn_ms".to_string(),
+                                t_attach_txn.elapsed().as_millis(),
+                            ),
+                        ]);
                     return Err(e);
                 }
             }

ceres/src/pack/mod.rs

@@ -5,6 +5,7 @@ use std::{
         Arc,
         atomic::{AtomicUsize, Ordering},
     },
+    time::Instant,
 };
 
 use async_trait::async_trait;
@@ -41,6 +42,20 @@ pub mod monorepo;
 pub trait RepoHandler: Send + Sync + 'static {
     fn is_monorepo(&self) -> bool;
 
+    /// Concurrency limit for `save_entry` batches during receive-pack.
+    /// `0` means unbounded.
+    fn save_entry_concurrency(&self) -> usize {
+        1
+    }
+
+    /// Optional extra timing key-values to be included in the final receive-pack report.
+    /// Implementations may return an empty vec.
+    ///
+    /// Intended for repo-type-specific breakdown (e.g. import attach lock waits).
+    fn receive_pack_extra_timings_ms(&self) -> Vec<(String, u128)> {
+        Vec::new()
+    }
+
     /// Replace the handler's ref-command snapshot after unpack and the protocol status loop
     /// (`default_branch`, `status`, etc.). The handler is built from an early `commands.clone()`,
     /// so without this, metadata updated in `git_receive_pack_stream` would be stale at finalize.
@@ -53,33 +68,72 @@ pub trait RepoHandler: Send + Sync + 'static {
         mut rx: UnboundedReceiver<MetaAttached<Entry, EntryMeta>>,
         _rx_pack_id: UnboundedReceiver<ObjectHash>,
     ) -> Result<(), MegaError> {
+        let t0 = Instant::now();
         let mut entry_list = vec![];
-        let semaphore = Arc::new(Semaphore::new(1)); //这里暂时改动
+        let limit = self.save_entry_concurrency();
+        let semaphore = if limit == 0 {
+            None
+        } else {
+            Some(Arc::new(Semaphore::new(limit)))
+        };
         let mut join_tasks = vec![];
 
         //let temp_pack_id = Uuid::new_v4().to_string();
         let temp_pack_id = String::new();
 
+        let mut total_entries: usize = 0;
+        let mut flush_batches: usize = 0;
         while let Some(mut entry) = rx.recv().await {
             self.check_entry(&entry.inner).await?;
             entry.meta.set_pack_id(temp_pack_id.clone());
             entry_list.push(entry);
+            total_entries += 1;
             if entry_list.len() >= 1000 {
-                let acquired = semaphore.clone().acquire_owned().await.unwrap();
+                flush_batches += 1;
                 let entries = std::mem::take(&mut entry_list);
                 let shared = self.clone();
+                let sem = semaphore.clone();
                 let handle = tokio::spawn(async move {
-                    let _acquired = acquired;
-                    shared.save_entry(entries).await
+                    let _permit = match sem {
+                        Some(s) => Some(s.acquire_owned().await.expect("semaphore closed")),
+                        None => None,
+                    };
+                    let t = Instant::now();
+                    let len = entries.len();
+                    shared.save_entry(entries).await.map(|_| {
+                        tracing::debug!(
+                            service = "git-receive-pack",
+                            batch_entries = len,
+                            elapsed_ms = t.elapsed().as_millis(),
+                            "save_entry batch completed"
+                        );
+                    })
                 });
                 join_tasks.push(handle);
             }
         }
         // process left entries
         if !entry_list.is_empty() {
+            flush_batches += 1;
             let handler = self.clone();
             let entries = std::mem::take(&mut entry_list);
-            let handle = tokio::spawn(async move { handler.save_entry(entries).await });
+            let sem = semaphore.clone();
+            let handle = tokio::spawn(async move {
+                let _permit = match sem {
+                    Some(s) => Some(s.acquire_owned().await.expect("semaphore closed")),
+                    None => None,
+                };
+                let t = Instant::now();
+                let len = entries.len();
+                handler.save_entry(entries).await.map(|_| {
+                    tracing::debug!(
+                        service = "git-receive-pack",
+                        batch_entries = len,
+                        elapsed_ms = t.elapsed().as_millis(),
+                        "save_entry batch completed"
+                    );
+                })
+            });
             join_tasks.push(handle);
         }
 
@@ -120,6 +174,14 @@ pub trait RepoHandler: Send + Sync + 'static {
         //     }
         // }
 
+        tracing::debug!(
+            service = "git-receive-pack",
+            total_entries = total_entries,
+            flush_batches = flush_batches,
+            save_entry_concurrency = limit,
+            elapsed_ms = t0.elapsed().as_millis(),
+            "receiver_handler done"
+        );
         Ok(())
     }
 

ceres/src/pack/monorepo.rs

@@ -70,6 +70,10 @@ impl RepoHandler for MonoRepo {
         true
     }
 
+    fn save_entry_concurrency(&self) -> usize {
+        self.storage.config().pack.save_entry_concurrency
+    }
+
     fn sync_commands_after_unpack(&self, commands: &[RefCommand]) {
         *self
             .command_list

ceres/src/protocol/mod.rs

@@ -187,7 +187,9 @@ impl SmartSession {
 
             let unpack_redlock = Arc::new(RedLock::new(
                 state.git_object_cache.connection.clone(),
-                "git:receive-pack:lock",
+                // Serialize monorepo root mega_refs update across concurrent import attaches.
+                // Filepath updates and per-repo work should not be blocked by this lock.
+                "git:receive-pack:lock:monorepo-root".to_string(),
                 30_000, // 30s TTL
             ));
             Ok(Arc::new(ImportRepo {
@@ -196,6 +198,7 @@ impl SmartSession {
                 repo,
                 command_list: Mutex::new(commands),
                 unpack_redlock,
+                receive_pack_extra_timings_ms: Mutex::new(Vec::new()),
             }) as Arc<dyn RepoHandler>)
         } else {
             let mut res = MonoRepo {