Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
19 kB · 489 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490use std::net::SocketAddr;use std::path::PathBuf;use std::pin::Pin;
use gix::bstr::ByteSlice as _;use gix::revision::plumbing::Spec as RevSpec;use gix::revision::walk::Sorting;use tempfile::TempDir;use tokio::sync::mpsc;use tokio_stream::Stream;use tokio_stream::wrappers::ReceiverStream;use tonic::transport::Server;use tonic::{Request, Response, Status};use tracing::{debug, info};
use crate::diff::{self, prepare_interdiff};use crate::protocol::v1::diff_request::ComparisonType;use crate::protocol::v1::git_mirror_service_server::{GitMirrorService, GitMirrorServiceServer};use crate::protocol::v1::{ BlobChunk, CommitLogRequest, CommitLogResponse, DiffRequest, FileDiff, GetBlobRequest, GitCommit, GitSignature, InterdiffRequest, MergeCheckRequest, MergeCheckResponse,};
type FileDiffStream = Pin<Box<dyn Stream<Item = Result<FileDiff, Status>> + Send>>;type BlobChunkStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;type CommitLogResponseStream = Pin<Box<dyn Stream<Item = Result<CommitLogResponse, Status>> + Send>>;
/// Blob bytes are streamed in chunks of this size.const BLOB_CHUNK_SIZE: usize = 64 * 1024;/// GitCommits are streamed in batches of this size.const COMMIT_LOG_BATCH_SIZE: usize = 128;
struct GitMirror { repo_base: PathBuf,}
/// A [`gix::Repository`] backed by a temporary directory that is removed on drop. Derefs to/// `gix::Repository`, so it is used just like one; keeping it alive keeps the scratch dir alive.struct TempRepository { repo: gix::Repository, // Declared AFTER `repo` so `repo` drops first: any file handles into the scratch dir close // before the dir itself is removed (Rust drops struct fields in declaration order). _dir: TempDir,}
impl TempRepository { fn new(repo: gix::Repository, dir: TempDir) -> Self { Self { repo, _dir: dir } }}
impl std::ops::Deref for TempRepository { type Target = gix::Repository; fn deref(&self) -> &gix::Repository { &self.repo }}
impl GitMirror { fn open_repo(&self, did: &str) -> Result<gix::Repository, Status> { if !did.starts_with("did:") { return Err(Status::invalid_argument("repo must be a DID")); } gix::open(self.repo_base.join(did)) .map_err(|e| Status::not_found(format!("repo not found: {e}"))) }
/// Build a throwaway bare repo whose `objects/info/alternates` points read-only at both the /// `target` and `source` mirrors, so a single `gix::Repository` can see objects from both /// without ever mutating either mirror. Opened with `.with_object_memory()` so any objects /// the merge writes stay in RAM. Returns a self-cleaning [`TempRepository`]: keep it alive for /// the whole merge; dropping it removes the scratch dir. fn open_merge_scratch( &self, target_did: &str, source_did: &str, ) -> Result<TempRepository, Status> { let object_dir = |did: &str| -> Result<PathBuf, Status> { if !did.starts_with("did:") { return Err(Status::invalid_argument("repo must be a DID")); } std::fs::canonicalize(self.repo_base.join(did).join("objects")) .map_err(|_| Status::not_found(format!("repo not found: {did}"))) }; let target_objects = object_dir(target_did)?; let source_objects = object_dir(source_did)?;
let scratch = tempfile::tempdir().map_err(|e| Status::internal(e.to_string()))?; gix::init_bare(scratch.path()).map_err(|e| Status::internal(e.to_string()))?;
let info_dir = scratch.path().join("objects").join("info"); std::fs::create_dir_all(&info_dir).map_err(|e| Status::internal(e.to_string()))?; let alternates = { let mut alternates = vec![target_objects.display().to_string()]; if source_objects != target_objects { alternates.push(source_objects.display().to_string()); } format!("{}\n", alternates.join("\n")) }; std::fs::write(info_dir.join("alternates"), alternates) .map_err(|e| Status::internal(e.to_string()))?;
let repo = gix::open(scratch.path()) .map_err(|e| Status::internal(e.to_string()))? .with_object_memory(); Ok(TempRepository::new(repo, scratch)) }}
#[tonic::async_trait]impl GitMirrorService for GitMirror { type DiffStream = FileDiffStream;
async fn diff( &self, request: Request<DiffRequest>, ) -> Result<Response<Self::DiffStream>, Status> { let req = request.into_inner(); let repo = self.open_repo(&req.repo)?; // Resolve refs up-front so bad input surfaces as the RPC status, then move the // owned repo + oids into a blocking task for the (synchronous) diff loop. let base_oid = req .base_rev_spec .as_deref() .map(|b| resolve_commit_oid(&repo, b)) .transpose()?; let head_oid = resolve_commit_oid(&repo, &req.head_rev_spec)?; let comparison_type = ComparisonType::try_from(req.comparison_type).unwrap_or(ComparisonType::Unspecified); let safe = repo.into_sync();
let (tx, rx) = mpsc::channel::<Result<FileDiff, Status>>(16); tokio::task::spawn_blocking(move || { let repo = safe.to_thread_local();
let run = || -> anyhow::Result<()> { let head_commit = repo.find_commit(head_oid)?; let old = match base_oid { Some(base_oid) => match comparison_type { ComparisonType::OnlyInHead => { repo.find_tree(repo.find_commit(base_oid)?.tree_id()?)? } ComparisonType::Unspecified | ComparisonType::Intersection => { let merge_base = repo.merge_base(base_oid, head_oid)?.detach(); repo.find_tree(repo.find_commit(merge_base)?.tree_id()?)? } }, None => match head_commit.parent_ids().next() { Some(parent_id) => { repo.find_tree(repo.find_commit(parent_id.detach())?.tree_id()?)? } None => repo.empty_tree(), }, }; let new = repo.find_tree(head_commit.tree_id()?)?; for item in diff::diff(&repo, &old, &new, false)? { if tx.blocking_send(Ok(item?.into())).is_err() { break; // client hung up } } Ok(()) }; if let Err(e) = run() { let _ = tx.blocking_send(Err(Status::internal(e.to_string()))); } });
Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) }
type InterdiffStream = FileDiffStream;
async fn interdiff( &self, request: Request<InterdiffRequest>, ) -> Result<Response<Self::InterdiffStream>, Status> { let req = request.into_inner(); let repo = self.open_repo(&req.repo)?; let from_base_id = resolve_commit_oid(&repo, &req.from_base)?; let from_head_id = resolve_commit_oid(&repo, &req.from_head)?; let to_base_id = resolve_commit_oid(&repo, &req.to_base)?; let to_head_id = resolve_commit_oid(&repo, &req.to_head)?; let safe = repo.into_sync();
let (tx, rx) = mpsc::channel::<Result<FileDiff, Status>>(16); tokio::task::spawn_blocking(move || { let repo = safe.to_thread_local(); let run = || -> anyhow::Result<()> { let to_head = repo.find_commit(to_head_id)?; let to_head_tree = to_head.tree()?; let rebased_tree = prepare_interdiff(&repo, (from_base_id, from_head_id), to_base_id)?; for item in diff::diff(&repo, &rebased_tree, &to_head_tree, true)? { if tx.blocking_send(Ok(item?.into())).is_err() { break; // client hung up } } Ok(()) }; if let Err(e) = run() { let _ = tx.blocking_send(Err(Status::internal(e.to_string()))); } });
Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) }
type GetBlobStream = BlobChunkStream;
async fn get_blob( &self, request: Request<GetBlobRequest>, ) -> Result<Response<Self::GetBlobStream>, Status> { let req = request.into_inner(); let repo = self.open_repo(&req.repo)?; let oid = gix::ObjectId::from_hex(req.oid.as_bytes()) .map_err(|e| Status::invalid_argument(format!("bad oid '{}': {e}", req.oid)))?; let safe = repo.into_sync();
let (tx, rx) = mpsc::channel::<Result<BlobChunk, Status>>(16); tokio::task::spawn_blocking(move || { let repo = safe.to_thread_local(); let run = || -> anyhow::Result<()> { let blob = repo.find_object(oid)?.try_into_blob()?; for chunk in blob.data.chunks(BLOB_CHUNK_SIZE) { let msg = BlobChunk { data: chunk.to_vec(), }; if tx.blocking_send(Ok(msg)).is_err() { break; // client hung up } } Ok(()) }; if let Err(e) = run() { let _ = tx.blocking_send(Err(Status::internal(e.to_string()))); } });
Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) }
type CommitLogStream = CommitLogResponseStream;
async fn commit_log( &self, request: Request<CommitLogRequest>, ) -> Result<Response<Self::CommitLogStream>, Status> { let req = request.into_inner(); debug!( repo = %req.repo, all_refs = req.all_refs, ranges = ?req.ranges, after = ?req.after, before = ?req.before, max_commits = req.max_commits, skip = req.skip, "commit_log request", ); if req.all_refs != req.ranges.is_empty() { return Err(Status::invalid_argument( "specify exactly one of `ranges` or `all_refs`", )); } let repo = self.open_repo(&req.repo)?; let safe = repo.into_sync();
let (tx, rx) = mpsc::channel::<Result<CommitLogResponse, Status>>(16); tokio::task::spawn_blocking(move || { let repo = safe.to_thread_local(); let run = || -> anyhow::Result<()> { let after = req.after.map(|t| t.seconds); let before = req.before.map(|t| t.seconds); let max_commits = req.max_commits as usize; let mut skip = req.skip as usize; let mut sent = 0usize; let mut batch = Vec::with_capacity(COMMIT_LOG_BATCH_SIZE);
for info in commit_log_walk(&repo, &req)? { let info = info?; let commit_time = info.commit_time(); if before.is_some_and(|b| commit_time > b) { continue; } if after.is_some_and(|a| commit_time < a) { break; // newest-first order: nothing older will match either } if skip > 0 { skip -= 1; continue; } if max_commits != 0 && sent >= max_commits { break; } batch.push(GitCommit::try_from(&info.object()?)?); sent += 1; if batch.len() == COMMIT_LOG_BATCH_SIZE { let msg = CommitLogResponse { commits: std::mem::take(&mut batch), }; if tx.blocking_send(Ok(msg)).is_err() { return Ok(()); // client hung up } } } if !batch.is_empty() { let _ = tx.blocking_send(Ok(CommitLogResponse { commits: batch })); } Ok(()) }; if let Err(e) = run() { let _ = tx.blocking_send(Err(Status::internal(e.to_string()))); } });
Ok(Response::new(Box::pin(ReceiverStream::new(rx)))) }
async fn merge_check( &self, request: Request<MergeCheckRequest>, ) -> Result<Response<MergeCheckResponse>, Status> { let req = request.into_inner(); let target = req .target .ok_or_else(|| Status::invalid_argument("target is required"))?; let source = req .source .ok_or_else(|| Status::invalid_argument("source is required"))?; debug!( target_repo = %target.repo, target_commit = %target.commit.as_bstr(), source_repo = %source.repo, source_commit = %source.commit.as_bstr(), "merge_check request", );
let scratch = self.open_merge_scratch(&target.repo, &source.repo)?;
let response = tokio::task::spawn_blocking(move || -> Result<MergeCheckResponse, Status> { let target_oid = find_commit_by_sha(&scratch, &target.commit)?; let source_oid = find_commit_by_sha(&scratch, &source.commit)?; crate::merge::merge_check(&scratch, target_oid, source_oid) .map_err(|e| Status::internal(e.to_string())) }) .await .map_err(|e| Status::internal(format!("merge task panicked: {e}")))??;
Ok(Response::new(response)) }}
/// Resolve a commit-ish ref string (oid, short oid, branch/tag) to its commit oid.fn resolve_commit_oid(repo: &gix::Repository, spec: &[u8]) -> Result<gix::ObjectId, Status> { let commit = repo .rev_parse_single(spec.as_bstr()) .map_err(|e| { Status::invalid_argument(format!("bad rev '{}': {e}", String::from_utf8_lossy(spec))) })? .object() .map_err(|e| Status::internal(e.to_string()))? .peel_to_commit() .map_err(|e| Status::internal(e.to_string()))?; Ok(commit.id)}
/// Parse a full commit object id from its hex bytes and confirm it exists as a commit. Unlike/// [`resolve_commit_oid`], this accepts only a concrete oid — never a revspec/branch/tag.fn find_commit_by_sha(repo: &gix::Repository, sha: &[u8]) -> Result<gix::ObjectId, Status> { let oid = gix::ObjectId::from_hex(sha).map_err(|e| { Status::invalid_argument(format!("bad commit sha '{}': {e}", sha.as_bstr())) })?; repo.find_commit(oid) .map_err(|_| Status::not_found(format!("commit not found: {oid}")))?; Ok(oid)}
/// Resolve tips/hidden commits from a [`CommitLogRequest`] and return the resulting gix walk,/// sorted newest-first to match `git log`'s default order.fn commit_log_walk<'repo>( repo: &'repo gix::Repository, req: &CommitLogRequest,) -> anyhow::Result<gix::revision::Walk<'repo>> { let mut tips = Vec::new(); let mut hidden = Vec::new();
if req.all_refs { for r in repo.references()?.all()? { let mut r = r.map_err(|e| anyhow::anyhow!(e))?; if let Ok(commit) = r.peel_to_commit() { tips.push(commit.id); } } } else { for range in &req.ranges { let revspec = repo.rev_parse(range.as_bstr())?; let spec = revspec.detach(); match spec { RevSpec::Include(id) => tips.push(id), RevSpec::Range { from, to } => { tips.push(to); hidden.push(from); } _ => { anyhow::bail!("The spec isn't currently supported: {spec:?}") } } } }
Ok(repo .rev_walk(tips) .sorting(Sorting::ByCommitTime(Default::default())) .with_hidden(hidden) .all()?)}
impl TryFrom<&gix::Commit<'_>> for GitCommit { type Error = anyhow::Error;
fn try_from(commit: &gix::Commit<'_>) -> anyhow::Result<Self> { let decoded = commit.decode()?; Ok(GitCommit { oid: commit.id.to_string(), author: Some(decoded.author()?.try_into()?), committer: Some(decoded.committer()?.try_into()?), message: decoded.message.to_vec(), parents: decoded .parents .into_iter() .map(|id| id.to_string()) .collect(), extra_headers: decoded .extra_headers .iter() .map(|(k, v)| (k.to_string(), v.to_string())) .collect(), }) }}
impl TryFrom<gix::actor::SignatureRef<'_>> for GitSignature { type Error = anyhow::Error;
fn try_from(sig: gix::actor::SignatureRef<'_>) -> anyhow::Result<Self> { let time = sig.time()?; Ok(GitSignature { name: sig.name.to_vec(), email: sig.email.to_vec(), date: Some(prost_types::Timestamp { seconds: time.seconds, nanos: 0, }), }) }}
/// Build the gRPC service and serve it until the process is stopped.pub async fn serve(addr: SocketAddr, repo_base: PathBuf) -> anyhow::Result<()> { let service = GitMirror { repo_base }; info!(addr = %addr, "gitmirror gRPC server listening");
Server::builder() .add_service(GitMirrorServiceServer::new(service)) .serve(addr) .await?;
Ok(())}
#[cfg(test)]mod tests { use super::*;
#[test] fn temp_repository_removes_scratch_dir_on_drop() { let dir = tempfile::tempdir().unwrap(); gix::init_bare(dir.path()).unwrap(); let repo = gix::open(dir.path()).unwrap().with_object_memory(); let temp = TempRepository::new(repo, dir); let path = temp.git_dir().to_path_buf(); assert!(path.exists()); // Derefs to gix::Repository. assert!(temp.object_hash() == gix::hash::Kind::Sha1); drop(temp); assert!(!path.exists(), "scratch dir should be gone after drop"); }}