[ceres] feat: improve performance when uploading LFS object.

Cargo.toml

@@ -54,7 +54,6 @@ bytes = "1.7.1"
 memchr = "2.7.4"
 chrono = "0.4.38"
 sha1 = "0.10.6"
-sha256 = "1.5"
 futures = "0.3.30"
 futures-util = "0.3.30"
 go-defer = "0.1.0"
@@ -83,3 +82,4 @@ ctrlc = "3.4.4"
 git2 = "0.19.0"
 tempfile = "3.10.1"
 home = "0.5.9"
+ring = "0.17.8"

ceres/Cargo.toml

@@ -29,5 +29,6 @@ futures = { workspace = true }
 bytes = { workspace = true }
 async-trait = { workspace = true }
 rand = { workspace = true }
-sha256 = { workspace = true }
 sea-orm = { workspace = true }
+ring = { workspace = true }
+hex ={ workspace = true}

ceres/src/lfs/handler.rs

@@ -1,23 +1,22 @@
 use std::cmp::min;
 use std::collections::HashMap;
 
-use anyhow::Result;
-use bytes::Bytes;
-use chrono::{prelude::*, Duration};
-use rand::prelude::*;
-use sea_orm::ActiveValue::Set;
-use sea_orm::IntoActiveModel;
-use callisto::{lfs_locks, lfs_objects, lfs_split_relations};
-use common::errors::{GitLFSError, MegaError};
-use jupiter::context::Context;
-use jupiter::storage::lfs_db_storage::LfsDbStorage;
-
 use crate::lfs::lfs_structs::ChunkRepresentation;
 use crate::lfs::lfs_structs::{
     BatchRequest, LockList, LockRequest, ObjectError, UnlockRequest, VerifiableLockList,
     VerifiableLockRequest,
 };
 use crate::lfs::lfs_structs::{Link, Lock, LockListQuery, MetaObject, Representation, RequestVars};
