~//! `txz`, or `zip`) for the given ref. `tar.xz` and `txz` are produced by~//! piping `git archive --format=tar` through an external `xz` process.~-use std::io::Read;-use std::process::{Command, Stdio};+use std::process::Stdio;~+use axum::body::Body;~use axum::extract::{Path, State};~use axum::http::HeaderValue;~use axum::http::header;~use axum::response::IntoResponse;~use serde::Deserialize;-use tokio::task::spawn_blocking;+use tokio::io::AsyncRead;+use tokio::process::{Child, Command};+use tokio_util::io::ReaderStream;~~use crate::error::{Error, Result};~use crate::git;~ }~}~-/// Runs `git archive` and returns (stdout, success, stderr).-fn run_archive(repo_path: &str, git_format: &str, ref_: &str) -> (Vec<u8>, bool, String) {- let output = Command::new("git")+/// Spawns `git archive` and streams its stdout directly into the response body.+fn stream_archive(repo_path: &str, git_format: &str, ref_: &str) -> Result<Body> {+ let mut child = Command::new("git")~ .arg("-C")~ .arg(repo_path)~ .arg("archive")~ .arg(format!("--format={git_format}"))~ .arg(ref_)- .output();-- match output {- Ok(o) => (- o.stdout,- o.status.success(),- String::from_utf8_lossy(&o.stderr).into_owned(),- ),- Err(e) => (Vec::new(), false, e.to_string()),- }+ .stdout(Stdio::piped())+ .stderr(Stdio::piped())+ .spawn()+ .map_err(Error::Io)?;++ let stdout = child.stdout.take().expect("stdout configured");+ let stderr = child.stderr.take().expect("stderr configured");++ tokio::spawn(reaper(child, stderr, "git archive"));++ let stream = ReaderStream::new(stdout);+ Ok(Body::from_stream(stream))~}~-/// Pipes `git archive --format=tar` through `xz`, returning (stdout, success, stderr).-fn run_archive_xz(repo_path: &str, ref_: &str) -> (Vec<u8>, bool, String) {- let mut git = match Command::new("git")+/// Pipes `git archive --format=tar` through `xz` and streams the compressed+/// output.+///+/// Tokio's `ChildStdout` does not implement `Into<Stdio>`, so we cannot use+/// the direct `.stdin(git_stdout)` approach that `std::process::Command`+/// supports. Instead we relay data between the two processes in a background+/// task via `tokio::io::copy`.+fn stream_archive_xz(repo_path: &str, ref_: &str) -> Result<Body> {+ let mut git = Command::new("git")~ .arg("-C")~ .arg(repo_path)~ .arg("archive")~ .stdout(Stdio::piped())~ .stderr(Stdio::piped())~ .spawn()- {- Ok(c) => c,- Err(e) => return (Vec::new(), false, format!("spawn git: {e}")),- };+ .map_err(Error::Io)?;~~ let git_stdout = git.stdout.take().expect("stdout configured");- let xz = match Command::new("xz")- .stdin(git_stdout)+ let git_stderr = git.stderr.take().expect("stderr configured");++ let mut xz = Command::new("xz")+ .arg("-c")+ .stdin(Stdio::piped())~ .stdout(Stdio::piped())~ .stderr(Stdio::piped())~ .spawn()- {- Ok(c) => c,- Err(e) => return (Vec::new(), false, format!("spawn xz: {e}")),- };-- // Wait for xz to finish (consumes all piped git output).- let xz_output = match xz.wait_with_output() {- Ok(o) => o,- Err(e) => return (Vec::new(), false, format!("xz wait: {e}")),- };-- // Check git's exit status. Report git errors first since xz failures- // are often a downstream consequence of git producing no output.- let git_status = match git.wait() {- Ok(s) => s,- Err(e) => return (Vec::new(), false, format!("git wait: {e}")),- };-- if !git_status.success() {- let msg = git- .stderr- .take()- .and_then(|mut s| {- let mut buf = String::new();- s.read_to_string(&mut buf).ok()?;- Some(buf)- })- .unwrap_or_default();- let msg = if msg.is_empty() {- "git archive failed".into()- } else {- msg- };- return (Vec::new(), false, msg);- }+ .map_err(Error::Io)?;++ let mut xz_stdin = xz.stdin.take().expect("stdin configured");+ let xz_stdout = xz.stdout.take().expect("stdout configured");+ let xz_stderr = xz.stderr.take().expect("stderr configured");++ // Relay git's stdout to xz's stdin in a background task.+ tokio::spawn(async move {+ use tokio::io::AsyncWriteExt;+ if let Err(e) =+ tokio::io::copy(&mut tokio::io::BufReader::new(git_stdout), &mut xz_stdin).await+ {+ eprintln!("pipe git->xz: {e}");+ }+ let _ = xz_stdin.shutdown().await;+ // Reap git (its stdout is fully consumed now).+ reaper(git, git_stderr, "git archive").await;+ });++ tokio::spawn(reaper(xz, xz_stderr, "xz compression"));++ let stream = ReaderStream::new(xz_stdout);+ Ok(Body::from_stream(stream))+}~- if !xz_output.status.success() {- let msg = String::from_utf8_lossy(&xz_output.stderr);- let msg = if msg.is_empty() {- "xz compression failed".into()- } else {- msg.into_owned()- };- return (Vec::new(), false, msg);+/// Background task that reads stderr and waits for the child process to exit.+/// Errors are printed to stderr — by this point we have already started+/// streaming the response and cannot change the HTTP status.+async fn reaper(mut child: Child, mut stderr: impl AsyncRead + Unpin, label: &'static str) {+ use tokio::io::AsyncReadExt;+ let mut buf = String::new();+ let _ = stderr.read_to_string(&mut buf).await;+ let ok = child.wait().await.is_ok_and(|s| s.success());+ if !ok || !buf.is_empty() {+ eprintln!("{label} failed: {buf}");~ }-- (xz_output.stdout, true, String::new())~}~~/// Streams a repository archive for the given ref in the requested format.~ let git_repo = git::open_repo(&state.root, &name)?;~ if git_repo.rev_parse_single(params.ref_.as_str()).is_err() {~ return Err(Error::NotFound(format!("ref not found: {}", params.ref_)));- }-- let repo_path_str = repo_path.to_string_lossy().into_owned();- let archive_ref = params.ref_.clone();- let ref_for_closure = archive_ref.clone();-- let (stdout, success, stderr) = spawn_blocking(move || {- if cfg.uses_xz {- run_archive_xz(&repo_path_str, &ref_for_closure)- } else {- run_archive(&repo_path_str, cfg.git_format, &ref_for_closure)- }- })- .await- .map_err(|e| Error::Io(std::io::Error::other(e)))?;-- if !success {- return Err(Error::GitCorrupt(format!("archive failed: {stderr}")));~ }~~ // Strip refs/heads/ or refs/tags/ prefix for a cleaner filename.- let short_ref = archive_ref+ let short_ref = params+ .ref_~ .strip_prefix("refs/heads/")- .or_else(|| archive_ref.strip_prefix("refs/tags/"))- .unwrap_or(&archive_ref);+ .or_else(|| params.ref_.strip_prefix("refs/tags/"))+ .unwrap_or(¶ms.ref_);~ let filename = format!("{name}-{short_ref}.{}", cfg.extension);~ let disposition = HeaderValue::from_str(&format!("attachment; filename=\"{filename}\""))~ .map_err(|_| Error::BadRequest("invalid header value".into()))?;~~ let content_type = HeaderValue::from_static(cfg.content_type);++ let repo_path = repo_path.to_string_lossy().into_owned();+ let archive_ref = params.ref_.clone();++ let body = if cfg.uses_xz {+ stream_archive_xz(&repo_path, &archive_ref)?+ } else {+ stream_archive(&repo_path, cfg.git_format, &archive_ref)?+ };~~ Ok((~ [~ (header::CONTENT_TYPE, content_type),~ (header::CONTENT_DISPOSITION, disposition),~ ],- stdout,+ body,~ ))~}