Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363use 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 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,};
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,}
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}"))) }}
#[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)))) }}
/// 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)}
/// 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(())}