Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381use std::path::Path;use std::sync::atomic::AtomicBool;
use gitmirror_db::RepoRow;use gitmirror_git::layout::Layout;use gitmirror_types::jacquard_ext::DidDocumentExt as _;use gitmirror_types::{KnotHost, KnotPolicy};use gix::refs::transaction::{Change, LogChange, PreviousValue, RefEdit, RefLog};use gix::refs::{FullName, Target};use gix_hash::ObjectId;use jacquard_common::types::did::Did;use jacquard_identity::resolver::IdentityResolver as _;use thiserror::Error;use tracing::debug;
use crate::admission::BackfillAdmission;use crate::Directory;
const AGENT: &str = concat!("gitmirror/", env!("CARGO_PKG_VERSION"));
#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum TaskDisposition { Finished { default_branch_commit: Option<ObjectId>, }, Deferred,}
#[derive(Debug, Error)]pub enum BackfillError { #[error("resolve {0}: {1}")] Resolve(Did, String), #[error("{0} declares no knot service")] NoKnotEndpoint(Did), #[error("{0} declares an unusable knot endpoint: {1}")] UnsafeKnotEndpoint(Did, &'static str), #[error("malformed refs record: {0}")] MalformedRefs(String), #[error("git: {0}")] Git(#[from] BackfillGitError), #[error("fetch task panicked")] Panicked,}
#[derive(Debug, Error)]pub enum BackfillGitError { #[error("init repo: {0}")] Init(#[from] gix::init::Error), #[error("connect host: {0}")] Connect(#[from] gix_transport::client::blocking_io::connect::Error), #[error("handshake host: {0}")] Handshake(#[from] gix_protocol::handshake::Error), #[error("fetch repo: {0}")] Fetch(#[from] gix_protocol::fetch::Error), #[error("edit references: {0}")] RefsEdit(#[from] gix::reference::edit::Error), #[error("unsupported object format: {0}")] ObjectFormat(String), #[error("object format mismatch: local {local}, knot {remote}")] ObjectFormatMismatch { local: gix_hash::Kind, remote: gix_hash::Kind, },}
/// syncs git repositorypub async fn did_task( _http: &reqwest::Client, directory: &Directory, layout: &Layout, policy: KnotPolicy, repo: &RepoRow, admission: &BackfillAdmission, _permit: tokio::sync::OwnedSemaphorePermit,) -> Result<TaskDisposition, BackfillError> { // git-fetch is usually done in three steps: // 1. handshake (host capability check) // 2. advertisement (ls-refs) // 3. git-upload-pack // // The handshake result will be cached per-host, and we already know the refs list (in db). // So we can go straight to git-upload-pack call.
let knot = knot_host(directory, &repo.did, policy).await?; let knot = knot.url(); tracing::Span::current().record("knot", knot.as_str());
let Some(_knot_permit) = admission.try_acquire(knot).await else { return Ok(TaskDisposition::Deferred); };
let default_branch = FullName::try_from(format!("refs/heads/{}", repo.refs.default_branch)) .map_err(|e| BackfillError::MalformedRefs(e.to_string()))?;
let url = format!("{knot}{}", repo.did.as_str()); let path = layout.repo_path(&repo.did); let refs = repo.refs.refs.clone(); let default_branch_commit = tokio::task::spawn_blocking(move || fetch_into(&url, &path, &refs, default_branch)) .await .map_err(|_| BackfillError::Panicked)??;
Ok(TaskDisposition::Finished { default_branch_commit, })}
/// The repo's own knot, from its DID document. Re-resolved every task: a repo can migrate.async fn knot_host( directory: &Directory, did: &Did, policy: KnotPolicy,) -> Result<KnotHost, BackfillError> { let document = directory .resolve_did_doc_owned(did) .await .map_err(|e| BackfillError::Resolve(did.clone(), e.to_string()))?;
let endpoint = document .knot_endpoint() // TODO(post-1.0): remove legacy `#atproto_pds` fallback .or_else(|| document.pds_endpoint()) .ok_or_else(|| BackfillError::NoKnotEndpoint(did.clone()))?;
let host = policy .safe_knot_host(endpoint.as_ref()) .map_err(|_reason| { BackfillError::UnsafeKnotEndpoint(did.clone(), "is invalid knot service endpoint") })?;
Ok(host)}
// TODO: move this to gitmirror-db & gitmirror-ingest maybe, so we can verify on ingestfn parse_refs( blob: &[u8], kind: gix_hash::Kind,) -> Result<Vec<(ObjectId, FullName)>, BackfillError> { let malformed = |e: &dyn std::fmt::Display| BackfillError::MalformedRefs(e.to_string()); gix::refs::packed::Iter::new(blob, kind) .map_err(|e| malformed(&e))? .map(|entry| { let entry = entry.map_err(|e| malformed(&e))?; Ok((entry.target(), entry.name.to_owned())) }) .collect()}
fn fetch_into( url: &str, path: &Path, refs: &[u8], default_branch: FullName,) -> Result<Option<ObjectId>, BackfillError> { let mut transport = gix_transport::client::blocking_io::connect::connect( url, gix_transport::client::blocking_io::connect::Options { version: gix_transport::Protocol::V2, ssh: Default::default(), trace: false, }, ) .map_err(BackfillGitError::from)?;
let mut progress = gix::progress::Discard; let should_interrupt = AtomicBool::new(false);
// TODO: cache per knot. We ignore `handshake.refs` which will be None for v2 protocol anyhow let mut handshake = gix_protocol::handshake( &mut transport, gix_transport::Service::UploadPack, // no auth |_| Ok(None), Default::default(), &mut progress, ) .map_err(BackfillGitError::from)?;
let object_hash = match handshake .capabilities .capability("object-format") .and_then(|c| c.value()) { Some(format) => format .to_string() .parse() .map_err(BackfillGitError::ObjectFormat)?, None => gix_hash::Kind::Sha1, };
let refs = parse_refs(refs, object_hash)?; let default_branch_commit = refs .iter() .find(|(_, name)| *name == default_branch) .map(|(oid, _)| *oid);
let repo = if path.join("HEAD").exists() { gix::open(path).map_err(gix::init::Error::Open) } else { gix::ThreadSafeRepository::init( path, gix::create::Kind::Bare, gix::create::Options { object_hash: Some(object_hash), ..Default::default() }, ) .map(Into::into) } .map_err(BackfillGitError::from)?; if repo.object_hash() != object_hash { return Err(BackfillGitError::ObjectFormatMismatch { local: repo.object_hash(), remote: object_hash, } .into()); }
let pack_dir = repo.path().join("objects").join("pack"); let odb = repo.objects.clone();
let cache = repo.commit_graph_if_enabled().ok().flatten(); let mut negotiate = StoredRefs { repo: &repo, wants: &refs, negotiator: gix_negotiate::Algorithm::Consecutive.into_negotiator(), graph: repo.revision_graph(cache.as_ref()), };
gix_protocol::fetch( &mut negotiate, |pack, progress, should_interrupt| { gix_pack::Bundle::write_to_directory( pack, Some(&pack_dir), progress, should_interrupt, Some(odb.clone()), gix_pack::bundle::write::Options { object_hash, ..Default::default() }, ) .map(|_| true) }, &mut progress, &should_interrupt, gix_protocol::fetch::Context { handshake: &mut handshake, transport: &mut transport, user_agent: ("agent", Some(AGENT.into())), trace_packetlines: false, }, gix_protocol::fetch::Options { shallow_file: path.join("shallow"), shallow: &gix_protocol::fetch::Shallow::NoChange, tags: gix_protocol::fetch::Tags::None, reject_shallow_remote: true, }, ) .map_err(BackfillGitError::from)?;
// TODO(boltless): dump packed-refs instead let ref_edit = |name: FullName, new: Target| RefEdit { change: Change::Update { log: LogChange { mode: RefLog::AndReference, force_create_reflog: false, message: "gitmirror: sync".into(), }, expected: PreviousValue::Any, new, }, name, deref: false, }; let edits = refs .iter() .map(|(oid, name)| ref_edit(name.clone(), Target::Object(*oid))) .chain([ref_edit( "HEAD".try_into().expect("valid ref name"), Target::Symbolic(default_branch), )]); repo.edit_references(edits) .map_err(BackfillGitError::from)?;
debug!(refs = refs.len(), "fetched and updated refs"); Ok(default_branch_commit)}
struct StoredRefs<'a, 'cache> { repo: &'a gix::Repository, wants: &'a [(ObjectId, FullName)], negotiator: Box<dyn gix_negotiate::Negotiator>, graph: gix_negotiate::Graph<'a, 'cache>,}
impl gix_protocol::fetch::Negotiate for StoredRefs<'_, '_> { fn mark_complete_and_common_ref( &mut self, ) -> Result<gix_protocol::fetch::negotiate::Action, gix_protocol::fetch::negotiate::Error> { use gix::refs::file::ReferenceExt as _; use gix_object::Exists as _; use gix_protocol::fetch::negotiate::Action;
let (repo, wants) = (self.repo, self.wants); if wants.is_empty() { return Ok(Action::NoChange); }
let known: Vec<bool> = wants .iter() .map(|(oid, _)| repo.objects.exists(oid)) .collect(); if known.iter().all(|k| *k) { return Ok(Action::SkipToRefUpdate); }
for (oid, _) in wants.iter().zip(&known).filter_map(|(w, k)| k.then_some(w)) { // TODO(boltless): load known commit to find max commit_time (used for cutoff_date) self.negotiator.known_common(*oid, &mut self.graph)?; }
// TODO(boltless): walk parents back like // [`gix_protocol::fetch::negotiate::mark_recent_complete_commits`] does. let store = &repo.refs; let packed = store.cached_packed_buffer()?; let mut tips = Vec::new(); for local_ref in store.iter()?.all()? { let mut local_ref = local_ref?; let id = local_ref.peel_to_id_packed( store, &repo.objects, packed.as_ref().map(|b| &***b), )?; tips.push(id); } for tip in tips { self.negotiator.add_tip(tip, &mut self.graph)?; }
Ok(Action::MustNegotiate { remote_ref_target_known: known, }) }
fn add_wants( &mut self, arguments: &mut gix_protocol::fetch::Arguments, remote_ref_target_known: &[bool], ) -> bool { let mut has_want = false; for (oid, _) in self .wants .iter() .zip(remote_ref_target_known) .filter_map(|(want, known)| (!known).then_some(want)) { arguments.want(oid); has_want = true; } has_want }
fn one_round( &mut self, state: &mut gix_protocol::fetch::negotiate::one_round::State, arguments: &mut gix_protocol::fetch::Arguments, previous_response: Option<&gix_protocol::fetch::Response>, ) -> Result<(gix_protocol::fetch::negotiate::Round, bool), gix_protocol::fetch::negotiate::Error> { gix_protocol::fetch::negotiate::one_round( &mut *self.negotiator, &mut self.graph, state, arguments, previous_response, ) }}