mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fetch: integrate the latest `gix-protocol` into `radicle-fetch` 

Signed-off-by: Sebastian Thiel <sebastian.thiel@icloud.com>
This commit is contained in:
Sebastian Thiel 2024-12-17 13:55:18 +01:00 committed by Lorenz Leutgeb
parent 96637aca7e
commit 72913b8c15
No known key found for this signature in database
6 changed files with 671 additions and 524 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

731
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

1
deny.toml
View File

@ -47,6 +47,7 @@ version = 2
allow = [
"MIT",
"Apache-2.0",
"Unlicense",
"Unicode-DFS-2016",
"Unicode-3.0",
"BSD-3-Clause",

13
radicle-fetch/Cargo.toml
View File

@ -10,13 +10,12 @@ edition = "2021"
[dependencies]
bstr = { version = "1.3" }
either = { version = "1.9.0" }
gix-actor = { version = "0.31.1" }
gix-features = { version = "0.38.1", features = ["progress"] }
gix-hash = { version = "0.14.2" }
gix-odb = { version = "0.61.0" }
gix-pack = { version = "0.51.0" }
gix-protocol = { version = "0.45", features = ["blocking-client"] }
gix-transport = { version = "0.42", features = ["blocking-client"] }
gix-features = { version = "0.39.1", features = ["progress"] }
gix-hash = { version = "0.15.1" }
gix-odb = { version = "0.66.0" }
gix-pack = { version = "0.56.0" }
gix-protocol = { version = "0.47.0", features = ["blocking-client"] }
gix-transport = { version = "0.44.0", features = ["blocking-client"] }
log = { version = "0.4.17", features = ["std"] }
nonempty = { version = "0.9.0" }
radicle-git-ext = { version = "0.8.0", features = ["bstr"] }

27
radicle-fetch/src/transport.rs
View File

@ -10,9 +10,7 @@ use std::sync::Arc;
use bstr::BString;
use gix_features::progress::prodash::progress;
use gix_protocol::handshake;
use gix_protocol::FetchConnection;
use gix_transport::client;
use gix_transport::client::TransportWithoutIO as _;
use gix_transport::Protocol;
use gix_transport::Service;
use radicle::git::Oid;
@ -85,7 +83,7 @@ where
log::trace!(target: "fetch", "Performing handshake for {}", self.repo);
let (read, write) = self.stream.open().map_err(io_other)?;
gix_protocol::fetch::handshake(
&mut Connection::new(read, write, FetchConnection::AllowReuse, self.repo.clone()),
&mut Connection::new(read, write, self.repo.clone()),
|_| Ok(None),
vec![],
&mut progress::Discard,
@ -105,11 +103,10 @@ where
ls_refs::run(
ls_refs::Config {
prefixes,
extra_params: vec![],
repo: self.repo.clone(),
},
handshake,
Connection::new(read, write, FetchConnection::AllowReuse, self.repo.clone()),
Connection::new(read, write, self.repo.clone()),
&mut progress::Discard,
)
.map_err(io_other)
@ -137,7 +134,7 @@ where
interrupt,
},
handshake,
Connection::new(read, write, FetchConnection::AllowReuse, self.repo.clone()),
Connection::new(read, write, self.repo.clone()),
&mut progress::Discard,
)
.map_err(io_other)?
@ -184,7 +181,6 @@ where
pub(crate) struct Connection<R, W> {
inner: client::git::Connection<R, W>,
mode: FetchConnection,
}
impl<R, W> Connection<R, W>
@ -192,7 +188,7 @@ where
R: io::Read,
W: io::Write,
{
pub fn new(read: R, write: W, mode: FetchConnection, repo: BString) -> Self {
pub fn new(read: R, write: W, repo: BString) -> Self {
Self {
inner: client::git::Connection::new(
read,
@ -203,7 +199,6 @@ where
client::git::ConnectMode::Daemon,
false,
),
mode,
}
}
}
@ -256,20 +251,6 @@ where
}
}
fn indicate_end_of_interaction<R, W>(transport: &mut Connection<R, W>) -> Result<(), client::Error>
where
R: io::Read,
W: io::Write,
{
// An empty request marks the (early) end of the interaction. Only relevant in stateful transports though.
if transport.connection_persists_across_multiple_requests() {
transport
.request(client::WriteMode::Binary, client::MessageKind::Flush, false)?
.into_read()?;
}
Ok(())
}
fn io_other(err: impl std::error::Error + Send + Sync + 'static) -> io::Error {
io::Error::new(io::ErrorKind::Other, err)
}

