diff --git a/src/archive/helper.rs b/src/archive/helper.rs index 79f2d79..ffbca13 100644 --- a/src/archive/helper.rs +++ b/src/archive/helper.rs @@ -1,9 +1,230 @@ -use std::path::Path; +use std::io; +use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::task::{Context, Poll}; +use serde::Serialize; +use tokio::io::AsyncWriteExt; + +use crate::archive::types::{ArchiveFormat, BundleRequest, GetArchiveRequest}; use crate::command::cmd::Cmd; use crate::command::env::Env; use crate::error::BabyError; +/// 缓存目录名,位于仓库根下(裸仓库无工作区,无污染风险)。 +const ARCHIVE_DIR_NAME: &str = "archive"; + +/// 已打包产物在仓库内的落盘目录。 +pub fn archive_store_dir(repo_dir: &Path) -> PathBuf { + repo_dir.join(ARCHIVE_DIR_NAME) +} + +/// 归档格式对应的文件扩展名(bundle 固定为 `bundle`)。 +pub fn format_ext(format: ArchiveFormat) -> &'static str { + match format { + ArchiveFormat::Tar => "tar", + ArchiveFormat::TarGz => "tar.gz", + ArchiveFormat::TarBz2 => "tar.bz2", + ArchiveFormat::Zip => "zip", + } +} + +/// 缓存文件最终路径:`/archive/.`。 +pub fn archive_file_path(repo_dir: &Path, key: &str, ext: &str) -> PathBuf { + archive_store_dir(repo_dir).join(format!("{}.{}", key, ext)) +} + +/// 写入过程中的临时路径:`.part.`,pid 后缀避免并发冲突。 +pub fn tmp_archive_path(final_path: &Path) -> PathBuf { + let mut os = final_path.as_os_str().to_os_string(); + os.push(format!(".part.{}", std::process::id())); + PathBuf::from(os) +} + +fn sha1_hex(data: &[u8]) -> Result { + let mut hasher = gix::hash::hasher(gix::hash::Kind::Sha1); + hasher.update(data); + let oid = hasher + .try_finalize() + .map_err(|e| BabyError::Custom(format!("archive cache key hash failed: {}", e)))?; + Ok(oid.to_string()) +} + +#[derive(Serialize)] +struct GetArchiveCacheKey<'a> { + oid: String, + format: &'a str, + prefix: Option<&'a str>, + path: Option<&'a str>, + exclude: Vec<&'a str>, +} + +/// get_archive 缓存键:commit 解析为 oid(防分支移动命中旧包),与其余参数规范化后哈希。 +/// exclude 仅参与求键时排序,命令参数保持原序。 +pub async fn get_archive_cache_key( + gitbaby: &crate::GitBaby, + req: &GetArchiveRequest, +) -> Result { + let oid = gitbaby.resolve_revision(&req.commit).await?; + let mut exclude: Vec<&str> = req.exclude.iter().map(String::as_str).collect(); + exclude.sort_unstable(); + let payload = GetArchiveCacheKey { + oid: oid.to_string(), + format: &req.format.to_string(), + prefix: req.prefix.as_deref(), + path: req.path.as_deref(), + exclude, + }; + let text = serde_json::to_vec(&payload) + .map_err(|e| BabyError::Custom(format!("archive cache key serialization failed: {}", e)))?; + sha1_hex(&text) +} + +/// create_bundle 缓存键:refs 以原顺序拼接待哈希,不解析。 +pub fn bundle_cache_key(req: &BundleRequest) -> Result { + sha1_hex(req.refs.join("\n").as_bytes()) +} + +pub async fn open_cached_file(path: &Path) -> Result, BabyError> { + match tokio::fs::File::open(path).await { + Ok(file) => Ok(Some(file)), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(BabyError::Io { + path: path.to_path_buf(), + source: e, + }), + } +} + +type PersistFut = + Pin, io::Result<()>)> + Send>>; + +/// 流式透写读取器:每块数据在返回给调用方之前先落盘, +/// 读尽后 flush 并以原子 rename 移到最终路径;任何写/rename 失败都在流上显式报错。 +pub struct PersistReader { + inner: R, + file: Option, + final_path: PathBuf, + tmp_path: PathBuf, + pending: Vec, + write_fut: Option, + scratch: Vec, + eof: bool, + finalized: bool, + error: Option, +} + +impl PersistReader { + pub fn new(inner: R, file: tokio::fs::File, tmp_path: PathBuf, final_path: PathBuf) -> Self { + Self { + inner, + file: Some(file), + final_path, + tmp_path, + pending: Vec::new(), + write_fut: None, + scratch: vec![0u8; 16 * 1024], + eof: false, + finalized: false, + error: None, + } + } +} + +impl tokio::io::AsyncRead for PersistReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + let this = self.as_mut().get_mut(); + loop { + if let Some(e) = this.error.take() { + return Poll::Ready(Err(e)); + } + if this.finalized { + return Poll::Ready(Ok(())); + } + if let Some(mut fut) = this.write_fut.take() { + match fut.as_mut().poll(cx) { + Poll::Pending => { + this.write_fut = Some(fut); + return Poll::Pending; + } + Poll::Ready((file, result)) => { + this.file = file; + if let Err(e) = result { + let _ = std::fs::remove_file(&this.tmp_path); + this.error = Some(e); + } + continue; + } + } + } + if !this.pending.is_empty() { + let n = buf.remaining().min(this.pending.len()); + if n == 0 { + return Poll::Pending; + } + buf.put_slice(&this.pending[..n]); + if n == this.pending.len() { + this.pending.clear(); + } else { + this.pending.drain(..n); + } + return Poll::Ready(Ok(())); + } + if this.eof { + if let Some(file) = this.file.take() { + let tmp_path = this.tmp_path.clone(); + let final_path = this.final_path.clone(); + this.write_fut = Some(Box::pin(async move { + let mut file = file; + if let Err(e) = file.flush().await { + let _ = std::fs::remove_file(&tmp_path); + return (None, Err(e)); + } + drop(file); + if let Err(e) = tokio::fs::rename(&tmp_path, &final_path).await { + let _ = std::fs::remove_file(&tmp_path); + return (None, Err(e)); + } + (None, Ok(())) + })); + continue; + } + this.finalized = true; + return Poll::Ready(Ok(())); + } + let mut scratch_buf = tokio::io::ReadBuf::new(&mut this.scratch); + match Pin::new(&mut this.inner).poll_read(cx, &mut scratch_buf) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(e)) => { + let _ = std::fs::remove_file(&this.tmp_path); + return Poll::Ready(Err(e)); + } + Poll::Ready(Ok(())) => { + let n = scratch_buf.filled().len(); + if n == 0 { + this.eof = true; + continue; + } + this.pending = scratch_buf.filled().to_vec(); + if let Some(file) = this.file.take() { + let chunk = this.pending.clone(); + this.write_fut = Some(Box::pin(async move { + let mut file = file; + let result = file.write_all(&chunk).await; + (Some(file), result) + })); + } + continue; + } + } + } + } +} + pub fn build_create_archive_cmd( dir: &Path, env: Env, diff --git a/src/archive/usecase.rs b/src/archive/usecase.rs index 3d3d31d..8fff15f 100644 --- a/src/archive/usecase.rs +++ b/src/archive/usecase.rs @@ -39,59 +39,83 @@ impl GitBaby { pub async fn get_archive( &self, req: GetArchiveRequest, - ) -> Result>>, BabyError> - { + ) -> Result>>, BabyError> { let (dir, env) = build_env(self).await?; + let key = helper::get_archive_cache_key(self, &req).await?; + let final_path = helper::archive_file_path(&dir, &key, helper::format_ext(req.format)); + if let Some(file) = helper::open_cached_file(&final_path).await? { + return Ok(ChildArchiveReader::new( + BufReader::new(Box::new(file) as Box), + None, + )); + } + let tmp_path = helper::tmp_archive_path(&final_path); + tokio::fs::create_dir_all(helper::archive_store_dir(&dir)) + .await + .map_err(|source| BabyError::Io { + path: helper::archive_store_dir(&dir), + source, + })?; + let file = tokio::fs::File::create(&tmp_path) + .await + .map_err(|source| BabyError::Io { + path: tmp_path.clone(), + source, + })?; let mut cmd = helper::build_get_archive_cmd(&dir, env, &req)?; cmd.spawn().await.map_err(BabyError::from)?; - let child = match cmd.cmd.as_mut() { - Some(c) => c, - None => { - return Err(BabyError::ArchiveFailed( - "git archive child not spawned".to_string(), - )); - } - }; - let stdout: Box = - Box::new(child.stdout.take().ok_or_else(|| { - BabyError::ArchiveFailed("git archive stdout not piped".to_string()) - })?); - let _ = std::mem::replace( + let stdout: Box = std::mem::replace( &mut cmd.stdout, Box::new(tokio::io::empty()) as Box, ); - Ok(ChildArchiveReader::new(BufReader::new(stdout), None)) + let reader = helper::PersistReader::new(stdout, file, tmp_path, final_path); + Ok(ChildArchiveReader::new( + BufReader::new(Box::new(reader) as Box), + None, + )) } pub async fn create_bundle( &self, req: BundleRequest, - ) -> Result>>, BabyError> - { + ) -> Result>>, BabyError> { if req.refs.is_empty() { return Err(BabyError::Custom( "bundle requires at least one ref".to_string(), )); } let (dir, env) = build_env(self).await?; + let key = helper::bundle_cache_key(&req)?; + let final_path = helper::archive_file_path(&dir, &key, "bundle"); + if let Some(file) = helper::open_cached_file(&final_path).await? { + return Ok(ChildArchiveReader::new( + BufReader::new(Box::new(file) as Box), + None, + )); + } + let tmp_path = helper::tmp_archive_path(&final_path); + tokio::fs::create_dir_all(helper::archive_store_dir(&dir)) + .await + .map_err(|source| BabyError::Io { + path: helper::archive_store_dir(&dir), + source, + })?; + let file = tokio::fs::File::create(&tmp_path) + .await + .map_err(|source| BabyError::Io { + path: tmp_path.clone(), + source, + })?; let mut cmd = helper::build_bundle_cmd(&dir, env, &req)?; cmd.spawn().await.map_err(BabyError::from)?; - let child = match cmd.cmd.as_mut() { - Some(c) => c, - None => { - return Err(BabyError::ArchiveFailed( - "git bundle child not spawned".to_string(), - )); - } - }; - let stdout: Box = - Box::new(child.stdout.take().ok_or_else(|| { - BabyError::ArchiveFailed("git bundle stdout not piped".to_string()) - })?); - let _ = std::mem::replace( + let stdout: Box = std::mem::replace( &mut cmd.stdout, Box::new(tokio::io::empty()) as Box, ); - Ok(ChildArchiveReader::new(BufReader::new(stdout), None)) + let reader = helper::PersistReader::new(stdout, file, tmp_path, final_path); + Ok(ChildArchiveReader::new( + BufReader::new(Box::new(reader) as Box), + None, + )) } } diff --git a/tests/archive.rs b/tests/archive.rs new file mode 100644 index 0000000..324af4f --- /dev/null +++ b/tests/archive.rs @@ -0,0 +1,319 @@ +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; + +use async_trait::async_trait; + +use gitbaby::GitBaby; +use gitbaby::archive::{ArchiveFormat, BundleRequest, GetArchiveRequest}; +use gitbaby::error::BabyError; +use gitbaby::repo::RepositoryFacade; + +static COUNTER: AtomicU32 = AtomicU32::new(0); + +struct TestFacade { + dir: PathBuf, +} + +#[async_trait] +impl RepositoryFacade for TestFacade { + async fn git_repo_dir(&self) -> Result { + Ok(self.dir.clone()) + } + + async fn git_alternate_object_directories(&self) -> Result, BabyError> { + Ok(Vec::new()) + } + + async fn gix_repo(&self) -> Result { + gix::open(&self.dir).map_err(|e| BabyError::Custom(e.to_string())) + } +} + +fn temp_repo() -> PathBuf { + let n = COUNTER.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!("gitbaby-archive-test-{}-{n}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("mkdir repo"); + let out = Command::new("git") + .args(["init", "-q"]) + .current_dir(&dir) + .output() + .expect("git init failed"); + assert!(out.status.success(), "git init: {:?}", out); + dir +} + +fn cleanup(repo: &Path) { + let _ = std::fs::remove_dir_all(repo); +} + +fn git_commit(repo: &Path, msg: &str) { + let out = Command::new("git") + .args([ + "-c", + "user.name=T", + "-c", + "user.email=t@e", + "commit", + "-q", + "-m", + msg, + ]) + .current_dir(repo) + .output() + .expect("git commit failed"); + assert!(out.status.success(), "git commit `{msg}`: {:?}", out); +} + +fn write_file(repo: &Path, rel: &str, content: &str) { + let p = repo.join(rel); + std::fs::create_dir_all(p.parent().expect("parent")).expect("mkdir"); + std::fs::write(p, content).expect("write"); +} + +fn index_commit(repo: &Path, msg: &str) { + let out = Command::new("git") + .args(["add", "-A"]) + .current_dir(repo) + .output() + .expect("git add failed"); + assert!(out.status.success(), "git add: {:?}", out); + git_commit(repo, msg); +} + +async fn baby_of(repo: PathBuf) -> GitBaby { + GitBaby::new(Arc::new(TestFacade { dir: repo })) +} + +fn archive_dir(repo: &Path) -> PathBuf { + repo.join("archive") +} + +fn archive_files(repo: &Path) -> Vec { + let mut files: Vec = std::fs::read_dir(archive_dir(repo)) + .expect("read archive dir") + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .filter(|p| p.is_file()) + .collect(); + files.sort(); + files +} + +#[tokio::test] +async fn get_archive_persists_and_hits_cache() { + let repo = temp_repo(); + write_file(&repo, "a.txt", "one"); + write_file(&repo, "b.txt", "bee"); + index_commit(&repo, "first"); + + let baby = baby_of(repo.clone()).await; + let req = || GetArchiveRequest { + commit: "HEAD".to_string(), + format: ArchiveFormat::Tar, + prefix: None, + path: None, + exclude: Vec::new(), + }; + + let first_bytes = baby + .get_archive(req()) + .await + .expect("get_archive") + .read_to_end() + .await + .expect("read archive body"); + + let files = archive_files(&repo); + assert_eq!(files.len(), 1, "one cached archive expected"); + assert_eq!( + files[0].extension().and_then(|e| e.to_str()), + Some("tar"), + "cached file: {:?}", + files[0] + ); + let on_disk = std::fs::read(&files[0]).expect("read cached file"); + assert_eq!(on_disk, first_bytes, "cached bytes match streamed bytes"); + + let direct = Command::new("git") + .args(["archive", "--format=tar", "HEAD"]) + .current_dir(&repo) + .output() + .expect("git archive direct"); + assert!(direct.status.success(), "git archive direct: {:?}", direct); + assert_eq!(direct.stdout, first_bytes, "matches raw git archive output"); + + let second_bytes = baby + .get_archive(req()) + .await + .expect("get_archive cached") + .read_to_end() + .await + .expect("read archive body"); + assert_eq!(second_bytes, first_bytes, "cache hit returns same bytes"); + assert_eq!(archive_files(&repo).len(), 1); + + cleanup(&repo); +} + +#[tokio::test] +async fn get_archive_keys_differ_by_options() { + let repo = temp_repo(); + write_file(&repo, "a.txt", "one"); + index_commit(&repo, "first"); + + let baby = baby_of(repo.clone()).await; + let base = GetArchiveRequest { + commit: "HEAD".to_string(), + format: ArchiveFormat::Tar, + prefix: None, + path: None, + exclude: Vec::new(), + }; + + let plain = baby + .get_archive(base.clone()) + .await + .expect("plain") + .read_to_end() + .await + .expect("plain body"); + + let prefixed = baby + .get_archive(GetArchiveRequest { + prefix: Some("p/".to_string()), + ..base.clone() + }) + .await + .expect("prefixed") + .read_to_end() + .await + .expect("prefixed body"); + + let excluded = baby + .get_archive(GetArchiveRequest { + exclude: vec!["a.txt".to_string()], + ..base.clone() + }) + .await + .expect("excluded") + .read_to_end() + .await + .expect("excluded body"); + + assert_ne!(plain, prefixed); + assert_ne!(plain, excluded); + assert_ne!(prefixed, excluded); + + let files = archive_files(&repo); + assert_eq!(files.len(), 3, "three distinct keys cached"); + + let zipped = baby + .get_archive(GetArchiveRequest { + format: ArchiveFormat::Zip, + ..base + }) + .await + .expect("zip") + .read_to_end() + .await + .expect("zip body"); + let files = archive_files(&repo); + assert_eq!(files.len(), 4); + assert!( + files + .iter() + .any(|p| p.extension().and_then(|e| e.to_str()) == Some("zip")), + "zip file present: {:?}", + files + ); + assert!(zipped.starts_with(b"PK"), "zip magic"); + assert_ne!(zipped, plain); + + cleanup(&repo); +} + +#[tokio::test] +async fn get_archive_new_commit_new_key() { + let repo = temp_repo(); + write_file(&repo, "a.txt", "one"); + index_commit(&repo, "first"); + + let baby = baby_of(repo.clone()).await; + let req = || GetArchiveRequest { + commit: "HEAD".to_string(), + format: ArchiveFormat::Tar, + prefix: None, + path: None, + exclude: Vec::new(), + }; + + let _ = baby + .get_archive(req()) + .await + .expect("first revision") + .read_to_end() + .await + .expect("body"); + assert_eq!(archive_files(&repo).len(), 1); + + write_file(&repo, "a.txt", "two"); + index_commit(&repo, "second"); + + let _ = baby + .get_archive(req()) + .await + .expect("second revision") + .read_to_end() + .await + .expect("body"); + let files = archive_files(&repo); + assert_eq!(files.len(), 2, "new commit produces new cached key"); + + cleanup(&repo); +} + +#[tokio::test] +async fn create_bundle_persists_and_hits_cache() { + let repo = temp_repo(); + write_file(&repo, "a.txt", "one"); + index_commit(&repo, "first"); + + let baby = baby_of(repo.clone()).await; + let req = || BundleRequest { + refs: vec!["HEAD".to_string()], + }; + + let first = baby + .create_bundle(req()) + .await + .expect("create_bundle") + .read_to_end() + .await + .expect("bundle body"); + + let files = archive_files(&repo); + assert_eq!(files.len(), 1, "one cached bundle expected"); + assert_eq!( + files[0].extension().and_then(|e| e.to_str()), + Some("bundle"), + "cached file: {:?}", + files[0] + ); + let on_disk = std::fs::read(&files[0]).expect("read cached file"); + assert_eq!(on_disk, first, "cached bundle bytes match streamed bytes"); + + let second = baby + .create_bundle(req()) + .await + .expect("create_bundle cached") + .read_to_end() + .await + .expect("bundle body"); + assert_eq!(second, first, "cache hit returns same bundle bytes"); + assert_eq!(archive_files(&repo).len(), 1); + + cleanup(&repo); +}