diff --git a/ceres/Cargo.toml b/ceres/Cargo.toml index fa0ee9682..ec830fd97 100644 --- a/ceres/Cargo.toml +++ b/ceres/Cargo.toml @@ -29,3 +29,4 @@ bytes = { workspace = true } async-trait = { workspace = true } rand = { workspace = true } sha256 = { workspace = true } +sea-orm = { workspace = true } diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 12b3e6a48..f4d29adaa 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -5,7 +5,8 @@ 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; @@ -236,7 +237,7 @@ pub async fn lfs_process_batch( /// else return an error. pub async fn lfs_fetch_chunk_ids( context: &Context, - fetch_vars: &RequestVars, + oid: &String, ) -> Result, GitLFSError> { let config = context.config.lfs.clone(); @@ -247,13 +248,13 @@ pub async fn lfs_fetch_chunk_ids( } let storage = context.services.lfs_db_storage.clone(); - let meta = lfs_get_meta(storage.clone(), &fetch_vars.oid) + let meta = lfs_get_meta(storage.clone(), oid) .await .map_err(|_| GitLFSError::GeneralError("".to_string()))?; assert!(meta.splited, "database didn't match the split mode"); let relations = storage - .get_lfs_relations(fetch_vars.oid.clone()) + .get_lfs_relations(oid.to_owned()) .await .map_err(|_| GitLFSError::GeneralError("".to_string()))?; @@ -270,10 +271,7 @@ pub async fn lfs_fetch_chunk_ids( 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(), + ..Default::default() }; response_objects.push(ChunkRepresentation { sub_oid: relation.sub_oid, @@ -557,7 +555,7 @@ async fn lfs_add_lock( match result { // Update - Some(mut val) => { + Some(val) => { let d = val.data.to_owned(); let mut locks_from_data = if !d.is_empty() { let locks_from_data: Vec = serde_json::from_str(&d).unwrap(); @@ -575,7 +573,9 @@ async fn lfs_add_lock( }); let d = serde_json::to_string(&locks_from_data).unwrap(); - d.clone_into(&mut val.data); + // must turn into `ActiveModel` before modify, or update failed. + let mut val = val.into_active_model(); + val.data = Set(d); let res = storage.update_lock(val).await; match res.is_ok() { true => Ok(()), @@ -678,7 +678,7 @@ async fn delete_lock( let result = storage.get_lock_by_id(repo).await.unwrap(); match result { // Exist, then delete. - Some(mut val) => { + Some(val) => { let d = val.data.to_owned(); let locks_from_data = if !d.is_empty() { let locks_from_data: Vec = serde_json::from_str(&d).unwrap(); @@ -728,7 +728,8 @@ async fn delete_lock( // Update remaining locks. let data = serde_json::to_string(&new_locks).unwrap(); - data.clone_into(&mut val.data); + let mut val = val.into_active_model(); + val.data = Set(data); let res = storage.update_lock(val).await; match res.is_ok() { true => Ok(lock_to_delete), diff --git a/ceres/src/lfs/lfs_structs.rs b/ceres/src/lfs/lfs_structs.rs index da5ca27f2..71ad4df80 100644 --- a/ceres/src/lfs/lfs_structs.rs +++ b/ceres/src/lfs/lfs_structs.rs @@ -88,7 +88,6 @@ pub struct BatchRequest { pub transfers: Vec, pub objects: Vec, pub hash_algo: String, - pub enable_split: Option, } #[derive(Serialize, Deserialize)] @@ -105,7 +104,7 @@ pub struct FetchchunkResponse { pub chunks: Vec, } -#[derive(Serialize, Deserialize)] +#[derive(Serialize, Deserialize, Clone)] pub struct Link { pub href: String, pub header: HashMap, @@ -192,9 +191,14 @@ pub struct VerifiableLockList { #[derive(Serialize, Deserialize, Debug)] pub struct LockListQuery { + #[serde(default)] pub path: String, + #[serde(default)] pub id: String, + #[serde(default)] pub cursor: String, + #[serde(default)] pub limit: String, + #[serde(default)] pub refspec: String, } diff --git a/jupiter/src/storage/lfs_db_storage.rs b/jupiter/src/storage/lfs_db_storage.rs index ae858eb0a..7b3f5b0c7 100644 --- a/jupiter/src/storage/lfs_db_storage.rs +++ b/jupiter/src/storage/lfs_db_storage.rs @@ -129,9 +129,9 @@ impl LfsDbStorage { pub async fn update_lock( &self, - lfs_lock: lfs_locks::Model, + lfs_lock: lfs_locks::ActiveModel, ) -> Result { - Ok(lfs_locks::Entity::update(lfs_lock.into_active_model()) + Ok(lfs_locks::Entity::update(lfs_lock) .exec(self.get_connection()) .await .unwrap()) diff --git a/libra/Cargo.toml b/libra/Cargo.toml index 2c4e60a07..660bfdd88 100644 --- a/libra/Cargo.toml +++ b/libra/Cargo.toml @@ -38,12 +38,13 @@ indicatif = "0.17.8" wax = "0.6.0" lazy_static = { workspace = true } regex = { workspace = true } -sha2 = "0.10.8" +ring = "0.17.8" hex = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } async_static = "0.1.3" once_cell = "1.19.0" +byte-unit = "5.1.4" [target.'cfg(unix)'.dependencies] # only on Unix pager = "0.16.0" diff --git a/libra/src/command/lfs.rs b/libra/src/command/lfs.rs index 04a079072..ab4eb197a 100644 --- a/libra/src/command/lfs.rs +++ b/libra/src/command/lfs.rs @@ -3,9 +3,17 @@ use std::fs::{File, OpenOptions}; use std::io; use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; use std::path::Path; +use byte_unit::UnitType; +use reqwest::StatusCode; +use ceres::lfs::lfs_structs::LockListQuery; +use mercury::internal::index::Index; +use crate::command::{ask_basic_auth, status}; +use crate::internal::head::Head; +use crate::internal::protocol::lfs_client::LFS_CLIENT; use crate::utils::{lfs, path, util}; use crate::utils::path_ext::PathExt; +/// [Docs](https://github.com/git-lfs/git-lfs/tree/main/docs/man) #[derive(Subcommand, Debug)] pub enum LfsCmds { /// View or add LFS paths to Libra Attributes (root) @@ -16,6 +24,40 @@ pub enum LfsCmds { Untrack { path: Vec, }, + /// Lists currently locked files from the Libra LFS server. (Current Branch) + Locks { + #[clap(long, short)] + id: Option, + #[clap(long, short)] + path: Option, + #[clap(long, short)] + limit: Option, + }, + /// Set a file as "locked" on the Libra LFS server + Lock { + /// String path name of the locked file. This should be relative to the root of the repository working directory + path: String, + }, + /// Remove "locked" setting for a file on the Libra LFS server + Unlock { + path: String, + #[clap(long, short)] + force: bool, + #[clap(long, short)] + id: Option + }, + /// Show information about Git LFS files in the index and working tree (current branch) + LsFiles { + /// Show the entire 64 character OID, instead of just first 10. + #[clap(long, short)] + long: bool, + /// Show the size of the LFS object between parenthesis at the end of a line. + #[clap(long, short)] + size: bool, + /// Show only the lfs tracked file names. + #[clap(long, short)] + name_only: bool, + } } pub async fn execute(cmd: LfsCmds) { @@ -43,6 +85,131 @@ pub async fn execute(cmd: LfsCmds) { let path = convert_patterns_to_workdir(path); // untrack_lfs_patterns(&attr_path, path).unwrap(); } + LfsCmds::Locks { id, path, limit } => { + let refspec = current_refspec().await.unwrap(); + tracing::debug!("refspec: {}", refspec); + let query = LockListQuery { + id: id.unwrap_or_default(), + path: path.unwrap_or_default(), + limit: limit.map(|l| l.to_string()).unwrap_or_default(), + cursor: "".to_string(), + refspec, + }; + let locks = LFS_CLIENT.await.get_locks(query).await.locks; + if !locks.is_empty() { + let max_path_len = locks.iter().map(|l| l.path.len()).max().unwrap(); + for lock in locks { + println!("{: { + // Only check existence + if !Path::new(&path).exists() { + eprintln!("fatal: pathspec '{}' did not match any files", path); + return; + } + + let refspec = current_refspec().await.unwrap(); + let mut auth = None; + loop { + let code = LFS_CLIENT.await.lock(path.clone(), refspec.clone(), auth.clone()).await; + if code.is_success() { + println!("Locked {}", path); + } else if code == StatusCode::FORBIDDEN { + eprintln!("Forbidden: You must have push access to create a lock"); + auth = Some(ask_basic_auth()); + continue; + } else if code == StatusCode::CONFLICT { + eprintln!("Conflict: already created lock"); + } + break; + } + } + LfsCmds::Unlock { path, force, id } => { + if !force { + if !Path::new(&path).exists() { + eprintln!("fatal: pathspec '{}' did not match any files", path); + return; + } + if !status::is_clean().await { + eprintln!("fatal: working tree not clean"); + return; + } + } + let refspec = current_refspec().await.unwrap(); + let id = match id { + None => { + // get id by path + let locks = LFS_CLIENT.await.get_locks(LockListQuery { + refspec: refspec.clone(), + path: path.clone(), + id: "".to_string(), + cursor: "".to_string(), + limit: "".to_string(), + }).await.locks; + if locks.is_empty() { + eprintln!("fatal: no lock found for path '{}'", path); + return; + } + locks[0].id.clone() + } + Some(id) => id + }; + let mut auth = None; + loop { + let code = LFS_CLIENT.await.unlock(id.clone(), refspec.clone(), force, auth.clone()).await; + if code.is_success() { + println!("Unlocked {}", path); + } else if code == StatusCode::FORBIDDEN { + eprintln!("Forbidden: You must have push access to unlock"); + auth = Some(ask_basic_auth()); + continue; + } + break; + } + } + LfsCmds::LsFiles { long, size, name_only} => { + let idx_file = path::index(); + let index = Index::load(&idx_file).unwrap(); + let entries = index.tracked_entries(0); + let storage = util::objects_storage(); + for entry in entries { + let path_abs = util::workdir_to_absolute(&entry.name); + if lfs::is_lfs_tracked(&path_abs) { + let data = storage.get(&entry.hash).unwrap(); + if let Some((oid, lfs_size)) = lfs::parse_pointer_data(&data) { + let is_pointer = lfs::parse_pointer_file(&path_abs).is_ok(); + // An asterisk (*) after the OID indicates a full object, a minus (-) indicates an LFS pointer. + // or not exists (-) + let _type = if is_pointer || !path_abs.exists() { "-" } else { "*" }; + let oid = if long { oid } else { oid[..10].to_owned() }; + let tail = if size { + let byte = byte_unit::Byte::from(lfs_size); + let byte = byte.get_appropriate_unit(UnitType::Decimal); + format!(" ({byte:.2})") + } else { + "".to_string() + }; + if name_only { + println!("{}{}", entry.name, tail); + } else { + println!("{} {} {}{}", oid, _type, entry.name, tail); + } + } + } + } + } + } +} + +pub(crate) async fn current_refspec() -> Option { + match Head::current().await { + Head::Branch(name) => Some(format!("refs/heads/{}", name)), + Head::Detached(_) => { + println!("fatal: HEAD is detached"); + None + } } } diff --git a/libra/src/command/push.rs b/libra/src/command/push.rs index 4c44f6837..5d4e59cd5 100644 --- a/libra/src/command/push.rs +++ b/libra/src/command/push.rs @@ -120,7 +120,11 @@ pub async fn execute(args: PushArgs) { { // upload lfs files let client = LFSClient::from_url(&url); - client.push_objects(&objs, auth.clone()).await; + let res = client.push_objects(&objs, auth.clone()).await; + if res.is_err() { + eprintln!("fatal: LFS files upload failed, stop pushing"); + return; + } } // let (tx, rx) = mpsc::channel::(); diff --git a/libra/src/command/status.rs b/libra/src/command/status.rs index 0aa75e706..3303e7151 100644 --- a/libra/src/command/status.rs +++ b/libra/src/command/status.rs @@ -108,6 +108,13 @@ pub async fn execute() { } } +/// Check if the working tree is clean +pub async fn is_clean() -> bool { + let staged = changes_to_be_committed().await; + let unstaged = changes_to_be_staged(); + staged.is_empty() && unstaged.is_empty() +} + /** * Compare the difference between `index` and the last `Commit Tree` */ diff --git a/libra/src/internal/config.rs b/libra/src/internal/config.rs index 0929b8e97..287445951 100644 --- a/libra/src/internal/config.rs +++ b/libra/src/internal/config.rs @@ -56,6 +56,7 @@ impl Config { } /// Get remote repo name by branch name + /// - You may need to `[branch::set-upstream]` if return `None` pub async fn get_remote(branch: &str) -> Option { // e.g. [branch "master"].remote = origin Config::get("branch", Some(branch), "remote").await diff --git a/libra/src/internal/protocol/lfs_client.rs b/libra/src/internal/protocol/lfs_client.rs index 77c21a04d..ddf1cdd4f 100644 --- a/libra/src/internal/protocol/lfs_client.rs +++ b/libra/src/internal/protocol/lfs_client.rs @@ -1,13 +1,16 @@ +use std::collections::HashSet; use std::path::Path; use async_static::async_static; use futures_util::StreamExt; -use reqwest::Client; +use reqwest::{Client, StatusCode}; +use ring::digest::{Context, SHA256}; use serde::{Deserialize, Serialize}; use tokio::io::AsyncWriteExt; use url::Url; -use ceres::lfs::lfs_structs::{BatchRequest, Representation, RequestVars}; +use ceres::lfs::lfs_structs::{BatchRequest, FetchchunkResponse, Link, LockList, LockListQuery, LockRequest, Ref, Representation, RequestVars, UnlockRequest, VerifiableLockList, VerifiableLockRequest}; use mercury::internal::object::types::ObjectType; use mercury::internal::pack::entry::Entry; +use crate::command; use crate::internal::config::Config; use crate::internal::protocol::https_client::BasicAuth; use crate::internal::protocol::ProtocolClient; @@ -18,7 +21,8 @@ async_static! { } pub struct LFSClient { - pub url: Url, + pub batch_url: Url, + pub lfs_url: Url, pub client: Client, } @@ -40,7 +44,8 @@ impl ProtocolClient for LFSClient { .build() .unwrap(); Self { - url: lfs_server.join("/objects/batch").unwrap(), + batch_url: lfs_server.join("/objects/batch").unwrap(), + lfs_url: lfs_server, client, } } @@ -57,7 +62,7 @@ impl LFSClient { } /// push LFS objects to remote server - pub async fn push_objects<'a, I>(&self, objs: I, auth: Option) + pub async fn push_objects<'a, I>(&self, objs: I, auth: Option) -> Result<(), ()> where I: IntoIterator { @@ -75,7 +80,7 @@ impl LFSClient { let path = lfs::lfs_object_path(oid); if !path.exists() { eprintln!("fatal: LFS object not found: {}", oid); - return; + continue; } let size = path.metadata().unwrap().len() as i64; lfs_objs.push(RequestVars { @@ -85,15 +90,57 @@ impl LFSClient { }) } + { // verify locks + let (code, locks) = self.verify_locks(VerifiableLockRequest { + refs: Ref { name: command::lfs::current_refspec().await.unwrap() }, + ..Default::default() + }, auth.clone()).await; + + if code == StatusCode::FORBIDDEN { + eprintln!("fatal: Forbidden: You must have push access to verify locks"); + return Err(()); + } else if code == StatusCode::NOT_FOUND { + // By default, an LFS server that doesn't implement any locking endpoints should return 404. + // This response will not halt any Git pushes. + } else if !code.is_success() { + eprintln!("fatal: LFS verify locks failed. Status: {}", code); + return Err(()); + } else { + // success + tracing::debug!("LFS verify locks response:\n {:?}", locks); + let oids: HashSet = lfs_oids.iter().map(|(oid, _)| oid.clone()).collect(); + let ours = locks.ours.iter().filter(|l| { + let oid = lfs::get_oid_by_path(&l.path); + oids.contains(&oid) + }).collect::>(); + if !ours.is_empty() { + println!("The following files are locked by you, consider unlocking them:"); + for lock in ours { + println!(" - {}", lock.path); + } + } + let theirs = locks.theirs.iter().filter(|l| { + let oid = lfs::get_oid_by_path(&l.path); + oids.contains(&oid) + }).collect::>(); + if !theirs.is_empty() { + eprintln!("Locking failed: The following files are locked by another user:"); + for lock in theirs { + eprintln!(" - {}", lock.path); + } + return Err(()); + } + } + } + let batch_request = BatchRequest { operation: "upload".to_string(), transfers: vec![lfs::LFS_TRANSFER_API.to_string()], objects: lfs_objs, hash_algo: lfs::LFS_HASH_ALGO.to_string(), - enable_split: None, }; - let mut request = self.client.post(self.url.clone()).json(&batch_request); + let mut request = self.client.post(self.batch_url.clone()).json(&batch_request); if let Some(auth) = auth { request = request.basic_auth(auth.username, Some(auth.password)); } @@ -108,6 +155,7 @@ impl LFSClient { self.upload_object(obj).await; } println!("LFS objects push completed."); + Ok(()) } /// upload (PUT) one LFS file to remote server @@ -159,10 +207,9 @@ impl LFSClient { ..Default::default() }], hash_algo: lfs::LFS_HASH_ALGO.to_string(), - enable_split: None, }; - let request = self.client.post(self.url.clone()).json(&batch_request); + let request = self.client.post(self.batch_url.clone()).json(&batch_request); let response = request.send().await.unwrap(); let resp = response.json::().await.unwrap(); @@ -170,26 +217,156 @@ impl LFSClient { let link = resp.objects[0].actions.as_ref().unwrap().get("download").unwrap(); - let mut request = self.client.get(link.href.clone()); - for (k, v) in &link.header { - request = request.header(k, v); + let mut is_chunked = false; + // Chunk API + let links = match self.fetch_chunk_links(&link.href).await { + Ok(chunks) => { + is_chunked = true; + tracing::info!("LFS Chunk API supported."); + chunks + }, + Err(_) => vec![link.clone()], + }; + + let mut file = tokio::fs::File::create(path).await.unwrap(); + let mut checksum = Context::new(&SHA256); + println!("Downloading LFS file: {}", oid); + let mut cnt = 0; + let total = links.len(); + for link in links { + cnt += 1; + if is_chunked { + println!("- part: {}/{}", cnt, total); + } + + let mut request = self.client.get(&link.href); + for (k, v) in &link.header { + request = request.header(k, v); + } + + let response = request.send().await.unwrap(); + if !response.status().is_success() { + eprintln!("fatal: LFS download failed. Status: {}, Message: {}", response.status(), response.text().await.unwrap()); + return; + } + + let mut stream = response.bytes_stream(); + + while let Some(chunk) = stream.next().await { + let chunk = chunk.unwrap(); + file.write_all(&chunk).await.unwrap(); + checksum.update(&chunk); + } + } + let checksum = hex::encode(checksum.finish().as_ref()); + if checksum == oid { + println!("Downloaded."); + } else { + eprintln!("fatal: LFS download failed. Checksum mismatch: {} != {}", checksum, oid); + } + } + + /// Only for MonoRepo (mega) + async fn fetch_chunk_links(&self, obj_link: &str) -> Result, ()> { + let request = self.client.get(obj_link.to_owned() + "/chunks"); + let resp = request.send().await.unwrap(); + let code = resp.status(); + if code == StatusCode::NOT_FOUND { + tracing::info!("Remote LFS Server not support Chunks API"); + return Err(()); + } else if !code.is_success() { + tracing::debug!("fatal: LFS get chunk hrefs failed. Status: {}, Message: {}", code, resp.text().await.unwrap()); + return Err(()); } + let mut res = resp.json::().await.unwrap(); + // sort by offset + res.chunks.sort_by(|a, b| a.offset.cmp(&b.offset)); + Ok(res.chunks.into_iter().map(|c| c.link).collect()) + } +} +// LFS locks API +impl LFSClient { + pub async fn get_locks(&self, query: LockListQuery) -> LockList { + let url = self.lfs_url.join("/locks").unwrap(); + let mut request = self.client.get(url); + request = request.query(&[ + ("id", query.id), + ("path", query.path), + ("limit", query.limit), + ("cursor", query.cursor), + ("refspec", query.refspec) + ]); let response = request.send().await.unwrap(); if !response.status().is_success() { - eprintln!("fatal: LFS download failed. Status: {}, Message: {}", response.status(), response.text().await.unwrap()); - return; + eprintln!("fatal: LFS get locks failed. Status: {}, Message: {}", response.status(), response.text().await.unwrap()); + return LockList { + locks: Vec::new(), + next_cursor: String::default(), + }; } - println!("Downloading LFS file: {}", oid); - let mut file = tokio::fs::File::create(path).await.unwrap(); - let mut stream = response.bytes_stream(); + response.json::().await.unwrap() + } - while let Some(chunk) = stream.next().await { - let chunk = chunk.unwrap(); - file.write_all(&chunk).await.unwrap(); + /// lock an LFS file + /// - `refspec` is must in Mega Server, but optional in Git Doc + pub async fn lock(&self, path: String, refspec: String, basic_auth: Option) -> StatusCode { + let url = self.lfs_url.join("/locks").unwrap(); + let mut request = self.client.post(url).json(&LockRequest { + path, + refs: Ref { name: refspec }, + }); + if let Some(auth) = basic_auth { + request = request.basic_auth(auth.username, Some(auth.password)); + } + let resp = request.send().await.unwrap(); + let code = resp.status(); + if !resp.status().is_success() && code != StatusCode::FORBIDDEN { + eprintln!("fatal: LFS lock failed. Status: {}, Message: {}", code, resp.text().await.unwrap()); + } + code + } + + pub async fn unlock(&self, id: String, refspec: String, force: bool, basic_auth: Option) -> StatusCode { + let url = self.lfs_url.join(&format!("/locks/{}/unlock", id)).unwrap(); + let mut request = self.client.post(url).json(&UnlockRequest { + force: Some(force), + refs: Ref { name: refspec }, + }); + if let Some(auth) = basic_auth.clone() { + request = request.basic_auth(auth.username, Some(auth.password)); + } + let resp = request.send().await.unwrap(); + let code = resp.status(); + if !resp.status().is_success() && code != StatusCode::FORBIDDEN { + eprintln!("fatal: LFS unlock failed. Status: {}, Message: {}", code, resp.text().await.unwrap()); + } + code + } + + /// List Locks for Verification + pub async fn verify_locks(&self, query: VerifiableLockRequest, basic_auth: Option) + -> (StatusCode, VerifiableLockList) + { + let url = self.lfs_url.join("/locks/verify").unwrap(); + let mut request = self.client.post(url).json(&query); + if let Some(auth) = basic_auth { + request = request.basic_auth(auth.username, Some(auth.password)); + } + let resp = request.send().await.unwrap(); + let code = resp.status(); + // By default, an LFS server that doesn't implement any locking endpoints should return 404. + // This response will not halt any Git pushes. + if !code.is_success() && code != StatusCode::NOT_FOUND && code != StatusCode::FORBIDDEN { + eprintln!("fatal: LFS verify locks failed. Status: {}, Message: {}", code, resp.text().await.unwrap()); + return (code, VerifiableLockList { + ours: Vec::new(), + theirs: Vec::new(), + next_cursor: String::default(), + }); } - println!("Downloaded."); // TODO: checksum + (code, resp.json::().await.unwrap()) } } \ No newline at end of file diff --git a/libra/src/utils/lfs.rs b/libra/src/utils/lfs.rs index 3f54efb1c..6d341ba39 100644 --- a/libra/src/utils/lfs.rs +++ b/libra/src/utils/lfs.rs @@ -6,8 +6,10 @@ use lazy_static::lazy_static; use path_abs::{PathInfo, PathOps}; use regex::Regex; use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE}; -use sha2::{Digest, Sha256}; +use ring::digest::{Context, SHA256}; +use url::Url; use wax::Pattern; +use mercury::internal::index::Index; use crate::utils::{path, util}; use crate::utils::path_ext::PathExt; @@ -28,7 +30,7 @@ lazy_static! { /// Check if a file is LFS tracked /// - support Glob pattern matching (TODO: support .gitignore patterns) /// - only check root attributes file now, should check all attributes files in sub-dirs -/// - absolute path or relative path to workdir +/// - absolute path pub fn is_lfs_tracked

(path: P) -> bool where P: AsRef, @@ -67,7 +69,9 @@ pub fn generate_pointer_file(path: impl AsRef) -> (String, String) { /// [doc: server-discovery](https://github.com/git-lfs/git-lfs/blob/main/docs/api/server-discovery.md) /// - like https://git-server.com/foo/bar.git/info/lfs /// - support ssh & https & git@ format -pub fn generate_lfs_server_url(mut url: String) -> String { +#[deprecated(note = "It's for git, not monorepo")] +#[allow(dead_code)] +pub fn generate_git_lfs_server_url(mut url: String) -> String { if url.ends_with('/') { url.pop(); } @@ -87,6 +91,24 @@ pub fn generate_lfs_server_url(mut url: String) -> String { url } +/// Generate Mono LFS Server Url from repo Url. +/// - Just get domain with port +/// ### Example +/// https://github.com/git-lfs/git-lfs/blob/main/docs/api/locking.md -> https://github.com +/// +/// http://localhost:8000/xxx/yyy -> http://localhost:8000 +pub fn generate_lfs_server_url(url: String) -> String { + let url = Url::parse(&url).unwrap(); + match url.port() { + None => { + format!("{}://{}", url.scheme(), url.host().unwrap()) + } + Some(port) => { + format!("{}://{}:{}", url.scheme(), url.host().unwrap(), port) + } + } +} + /// Generate LFS cache path, in `.libra/lfs/objects` pub fn lfs_object_path(oid: &str) -> PathBuf { util::storage_path() @@ -96,6 +118,17 @@ pub fn lfs_object_path(oid: &str) -> PathBuf { .join(oid) } +/// Get LFS file oid by path (through `Index`), NOT re-calculate +pub fn get_oid_by_path(path: &str) -> String { + let index_file = path::index(); + let index = Index::load(&index_file).unwrap(); + let hash = index.get_hash(path, 0).unwrap(); + let storage = util::objects_storage(); + let data = storage.get(&hash).unwrap(); + let (oid, _) = parse_pointer_data(&data).unwrap(); + oid +} + /// Copy LFS file to `.libra/lfs/objects` /// - absolute path pub fn backup_lfs_file

(path: P, oid: &str) -> io::Result<()> @@ -112,13 +145,13 @@ where } /// SHA256 without type -/// TODO: performance optimization, 200MB 4s now, slower than `sha256sum` +// `ring` crate is much faster than `sha2` crate ( > 10 times) pub fn calc_lfs_file_hash

(path: P) -> io::Result where P: AsRef, { let path = path.as_ref(); - let mut hash = Sha256::new(); + let mut hash = Context::new(&SHA256); let file = File::open(path)?; let mut reader = BufReader::new(file); let mut buffer = [0; 65536]; @@ -129,11 +162,11 @@ where } hash.update(&buffer[..n]); } - let file_hash = hex::encode(hash.finalize()); + let file_hash = hex::encode(hash.finish().as_ref()); Ok(file_hash) } -/// Check if `data` is an LFS pointer, return `oid` +/// Check if `data` is an LFS pointer, return `oid` & `size` pub fn parse_pointer_data(data: &[u8]) -> Option<(String, u64)> { if data.len() > LFS_POINTER_MAX_SIZE { return None; @@ -154,6 +187,17 @@ pub fn parse_pointer_data(data: &[u8]) -> Option<(String, u64)> { None } +/// Read max LFS_POINTER_MAX_SIZE bytes +pub fn parse_pointer_file(path: impl AsRef) -> io::Result<(String, u64)> { + let mut file = File::open(path)?; + let mut buffer = [0; LFS_POINTER_MAX_SIZE]; + let bytes_read = file.read(&mut buffer)?; + if let Some((oid, size)) = parse_pointer_data(&buffer[..bytes_read]) { + return Ok((oid, size)); + } + Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid LFS pointer file")) +} + /// Extract LFS patterns from `.libra_attributes` file pub fn extract_lfs_patterns(file_path: &str) -> io::Result> { let path = Path::new(file_path); @@ -202,7 +246,7 @@ mod tests { } #[test] - fn test_gen_lfs_server_url() { + fn test_gen_git_lfs_server_url() { const LFS_SERVER_URL: &str = "https://github.com/web3infra-foundation/mega.git/info/lfs"; let url = "https://github.com/web3infra-foundation/mega".to_owned(); assert_eq!(generate_lfs_server_url(url), LFS_SERVER_URL); @@ -217,6 +261,14 @@ mod tests { assert_eq!(generate_lfs_server_url(url), LFS_SERVER_URL); } + #[test] + fn test_gen_mono_lfs_server_url() { + const LFS_SERVER_URL: &str = "https://github.com/web3infra-foundation/mega.git/info/lfs"; + assert_eq!(generate_lfs_server_url(LFS_SERVER_URL.to_owned()), "https://github.com"); + const LOCAL_LFS_SERVER_URL: &str = "http://localhost:8000/xxx/yyy"; + assert_eq!(generate_lfs_server_url(LOCAL_LFS_SERVER_URL.to_owned()), "http://localhost:8000"); + } + #[test] fn test_parse_pointer_data() { let data = r#"version https://git-lfs.github.com/spec/v1 diff --git a/mega/tests/lfs_test.rs b/mega/tests/lfs_test.rs index b36648743..9e9d72d37 100644 --- a/mega/tests/lfs_test.rs +++ b/mega/tests/lfs_test.rs @@ -11,6 +11,7 @@ use rand::Rng; use tempfile::TempDir; const PORT: u16 = 8000; // mega server port +const LARGE_FILE_SIZE_MB: usize = 60; /// check if git lfs is installed fn check_git_lfs() -> bool { let status = Command::new("git") @@ -22,12 +23,24 @@ fn check_git_lfs() -> bool { } fn run_git_cmd(args: &[&str]) { - let status = Command::new("git") + let output = Command::new("git") .args(args) - .status() + .output() .unwrap(); - assert!(status.success(), "Git command failed: git {}", args.join(" ")); + let status = output.status; + // assert!(status.success(), "Git command failed: git {}", args.join(" ")); + if !status.success() { + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + panic!( + "Git command failed: git {}\nStatus: {}\nStdout: {}\nStderr: {}", + args.join(" "), + status, + stdout, + stderr + ); + } } fn is_port_in_use(port: u16) -> bool { @@ -79,7 +92,7 @@ fn lfs_push(url: &str) -> io::Result<()> { run_git_cmd(&["lfs", "track", "*.bin"]); // create large file - generate_large_file("large_file.bin", 60)?; + generate_large_file("large_file.bin", LARGE_FILE_SIZE_MB)?; // add & commit run_git_cmd(&["add", "."]); @@ -99,7 +112,9 @@ fn lfs_clone(url: &str) -> io::Result<()> { // git clone url run_git_cmd(&["clone", url]); - assert!(Path::new("lfs/large_file.bin").exists(), "Failed to clone large file"); + let file = Path::new("lfs/large_file.bin"); + assert!(file.exists(), "Failed to clone large file"); + assert_eq!(file.metadata()?.len(), LARGE_FILE_SIZE_MB as u64 * 1024 * 1024); Ok(()) } @@ -112,6 +127,10 @@ fn lfs_split_with_git() { // start mega server at background run_mega_server(); + // MonoRepo (mega)'s lfs.url is not compatible with git-lfs + let lfs_url = format!("http://localhost:{}", PORT); + run_git_cmd(&["config", "--global", "lfs.url", &lfs_url]); + let url = &format!("http://localhost:{}/third-part/lfs.git", PORT); lfs_push(url).expect("Failed to push large file to mega server"); lfs_clone(url).expect("Failed to clone large file from mega server"); diff --git a/mono/src/api/lfs/lfs_router.rs b/mono/src/api/lfs/lfs_router.rs index 7f4bef491..582a7c49b 100644 --- a/mono/src/api/lfs/lfs_router.rs +++ b/mono/src/api/lfs/lfs_router.rs @@ -207,9 +207,8 @@ pub async fn lfs_process_batch( pub async fn lfs_fetch_chunk_ids( state: State, Path(oid): Path, - Json(json): Json, ) -> Result, (StatusCode, String)> { - let result = handler::lfs_fetch_chunk_ids(&state.context, &json).await; + let result = handler::lfs_fetch_chunk_ids(&state.context, &oid).await; match result { Ok(response) => { let size = response.iter().fold(0, |acc, chunk| acc + chunk.size);