From b59dcfdca7cd94068ba2ffad7f9bf79beb73a296 Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Sun, 30 Jun 2024 16:55:38 +0800 Subject: [PATCH 1/7] add LFS config Signed-off-by: HouXiaoxuan --- common/src/config.rs | 16 ++++++++++++++++ mega/config.toml | 6 +++++- 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/common/src/config.rs b/common/src/config.rs index 206173000..f3e40c318 100644 --- a/common/src/config.rs +++ b/common/src/config.rs @@ -12,6 +12,7 @@ pub struct Config { pub monorepo: MonoConfig, pub pack: PackConfig, pub ztm: ZTMConfig, + pub lfs: LFSConfig, } impl Config { @@ -152,3 +153,18 @@ impl Default for ZTMConfig { } } } + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct LFSConfig { + pub enable_split: bool, + pub split_size: usize, +} + +impl Default for LFSConfig { + fn default() -> Self { + Self { + enable_split: false, + split_size: 1024*1024*20, // 20MB + } + } +} \ No newline at end of file diff --git a/mega/config.toml b/mega/config.toml index c993e83cc..cd18828bb 100644 --- a/mega/config.toml +++ b/mega/config.toml @@ -72,4 +72,8 @@ channel_message_size = 1_000_000 [ztm] ca = "http://127.0.0.1:9999" hub = "http://127.0.0.1:8888" -agent = "http://127.0.0.1:7777" \ No newline at end of file +agent = "http://127.0.0.1:7777" + +[lfs] +enable_split = true +split_size = 20971520 # 20MB \ No newline at end of file From 48bcaa55094761a3634c0cd8b0092ef3febc2856 Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Mon, 1 Jul 2024 00:00:53 +0800 Subject: [PATCH 2/7] add LFS split model --- ceres/src/lfs/handler.rs | 27 +++++++++++++-- ceres/src/lfs/lfs_structs.rs | 2 ++ ceres/src/lfs/mod.rs | 4 +++ gateway/src/https_server.rs | 6 +++- jupiter/callisto/src/lfs_objects.rs | 1 + jupiter/callisto/src/lfs_split_relation.rs | 18 ++++++++++ jupiter/callisto/src/lib.rs | 1 + jupiter/src/storage/lfs_storage.rs | 39 ++++++++++++++++++++-- sql/postgres/pg_20240205__init.sql | 12 +++++-- 9 files changed, 103 insertions(+), 7 deletions(-) create mode 100644 jupiter/callisto/src/lfs_split_relation.rs diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 15399219e..9825f1828 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use anyhow::Result; use bytes::Bytes; use chrono::{prelude::*, Duration}; +use common::config::LFSConfig; use jupiter::storage::lfs_storage::LfsStorage; use rand::prelude::*; @@ -175,6 +176,12 @@ pub async fn lfs_delete_lock( } } +/// Process batch request. +/// if operation is "download": +/// 1. if server enable splite and client support it, then return a list of links to download the object. +/// 2. if server enable splite and client not support it, then return a single link to download the object. +/// 3. if server not enable splite, client support it, report error. +/// if operation is "upload", then return a link to upload the object. pub async fn lfs_process_batch( config: &LfsConfig, mut batch_vars: BatchRequest, @@ -187,7 +194,7 @@ pub async fn lfs_process_batch( let server_url = format!("http://{}:{}", config.host, config.port); let storage = config.context.services.lfs_storage.clone(); - + for object in &batch_vars.objects { let meta = lfs_get_meta(storage.clone(), object).await; // Found @@ -199,7 +206,7 @@ pub async fn lfs_process_batch( } // Not found if batch_vars.operation == "upload" { - meta = lfs_put_meta(storage.clone(), object).await.unwrap(); + meta = lfs_put_meta(storage.clone(), object, config.enable_split).await.unwrap(); response_objects.push(represent(object, &meta, false, true, false, &server_url).await); } else { let rep = Representation { @@ -218,6 +225,14 @@ pub async fn lfs_process_batch( Ok(response_objects) } +/// if server enable split, then return a list of chunk ids. +/// else return an error. +pub async fn lfs_fetch_chunk_ids(config: &LFSConfig, oid: &str) -> Result<(), GitLFSError> { + unimplemented!(); +} + +/// Upload object to storage. +/// if server enable split, split the object and upload each part to storage, save the relationship to database. pub async fn lfs_upload_object( config: &LfsConfig, request_vars: &RequestVars, @@ -226,6 +241,7 @@ pub async fn lfs_upload_object( let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars) .await .unwrap(); + // TODO: splite let res = config .lfs_storage .put_object(&config.repo_name,&meta.oid, body_bytes) @@ -241,6 +257,8 @@ pub async fn lfs_upload_object( Ok(()) } +/// Download object from storage. +/// when server enable split, if OID is a complete object, then splice the object and return it. pub async fn lfs_download_object( config: &LfsConfig, request_vars: &RequestVars, @@ -457,6 +475,7 @@ async fn lfs_get_meta( oid: val.oid, size: val.size, exist: val.exist, + splited: val.splited, }), None => Err(GitLFSError::GeneralError("".to_string())), } @@ -465,6 +484,7 @@ async fn lfs_get_meta( async fn lfs_put_meta( storage: Arc, v: &RequestVars, + splited: bool, ) -> Result { // Check if already exist. let result = storage.get_lfs_object(v.oid.clone()).await.unwrap(); @@ -473,6 +493,7 @@ async fn lfs_put_meta( oid: result.oid, size: result.size, exist: true, + splited: result.splited, }); } @@ -481,12 +502,14 @@ async fn lfs_put_meta( oid: v.oid.to_string(), size: v.size, exist: true, + splited }; let meta_to = lfs_objects::Model { oid: meta.oid.to_owned(), size: meta.size.to_owned(), exist: true, + splited }; let res = storage.new_lfs_object(meta_to).await; diff --git a/ceres/src/lfs/lfs_structs.rs b/ceres/src/lfs/lfs_structs.rs index 270a1b887..137c12495 100644 --- a/ceres/src/lfs/lfs_structs.rs +++ b/ceres/src/lfs/lfs_structs.rs @@ -16,6 +16,7 @@ pub struct MetaObject { pub oid: String, pub size: i64, pub exist: bool, + pub splited: bool, } #[derive(Serialize, Deserialize, Debug, Default)] @@ -87,6 +88,7 @@ pub struct BatchRequest { pub transfers: Vec, pub objects: Vec, pub hash_algo: String, + pub enable_split: Option, } #[derive(Serialize, Deserialize)] diff --git a/ceres/src/lfs/mod.rs b/ceres/src/lfs/mod.rs index 8f5cf7046..67a4c3fa1 100644 --- a/ceres/src/lfs/mod.rs +++ b/ceres/src/lfs/mod.rs @@ -16,4 +16,8 @@ pub struct LfsConfig { pub lfs_storage: Arc, pub repo_name: String, + + pub enable_split: bool, + + pub split_size: usize, } diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index d6443fdbd..d6eb3c290 100644 --- a/gateway/src/https_server.rs +++ b/gateway/src/https_server.rs @@ -73,6 +73,8 @@ impl From for LfsConfig { value.context.config.storage.lfs_obj_local_path, )), repo_name: String::from("repo_name"), + enable_split: value.context.config.lfs.enable_split, + split_size: value.context.config.lfs.split_size, } } } @@ -197,7 +199,9 @@ async fn post_method_router( return lfs::lfs_delete_lock(state, &lfs_config, uri.path(), req).await; } else if Regex::new(r"/objects/batch$").unwrap().is_match(uri.path()) { return lfs::lfs_process_batch(state, &lfs_config, req).await; - } else if Regex::new(r"/git-upload-pack$") + } + // Routing git services. + else if Regex::new(r"/git-upload-pack$") .unwrap() .is_match(uri.path()) { diff --git a/jupiter/callisto/src/lfs_objects.rs b/jupiter/callisto/src/lfs_objects.rs index e82b6b653..c2a51b8f4 100644 --- a/jupiter/callisto/src/lfs_objects.rs +++ b/jupiter/callisto/src/lfs_objects.rs @@ -9,6 +9,7 @@ pub struct Model { pub oid: String, pub size: i64, pub exist: bool, + pub splited: bool, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/jupiter/callisto/src/lfs_split_relation.rs b/jupiter/callisto/src/lfs_split_relation.rs new file mode 100644 index 000000000..aa3575cab --- /dev/null +++ b/jupiter/callisto/src/lfs_split_relation.rs @@ -0,0 +1,18 @@ +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "lfs_split_relations")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub ori_oid: String, + #[sea_orm(primary_key, auto_increment = false)] + pub sub_oid: String, + #[sea_orm(primary_key, auto_increment = false)] + pub offset: i64, + pub size: i64, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} \ No newline at end of file diff --git a/jupiter/callisto/src/lib.rs b/jupiter/callisto/src/lib.rs index 1ac2ab7af..1271d6da4 100644 --- a/jupiter/callisto/src/lib.rs +++ b/jupiter/callisto/src/lib.rs @@ -23,3 +23,4 @@ pub mod mega_refs; pub mod mega_tag; pub mod mega_tree; pub mod raw_blob; +pub mod lfs_split_relation; \ No newline at end of file diff --git a/jupiter/src/storage/lfs_storage.rs b/jupiter/src/storage/lfs_storage.rs index fea30de39..cb42414e8 100644 --- a/jupiter/src/storage/lfs_storage.rs +++ b/jupiter/src/storage/lfs_storage.rs @@ -1,7 +1,9 @@ use std::sync::Arc; -use callisto::{lfs_locks, lfs_objects}; -use sea_orm::{DatabaseConnection, EntityTrait, InsertResult, IntoActiveModel}; +use callisto::{lfs_locks, lfs_objects, lfs_split_relation}; +use sea_orm::{ + ColumnTrait, DatabaseConnection, EntityTrait, InsertResult, IntoActiveModel, QueryFilter, +}; use common::errors::MegaError; @@ -35,6 +37,18 @@ impl LfsStorage { .unwrap()) } + pub async fn new_lfs_relation( + &self, + relation: lfs_split_relation::Model, + ) -> Result, MegaError> { + Ok( + lfs_split_relation::Entity::insert(relation.into_active_model()) + .exec(self.get_connection()) + .await + .unwrap(), + ) + } + pub async fn get_lfs_object( &self, oid: String, @@ -46,6 +60,27 @@ impl LfsStorage { Ok(result) } + pub async fn get_lfs_relations( + &self, + oid: String, + ) -> Result, MegaError> { + let obj = self.get_lfs_object(oid.clone()).await?; + if obj.is_none() { + return Err(MegaError::with_message("Object not found")); + } + let result = lfs_split_relation::Entity::find() + .filter(lfs_split_relation::Column::OriOid.eq(oid)) + .all(self.get_connection()) + .await + .unwrap(); + if result.is_empty() { + return Err(MegaError::with_message( + "Object relation not found, maybe have not been uploaded yet", + )); + } + Ok(result) + } + pub async fn delete_lfs_object(&self, oid: String) -> Result<(), MegaError> { lfs_objects::Entity::delete_by_id(oid) .exec(self.get_connection()) diff --git a/sql/postgres/pg_20240205__init.sql b/sql/postgres/pg_20240205__init.sql index 0909b7be5..96f27014e 100644 --- a/sql/postgres/pg_20240205__init.sql +++ b/sql/postgres/pg_20240205__init.sql @@ -215,5 +215,13 @@ CREATE TABLE IF NOT EXISTS "lfs_locks" ( CREATE TABLE IF NOT EXISTS "lfs_objects" ( "oid" VARCHAR(64) PRIMARY KEY, "size" BIGINT NOT NULL, - "exist" BOOLEAN NOT NULL -); \ No newline at end of file + "exist" BOOLEAN NOT NULL, + "splited" BOOLEAN NOT NULL, +); +CREATE TABLE IF NOT EXISTS "lfs_split_relations" ( + "ori_oid" VARCHAR(64) NOT NULL, + "sub_oid" VARCHAR(64) NOT NULL, + "offset" BIGINT NOT NULL, + "size" BIGINT NOT NULL, + PRIMARY KEY ("ori_oid", "sub_oid", "offset") +) \ No newline at end of file From 5c02d1b1499f3f0b50f15fcd8c0cbcf4d0e12a45 Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Mon, 1 Jul 2024 21:25:02 +0800 Subject: [PATCH 3/7] fix sql error Signed-off-by: HouXiaoxuan --- sql/postgres/pg_20240205__init.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/postgres/pg_20240205__init.sql b/sql/postgres/pg_20240205__init.sql index 96f27014e..cde3b8c02 100644 --- a/sql/postgres/pg_20240205__init.sql +++ b/sql/postgres/pg_20240205__init.sql @@ -216,7 +216,7 @@ CREATE TABLE IF NOT EXISTS "lfs_objects" ( "oid" VARCHAR(64) PRIMARY KEY, "size" BIGINT NOT NULL, "exist" BOOLEAN NOT NULL, - "splited" BOOLEAN NOT NULL, + "splited" BOOLEAN NOT NULL ); CREATE TABLE IF NOT EXISTS "lfs_split_relations" ( "ori_oid" VARCHAR(64) NOT NULL, From b95323cdfdf7779ba1560ad84cc7488a4626d1f4 Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Mon, 1 Jul 2024 23:12:27 +0800 Subject: [PATCH 4/7] split lfs: upload & download --- ceres/Cargo.toml | 1 + ceres/src/lfs/handler.rs | 164 ++++++++++++++++++++++++----- jupiter/src/storage/lfs_storage.rs | 12 +++ 3 files changed, 150 insertions(+), 27 deletions(-) diff --git a/ceres/Cargo.toml b/ceres/Cargo.toml index 39283641b..cacc82afc 100644 --- a/ceres/Cargo.toml +++ b/ceres/Cargo.toml @@ -30,3 +30,4 @@ futures = { workspace = true } bytes = { workspace = true } async-trait = { workspace = true } rand = { workspace = true } +sha256 = { workspace = true } diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 9825f1828..030b67dcb 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -9,7 +9,7 @@ use common::config::LFSConfig; use jupiter::storage::lfs_storage::LfsStorage; use rand::prelude::*; -use callisto::{lfs_locks, lfs_objects}; +use callisto::{lfs_locks, lfs_objects, lfs_split_relation}; use common::errors::{GitLFSError, MegaError}; use crate::lfs::lfs_structs::{ @@ -178,10 +178,8 @@ pub async fn lfs_delete_lock( /// Process batch request. /// if operation is "download": -/// 1. if server enable splite and client support it, then return a list of links to download the object. -/// 2. if server enable splite and client not support it, then return a single link to download the object. -/// 3. if server not enable splite, client support it, report error. -/// if operation is "upload", then return a link to upload the object. +/// if server enable splite and client support it, report error, use `lfs_fetch_chunk_ids` to get chunk ids. +/// else use original method to download object. pub async fn lfs_process_batch( config: &LfsConfig, mut batch_vars: BatchRequest, @@ -194,19 +192,26 @@ pub async fn lfs_process_batch( let server_url = format!("http://{}:{}", config.host, config.port); let storage = config.context.services.lfs_storage.clone(); - + for object in &batch_vars.objects { let meta = lfs_get_meta(storage.clone(), object).await; // Found let found = meta.is_ok(); let mut meta = meta.unwrap_or_default(); - if found && config.lfs_storage.exist_object(&config.repo_name, &meta.oid) { + if found + && config + .lfs_storage + .exist_object(&config.repo_name, &meta.oid) + { + // originla download method, split mode use `` response_objects.push(represent(object, &meta, true, false, false, &server_url).await); continue; } // Not found if batch_vars.operation == "upload" { - meta = lfs_put_meta(storage.clone(), object, config.enable_split).await.unwrap(); + meta = lfs_put_meta(storage.clone(), object, config.enable_split) + .await + .unwrap(); response_objects.push(represent(object, &meta, false, true, false, &server_url).await); } else { let rep = Representation { @@ -241,18 +246,53 @@ pub async fn lfs_upload_object( let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars) .await .unwrap(); - // TODO: splite - let res = config - .lfs_storage - .put_object(&config.repo_name,&meta.oid, body_bytes) - .await; - if res.is_err() { - lfs_delete_meta(config.context.services.lfs_storage.clone(), request_vars) - .await - .unwrap(); - return Err(GitLFSError::GeneralError(String::from( - "Header not acceptable!", - ))); + if config.enable_split && meta.splited { + assert!(request_vars.size == body_bytes.len() as i64); + // split object to blocks + let mut sub_ids = vec![]; + + for chunk in body_bytes.chunks(config.split_size) { + // sha256 + let sub_id = sha256::digest(chunk); + let res = config + .lfs_storage + .put_object(&config.repo_name, &sub_id, chunk) + .await; + if res.is_err() { + lfs_delete_meta(config.context.services.lfs_storage.clone(), 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); + } + // save the relationship to database + let mut offset = 0; + for sub_id in sub_ids { + let size = min(config.split_size as i64, body_bytes.len() as i64 - offset); + let db = config.context.services.lfs_storage.clone(); + lfs_put_relation(db, &meta.oid, &sub_id, offset, size) + .await + .unwrap(); + offset += size; + } + unimplemented!(); // Split object and upload each part to storage. + } else { + let res = config + .lfs_storage + .put_object(&config.repo_name, &meta.oid, body_bytes) + .await; + if res.is_err() { + lfs_delete_meta(config.context.services.lfs_storage.clone(), request_vars) + .await + .unwrap(); + return Err(GitLFSError::GeneralError(String::from( + "Header not acceptable!", + ))); + } } Ok(()) } @@ -263,11 +303,60 @@ pub async fn lfs_download_object( config: &LfsConfig, request_vars: &RequestVars, ) -> Result { - let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars) - .await - .unwrap(); - let bytes = config.lfs_storage.get_object(&config.repo_name,&meta.oid).await.unwrap(); - Ok(bytes) + + if config.enable_split { + let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars).await; + let relation_db = config.context.services.lfs_storage.clone(); + + match meta { + Ok(meta) => { + // client didn't support split, splice the object and return it. + let relations = relation_db.get_lfs_relations(meta.oid).await.unwrap(); + let mut bytes = vec![0u8; meta.size as usize]; + for relation in relations { + let sub_bytes = config + .lfs_storage + .get_object(&config.repo_name, &relation.sub_oid) + .await + .unwrap(); + let offset = relation.offset as usize; + let size = relation.size as usize; + bytes[offset..offset + size].copy_from_slice(&sub_bytes); + } + Ok(Bytes::from(bytes)) + } + Err(_) => { + // check if the oid is a part of a split object, if so, return the part. + let sub_oid = request_vars.oid.clone(); + let ori_oid = relation_db + .get_lfs_relations_ori_oid(&sub_oid) + .await + .unwrap(); + if ori_oid.is_none() { + return Err(GitLFSError::GeneralError( + "oid didn't belong to any object".to_string(), + )); + } + + let bytes = config + .lfs_storage + .get_object(&config.repo_name, &sub_oid) + .await + .unwrap(); + Ok(bytes) + } + } + } else { + let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars) + .await + .unwrap(); + let bytes = config + .lfs_storage + .get_object(&config.repo_name, &meta.oid) + .await + .unwrap(); + Ok(bytes) + } } pub async fn represent( @@ -502,14 +591,14 @@ async fn lfs_put_meta( oid: v.oid.to_string(), size: v.size, exist: true, - splited + splited, }; let meta_to = lfs_objects::Model { oid: meta.oid.to_owned(), size: meta.size.to_owned(), exist: true, - splited + splited, }; let res = storage.new_lfs_object(meta_to).await; @@ -598,3 +687,24 @@ async fn delete_lock( None => Err(GitLFSError::GeneralError("".to_string())), } } + +async fn lfs_put_relation( + storage: Arc, + ori_oid: &String, + sub_oid: &String, + offset: i64, + size: i64, +) -> Result<(), GitLFSError> { + let relation = lfs_split_relation::Model { + ori_oid: ori_oid.to_owned(), + sub_oid: sub_oid.to_owned(), + offset, + size, + }; + + let res = storage.new_lfs_relation(relation).await; + match res.is_ok() { + true => Ok(()), + false => Err(GitLFSError::GeneralError("".to_string())), + } +} diff --git a/jupiter/src/storage/lfs_storage.rs b/jupiter/src/storage/lfs_storage.rs index cb42414e8..535e25d8e 100644 --- a/jupiter/src/storage/lfs_storage.rs +++ b/jupiter/src/storage/lfs_storage.rs @@ -81,6 +81,18 @@ impl LfsStorage { Ok(result) } + pub async fn get_lfs_relations_ori_oid(&self, sub_oid: &String) -> Result, MegaError> { + let result = lfs_split_relation::Entity::find() + .filter(lfs_split_relation::Column::SubOid.eq(sub_oid)) + .all(self.get_connection()) + .await + .unwrap(); + if result.is_empty() { + return Ok(None); + } + Ok(Some(result[0].ori_oid.clone())) + } + pub async fn delete_lfs_object(&self, oid: String) -> Result<(), MegaError> { lfs_objects::Entity::delete_by_id(oid) .exec(self.get_connection()) From b0ba6fa5b922ac47ea1689d4015944d19bf176ec Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Tue, 2 Jul 2024 23:15:44 +0800 Subject: [PATCH 5/7] split lfs: lfs_fetch_chunk_ids Signed-off-by: HouXiaoxuan --- ceres/src/lfs/handler.rs | 58 +++++++++++++++++++++++++++++++++--- ceres/src/lfs/lfs_structs.rs | 17 ++++++++++- gateway/src/https_server.rs | 2 ++ gateway/src/lfs.rs | 24 +++++++++++++++ 4 files changed, 96 insertions(+), 5 deletions(-) diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 030b67dcb..072eec12a 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -5,7 +5,6 @@ use std::sync::Arc; use anyhow::Result; use bytes::Bytes; use chrono::{prelude::*, Duration}; -use common::config::LFSConfig; use jupiter::storage::lfs_storage::LfsStorage; use rand::prelude::*; @@ -19,6 +18,8 @@ use crate::lfs::lfs_structs::{ use crate::lfs::lfs_structs::{Link, Lock, LockListQuery, MetaObject, Representation, RequestVars}; use crate::lfs::LfsConfig; +use super::lfs_structs::ChunkRepresentation; + pub async fn lfs_retrieve_lock( config: &LfsConfig, query: LockListQuery, @@ -232,8 +233,58 @@ pub async fn lfs_process_batch( /// if server enable split, then return a list of chunk ids. /// else return an error. -pub async fn lfs_fetch_chunk_ids(config: &LFSConfig, oid: &str) -> Result<(), GitLFSError> { - unimplemented!(); +pub async fn lfs_fetch_chunk_ids( + config: &LfsConfig, + fetch_vars: &RequestVars, +) -> Result, GitLFSError> { + if !config.enable_split { + return Err(GitLFSError::GeneralError( + "Server didn't run in `split` mode, didn't support chunk ids".to_string(), + )); + } + + let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), fetch_vars) + .await + .map_err(|_| GitLFSError::GeneralError("".to_string()))?; + assert!(meta.splited, "database didn't match the split mode"); + + let relations = config + .context + .services + .lfs_storage + .get_lfs_relations(fetch_vars.oid.clone()) + .await + .map_err(|_| GitLFSError::GeneralError("".to_string()))?; + + if relations.is_empty() { + return Err(GitLFSError::GeneralError( + "oid didn't have chunks".to_string(), + )); + } + let mut response_objects = Vec::::new(); + let server_url = format!("http://{}:{}", config.host, config.port); + + for relation in relations { + // Reuse RequestArgs to create a link + let tmp_request_vars = RequestVars { + oid: relation.sub_oid.clone(), + size: relation.size, + authorization: fetch_vars.authorization.clone(), + password: fetch_vars.password.clone(), + user: fetch_vars.user.clone(), + repo: fetch_vars.repo.clone(), + }; + response_objects.push(ChunkRepresentation { + sub_oid: relation.sub_oid, + size: relation.size, + offset: relation.offset, + link: create_link( + &tmp_request_vars.download_link(server_url.to_string()).await, + &HashMap::new(), + ), + }); + } + Ok(response_objects) } /// Upload object to storage. @@ -303,7 +354,6 @@ pub async fn lfs_download_object( config: &LfsConfig, request_vars: &RequestVars, ) -> Result { - if config.enable_split { let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars).await; let relation_db = config.context.services.lfs_storage.clone(); diff --git a/ceres/src/lfs/lfs_structs.rs b/ceres/src/lfs/lfs_structs.rs index 137c12495..da5ca27f2 100644 --- a/ceres/src/lfs/lfs_structs.rs +++ b/ceres/src/lfs/lfs_structs.rs @@ -8,7 +8,7 @@ pub enum TransferMode { BASIC, MULTIPART, //not implement yet - STREAMING + STREAMING, } #[derive(Debug, Default)] @@ -98,6 +98,13 @@ pub struct BatchResponse { pub hash_algo: String, } +#[derive(Serialize, Deserialize)] +pub struct FetchchunkResponse { + pub oid: String, + pub size : i64, + pub chunks: Vec, +} + #[derive(Serialize, Deserialize)] pub struct Link { pub href: String, @@ -123,6 +130,14 @@ pub struct Representation { pub error: Option, } +#[derive(Serialize, Deserialize)] +pub struct ChunkRepresentation { + pub sub_oid: String, + pub offset: i64, + pub size: i64, + pub link: Link, +} + #[derive(Serialize, Deserialize, Debug, Default)] pub struct Ref { pub name: String, diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index d6eb3c290..ff9acf2c9 100644 --- a/gateway/src/https_server.rs +++ b/gateway/src/https_server.rs @@ -199,6 +199,8 @@ async fn post_method_router( return lfs::lfs_delete_lock(state, &lfs_config, uri.path(), req).await; } else if Regex::new(r"/objects/batch$").unwrap().is_match(uri.path()) { return lfs::lfs_process_batch(state, &lfs_config, req).await; + } else if Regex::new(r"objects/chunkids$").unwrap().is_match(uri.path()) { + return lfs::lfs_fetch_chunk_ids(state, &lfs_config, req).await; } // Routing git services. else if Regex::new(r"/git-upload-pack$") diff --git a/gateway/src/lfs.rs b/gateway/src/lfs.rs index eb9538096..adec6a565 100644 --- a/gateway/src/lfs.rs +++ b/gateway/src/lfs.rs @@ -217,6 +217,30 @@ pub async fn lfs_process_batch( } } +pub async fn lfs_fetch_chunk_ids( + state: State, + config: &LfsConfig, + req: Request, +) -> Result, (StatusCode, String)> { + let request = Json::from_request(req, &state).await.unwrap(); + let result = handler::lfs_fetch_chunk_ids(config, &request).await; + match result { + Ok(response) => { + let body = serde_json::to_string(&response).unwrap_or_default(); + Ok(Response::builder() + .header("Content-Type", LFS_CONTENT_TYPE) + .body(Body::from(body)) + .unwrap()) + } + Err(err) => Ok({ + Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(Body::from(format!("Error: {}", err))) + .unwrap() + }), + } +} + pub async fn lfs_download_object( config: &LfsConfig, path: &str, From cfbecded7051ee83d6b2f3cec8399872d90c335e Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Wed, 3 Jul 2024 01:12:13 +0800 Subject: [PATCH 6/7] split lfs: fix bugs 1. parse request args error 2. put_relation ignore duplicate Signed-off-by: HouXiaoxuan --- ceres/src/lfs/handler.rs | 55 ++++++++++++++++++++++-------- gateway/src/lfs.rs | 21 +++++++++--- jupiter/src/storage/lfs_storage.rs | 32 +++++++++++------ 3 files changed, 78 insertions(+), 30 deletions(-) diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 072eec12a..dce0b3d54 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -178,9 +178,6 @@ pub async fn lfs_delete_lock( } /// Process batch request. -/// if operation is "download": -/// if server enable splite and client support it, report error, use `lfs_fetch_chunk_ids` to get chunk ids. -/// else use original method to download object. pub async fn lfs_process_batch( config: &LfsConfig, mut batch_vars: BatchRequest, @@ -298,10 +295,10 @@ pub async fn lfs_upload_object( .await .unwrap(); if config.enable_split && meta.splited { - assert!(request_vars.size == body_bytes.len() as i64); + // assert!(request_vars.size == body_bytes.len() as i64, "size didn't match: {} != {}", request_vars.size, body_bytes.len()); // TODO: git client, request_vars.size is `0`!! // split object to blocks - let mut sub_ids = vec![]; + let mut sub_ids = vec![]; for chunk in body_bytes.chunks(config.split_size) { // sha256 let sub_id = sha256::digest(chunk); @@ -323,15 +320,15 @@ pub async fn lfs_upload_object( // save the relationship to database let mut offset = 0; for sub_id in sub_ids { - let size = min(config.split_size as i64, body_bytes.len() as i64 - offset); let db = config.context.services.lfs_storage.clone(); + let size = min(config.split_size as i64, body_bytes.len() as i64 - offset); lfs_put_relation(db, &meta.oid, &sub_id, offset, size) .await .unwrap(); offset += size; } - unimplemented!(); // Split object and upload each part to storage. } else { + // normal mode let res = config .lfs_storage .put_object(&config.repo_name, &meta.oid, body_bytes) @@ -378,11 +375,10 @@ pub async fn lfs_download_object( Err(_) => { // check if the oid is a part of a split object, if so, return the part. let sub_oid = request_vars.oid.clone(); - let ori_oid = relation_db - .get_lfs_relations_ori_oid(&sub_oid) + if !lfs_check_sub_oid_exist(relation_db, &sub_oid) .await - .unwrap(); - if ori_oid.is_none() { + .unwrap() + { return Err(GitLFSError::GeneralError( "oid didn't belong to any object".to_string(), )); @@ -660,6 +656,9 @@ async fn lfs_put_meta( async fn lfs_delete_meta(storage: Arc, v: &RequestVars) -> Result<(), GitLFSError> { let res = storage.delete_lfs_object(v.oid.to_owned()).await; + lfs_delete_all_relations(storage.clone(), &v.oid) + .await + .unwrap(); match res { Ok(_) => Ok(()), Err(_) => Err(GitLFSError::GeneralError("".to_string())), @@ -738,6 +737,7 @@ async fn delete_lock( } } +/// put relation, ignore if already exist. async fn lfs_put_relation( storage: Arc, ori_oid: &String, @@ -751,10 +751,35 @@ async fn lfs_put_relation( offset, size, }; - let res = storage.new_lfs_relation(relation).await; - match res.is_ok() { - true => Ok(()), - false => Err(GitLFSError::GeneralError("".to_string())), + match res { + Ok(_) => Ok(()), + Err(e) => { + if e.to_string().contains("duplicate key value") { + Ok(()) + } else { + Err(GitLFSError::GeneralError(e.to_string())) + } + } } } + +/// delete all relations of an object if it exists. do nothing if not. +async fn lfs_delete_all_relations( + storage: Arc, + ori_oid: &String, +) -> Result<(), GitLFSError> { + let relations = storage.get_lfs_relations(ori_oid.to_owned()).await.unwrap(); + for relation in relations { + let _ = storage.delete_lfs_relation(relation).await; + } + Ok(()) +} + +async fn lfs_check_sub_oid_exist( + storage: Arc, + sub_oid: &String, +) -> Result { + let result = storage.get_lfs_relations_ori_oid(sub_oid).await.unwrap(); + Ok(!result.is_empty()) +} diff --git a/gateway/src/lfs.rs b/gateway/src/lfs.rs index adec6a565..ebd02fe1e 100644 --- a/gateway/src/lfs.rs +++ b/gateway/src/lfs.rs @@ -52,8 +52,8 @@ use axum::{ use ceres::lfs::{ handler, lfs_structs::{ - BatchResponse, LockList, LockListQuery, LockRequest, LockResponse, RequestVars, - UnlockRequest, UnlockResponse, VerifiableLockRequest, + BatchResponse, FetchchunkResponse, LockList, LockListQuery, LockRequest, LockResponse, + RequestVars, UnlockRequest, UnlockResponse, VerifiableLockRequest, }, LfsConfig, }; @@ -209,6 +209,8 @@ pub async fn lfs_process_batch( .unwrap()) } Err(err) => Ok({ + tracing::error!("Error: {}", err); + Response::builder() .status(StatusCode::INTERNAL_SERVER_ERROR) .body(Body::from(format!("Error: {}", err))) @@ -222,17 +224,28 @@ pub async fn lfs_fetch_chunk_ids( config: &LfsConfig, req: Request, ) -> Result, (StatusCode, String)> { - let request = Json::from_request(req, &state).await.unwrap(); + let request = Json::from_request(req, &state).await; + if request.is_err() { + return Err((StatusCode::BAD_REQUEST, "Invalid request".to_string())); + } + let request = request.unwrap(); let result = handler::lfs_fetch_chunk_ids(config, &request).await; match result { Ok(response) => { - let body = serde_json::to_string(&response).unwrap_or_default(); + let size = response.iter().fold(0, |acc, chunk| acc + chunk.size); + let fetch_response = FetchchunkResponse { + oid: request.oid.clone(), + size, + chunks: response, + }; + let body = serde_json::to_string(&fetch_response).unwrap_or_default(); Ok(Response::builder() .header("Content-Type", LFS_CONTENT_TYPE) .body(Body::from(body)) .unwrap()) } Err(err) => Ok({ + tracing::error!("Error: {}", err); Response::builder() .status(StatusCode::INTERNAL_SERVER_ERROR) .body(Body::from(format!("Error: {}", err))) diff --git a/jupiter/src/storage/lfs_storage.rs b/jupiter/src/storage/lfs_storage.rs index 535e25d8e..30ac480e3 100644 --- a/jupiter/src/storage/lfs_storage.rs +++ b/jupiter/src/storage/lfs_storage.rs @@ -41,12 +41,10 @@ impl LfsStorage { &self, relation: lfs_split_relation::Model, ) -> Result, MegaError> { - Ok( - lfs_split_relation::Entity::insert(relation.into_active_model()) - .exec(self.get_connection()) - .await - .unwrap(), - ) + lfs_split_relation::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( @@ -81,16 +79,16 @@ impl LfsStorage { Ok(result) } - pub async fn get_lfs_relations_ori_oid(&self, sub_oid: &String) -> Result, MegaError> { + pub async fn get_lfs_relations_ori_oid( + &self, + sub_oid: &String, + ) -> Result, MegaError> { let result = lfs_split_relation::Entity::find() .filter(lfs_split_relation::Column::SubOid.eq(sub_oid)) .all(self.get_connection()) .await .unwrap(); - if result.is_empty() { - return Ok(None); - } - Ok(Some(result[0].ori_oid.clone())) + Ok(result.iter().map(|r| r.ori_oid.clone()).collect()) } pub async fn delete_lfs_object(&self, oid: String) -> Result<(), MegaError> { @@ -101,6 +99,18 @@ impl LfsStorage { Ok(()) } + pub async fn delete_lfs_relation( + &self, + object: lfs_split_relation::Model, + ) -> Result<(), MegaError> { + let r: lfs_split_relation::ActiveModel = object.into(); + lfs_split_relation::Entity::delete(r) + .exec(self.get_connection()) + .await + .unwrap(); + Ok(()) + } + pub async fn new_lock( &self, lfs_lock: lfs_locks::Model, From 906405649681018c39b6d76619dd0fc11ce709b0 Mon Sep 17 00:00:00 2001 From: HouXiaoxuan Date: Wed, 3 Jul 2024 14:03:10 +0800 Subject: [PATCH 7/7] split lfs: change default method Signed-off-by: HouXiaoxuan --- common/src/config.rs | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/common/src/config.rs b/common/src/config.rs index f3e40c318..10ad4f449 100644 --- a/common/src/config.rs +++ b/common/src/config.rs @@ -154,17 +154,13 @@ impl Default for ZTMConfig { } } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct LFSConfig { pub enable_split: bool, + #[serde(default = "default_split_size")] pub split_size: usize, } -impl Default for LFSConfig { - fn default() -> Self { - Self { - enable_split: false, - split_size: 1024*1024*20, // 20MB - } - } -} \ No newline at end of file +fn default_split_size() -> usize { + 1024 * 1024 * 20 // 20MB +}