194c44c
use std::path::Path;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use futures::StreamExt;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::process::Command;
use tokio::sync::{Mutex, oneshot};
use tokio_util::io::ReaderStream;
use crate::smart::error::{SmartError, SmartResult};
pub type RpcStream = futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>>;
const MAX_STDERR_BYTES: usize = 64 * 1024 * 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GitService {
UploadPack,
ReceivePack,
}
impl GitService {
pub fn from_name(name: &str) -> SmartResult<Self> {
match name {
"git-upload-pack" => Ok(Self::UploadPack),
"git-receive-pack" => Ok(Self::ReceivePack),
_ => Err(SmartError::BadRequest(format!(
"unsupported git service: {name}"
))),
}
}
pub fn name(self) -> &'static str {
match self {
Self::UploadPack => "upload-pack",
Self::ReceivePack => "receive-pack",
}
}
pub fn dashed_name(self) -> &'static str {
match self {
Self::UploadPack => "git-upload-pack",
Self::ReceivePack => "git-receive-pack",
}
}
pub fn advertisement_content_type(self) -> &'static str {
match self {
Self::UploadPack => "application/x-git-upload-pack-advertisement",
Self::ReceivePack => "application/x-git-receive-pack-advertisement",
}
}
pub fn result_content_type(self) -> &'static str {
match self {
Self::UploadPack => "application/x-git-upload-pack-result",
Self::ReceivePack => "application/x-git-receive-pack-result",
}
}
}
fn build_command(bare_path: &Path, service: GitService, advertise: bool) -> Command {
let mut cmd = Command::new("git");
cmd.arg(service.name()).arg("--stateless-rpc");
if advertise {
cmd.arg("--advertise-refs");
}
cmd.arg(bare_path)
.env("GIT_DIR", bare_path)
.env("GIT_PROTOCOL", "version=2")
.env("GIT_TERMINAL_PROMPT", "0")
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("GIT_NO_REPLACE_OBJECTS", "1")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
cmd
}
#[allow(clippy::type_complexity)]
fn spawn_with_supervisor(
bare_path: &Path,
service: GitService,
advertise: bool,
input: Bytes,
timeout_secs: u64,
) -> SmartResult<(
RpcStream,
Option<tokio::sync::oneshot::Receiver<SmartResult<()>>>,
)> {
let mut cmd = build_command(bare_path, service, advertise);
let mut child = cmd.spawn().map_err(|e| {
SmartError::Internal(format!("failed to spawn git {}: {e}", service.name()))
})?;
tracing::debug!(
svc = service.name(),
advertise,
bare = %bare_path.display(),
input_bytes = input.len(),
timeout_secs,
pid = ?child.id(),
"spawned git rpc process"
);
let mut stdin = child
.stdin
.take()
.ok_or_else(|| SmartError::Internal("stdin not piped".to_string()))?;
let stdout = child
.stdout
.take()
.ok_or_else(|| SmartError::Internal("stdout not piped".to_string()))?;
let mut stderr = child
.stderr
.take()
.ok_or_else(|| SmartError::Internal("stderr not piped".to_string()))?;
let child_arc = Arc::new(Mutex::new(Some(child)));
{
let child_arc = child_arc.clone();
tokio::spawn(async move {
if stdin.write_all(&input).await.is_err() {
tracing::warn!(
svc = service.name(),
"writing rpc input to git stdin failed; killing process"
);
if let Some(mut ch) = child_arc.lock().await.take() {
let _ = ch.start_kill();
let _ = ch.wait().await;
}
}
let _ = stdin.shutdown().await;
});
}
let (tx, rx) = oneshot::channel();
{
let child_arc = child_arc.clone();
tokio::spawn(async move {
let status = {
let mut guard = child_arc.lock().await;
match guard.as_mut() {
None => None,
Some(ch) => {
match tokio::time::timeout(Duration::from_secs(timeout_secs), ch.wait())
.await
{
Ok(Ok(s)) => Some(s),
_ => {
tracing::warn!(
svc = service.name(),
timeout_secs,
"git rpc process timed out; killing"
);
let _ = ch.start_kill();
let _ = ch.wait().await;
None
}
}
}
}
};
let mut buf = vec![0u8; 8192];
let mut total = 0usize;
let mut text = Vec::new();
loop {
match stderr.read(&mut buf).await {
Ok(0) | Err(_) => break,
Ok(n) => {
total += n;
if total <= MAX_STDERR_BYTES {
text.extend_from_slice(&buf[..n]);
}
}
}
}
let verdict = match status {
Some(s) if s.success() => {
tracing::debug!(
svc = service.name(),
stderr_bytes = total,
"git rpc process exited successfully"
);
Ok(())
}
Some(s) => {
tracing::warn!(
svc = service.name(),
exit = %s,
stderr = %String::from_utf8_lossy(text.as_slice()).trim(),
"git rpc process exited with error"
);
Err(SmartError::Internal(format!(
"git {} exited with {s}: {}",
service.name(),
String::from_utf8_lossy(text.as_slice()).trim()
)))
}
None => Err(SmartError::Internal(format!(
"git {} timed out after {timeout_secs}s",
service.name()
))),
};
let _ = tx.send(verdict);
});
}
let stream: RpcStream = Box::pin(ReaderStream::new(stdout));
Ok((stream, Some(rx)))
}
pub async fn run_rpc_stream(
bare_path: &Path,
service: GitService,
advertise: bool,
input: Bytes,
timeout_secs: u64,
) -> SmartResult<RpcStream> {
let (stream, _rx) = spawn_with_supervisor(bare_path, service, advertise, input, timeout_secs)?;
Ok(stream)
}
pub async fn run_rpc_bytes(
bare_path: &Path,
service: GitService,
advertise: bool,
input: Bytes,
timeout_secs: u64,
) -> SmartResult<Bytes> {
let (mut stream, rx) =
spawn_with_supervisor(bare_path, service, advertise, input, timeout_secs)?;
let mut out: Vec<u8> = Vec::new();
while let Some(chunk) = stream.next().await {
out.extend_from_slice(&chunk.map_err(|e| {
SmartError::Internal(format!("git {} stdout read failed: {e}", service.name()))
})?);
}
if let Some(rx) = rx {
let _ = rx.await;
}
tracing::debug!(
svc = service.name(),
output_bytes = out.len(),
"git rpc output collected"
);
Ok(Bytes::from(out))
}