Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374use std::net::SocketAddr;
use clap::Parser;use tonic::transport::Server;use tonic::{Request, Response, Status};use tracing::info;
mod diff;mod protocol;
pub mod gitmirror { pub mod v1 { tonic::include_proto!("gitmirror.v1"); }}
use std::path::PathBuf;use std::pin::Pin;
use gix::bstr::ByteSlice as _;use tokio::sync::mpsc;use tokio_stream::wrappers::ReceiverStream;use tokio_stream::Stream;
use gix::revision::plumbing::Spec as RevSpec;use gix::revision::walk::Sorting;
use gitmirror::v1::git_mirror_service_server::{GitMirrorService, GitMirrorServiceServer};use gitmirror::v1::{ BlobChunk, ByteChanges, CommitLogRequest, CommitLogResponse, CommitsRequest, CommitsResponse, DiffRequest, FileContent, FileDiff, GetBlobRequest, GitCommit, GitSignature, Hunk, InterdiffRequest, LinePair,};
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;
#[derive(Parser)]#[command(name = "gitmirror", about = "Git mirror gRPC service")]struct Args { /// Address to bind the gRPC server to. #[arg(long, env = "GITMIRROR_ADDR", default_value = "127.0.0.1:9000")] addr: SocketAddr,
/// Base directory holding bare mirror repos, one per DID (<base>/<did>). #[arg(long, env = "GITMIRROR_REPO_BASE", default_value = "repos")] repo_base: PathBuf,}
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}"))) }
}
/// 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()?)}
#[tonic::async_trait]impl GitMirrorService for GitMirror { async fn commits( &self, _request: Request<CommitsRequest>, ) -> Result<Response<CommitsResponse>, Status> { Err(Status::unimplemented("not implemented")) }
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 old_oid = resolve_tree_oid(&repo, &req.base)?; let new_oid = resolve_tree_oid(&repo, &req.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 old = repo.find_tree(old_oid)?; let new = repo.find_tree(new_oid)?; for item in diff::diff(&repo, &old, &new)? { let proto = to_proto(item?); if tx.blocking_send(Ok(proto)).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> { Err(Status::unimplemented("not implemented")) }
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(); 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 tree oid.fn resolve_tree_oid(repo: &gix::Repository, spec: &str) -> Result<gix::ObjectId, Status> { let tree = repo .rev_parse_single(spec.as_bytes().as_bstr()) .map_err(|e| Status::invalid_argument(format!("bad rev '{spec}': {e}")))? .object() .map_err(|e| Status::internal(e.to_string()))? .peel_to_tree() .map_err(|e| Status::internal(e.to_string()))?; Ok(tree.id)}
fn to_proto(d: diff::Diff) -> FileDiff { FileDiff { lhs_src: Some(file_content(d.lhs_src)), rhs_src: Some(file_content(d.rhs_src)), hunks: d.hunks.into_iter().map(hunk).collect(), has_byte_changes: d.has_byte_changes.map(|(lhs, rhs)| ByteChanges { lhs: lhs as u64, rhs: rhs as u64, }), has_syntactic_changes: d.has_syntactic_changes, }}
fn file_content(f: diff::FileContent) -> FileContent { FileContent { path: f.path, oid: f.oid, size: f.size as u64, is_binary: f.is_binary, is_submodule: f.is_submodule, }}
impl TryFrom<&gix::Commit<'_>> for GitCommit { type Error = anyhow::Error;
fn try_from(commit: &gix::Commit<'_>) -> anyhow::Result<Self> { Ok(GitCommit { oid: commit.id.to_string(), author: Some(commit.author()?.try_into()?), committer: Some(commit.committer()?.try_into()?), message: commit.message_raw()?.to_vec(), parents: commit.parent_ids().map(|id| id.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, }), }) }}
fn hunk(h: diff::Hunk) -> Hunk { // Sort the novel line sets for deterministic output. let sorted = |set: rustc_hash::FxHashSet<line_numbers::LineNumber>| -> Vec<u32> { let mut v: Vec<u32> = set.into_iter().map(|n| n.0).collect(); v.sort_unstable(); v }; Hunk { novel_lhs: sorted(h.novel_lhs), novel_rhs: sorted(h.novel_rhs), lines: h .lines .into_iter() .map(|(lhs, rhs)| LinePair { lhs: lhs.map(|n| n.0), rhs: rhs.map(|n| n.0), }) .collect(), }}
#[tokio::main]async fn main() -> Result<(), Box<dyn std::error::Error>> { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()), ) .init();
let args = Args::parse(); let service = GitMirror { repo_base: args.repo_base, }; info!(addr = %args.addr, "gitmirror gRPC server listening");
Server::builder() .add_service(GitMirrorServiceServer::new(service)) .serve(args.addr) .await?;
Ok(())}