git.rs
raw
//! The git processes the site runs: `git http-backend` for clones and
//! fetches, and `git cat-file` for the raw view. libgit2 answers the
//! site's questions in-process, but it cannot stream a packed object and
//! pack generation is git's job anyway, so bytes that leave the site in
//! bulk come from git itself. Every such process occupies one of a
//! bounded number of slots, and is killed once nothing has moved between
//! it and its client for longer than the idle timeout, so that a crowd of
//! slow or silent clients cannot pin processes for good.
use std::{
io,
path::{Path, PathBuf},
process::Stdio,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::{Duration, Instant},
};
use anyhow::{Context as _, Error, anyhow};
use poem::{
Body, IntoResponse,
http::{StatusCode, header},
};
use tokio::{
io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, DuplexStream},
process::{Child, ChildStdin, ChildStdout, Command},
sync::{OwnedSemaphorePermit, Semaphore},
time::timeout,
};
// Bytes moved per read or write between a process and its client.
const CHUNK: usize = 64 * 1024;
/// Where git processes are started from: the store they may touch, and
/// the bounds every one of them runs under.
pub struct Git {
root: PathBuf,
slots: Arc<Semaphore>,
idle: Duration,
}
impl Git {
/// Runs git against the repositories under `root`, at most `processes`
/// at a time, none of them left stalled for longer than `idle`.
pub fn new(root: PathBuf, processes: u32, idle: Duration) -> Self {
Git {
root,
slots: Arc::new(Semaphore::new(processes as usize)),
idle,
}
}
/// The directory of repositories.
pub fn root(&self) -> &Path {
&self.root
}
/// Starts `git` in a free slot, waiting up to the idle timeout for
/// one. `configure` adds the arguments and environment of the
/// particular command to a process that otherwise sees nothing but
/// `PATH` and the store declared safe. `input`, when given, is fed to
/// the process as it arrives; otherwise the process has no input.
/// `what` names the command in the journal.
pub async fn start(
&self,
what: &'static str,
input: Option<Body>,
configure: impl FnOnce(&mut Command),
) -> Result<Process, StartError> {
let slot = match timeout(self.idle, self.slots.clone().acquire_owned()).await {
Ok(Ok(slot)) => slot,
Ok(Err(closed)) => return Err(StartError::Failed(closed.into())),
Err(_) => {
return Err(StartError::Busy {
retry_after: self.idle,
});
}
};
let mut command = Command::new("git");
command
.env_clear()
.env("PATH", std::env::var_os("PATH").unwrap_or_default())
// The repositories belong to the hosting user, not to the one
// serving them, and git refuses to touch a repository owned by
// someone else unless it is declared safe. The configured
// store is trusted by definition.
.env("GIT_CONFIG_COUNT", "1")
.env("GIT_CONFIG_KEY_0", "safe.directory")
.env("GIT_CONFIG_VALUE_0", format!("{}/*", self.root.display()))
.stdin(if input.is_some() {
Stdio::piped()
} else {
Stdio::null()
})
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.kill_on_drop(true);
configure(&mut command);
let mut child = command
.spawn()
.with_context(|| format!("starting git {what}"))?;
let stdout = child
.stdout
.take()
.ok_or_else(|| anyhow!("no output pipe for git {what}"))?;
let activity = Arc::new(Activity::new());
if let Some(body) = input {
let stdin = child
.stdin
.take()
.ok_or_else(|| anyhow!("no input pipe for git {what}"))?;
tokio::spawn(feed(body, stdin, activity.clone(), self.idle, what));
}
Ok(Process {
child,
output: BufReader::with_capacity(CHUNK, stdout),
activity,
_slot: slot,
idle: self.idle,
what,
})
}
}
/// Why a git process could not be started.
#[derive(Debug)]
pub enum StartError {
/// Every slot stayed taken for the whole idle timeout.
Busy {
retry_after: Duration,
},
Failed(Error),
}
impl From<Error> for StartError {
fn from(err: Error) -> Self {
StartError::Failed(err)
}
}
// Being refused for lack of a slot is an answer to the client, not a
// failure of the site, so it travels as a complete response: a plain 503
// naming the wait the client should observe.
impl From<StartError> for poem::Error {
fn from(err: StartError) -> Self {
match err {
StartError::Busy { retry_after } => poem::Error::from_response(
"The site is serving as many clones and downloads as it can at once. Try again shortly.\n"
.with_status(StatusCode::SERVICE_UNAVAILABLE)
.with_header(header::RETRY_AFTER, retry_after.as_secs().max(1).to_string())
.into_response(),
),
StartError::Failed(err) => err.into(),
}
}
}
/// A running git process in its slot. Dropping it kills the process and
/// frees the slot.
pub struct Process {
child: Child,
output: BufReader<ChildStdout>,
activity: Arc<Activity>,
_slot: OwnedSemaphorePermit,
idle: Duration,
what: &'static str,
}
impl Process {
/// The next line of output without its line ending, or `None` once
/// the output is over.
pub async fn read_line(&mut self) -> Result<Option<String>, Error> {
let mut line = String::new();
let read = within(self.idle, self.what, self.output.read_line(&mut line)).await?;
if read == 0 {
return Ok(None);
}
line.truncate(line.trim_end_matches(['\r', '\n']).len());
Ok(Some(line))
}
/// The output produced so far, left in place for whatever reads next;
/// empty once the output is over.
pub async fn peek(&mut self) -> Result<&[u8], Error> {
within(self.idle, self.what, self.output.fill_buf()).await
}
/// The rest of the output, as a response body. The process is reaped
/// once its output is over, and killed if it or the client stalls.
pub fn into_body(self) -> Body {
// A pipe whose ends notice each other going away: the body sees
// the end of the output once the pump drops its end, and the pump
// fails to write once the body has been dropped.
let (ours, theirs) = tokio::io::duplex(2 * CHUNK);
tokio::spawn(self.pump(ours));
Body::from_async_read(theirs)
}
// Moves output to the client until it is over, then sees the process
// out.
async fn pump(mut self, mut sink: DuplexStream) {
let outcome = self.copy_output(&mut sink).await;
drop(sink);
let stall = match outcome {
Ok(()) => return self.finish().await,
Err(stall) => stall,
};
let seconds = self.idle.as_secs();
match stall {
Stall::Silent => tracing::warn!(
command = self.what,
"nothing moved for {seconds}s; killing the process"
),
Stall::ClientNotReading => tracing::info!(
command = self.what,
"client read nothing for {seconds}s; killing the process"
),
Stall::ClientGone => {
tracing::debug!(command = self.what, "client went away; killing the process")
}
Stall::ReadFailed(err) => tracing::warn!(
command = self.what,
error = %err,
"reading git output failed; killing the process"
),
}
self.kill().await;
}
async fn copy_output(&mut self, sink: &mut DuplexStream) -> Result<(), Stall> {
let mut buf = vec![0u8; CHUNK];
loop {
let read = loop {
match timeout(self.idle, self.output.read(&mut buf)).await {
Ok(Ok(read)) => break read,
Ok(Err(err)) => return Err(Stall::ReadFailed(err)),
// Silence on the output is fine while input is still
// arriving: the process is working on it.
Err(_) if self.activity.idle_for() < self.idle => {}
Err(_) => return Err(Stall::Silent),
}
};
if read == 0 {
return Ok(());
}
match timeout(self.idle, sink.write_all(&buf[..read])).await {
Ok(Ok(())) => self.activity.touch(),
Ok(Err(_)) => return Err(Stall::ClientGone),
Err(_) => return Err(Stall::ClientNotReading),
}
}
}
// Reaps a process whose output is over. One that keeps running
// regardless is killed.
async fn finish(mut self) {
let waited = timeout(self.idle, self.child.wait()).await;
match waited {
Ok(Ok(status)) if status.success() => {}
Ok(Ok(status)) => tracing::warn!(command = self.what, %status, "git failed"),
Ok(Err(err)) => tracing::warn!(command = self.what, error = %err, "waiting for git"),
Err(_) => {
tracing::warn!(
command = self.what,
"git kept running for {}s after its output ended; killing it",
self.idle.as_secs()
);
self.kill().await;
}
}
}
async fn kill(mut self) {
if let Err(err) = self.child.kill().await {
tracing::debug!(command = self.what, error = %err, "killing git");
}
}
}
// How the output pump stopped short.
enum Stall {
Silent,
ClientNotReading,
ClientGone,
ReadFailed(io::Error),
}
// Waits for an operation on the process's output, for at most the idle
// timeout.
async fn within<T>(
idle: Duration,
what: &str,
operation: impl Future<Output = io::Result<T>>,
) -> Result<T, Error> {
match timeout(idle, operation).await {
Ok(result) => Ok(result?),
Err(_) => Err(anyhow!(
"git {what} produced no output for {}s",
idle.as_secs()
)),
}
}
// Feeds a request body to a process as it arrives, closing the process's
// input once the body is over or the client has sent nothing for the idle
// timeout. A process whose input ends early fails on its own, and the
// output side sees it out.
async fn feed(
body: Body,
mut stdin: ChildStdin,
activity: Arc<Activity>,
idle: Duration,
what: &'static str,
) {
let mut source = body.into_async_read();
let mut buf = vec![0u8; CHUNK];
loop {
let read = match timeout(idle, source.read(&mut buf)).await {
Ok(Ok(0)) => return,
Ok(Ok(read)) => read,
Ok(Err(err)) => {
tracing::debug!(command = what, error = %err, "request body ended early");
return;
}
Err(_) => {
tracing::info!(
command = what,
"client sent no input for {}s; closing the process's input",
idle.as_secs()
);
return;
}
};
activity.touch();
match timeout(idle, stdin.write_all(&buf[..read])).await {
Ok(Ok(())) => {}
// The process is gone; the output side reports why.
Ok(Err(_)) => return,
Err(_) => {
tracing::warn!(
command = what,
"git read no input for {}s; closing its input",
idle.as_secs()
);
return;
}
}
}
}
// When bytes last moved in either direction, shared by both sides of a
// process.
struct Activity {
started: Instant,
last_millis: AtomicU64,
}
impl Activity {
fn new() -> Self {
Activity {
started: Instant::now(),
last_millis: AtomicU64::new(0),
}
}
fn touch(&self) {
self.last_millis
.store(self.started.elapsed().as_millis() as u64, Ordering::Relaxed);
}
fn idle_for(&self) -> Duration {
let last = Duration::from_millis(self.last_millis.load(Ordering::Relaxed));
self.started.elapsed().saturating_sub(last)
}
}