diff --git a/Cargo.lock b/Cargo.lock index 1c22f001a..752796b37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2851,6 +2851,7 @@ dependencies = [ "futures-lite", "gitmirror-git", "gix", + "gix-archive", "gix-pack", "gix-receive-pack", "gix-transport", diff --git a/gitmirror/crates/gitmirror-xrpc/Cargo.toml b/gitmirror/crates/gitmirror-xrpc/Cargo.toml index 39c07c2e7..2cb4a8f48 100644 --- a/gitmirror/crates/gitmirror-xrpc/Cargo.toml +++ b/gitmirror/crates/gitmirror-xrpc/Cargo.toml @@ -28,16 +28,17 @@ thiserror = { workspace = true } tracing = "0.1" line-numbers = "0.4.0" rustc-hash = "2.1.2" -gix = { version = "0.84", features = ["parallel", "blob-diff", "blame", "merge", "sha1", "sha256", "revision", "tree-editor"] } +gix = { version = "0.84", features = ["parallel", "blob-diff", "blame", "merge", "sha1", "sha256", "revision", "tree-editor", "worktree-stream"] } +gix-archive = { workspace = true } gix-pack = { workspace = true } gix-transport = { workspace = true, features = ["async-client"] } quick_cache = { workspace = true } futures-lite = { workspace = true } reqwest = { workspace = true } sha2 = { workspace = true } +tempfile = "3" [dev-dependencies] -tempfile = "3" tokio = { workspace = true, features = ["macros", "io-util"] } [lints] diff --git a/gitmirror/crates/gitmirror-xrpc/src/archive.rs b/gitmirror/crates/gitmirror-xrpc/src/archive.rs new file mode 100644 index 000000000..1f46cf96f --- /dev/null +++ b/gitmirror/crates/gitmirror-xrpc/src/archive.rs @@ -0,0 +1,170 @@ +//! ported from `knot2/crates/knot-git/src/archive.rs` + +use std::io::{Seek, SeekFrom, Write}; + +use gix::ObjectId; +use gix::bstr::BString; +use gix::error::ResultExt as _; +use gix::objs::tree::EntryKind; + +use crate::error::XrpcError; + +const TAR_BLOCK: u64 = 512; +const MAX_TREE_DEPTH: usize = 1024; +const MAX_PREFIX_BYTES: usize = 255; + +pub(crate) fn validate_prefix(value: &str) -> Result<(), XrpcError> { + let safe = !value.is_empty() + && value.len() <= MAX_PREFIX_BYTES + && !value.contains('\0') + && !value.starts_with(['/', '\\']) + && value.split(['/', '\\']).all(|component| component != ".."); + match safe { + true => Ok(()), + false => Err(XrpcError::InvalidRequest(format!( + "invalid archive prefix {value:?}" + ))), + } +} + +pub(crate) fn bound_source( + repo: &gix::Repository, + tree: ObjectId, + limit: u64, +) -> Result<(), XrpcError> { + bound_tree(repo, tree, limit, MAX_TREE_DEPTH, &mut 0) +} + +fn bound_tree( + repo: &gix::Repository, + tree: ObjectId, + limit: u64, + nesting: usize, + spooled: &mut u64, +) -> Result<(), XrpcError> { + if nesting == 0 { + return Err(XrpcError::InvalidRequest( + "tree nesting is too deep to archive".to_owned(), + )); + } + if tree == ObjectId::empty_tree(repo.object_hash()) { + return Ok(()); + } + let internal = |e: &dyn std::fmt::Display| XrpcError::Internal(e.to_string()); + let object = repo.find_tree(tree).map_err(|e| internal(&e))?; + let decoded = object.decode().map_err(|e| internal(&e))?; + decoded.entries.iter().try_for_each(|entry| { + let oid = entry.oid.to_owned(); + *spooled = spooled.saturating_add(TAR_BLOCK); + match entry.mode.kind() { + _ if *spooled > limit => Err(XrpcError::ArchiveTooLarge { limit }), + EntryKind::Commit => Ok(()), + EntryKind::Tree => bound_tree(repo, oid, limit, nesting - 1, spooled), + _ => { + let content = repo.find_header(oid).map_err(|e| internal(&e))?.size(); + *spooled = spooled.saturating_add(content.next_multiple_of(TAR_BLOCK)); + match *spooled > limit { + true => Err(XrpcError::ArchiveTooLarge { limit }), + false => Ok(()), + } + } + } + }) +} + +pub(crate) fn write_tar_gz( + repo: &gix::Repository, + tree: ObjectId, + prefix: &str, + modification_time: i64, + limit: u64, + out: impl Write, +) -> anyhow::Result<()> { + let (mut stream, _index) = repo.worktree_stream(tree)?; + let out = Bounded { + inner: out, + written: 0, + limit, + }; + gix_archive::write_stream( + &mut stream, + |stream| stream.next_entry().or_erased(), + out, + gix_archive::Options { + format: gix_archive::Format::TarGz { + compression_level: None, + }, + tree_prefix: Some(BString::from(format!("{prefix}/"))), + modification_time, + }, + ) + .map_err(|e| anyhow::anyhow!("{e}"))?; + Ok(()) +} + +pub(crate) fn write_zip( + repo: &gix::Repository, + tree: ObjectId, + prefix: &str, + modification_time: i64, + limit: u64, + mut out: impl Write, +) -> anyhow::Result<()> { + let (mut stream, _index) = repo.worktree_stream(tree)?; + let mut spool = Bounded { + inner: tempfile::tempfile()?, + written: 0, + limit, + }; + gix_archive::write_stream_seek( + &mut stream, + |stream| stream.next_entry().or_erased(), + &mut spool, + gix_archive::Options { + format: gix_archive::Format::Zip { + compression_level: None, + }, + tree_prefix: Some(BString::from(format!("{prefix}/"))), + modification_time, + }, + ) + .map_err(|e| anyhow::anyhow!("{e}"))?; + let mut file = spool.inner; + file.rewind()?; + std::io::copy(&mut file, &mut out)?; + out.flush()?; + Ok(()) +} + +struct Bounded { + inner: W, + written: u64, + limit: u64, +} + +impl Write for Bounded { + fn write(&mut self, data: &[u8]) -> std::io::Result { + let remaining = self.limit.saturating_sub(self.written); + if data.len() as u64 > remaining { + return Err(std::io::Error::other(format!( + "archive exceeds {} bytes", + self.limit + ))); + } + let written = self.inner.write(data)?; + self.written = self.written.saturating_add(written as u64); + Ok(written) + } + + fn flush(&mut self) -> std::io::Result<()> { + self.inner.flush() + } +} + +impl Seek for Bounded { + fn seek(&mut self, pos: SeekFrom) -> std::io::Result { + let position = self.inner.seek(pos)?; + self.written = position; + Ok(position) + } +} diff --git a/gitmirror/crates/gitmirror-xrpc/src/error.rs b/gitmirror/crates/gitmirror-xrpc/src/error.rs index b7e918394..488f9cb37 100644 --- a/gitmirror/crates/gitmirror-xrpc/src/error.rs +++ b/gitmirror/crates/gitmirror-xrpc/src/error.rs @@ -15,7 +15,6 @@ pub(crate) enum XrpcError { RepoNotFound { detail: String }, RefNotFound { rev: String, detail: String }, RevisionNotFound { rev: String }, - BranchNotFound { name: String }, EntryNotFound { path: String }, PathNotFound { path: String }, NoMergeBase { base: String, head: String }, @@ -23,6 +22,7 @@ pub(crate) enum XrpcError { FileNotFound { path: String }, BlameTooLarge, BlobTooLarge { limit: u64 }, + ArchiveTooLarge { limit: u64 }, BlameTimeout, MergeConflict(Vec), PushRejected(String), @@ -55,11 +55,6 @@ impl IntoResponse for XrpcError { "RevisionNotFound", format!("commit not found: {rev}"), ), - Self::BranchNotFound { name } => ( - StatusCode::NOT_FOUND, - "BranchNotFound", - format!("branch not found: {name}"), - ), Self::EntryNotFound { path } => ( StatusCode::NOT_FOUND, "EntryNotFound", @@ -93,6 +88,11 @@ impl IntoResponse for XrpcError { "BlobTooLarge", format!("file is too large to serve, limit is {limit} bytes"), ), + Self::ArchiveTooLarge { limit } => ( + StatusCode::PAYLOAD_TOO_LARGE, + "ArchiveTooLarge", + format!("archive is too large to serve, limit is {limit} bytes"), + ), Self::BlameTooLarge => ( StatusCode::PAYLOAD_TOO_LARGE, "BlameTooLarge", diff --git a/gitmirror/crates/gitmirror-xrpc/src/lib.rs b/gitmirror/crates/gitmirror-xrpc/src/lib.rs index 7d7994c1b..a1fc8d6fd 100644 --- a/gitmirror/crates/gitmirror-xrpc/src/lib.rs +++ b/gitmirror/crates/gitmirror-xrpc/src/lib.rs @@ -1,3 +1,4 @@ +mod archive; mod did_ext; mod diff; mod error; @@ -33,6 +34,7 @@ pub struct AppState { pub(crate) stats: Arc, pub(crate) redis: Option, pub(crate) max_blob_bytes: u64, + pub(crate) max_archive_bytes: u64, } impl AppState { @@ -45,6 +47,7 @@ impl AppState { clock: Arc, redis: Option, max_blob_bytes: u64, + max_archive_bytes: u64, ) -> Self { let layout = Arc::new(Layout::new((*repo_base).clone())); Self { @@ -56,6 +59,7 @@ impl AppState { stats: Arc::default(), redis, max_blob_bytes, + max_archive_bytes, } } } @@ -89,11 +93,14 @@ pub fn router(state: AppState) -> Router { .route("/xrpc/sh.tangled.git.temp2.mergeCheck", get(git::merge_check)) .route("/xrpc/sh.tangled.git.mergeCommit", post(git::merge_commit)) .route("/xrpc/org.tangled.temp.git.deleteBranch", post(git::delete_branch)) + .route("/xrpc/org.tangled.temp.git.getArchive", get(reads::get_archive)) .route("/xrpc/org.tangled.temp.git.getBlob", get(reads::get_blob)) - .route("/xrpc/org.tangled.temp.git.getBranch", get(reads::get_branch)) .route("/xrpc/org.tangled.temp.git.getEntry", get(reads::get_entry)) .route("/xrpc/org.tangled.temp.git.getMergeBase", get(reads::get_merge_base)) + .route("/xrpc/org.tangled.temp.git.getRef", get(reads::get_ref)) .route("/xrpc/org.tangled.temp.git.getTree", get(reads::get_tree)) + .route("/xrpc/org.tangled.temp.git.listBranches", get(reads::list_branches)) + .route("/xrpc/org.tangled.temp.git.listTags", get(reads::list_tags)) .layer(axum::middleware::from_fn(metrics::metrics_middleware)) .with_state(state) } diff --git a/gitmirror/crates/gitmirror-xrpc/src/routes/reads.rs b/gitmirror/crates/gitmirror-xrpc/src/routes/reads.rs index 4a435a2fd..4c4ccbcc6 100644 --- a/gitmirror/crates/gitmirror-xrpc/src/routes/reads.rs +++ b/gitmirror/crates/gitmirror-xrpc/src/routes/reads.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; +use std::io::{BufWriter, Write}; use std::time::Instant; +use axum::body::Body; use axum::extract::State; use axum::http::{HeaderMap, StatusCode, header}; use axum::response::{IntoResponse as _, Response}; @@ -9,14 +11,17 @@ use gix::bstr::ByteSlice as _; use gix::objs::tree::EntryKind; use jacquard_axum::{ExtractXrpc, XrpcResponse}; use jacquard_common::ToSmolStr; +use jacquard_common::deps::smol_str::SmolStr; use jacquard_common::types::did::Did; -use lexicons::org_tangled::temp::git::{get_blob, get_branch, get_entry, get_merge_base, get_tree}; -use lexicons::sh_tangled::git::temp; +use lexicons::org_tangled::temp::git::{ + get_archive, get_blob, get_entry, get_merge_base, get_ref, get_tree, list_branches, list_tags, +}; +use lexicons::sh_tangled::git::{self as sh_git, temp}; use sha2::{Digest as _, Sha256}; use tracing::warn; use crate::routes::git::{GixSignature, allocation_limited}; -use crate::{AppState, error::XrpcError, last_commit, metrics::BlockingKind, sniff}; +use crate::{AppState, archive, error::XrpcError, last_commit, metrics::BlockingKind, sniff}; const IMMUTABLE: &str = "public, max-age=31536000, immutable"; const MUTABLE: &str = "private, no-store"; @@ -24,6 +29,12 @@ const MUTABLE: &str = "private, no-store"; const BLOB_ETAG_DOMAIN: &str = "knotmirror.blob.v1"; const LARGE_BLOB: u64 = 1024 * 1024; +const ARCHIVE_ETAG_DOMAIN: &str = "gitutil.archive.v1"; +const ARCHIVE_CHUNK: usize = 64 * 1024; + +const DEFAULT_REFS_LIMIT: i64 = 50; +const MAX_REFS_LIMIT: i64 = 100; + fn cache_control(refs: &[&str]) -> &'static str { if !refs.is_empty() && refs.iter().all(|r| ObjectId::from_hex(r.as_bytes()).is_ok()) { IMMUTABLE @@ -159,43 +170,6 @@ async fn last_commits( last_commit::store(pool.as_ref(), &head, &tree_path, &walked).await; Ok(hydrated) } - -pub(crate) async fn get_branch( - State(state): State, - ExtractXrpc(args): ExtractXrpc, -) -> Result, XrpcError> { - tracing::Span::current().record("repo", args.repo.as_str()); - let repo = open(&state, &args.repo)?; - let name = args.name.to_string(); - - crate::metrics::blocking(BlockingKind::Git, move || { - let missing = || XrpcError::BranchNotFound { name: name.clone() }; - let mut reference = repo - .find_reference(format!("refs/heads/{name}").as_str()) - .map_err(|_| missing())?; - let commit = reference.peel_to_commit().map_err(|_| missing())?; - let decoded = commit - .decode() - .map_err(|e| XrpcError::Internal(e.to_string()))?; - Ok(get_branch::GetBranchOutput { - name: name.as_str().into(), - hash: commit.id.to_smolstr(), - message: decoded.message.to_smolstr(), - author: GixSignature( - decoded - .author() - .map_err(|e| XrpcError::Internal(e.to_string()))?, - ) - .try_into() - .map_err(|e: anyhow::Error| XrpcError::Internal(e.to_string()))?, - extra_data: Default::default(), - }) - }) - .await - .map_err(|e| XrpcError::Internal(e.to_string()))? - .map(XrpcResponse) -} - pub(crate) async fn get_merge_base( State(state): State, ExtractXrpc(args): ExtractXrpc, @@ -556,3 +530,392 @@ pub(crate) async fn get_blob( ) .into_response()) } + +fn internal(e: impl std::fmt::Display) -> XrpcError { + XrpcError::Internal(e.to_string()) +} + +fn page(limit: Option, cursor: Option<&str>) -> Result<(usize, usize), XrpcError> { + let limit = limit.unwrap_or(DEFAULT_REFS_LIMIT); + if !(1..=MAX_REFS_LIMIT).contains(&limit) { + return Err(XrpcError::InvalidRequest(format!( + "limit must be between 1 and {MAX_REFS_LIMIT}" + ))); + } + let offset = cursor + .map(|cursor| { + cursor + .parse::() + .map_err(|_| XrpcError::InvalidRequest(format!("invalid cursor {cursor:?}"))) + }) + .transpose()? + .unwrap_or(0); + Ok((offset, limit as usize)) +} + +fn paginate( + mut rows: Vec<(i64, String, T)>, + offset: usize, + limit: usize, +) -> (Vec, Option) { + rows.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(&b.1))); + let next = offset.saturating_add(limit); + let cursor = (next < rows.len()).then(|| next.to_smolstr()); + let rows = rows + .into_iter() + .skip(offset) + .take(limit) + .map(|(_, _, row)| row) + .collect(); + (rows, cursor) +} + +fn summarize( + commit: &gix::Commit<'_>, +) -> Result<(ObjectId, SmolStr, sh_git::Signature), XrpcError> { + let decoded = commit.decode().map_err(internal)?; + let author = GixSignature(decoded.author().map_err(internal)?) + .try_into() + .map_err(|e: anyhow::Error| internal(e))?; + Ok((commit.id, decoded.message.to_smolstr(), author)) +} + +pub(crate) async fn list_branches( + State(state): State, + ExtractXrpc(args): ExtractXrpc, +) -> Result, XrpcError> { + tracing::Span::current().record("repo", args.repo.as_str()); + let repo = open(&state, &args.repo)?; + let (offset, limit) = page(args.limit, args.cursor.as_deref())?; + + crate::metrics::blocking(BlockingKind::Git, move || { + let platform = repo.references().map_err(internal)?; + let mut rows = Vec::new(); + let mut total = 0; + for reference in platform.local_branches().map_err(internal)? { + let mut reference = reference.map_err(internal)?; + total += 1; + let Ok(commit) = reference.peel_to_commit() else { + continue; + }; + let time = commit.time().map_err(internal)?.seconds; + let name = reference.name().shorten().to_str_lossy().into_owned(); + rows.push((time, name.clone(), (name, commit.id))); + } + let default_branch = repo.head_name().map_err(internal)?.and_then(|name| { + match name.category_and_short_name() { + Some((gix::refs::Category::LocalBranch, short)) => { + Some(short.to_str_lossy().into()) + } + _ => None, + } + }); + let (rows, cursor) = paginate(rows, offset, limit); + let branches = rows + .into_iter() + .map(|(name, oid)| { + let commit = repo.find_commit(oid).map_err(internal)?; + let (oid, message, author) = summarize(&commit)?; + let committer = GixSignature(commit.committer().map_err(internal)?) + .try_into() + .map_err(|e: anyhow::Error| internal(e))?; + Ok(list_branches::BranchInfo { + name: name.as_str().into(), + oid: oid.to_smolstr(), + message, + author, + committer, + extra_data: Default::default(), + }) + }) + .collect::, XrpcError>>()?; + Ok(list_branches::ListBranchesOutput { + branches, + cursor, + total, + default_branch, + extra_data: Default::default(), + }) + }) + .await + .map_err(internal)? + .map(XrpcResponse) +} + +fn tag_info( + repo: &gix::Repository, + reference: &gix::Reference<'_>, +) -> Result, XrpcError> { + let Some(target) = reference.try_id() else { + return Ok(None); + }; + let object = repo.find_object(target).map_err(internal)?; + let (annotation, tagged_at) = match object.kind { + gix::object::Kind::Tag => { + let tag = object.clone().into_tag(); + let decoded = tag.decode().map_err(internal)?; + let tagger = decoded.tagger().map_err(internal)?; + let tagged_at = tagger + .map(|tagger| tagger.time().map(|time| time.seconds)) + .transpose() + .map_err(internal)?; + let tagger = tagger + .map(|tagger| GixSignature(tagger).try_into()) + .transpose() + .map_err(|e: anyhow::Error| internal(e))?; + let annotation = list_tags::Annotation { + message: decoded.message.to_smolstr(), + tagger, + extra_data: Default::default(), + }; + (Some(annotation), tagged_at) + } + _ => (None, None), + }; + let Ok(commit) = object + .peel_tags_to_end() + .map_err(internal)? + .try_into_commit() + else { + return Ok(None); + }; + let time = match tagged_at { + Some(time) => time, + None => commit.time().map_err(internal)?.seconds, + }; + let (oid, message, author) = summarize(&commit)?; + Ok(Some(( + time, + list_tags::TagInfo { + name: reference.name().shorten().to_str_lossy().as_ref().into(), + oid: oid.to_smolstr(), + message, + author, + annotation, + extra_data: Default::default(), + }, + ))) +} + +pub(crate) async fn list_tags( + State(state): State, + ExtractXrpc(args): ExtractXrpc, +) -> Result, XrpcError> { + tracing::Span::current().record("repo", args.repo.as_str()); + let repo = open(&state, &args.repo)?; + let (offset, limit) = page(args.limit, args.cursor.as_deref())?; + + crate::metrics::blocking(BlockingKind::Git, move || { + let platform = repo.references().map_err(internal)?; + let mut rows = Vec::new(); + let mut total = 0; + for reference in platform.tags().map_err(internal)? { + let reference = reference.map_err(internal)?; + total += 1; + if let Some((time, tag)) = tag_info(&repo, &reference)? { + rows.push((time, tag.name.to_string(), tag)); + } + } + let (tags, cursor) = paginate(rows, offset, limit); + Ok(list_tags::ListTagsOutput { + tags, + cursor, + total, + extra_data: Default::default(), + }) + }) + .await + .map_err(internal)? + .map(XrpcResponse) +} + +#[axum::debug_handler] +pub(crate) async fn get_ref( + State(state): State, + ExtractXrpc(args): ExtractXrpc, +) -> Result, XrpcError> { + tracing::Span::current().record("repo", args.repo.as_str()); + let repo = open(&state, &args.repo)?; + let name = args.r#ref.to_string(); + + crate::metrics::blocking(BlockingKind::Git, move || { + let not_found = |detail: &str| XrpcError::RefNotFound { + rev: name.clone(), + detail: detail.to_owned(), + }; + let mut reference = repo + .try_find_reference(name.as_str()) + .map_err(|e| not_found(&e.to_string()))? + .ok_or_else(|| not_found("the mirror holds no such reference"))?; + let (_, info) = tag_info(&repo, &reference)? + .ok_or_else(|| not_found("reference does not point at a commit"))?; + let total_commits = reference + .peel_to_commit() + .map_err(internal)? + .ancestors() + .all() + .map_err(internal)? + .try_fold(0, |count, info| info.map(|_| count + 1)) + .map_err(internal)?; + Ok(get_ref::GetRefOutput { + name: reference.name().as_bstr().to_str_lossy().as_ref().into(), + oid: info.oid, + message: info.message, + author: info.author, + annotation: info.annotation, + total_commits, + extra_data: Default::default(), + }) + }) + .await + .map_err(internal)? + .map(XrpcResponse) +} + +fn archive_etag(repo: &str, commit: ObjectId, format: &str, prefix: &str) -> String { + let digest = Sha256::digest( + [ + ARCHIVE_ETAG_DOMAIN, + repo, + &commit.to_string(), + format, + prefix, + ] + .join("\0"), + ); + let hex: String = digest.iter().map(|byte| format!("{byte:02x}")).collect(); + format!("\"{hex}\"") +} + +fn content_disposition(filename: &str) -> String { + let plain = filename + .bytes() + .all(|byte| (0x20..0x7f).contains(&byte) && byte != b'"'); + if plain { + return format!("attachment; filename=\"{filename}\""); + } + let encoded: String = filename + .bytes() + .map(|byte| match byte { + b'A'..=b'Z' + | b'a'..=b'z' + | b'0'..=b'9' + | b'!' + | b'#' + | b'$' + | b'&' + | b'+' + | b'-' + | b'.' + | b'^' + | b'_' + | b'`' + | b'|' + | b'~' => (byte as char).to_string(), + _ => format!("%{byte:02X}"), + }) + .collect(); + format!("attachment; filename*=utf-8''{encoded}") +} + +struct ChannelWriter(tokio::sync::mpsc::Sender>>); + +impl Write for ChannelWriter { + fn write(&mut self, data: &[u8]) -> std::io::Result { + self.0 + .blocking_send(Ok(data.to_vec())) + .map_err(|_| std::io::Error::from(std::io::ErrorKind::BrokenPipe))?; + Ok(data.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +pub(crate) async fn get_archive( + State(state): State, + headers: HeaderMap, + ExtractXrpc(args): ExtractXrpc, +) -> Result { + tracing::Span::current().record("repo", args.repo.as_str()); + let repo = open(&state, &args.repo)?; + let format = args.format.as_deref().unwrap_or("tar.gz").to_owned(); + if format != "tar.gz" && format != "zip" { + return Err(XrpcError::InvalidRequest(format!( + "unsupported archive format {format:?}" + ))); + } + let prefix = args.prefix.to_string(); + archive::validate_prefix(&prefix)?; + let refspec = args.r#ref.to_string(); + let limit = state.max_archive_bytes; + + let (repo, commit, tree, time) = crate::metrics::blocking(BlockingKind::Git, move || { + let commit = resolve_commit(&repo, &refspec)?; + let (tree, time) = { + let found = repo.find_commit(commit).map_err(internal)?; + let tree = found.tree_id().map_err(internal)?.detach(); + (tree, found.time().map_err(internal)?.seconds) + }; + archive::bound_source(&repo, tree, limit)?; + Ok::<_, XrpcError>((repo, commit, tree, time)) + }) + .await + .map_err(internal)??; + + let etag = archive_etag(args.repo.as_str(), commit, &format, &prefix); + if etag_matches(&headers, &etag) { + return Ok(( + StatusCode::NOT_MODIFIED, + [ + (header::ETAG, etag), + (header::CACHE_CONTROL, "no-cache".to_owned()), + ], + ) + .into_response()); + } + + let filename = format!("{}.{format}", prefix.replace(['/', '\\'], "-")); + let content_type = match format.as_str() { + "zip" => "application/zip", + _ => "application/gzip", + }; + // TODO(boltless): serve a spooled file with `tower_http::services::ServeFile` for resumable archives + let (tx, rx) = tokio::sync::mpsc::channel(8); + let errors = tx.clone(); + let tree_prefix = prefix.clone(); + tokio::spawn(async move { + let written = crate::metrics::blocking(BlockingKind::Git, move || { + let out = BufWriter::with_capacity(ARCHIVE_CHUNK, ChannelWriter(tx)); + match format.as_str() { + "zip" => archive::write_zip(&repo, tree, &tree_prefix, time, limit, out), + _ => archive::write_tar_gz(&repo, tree, &tree_prefix, time, limit, out), + } + }) + .await; + let error = match written { + Ok(Ok(())) => return, + Ok(Err(error)) => error.to_string(), + Err(error) => error.to_string(), + }; + warn!(%error, "archive stream failed"); + let _ = errors.send(Err(std::io::Error::other(error))).await; + }); + let body = Body::from_stream(futures_lite::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|chunk| (chunk, rx)) + })); + + Ok(( + StatusCode::OK, + [ + (header::CONTENT_TYPE, content_type.to_owned()), + (header::CONTENT_DISPOSITION, content_disposition(&filename)), + (header::ETAG, etag), + (header::CACHE_CONTROL, "no-cache".to_owned()), + (header::X_CONTENT_TYPE_OPTIONS, "nosniff".to_owned()), + ], + body, + ) + .into_response()) +} diff --git a/gitmirror/crates/gitmirror/src/config.rs b/gitmirror/crates/gitmirror/src/config.rs index 01c007cda..7b822794b 100644 --- a/gitmirror/crates/gitmirror/src/config.rs +++ b/gitmirror/crates/gitmirror/src/config.rs @@ -81,6 +81,9 @@ pub struct RepoConfig { /// a single request reads into memory. #[config(env = "GITMIRROR_MAX_BLOB_BYTES", default = 26_214_400)] pub max_blob_bytes: u64, + + #[config(env = "GITMIRROR_MAX_ARCHIVE_BYTES", default = 1_073_741_824)] + pub max_archive_bytes: u64, } #[derive(Debug, Config)] diff --git a/gitmirror/crates/gitmirror/src/main.rs b/gitmirror/crates/gitmirror/src/main.rs index bea4b3620..122c032a5 100644 --- a/gitmirror/crates/gitmirror/src/main.rs +++ b/gitmirror/crates/gitmirror/src/main.rs @@ -173,6 +173,7 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { .transpose()?; let max_blob_bytes = cfg.repo.max_blob_bytes; + let max_archive_bytes = cfg.repo.max_archive_bytes; let state = AppState::new( Arc::new(cfg.repo.scan_path), http, @@ -184,6 +185,7 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { Arc::new(SystemClock::new()), redis, max_blob_bytes, + max_archive_bytes, ); let app = router(state);