276
radicle-fetch/src/transport/fetch.rs
View File

@ -1,27 +1,19 @@
use std::borrow::Cow;
use std::io;
use std::io::BufRead;
use std::path::PathBuf;
use std::sync::{atomic::AtomicBool, Arc};
use gix_features::progress::NestedProgress;
use gix_features::progress::{DynNestedProgress, NestedProgress};
use gix_pack as pack;
use gix_protocol::fetch;
use gix_protocol::fetch::{Delegate, DelegateBlocking};
use gix_protocol::fetch::negotiate::one_round::State;
use gix_protocol::handshake;
use gix_protocol::handshake::Ref;
use gix_protocol::ls_refs;
use gix_protocol::FetchConnection;
use gix_transport::bstr::BString;
use gix_transport::client;
use gix_transport::client::{ExtendedBufRead, MessageKind};
use gix_transport::Protocol;
use crate::git::packfile;
use crate::git::{oid, packfile};
use super::{agent_name, indicate_end_of_interaction, Connection, WantsHaves};
use super::{agent_name, Connection, WantsHaves};
pub type Error = gix_protocol::fetch::Error;
pub type Error = fetch::Error;
pub mod error {
use std::io;
@ -50,15 +42,11 @@ pub struct PackWriter {
impl PackWriter {
/// Write the packfile read from `pack` to the `objects/pack`
/// directory.
pub fn write_pack<P>(
pub fn write_pack(
&self,
mut pack: impl BufRead,
mut progress: P,
) -> Result<pack::bundle::write::Outcome, error::PackWriter>
where
P: NestedProgress,
P::SubProgress: 'static,
{
pack: &mut dyn std::io::BufRead,
progress: &mut dyn DynNestedProgress,
) -> Result<pack::bundle::write::Outcome, error::PackWriter> {
let options = pack::bundle::write::Options {
// N.b. use all cores. Can make configurable if needed
// later.
@ -80,9 +68,9 @@ impl PackWriter {
)?);
let thickener = thickener.to_handle_arc();
Ok(pack::Bundle::write_to_directory(
&mut pack,
pack,
Some(&self.git_dir.join("objects").join("pack")),
&mut progress,
progress,
&self.interrupt,
Some(thickener),
options,
@ -92,10 +80,8 @@ impl PackWriter {
/// The fetch [`Delegate`] that negotiates the fetch with the
/// server-side.
pub struct Fetch {
pub struct Negotiate {
wants_haves: WantsHaves,
pack_writer: PackWriter,
out: FetchOut,
}
/// The result of running a fetch via [`run`].
@ -108,74 +94,50 @@ pub struct FetchOut {
// FIXME: the delegate pattern will be removed in the near future and
// we should look at the fetch code being used in gix to see how we
// can migrate to the proper form of fetching.
impl Delegate for &mut Fetch {
fn receive_pack(
impl fetch::Negotiate for Negotiate {
fn mark_complete_and_common_ref(
&mut self,
input: impl io::BufRead,
progress: impl NestedProgress + 'static,
_refs: &[handshake::Ref],
previous_response: &fetch::Response,
) -> io::Result<()> {
self.out
.refs
.extend(previous_response.wanted_refs().iter().map(
|fetch::response::WantedRef { id, path }| Ref::Direct {
full_ref_name: path.clone(),
object: *id,
},
));
let pack = self
.pack_writer
.write_pack(input, progress)
.map_err(|err| io::Error::new(io::ErrorKind::Other, err))?;
self.out.keepfile = pack.keep_path.as_ref().and_then(packfile::Keepfile::new);
self.out.pack = Some(pack);
Ok(())
}
) -> Result<fetch::negotiate::Action, fetch::negotiate::Error> {
Ok(fetch::negotiate::Action::MustNegotiate {
remote_ref_target_known: vec![],
})
}
impl DelegateBlocking for &mut Fetch {
fn negotiate(
fn add_wants(
&mut self,
_refs: &[handshake::Ref],
arguments: &mut fetch::Arguments,
_previous_response: Option<&fetch::Response>,
) -> io::Result<fetch::Action> {
use crate::git::oid;
_remote_ref_target_known: &[bool],
) -> bool {
let mut has_want = false;
for oid in &self.wants_haves.wants {
arguments.want(oid::to_object_id(*oid));
has_want = true;
}
has_want
}
/// We don't actually negotiate, just provides all our haves and wants, while telling the
/// server to make the best of it and just send a pack.
/// Real Git negotiation can be done with calls to [`fetch::negotiate::one_round()`], but that
/// requires a [`fetch::RefMap`] which can be instantiated with refspecs.
fn one_round(
&mut self,
_state: &mut State,
arguments: &mut fetch::Arguments,
_previous_response: Option<&fetch::Response>,
) -> Result<(fetch::negotiate::Round, bool), fetch::negotiate::Error> {
for oid in &self.wants_haves.haves {
arguments.have(oid::to_object_id(*oid));
}
// N.b. sends `done` packet
Ok(fetch::Action::Cancel)
}
fn prepare_ls_refs(
&mut self,
_server: &client::Capabilities,
_arguments: &mut Vec<BString>,
_features: &mut Vec<(&str, Option<Cow<'_, str>>)>,
) -> io::Result<ls_refs::Action> {
// N.b. we performed ls-refs before the fetch already.
Ok(ls_refs::Action::Skip)
}
fn prepare_fetch(
&mut self,
_version: Protocol,
_server: &client::Capabilities,
_features: &mut Vec<(&str, Option<Cow<'_, str>>)>,
_refs: &[handshake::Ref],
) -> io::Result<fetch::Action> {
if self.wants_haves.wants.is_empty() {
return Err(io::Error::new(io::ErrorKind::InvalidData, "empty fetch"));
}
Ok(fetch::Action::Continue)
let round = fetch::negotiate::Round {
haves_sent: self.wants_haves.haves.len(),
in_vain: 0,
haves_to_send: 0,
previous_response_had_at_least_one_in_common: false,
};
let is_done = true;
Ok((round, is_done))
}
}
@ -200,115 +162,59 @@ where
{
log::trace!(target: "fetch", "Performing fetch");
let mut delegate = Fetch {
wants_haves,
pack_writer,
out: FetchOut {
if wants_haves.wants.is_empty() {
return Err(Error::ReadRemainingBytes(io::Error::new(
io::ErrorKind::InvalidData,
"empty fetch",
)));
}
let mut out = FetchOut {
refs: Vec::new(),
pack: None,
keepfile: None,
},
};
let mut negotiate = Negotiate { wants_haves };
let agent = agent_name().map_err(Error::ReadRemainingBytes)?;
let handshake::Outcome {
server_protocol_version: protocol,
refs: _refs,
capabilities,
} = handshake;
let agent = agent_name()?;
let fetch = gix_protocol::Command::Fetch;
let mut pack_out = None;
let mut handshake = handshake.clone();
let fetch_out = gix_protocol::fetch(
&mut negotiate,
|read_pack, progress, _should_interrupt| -> Result<_, error::PackWriter> {
let res = pack_writer.write_pack(read_pack, progress)?;
pack_out = Some(res);
Ok(true)
},
progress,
&pack_writer.interrupt,
fetch::Context {
handshake: &mut handshake,
transport: &mut conn,
user_agent: ("agent", Some(agent.into())),
trace_packetlines: false,
},
fetch::Options {
shallow_file: "no shallow file required as we reject shallow remotes (and we aren't shallow ourselves)".into(),
reject_shallow_remote: true,
shallow: &fetch::Shallow::NoChange,
tags: fetch::Tags::None,
},
)?.expect("we always get a pack");
let mut features = fetch.default_features(*protocol, capabilities);
match (&mut delegate).prepare_fetch(*protocol, capabilities, &mut features, &[]) {
Ok(fetch::Action::Continue) => {
// FIXME: this is a private function in gitoxide
// fetch.validate_argument_prefixes_or_panic(protocol, &capabilities, &[], &features)
}
// N.b. we always return Action::Continue
Ok(fetch::Action::Cancel) => unreachable!(),
Err(err) => {
indicate_end_of_interaction(&mut conn)?;
return Err(err.into());
}
}
out.refs
.extend(fetch_out.last_response.wanted_refs().iter().map(
|fetch::response::WantedRef { id, path }| Ref::Direct {
full_ref_name: path.clone(),
object: *id,
},
));
let pack_out = pack_out.expect("we always get a pack");
out.keepfile = pack_out
.keep_path
.as_ref()
.and_then(packfile::Keepfile::new);
out.pack = Some(pack_out);
gix_protocol::fetch::Response::check_required_features(*protocol, &features)?;
let sideband_all = features.iter().any(|(n, _)| *n == "sideband-all");
features.push(("agent", Some(Cow::Owned(agent))));
let mut args = fetch::Arguments::new(*protocol, features, false);
let mut previous_response = None::<fetch::Response>;
let mut round = 1;
'negotiation: loop {
progress.step();
progress.set_name(format!("negotiate (round {round})"));
round += 1;
let action = (&mut delegate).negotiate(&[], &mut args, previous_response.as_ref())?;
let mut reader = args.send(&mut conn, action == fetch::Action::Cancel)?;
if sideband_all {
setup_remote_progress(progress, &mut reader);
}
let response = fetch::Response::from_line_reader(*protocol, &mut reader, true, false)?;
previous_response = if response.has_pack() {
progress.step();
if !sideband_all {
setup_remote_progress(progress, &mut reader);
}
let timer = std::time::Instant::now();
// TODO: remove delegate in favor of functional style to fix progress-hack,
// needed as it needs `'static`. As the top-level seems to pass `Discard`,
// there should be no repercussions right now.
(&mut delegate).receive_pack(
&mut reader,
progress.add_child("receiving pack"),
&[],
&response,
)?;
log::trace!(target: "fetch", "Received pack ({}ms)", timer.elapsed().as_millis());
assert_eq!(
reader.stopped_at(),
None,
"packs are read without 'overshooting', hence it never encountered EOF"
);
// Consume anything that might still be left on the wire - this is 'EOF' most of the time,
// but some tests have 'garbage' here as well.
std::io::copy(&mut reader, &mut std::io::sink())?;
assert_eq!(
reader.stopped_at(),
Some(MessageKind::Flush),
"the flush packet was now consumed"
);
break 'negotiation;
} else {
match action {
fetch::Action::Cancel => break 'negotiation,
fetch::Action::Continue => Some(response),
}
}
}
if matches!(protocol, Protocol::V2)
&& matches!(conn.mode, FetchConnection::TerminateOnSuccessfulCompletion)
{
log::trace!(target: "fetch", "Indicating end of interaction");
indicate_end_of_interaction(&mut conn)?;
}
log::trace!(target: "fetch", "fetched refs: {:?}", delegate.out.refs);
Ok(delegate.out)
}
fn setup_remote_progress<'a, P>(
progress: &mut P,
reader: &mut Box<dyn gix_transport::client::ExtendedBufRead<'a> + Unpin + 'a>,
) where
P: NestedProgress,
P::SubProgress: 'static,
{
reader.set_progress_handler(Some(Box::new({
let mut remote_progress = progress.add_child("remote");
move |is_err: bool, data: &[u8]| {
gix_protocol::RemoteProgress::translate_to_progress(is_err, data, &mut remote_progress);
gix_transport::packetline::read::ProgressAction::Continue
}
}) as gix_transport::client::HandleProgress<'a>));
log::trace!(target: "fetch", "fetched refs: {:?}", out.refs);
Ok(out)
}

135
radicle-fetch/src/transport/ls_refs.rs
View File

@ -1,16 +1,13 @@
use std::borrow::Cow;
use std::io::{self, BufRead};
use std::io;
use bstr::ByteSlice;
use gix_features::progress::Progress;
use gix_protocol::fetch::{self, Delegate, DelegateBlocking};
use gix_protocol::handshake::{self, Ref};
use gix_protocol::ls_refs;
use gix_protocol::transport::Protocol;
use gix_protocol::{ls_refs, Command};
use gix_transport::bstr::{BString, ByteVec};
use gix_transport::client::{self, TransportV2Ext};
use super::{agent_name, indicate_end_of_interaction, Connection};
use super::{agent_name, Connection};
/// Configuration for running an ls-refs process.
///
@ -19,84 +16,10 @@ pub struct Config {
/// The repository name, i.e. `/<rid>`.
#[allow(dead_code)]
pub repo: BString,
/// Extra parameters to pass to the ls-refs process.
pub extra_params: Vec<(String, Option<String>)>,
/// Ref prefixes for filtering the output of the ls-refs process.
pub prefixes: Vec<BString>,
}
/// The Gitoxide delegate for running the ls-refs process.
struct LsRefs {
/// Configuration for the ls-refs process.
config: Config,
/// The resulting references returned by the ls-refs process.
refs: Vec<Ref>,
}
impl LsRefs {
fn new(config: Config) -> Self {
Self {
config,
refs: Vec::new(),
}
}
}
// FIXME: the delegate pattern will be removed in the near future and
// we should look at the fetch code being used in gix to see how we
// can migrate to the proper form of fetching.
impl DelegateBlocking for LsRefs {
fn handshake_extra_parameters(&self) -> Vec<(String, Option<String>)> {
self.config.extra_params.clone()
}
fn prepare_ls_refs(
&mut self,
_caps: &client::Capabilities,
args: &mut Vec<BString>,
_: &mut Vec<(&str, Option<Cow<'_, str>>)>,
) -> io::Result<ls_refs::Action> {
for prefix in &self.config.prefixes {
let mut arg = BString::from("ref-prefix ");
arg.push_str(prefix);
args.push(arg)
}
Ok(ls_refs::Action::Continue)
}
fn prepare_fetch(
&mut self,
_: Protocol,
_: &client::Capabilities,
_: &mut Vec<(&str, Option<Cow<'_, str>>)>,
refs: &[Ref],
) -> io::Result<fetch::Action> {
self.refs.extend_from_slice(refs);
Ok(fetch::Action::Cancel)
}
fn negotiate(
&mut self,
_: &[Ref],
_: &mut fetch::Arguments,
_: Option<&fetch::Response>,
) -> io::Result<fetch::Action> {
unreachable!("`negotiate` called even though no `fetch` command was sent")
}
}
impl Delegate for LsRefs {
fn receive_pack(
&mut self,
_: impl BufRead,
_: impl Progress,
_: &[Ref],
_: &fetch::Response,
) -> io::Result<()> {
unreachable!("`receive_pack` called even though no `fetch` command was sent")
}
}
/// Run the ls-refs process using the provided `config`.
///
/// It is expected that the `handshake` was run outside of this
@ -116,7 +39,6 @@ where
W: io::Write,
{
log::trace!(target: "fetch", "Performing ls-refs: {:?}", config.prefixes);
let mut delegate = LsRefs::new(config);
let handshake::Outcome {
server_protocol_version: protocol,
capabilities,
@ -130,48 +52,21 @@ where
)));
}
let ls = Command::LsRefs;
let mut features = ls.default_features(Protocol::V2, capabilities);
// N.b. copied from gitoxide
let mut args = vec![
b"symrefs".as_bstr().to_owned(),
b"peel".as_bstr().to_owned(),
];
if capabilities
.capability("ls-refs")
.and_then(|cap| cap.supports("unborn"))
.unwrap_or_default()
{
args.push("unborn".into());
let refs = gix_protocol::ls_refs(
&mut conn,
capabilities,
|_caps, args, features| {
for prefix in &config.prefixes {
let mut arg = BString::from("ref-prefix ");
arg.push_str(prefix);
args.push(arg)
}
let refs = match delegate.prepare_ls_refs(capabilities, &mut args, &mut features) {
Ok(ls_refs::Action::Skip) => Vec::new(),
Ok(ls_refs::Action::Continue) => {
// FIXME: this is a private function
// ls.validate_argument_prefixes_or_panic(Protocol::V2, capabilities, &args, &features);
let agent = agent_name()?;
features.push(("agent", Some(Cow::Owned(agent))));
progress.step();
progress.set_name("list refs".into());
let mut remote_refs = conn.invoke(
ls.as_str(),
features.clone().into_iter(),
if args.is_empty() {
None
} else {
Some(args.into_iter())
features.push(("agent", Some(Cow::Owned(agent_name()?))));
Ok(gix_protocol::ls_refs::Action::Continue)
},
false,
progress,
false, /* trace packetlines */
)?;
handshake::refs::from_v2_refs(&mut remote_refs)?
}
Err(err) => {
indicate_end_of_interaction(&mut conn)?;
return Err(err.into());
}
};
Ok(refs)
}