351 lines
12 KiB
Rust
351 lines
12 KiB
Rust
use std::io;
|
|
use std::io::{Read, Write};
|
|
use std::process::{Command, ExitStatus, Stdio};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use radicle_fetch::{ByteSlice as _, RemoteProgress};
|
|
|
|
use radicle::identity::RepoId;
|
|
use radicle::node::events;
|
|
use radicle::node::events::Emitter;
|
|
use radicle::node::{Event, NodeId};
|
|
use radicle::storage::git::paths;
|
|
use radicle::Storage;
|
|
|
|
use crate::runtime::thread;
|
|
|
|
/// Perform the Git upload-pack process, given that the Git request
|
|
/// `header` has already been read and parsed.
|
|
///
|
|
/// N.b. The upload-pack process itself is strict, i.e. it will read
|
|
/// requests from the client indefinitely, and so the client side MUST
|
|
/// send the EOF file message.
|
|
pub fn upload_pack<R, W>(
|
|
nid: &NodeId,
|
|
remote: NodeId,
|
|
storage: &Storage,
|
|
emitter: &Emitter<Event>,
|
|
header: &pktline::GitRequest,
|
|
mut recv: R,
|
|
send: W,
|
|
timeout: Duration,
|
|
) -> io::Result<ExitStatus>
|
|
where
|
|
R: io::Read + Send,
|
|
W: io::Write + Send,
|
|
{
|
|
let timer = Instant::now();
|
|
let protocol_version = header
|
|
.extra
|
|
.iter()
|
|
.find_map(|kv| match kv {
|
|
(ref k, Some(v)) if k == "version" => {
|
|
let version = match v.as_str() {
|
|
"2" => 2,
|
|
"1" => 1,
|
|
_ => 0,
|
|
};
|
|
Some(version)
|
|
}
|
|
_ => None,
|
|
})
|
|
.unwrap_or(0);
|
|
|
|
if protocol_version != 2 {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"only Git protocol version 2 is supported",
|
|
));
|
|
}
|
|
|
|
let git_dir = paths::repository(storage, &header.repo);
|
|
let mut child = {
|
|
let mut cmd = Command::new("git");
|
|
cmd.current_dir(git_dir)
|
|
.env_clear()
|
|
.envs(std::env::vars().filter(|(key, _)| key == "PATH" || key.starts_with("GIT_TRACE")))
|
|
.env("GIT_PROTOCOL", format!("version={protocol_version}"))
|
|
.args([
|
|
"-c",
|
|
"uploadpack.allowAnySha1InWant=true",
|
|
"-c",
|
|
"uploadpack.allowRefInWant=true",
|
|
"-c",
|
|
"lsrefs.unborn=ignore",
|
|
"upload-pack",
|
|
"--strict",
|
|
format!("--timeout={}", timeout.as_secs()).as_str(),
|
|
".",
|
|
])
|
|
.stdout(Stdio::piped())
|
|
.stdin(Stdio::piped())
|
|
.stderr(Stdio::inherit());
|
|
|
|
cmd.spawn()?
|
|
};
|
|
|
|
let mut stdin = child.stdin.take().unwrap();
|
|
let mut stdout = io::BufReader::new(child.stdout.take().unwrap());
|
|
let reporter = std::sync::Mutex::new(Reporter::new(header.repo, remote, emitter.clone(), send));
|
|
|
|
thread::scope(|s| {
|
|
thread::spawn_scoped(nid, "upload-pack", s, || {
|
|
let mut buffer = [0; u16::MAX as usize + 1];
|
|
loop {
|
|
match stdout.read(&mut buffer) {
|
|
Ok(0) => break,
|
|
Ok(n) => {
|
|
let mut lock = reporter.lock().expect("FATAL: upload_pack poisoned lock");
|
|
if let Err(e) = lock.write_all(&buffer[..n]) {
|
|
log::warn!(target: "worker", "Error reading stdout to upload-pack reporter: {e}");
|
|
emitter.emit(events::UploadPack::error(header.repo, remote, e).into());
|
|
break;
|
|
}
|
|
drop(lock);
|
|
}
|
|
Err(e) => {
|
|
log::debug!(target: "worker", "Exiting upload-pack writer thread for {}: {e}", header.repo);
|
|
emitter.emit(events::UploadPack::error(header.repo, remote, e).into());
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
let reader = thread::spawn_scoped(nid, "upload-pack", s, || {
|
|
let mut buffer = [0; u16::MAX as usize + 1];
|
|
loop {
|
|
match recv.read(&mut buffer) {
|
|
Ok(0) => break,
|
|
Ok(n) => {
|
|
if let Err(e) = stdin.write_all(&buffer[..n]) {
|
|
log::warn!(target: "worker", "Error writing to upload-pack stdin: {e}");
|
|
break;
|
|
}
|
|
}
|
|
Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
|
|
log::debug!(target: "worker", "Exiting upload-pack reader thread for {}", header.repo);
|
|
break;
|
|
}
|
|
Err(e) if e.kind() == io::ErrorKind::TimedOut => {
|
|
log::warn!(target: "worker", "Read channel timed out for upload-pack {}", header.repo);
|
|
// N.b. if the read timed out, ensure that the sender isn't
|
|
// still sending messages.
|
|
let lock = reporter.lock().expect("FATAL: upload_pack poisoned lock");
|
|
if lock.is_timeout(timeout) {
|
|
break;
|
|
}
|
|
}
|
|
Err(e) => {
|
|
log::error!(target: "worker", "Error on upload-pack channel read for {}: {e}", header.repo);
|
|
emitter.emit(events::UploadPack::error(header.repo, remote, e).into());
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// N.b. we only care if the `reader` is finished. We then kill
|
|
// the child which will end the thread for the sender.
|
|
if let Err(e) = reader.join() {
|
|
log::warn!(target: "worker", "Upload pack thread panicked: {e:?}");
|
|
}
|
|
child.kill()?;
|
|
Ok::<_, io::Error>(())
|
|
})?;
|
|
|
|
let status = child.wait()?;
|
|
emitter.emit(events::UploadPack::done(header.repo, remote, status).into());
|
|
log::debug!(target: "worker", "Upload pack finished ({}ms)", timer.elapsed().as_millis());
|
|
Ok(status)
|
|
}
|
|
|
|
/// A combination of the upload-pack sender with an [`Emitter`] for reporting
|
|
/// the progress events to subscribers.
|
|
struct Reporter<W> {
|
|
rid: RepoId,
|
|
remote: NodeId,
|
|
emitter: Emitter<Event>,
|
|
send: W,
|
|
total: usize,
|
|
last_sent: Instant,
|
|
}
|
|
|
|
impl<W> Reporter<W> {
|
|
fn new(rid: RepoId, remote: NodeId, emitter: Emitter<Event>, send: W) -> Self {
|
|
Self {
|
|
rid,
|
|
remote,
|
|
emitter,
|
|
send,
|
|
total: 0,
|
|
last_sent: Instant::now(),
|
|
}
|
|
}
|
|
|
|
fn is_timeout(&self, timeout: Duration) -> bool {
|
|
Instant::now().duration_since(self.last_sent) > timeout
|
|
}
|
|
|
|
fn emit(&mut self, buf: &[u8]) {
|
|
let event = match Self::as_upload_pack_progress(buf) {
|
|
Some(progress) => events::UploadPack::write(self.rid, self.remote, progress),
|
|
None => {
|
|
self.total += buf.len();
|
|
events::UploadPack::pack_progress(self.rid, self.remote, self.total)
|
|
}
|
|
};
|
|
log::trace!(target: "worker", "upload-pack progress: {event:?}");
|
|
self.emitter.emit(event.into());
|
|
}
|
|
|
|
fn as_upload_pack_progress(buf: &[u8]) -> Option<events::upload_pack::Progress> {
|
|
use events::upload_pack::Progress::*;
|
|
let RemoteProgress {
|
|
action, step, max, ..
|
|
} = RemoteProgress::from_bytes(buf)?;
|
|
if action.contains_str("Counting objects") {
|
|
step.and_then(|processed| max.map(|total| Counting { processed, total }))
|
|
} else if action.contains_str("Compressing objects") {
|
|
step.and_then(|processed| max.map(|total| Compressing { processed, total }))
|
|
} else if action.contains_str("Enumerating objects") {
|
|
max.map(|total| Enumerating { total })
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<W> io::Write for Reporter<W>
|
|
where
|
|
W: io::Write,
|
|
{
|
|
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
|
let n = self.send.write(buf)?;
|
|
self.emit(buf);
|
|
self.last_sent = Instant::now();
|
|
Ok(n)
|
|
}
|
|
|
|
fn flush(&mut self) -> io::Result<()> {
|
|
self.send.flush()
|
|
}
|
|
}
|
|
|
|
pub(super) mod pktline {
|
|
use std::io;
|
|
use std::io::Read;
|
|
use std::str;
|
|
|
|
use radicle::prelude::RepoId;
|
|
|
|
pub const HEADER_LEN: usize = 4;
|
|
|
|
/// Read and parse the `GitRequest` data from the client side.
|
|
pub fn git_request<R>(reader: &mut R) -> io::Result<GitRequest>
|
|
where
|
|
R: io::Read,
|
|
{
|
|
let mut reader = Reader::new(reader);
|
|
let (header, _) = reader.read_request_pktline()?;
|
|
Ok(header)
|
|
}
|
|
|
|
struct Reader<'a, R> {
|
|
stream: &'a mut R,
|
|
}
|
|
|
|
impl<'a, R: io::Read> Reader<'a, R> {
|
|
/// Create a new packet-line reader.
|
|
pub fn new(stream: &'a mut R) -> Self {
|
|
Self { stream }
|
|
}
|
|
|
|
/// Parse a Git request packet-line.
|
|
///
|
|
/// Example: `0032git-upload-pack /project.git\0host=myserver.com\0`
|
|
///
|
|
fn read_request_pktline(&mut self) -> io::Result<(GitRequest, Vec<u8>)> {
|
|
let mut pktline = [0u8; 1024];
|
|
let length = self.read_pktline(&mut pktline)?;
|
|
let Some(cmd) = GitRequest::parse(&pktline[4..length]) else {
|
|
return Err(io::ErrorKind::InvalidInput.into());
|
|
};
|
|
Ok((cmd, Vec::from(&pktline[..length])))
|
|
}
|
|
|
|
/// Parse a Git packet-line.
|
|
fn read_pktline(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
|
self.read_exact(&mut buf[..HEADER_LEN])?;
|
|
|
|
let length = str::from_utf8(&buf[..HEADER_LEN])
|
|
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;
|
|
let length = usize::from_str_radix(length, 16)
|
|
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;
|
|
|
|
self.read_exact(&mut buf[HEADER_LEN..length])?;
|
|
|
|
Ok(length)
|
|
}
|
|
}
|
|
|
|
impl<R: io::Read> io::Read for Reader<'_, R> {
|
|
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
|
self.stream.read(buf)
|
|
}
|
|
}
|
|
|
|
/// The Git request packet-line for a Heartwood repository.
|
|
///
|
|
/// Example: `0032git-upload-pack /rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5.git\0host=myserver.com\0`
|
|
#[derive(Debug)]
|
|
pub struct GitRequest {
|
|
pub repo: RepoId,
|
|
#[allow(dead_code)]
|
|
pub path: String,
|
|
#[allow(dead_code)]
|
|
pub host: Option<(String, Option<u16>)>,
|
|
pub extra: Vec<(String, Option<String>)>,
|
|
}
|
|
|
|
impl GitRequest {
|
|
/// Parse a Git command from a packet-line.
|
|
fn parse(input: &[u8]) -> Option<Self> {
|
|
let input = str::from_utf8(input).ok()?;
|
|
let mut parts = input
|
|
.strip_prefix("git-upload-pack ")?
|
|
.split_terminator('\0');
|
|
|
|
let path = parts.next()?.to_owned();
|
|
let repo = path.strip_prefix('/')?.parse().ok()?;
|
|
let host = match parts.next() {
|
|
None | Some("") => None,
|
|
Some(host) => {
|
|
let host = host.strip_prefix("host=")?;
|
|
match host.split_once(':') {
|
|
None => Some((host.to_owned(), None)),
|
|
Some((host, port)) => {
|
|
let port = port.parse::<u16>().ok()?;
|
|
Some((host.to_owned(), Some(port)))
|
|
}
|
|
}
|
|
}
|
|
};
|
|
let extra = parts
|
|
.skip_while(|part| part.is_empty())
|
|
.map(|part| match part.split_once('=') {
|
|
None => (part.to_owned(), None),
|
|
Some((k, v)) => (k.to_owned(), Some(v.to_owned())),
|
|
})
|
|
.collect();
|
|
|
|
Some(Self {
|
|
repo,
|
|
path,
|
|
host,
|
|
extra,
|
|
})
|
|
}
|
|
}
|
|
}
|