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 15399219e..dce0b3d54 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -8,7 +8,7 @@ use chrono::{prelude::*, Duration}; 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::{ @@ -18,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, @@ -175,6 +177,7 @@ pub async fn lfs_delete_lock( } } +/// Process batch request. pub async fn lfs_process_batch( config: &LfsConfig, mut batch_vars: BatchRequest, @@ -193,13 +196,20 @@ pub async fn lfs_process_batch( // 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).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 +228,64 @@ 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, + 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. +/// 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,30 +294,115 @@ pub async fn lfs_upload_object( let meta = lfs_get_meta(config.context.services.lfs_storage.clone(), request_vars) .await .unwrap(); - 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, "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![]; + 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 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; + } + } else { + // normal mode + 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(()) } +/// 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, ) -> 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(); + if !lfs_check_sub_oid_exist(relation_db, &sub_oid) + .await + .unwrap() + { + 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( @@ -457,6 +610,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 +619,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 +628,7 @@ async fn lfs_put_meta( oid: result.oid, size: result.size, exist: true, + splited: result.splited, }); } @@ -481,12 +637,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; @@ -498,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())), @@ -575,3 +736,50 @@ async fn delete_lock( None => Err(GitLFSError::GeneralError("".to_string())), } } + +/// put relation, ignore if already exist. +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 { + 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/ceres/src/lfs/lfs_structs.rs b/ceres/src/lfs/lfs_structs.rs index 270a1b887..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)] @@ -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)] @@ -96,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, @@ -121,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/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/common/src/config.rs b/common/src/config.rs index 206173000..10ad4f449 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,14 @@ impl Default for ZTMConfig { } } } + +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct LFSConfig { + pub enable_split: bool, + #[serde(default = "default_split_size")] + pub split_size: usize, +} + +fn default_split_size() -> usize { + 1024 * 1024 * 20 // 20MB +} diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index 3c9654fe1..457868bce 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, } } } @@ -194,7 +196,11 @@ 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$") + } 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$") .unwrap() .is_match(uri.path()) { diff --git a/gateway/src/lfs.rs b/gateway/src/lfs.rs index eb9538096..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,43 @@ 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))) + .unwrap() + }), + } +} + +pub async fn lfs_fetch_chunk_ids( + state: State, + config: &LfsConfig, + req: Request, +) -> Result, (StatusCode, String)> { + 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 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/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..30ac480e3 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,16 @@ impl LfsStorage { .unwrap()) } + pub async fn new_lfs_relation( + &self, + relation: lfs_split_relation::Model, + ) -> Result, MegaError> { + 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( &self, oid: String, @@ -46,6 +58,39 @@ 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 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(); + Ok(result.iter().map(|r| r.ori_oid.clone()).collect()) + } + pub async fn delete_lfs_object(&self, oid: String) -> Result<(), MegaError> { lfs_objects::Entity::delete_by_id(oid) .exec(self.get_connection()) @@ -54,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, 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 diff --git a/sql/postgres/pg_20240205__init.sql b/sql/postgres/pg_20240205__init.sql index 0909b7be5..cde3b8c02 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