+use anyhow::Result;
+use bytes::Bytes;
+use callisto::{lfs_locks, lfs_objects, lfs_split_relations};
+use chrono::{prelude::*, Duration};
+use common::errors::{GitLFSError, MegaError};
+use jupiter::context::Context;
+use jupiter::storage::lfs_db_storage::LfsDbStorage;
+use rand::prelude::*;
+use sea_orm::ActiveValue::Set;
+use sea_orm::{DatabaseTransaction, EntityTrait, IntoActiveModel, TransactionTrait};
 
 pub async fn lfs_retrieve_lock(
     storage: LfsDbStorage,
@@ -86,10 +85,7 @@ pub async fn lfs_verify_lock(
     Ok(lock_list)
 }
 
-pub async fn lfs_create_lock(
-    storage: LfsDbStorage,
-    req: LockRequest,
-) -> Result<Lock, GitLFSError> {
+pub async fn lfs_create_lock(storage: LfsDbStorage, req: LockRequest) -> Result<Lock, GitLFSError> {
     let res = lfs_get_filtered_locks(
         storage.clone(),
         &req.refs.name,
@@ -308,37 +304,47 @@ pub async fn lfs_upload_object(
         let mut sub_ids = vec![];
         for chunk in body_bytes.chunks(config.split_size) {
             // sha256
-            let sub_id = sha256::digest(chunk);
+            let sub_id = hex::encode(ring::digest::digest(&ring::digest::SHA256, chunk));
             let res = lfs_storage.put_object(&sub_id, chunk).await;
             if res.is_err() {
-                lfs_delete_meta(storage.clone(), request_vars)
-                    .await
-                    .unwrap();
+                lfs_delete_meta(&storage, request_vars).await.unwrap();
                 // TODO: whether/how to delete the uploaded blocks.
                 return Err(GitLFSError::GeneralError(String::from(
                     "Header not acceptable!",
                 )));
             }
             sub_ids.push(sub_id);
         }
-        tracing::debug!("lfs object {} split into {} chunks", meta.oid, sub_ids.len());
+        tracing::debug!(
+            "lfs object {} split into {} chunks",
+            meta.oid,
+            sub_ids.len()
+        );
 
         // save the relationship to database
+        let con = storage.get_connection();
+        let tx = con.begin().await.unwrap();
         let mut offset = 0;
         for sub_id in sub_ids {
             let size = min(config.split_size as i64, body_bytes.len() as i64 - offset);
-            lfs_put_relation(storage.clone(), &meta.oid, &sub_id, offset, size)
-                .await
-                .unwrap();
+            let result = lfs_put_relation(&tx, &meta.oid, &sub_id, offset, size).await;
+            if result.is_err() {
+                tx.rollback().await.unwrap();
+                lfs_delete_meta(&storage, request_vars).await.unwrap();
+                tracing::error!("lfs object upload failed, failed to save split relationship");
+                return Err(GitLFSError::GeneralError(String::from(
+                    "Header not acceptable!",
+                )));
+            }
             offset += size;
         }
+        tx.commit().await.unwrap();
+        tracing::debug!("lfs object  split relationship saved");
     } else {
         // normal mode
         let res = lfs_storage.put_object(&meta.oid, body_bytes).await;
         if res.is_err() {
-            lfs_delete_meta(storage.clone(), request_vars)
-                .await
-                .unwrap();
+            lfs_delete_meta(&storage, request_vars).await.unwrap();
             return Err(GitLFSError::GeneralError(String::from(
                 "Header not acceptable!",
             )));
@@ -534,10 +540,7 @@ async fn lfs_get_filtered_locks(
     Ok((locks, next))
 }
 
-async fn lfs_get_locks(
-    storage: LfsDbStorage,
-    refspec: &str,
-) -> Result<Vec<Lock>, GitLFSError> {
+async fn lfs_get_locks(storage: LfsDbStorage, refspec: &str) -> Result<Vec<Lock>, GitLFSError> {
     let result = storage.get_lock_by_id(refspec).await.unwrap();
     match result {
         Some(val) => {
@@ -660,7 +663,7 @@ async fn lfs_put_meta(
     }
 }
 
-async fn lfs_delete_meta(storage: LfsDbStorage, v: &RequestVars) -> Result<(), GitLFSError> {
+async fn lfs_delete_meta(storage: &LfsDbStorage, v: &RequestVars) -> Result<(), GitLFSError> {
     let res = storage.delete_lfs_object(v.oid.to_owned()).await;
     lfs_delete_all_relations(storage.clone(), &v.oid)
         .await
@@ -746,7 +749,7 @@ async fn delete_lock(
 
 /// put relation, ignore if already exist.
 async fn lfs_put_relation(
-    storage: LfsDbStorage,
+    tx: &DatabaseTransaction,
     ori_oid: &String,
     sub_oid: &String,
     offset: i64,
@@ -758,8 +761,10 @@ async fn lfs_put_relation(
         offset,
         size,
     };
-    let res = storage.new_lfs_relation(relation).await;
-    match res {
+    let result = lfs_split_relations::Entity::insert(relation.into_active_model())
+        .exec(tx)
+        .await;
+    match result {
         Ok(_) => Ok(()),
         Err(e) => {
             if e.to_string().contains("duplicate key value") {

jupiter/src/storage/lfs_db_storage.rs

@@ -37,16 +37,6 @@ impl LfsDbStorage {
             .unwrap())
     }
 
-    pub async fn new_lfs_relation(
-        &self,
-        relation: lfs_split_relations::Model,
-    ) -> Result<InsertResult<lfs_split_relations::ActiveModel>, MegaError> {
-        lfs_split_relations::Entity::insert(relation.into_active_model())
-            .exec(self.get_connection())
-            .await
-            .map_err(|e| MegaError::with_message(e.to_string().as_str()))
-    }
-
     pub async fn get_lfs_object(
         &self,
         oid: String,

mono/config.toml

@@ -21,10 +21,10 @@ print_std = true
 db_type = "postgres"
 
 # used for sqlite
-db_path = "${base_dir}/mega.db"
+db_path = "${base_dir}/mono.db"
 
 # database connection url
-db_url = "postgres://mega:mega@localhost:5432/mega"
+db_url = "postgres://mono:mono@localhost:5432/mono"
 
 # db max connection, setting it to twice the number of CPU cores would be appropriate.
 max_connection = 32