gix_protocol/fetch/

function.rs

use std::{
    path::Path,
    sync::atomic::{AtomicBool, Ordering},
};

use gix_features::progress::DynNestedProgress;

use crate::fetch::{
    Arguments, Context, Error, Negotiate, NegotiateOutcome, Options, Outcome, ProgressId, Shallow, Tags, negotiate,
};
#[crate::bisync::only_async]
use crate::transport::client::async_io::{ExtendedBufRead, HandleProgress, Transport};
#[crate::bisync::only_sync]
use crate::transport::client::blocking_io::{ExtendedBufRead, HandleProgress, Transport};

/// Perform one fetch operation, relying on a `transport`.
/// `negotiate` is used to run the negotiation of objects that should be contained in the pack, *if* one is to be received.
/// `progress` and `should_interrupt` is passed to all potentially long-running parts of the operation.
///
/// `consume_pack(pack_read, progress, interrupt) -> bool` is always called to consume all bytes that are sent by the server, returning `true` if we should assure the pack is read to the end,
/// or `false` to do nothing. Dropping the reader without reading to EOF (i.e. returning `false`) is an offense to the server, and
/// `transport` won't be in the correct state to perform additional operations, or indicate the end of operation.
/// Note that the passed reader blocking as the pack-writing is blocking as well.
///
/// The `Context` and `Options` further define parts of this `fetch` operation.
///
/// As opposed to a full `git fetch`, this operation does *not*…
///
/// * …update local refs
/// * …end the interaction after the fetch
///
/// **Note that the interaction will never be ended**, even on error or failure, leaving it up to the caller to do that, maybe
/// with the help of [`SendFlushOnDrop`](crate::SendFlushOnDrop) which can wrap `transport`.
/// Generally, the `transport` is left in a state that allows for more commands to be run.
///
/// Return `Ok(None)` if there was nothing to do because all remote refs are at the same state as they are locally,
/// or there was nothing wanted, or `Ok(Some(outcome))` to inform about all the changes that were made.
#[crate::bisync::bisync]
pub async fn fetch<P, T, E>(
    negotiate: &mut impl Negotiate,
    consume_pack: impl FnOnce(&mut dyn std::io::BufRead, &mut dyn DynNestedProgress, &AtomicBool) -> Result<bool, E>,
    mut progress: P,
    should_interrupt: &AtomicBool,
    Context {
        handshake,
        transport,
        user_agent,
        trace_packetlines,
    }: Context<'_, T>,
    Options {
        shallow_file,
        shallow,
        tags,
        reject_shallow_remote,
    }: Options<'_>,
) -> Result<Option<Outcome>, Error>
where
    P: gix_features::progress::NestedProgress,
    P::SubProgress: 'static,
    T: Transport,
    E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
{
    let _span = gix_trace::coarse!("gix_protocol::fetch()");
    let v1_shallow_updates = handshake.v1_shallow_updates.take();
    let protocol_version = handshake.server_protocol_version;

    let fetch = crate::Command::Fetch;
    let fetch_features = {
        let mut f = fetch.default_features(protocol_version, &handshake.capabilities);
        f.push(user_agent);
        f
    };

    crate::fetch::Response::check_required_features(protocol_version, &fetch_features)?;
    let sideband_all = fetch_features.iter().any(|(n, _)| *n == "sideband-all");
    let mut arguments = Arguments::new(protocol_version, fetch_features, trace_packetlines);
    if matches!(tags, Tags::Included) {
        if !arguments.can_use_include_tag() {
            return Err(Error::MissingServerFeature {
                    feature: "include-tag",
                    description:
                    // NOTE: if this is an issue, we could probably do what's proposed here.
                    "To make this work we would have to implement another pass to fetch attached tags separately",
                });
        }
        arguments.use_include_tag();
    }
    let (shallow_commits, mut shallow_lock) = add_shallow_args(&mut arguments, shallow, &shallow_file)?;

    let negotiate_span = gix_trace::detail!(
        "negotiate",
        protocol_version = handshake.server_protocol_version as usize
    );
    let action = negotiate.mark_complete_and_common_ref()?;
    let mut previous_response = None::<crate::fetch::Response>;
    match &action {
        negotiate::Action::NoChange | negotiate::Action::SkipToRefUpdate => Ok(None),
        negotiate::Action::MustNegotiate {
            remote_ref_target_known,
        } => {
            if !negotiate.add_wants(&mut arguments, remote_ref_target_known) {
                return Ok(None);
            }
            let mut rounds = Vec::new();
            let is_stateless = arguments.is_stateless(!transport.connection_persists_across_multiple_requests());
            let mut state = negotiate::one_round::State::new(is_stateless);
            let reader = 'negotiation: loop {
                let _round = gix_trace::detail!("negotiate round", round = rounds.len() + 1);
                progress.step();
                progress.set_name(format!("negotiate (round {})", rounds.len() + 1));
                if should_interrupt.load(Ordering::Relaxed) {
                    return Err(Error::Negotiate(negotiate::Error::NegotiationFailed {
                        rounds: rounds.len(),
                    }));
                }

                let is_done = match negotiate.one_round(&mut state, &mut arguments, previous_response.as_ref()) {
                    Ok((round, is_done)) => {
                        rounds.push(round);
                        is_done
                    }
                    Err(err) => {
                        return Err(err.into());
                    }
                };
                let mut reader = arguments.send(transport, is_done).await?;
                if sideband_all {
                    setup_remote_progress(&mut progress, &mut reader, should_interrupt);
                }
                let response =
                    crate::fetch::Response::from_line_reader(protocol_version, &mut reader, is_done, !is_done).await?;
                let has_pack = response.has_pack();
                previous_response = Some(response);
                if has_pack {
                    progress.step();
                    progress.set_name("receiving pack".into());
                    if !sideband_all {
                        setup_remote_progress(&mut progress, &mut reader, should_interrupt);
                    }
                    break 'negotiation reader;
                }
            };
            // This needs drop if tracing is compiled in. We just don't know it.
            drop(negotiate_span);

            let mut previous_response = previous_response.expect("knowledge of a pack means a response was received");
            previous_response.append_v1_shallow_updates(v1_shallow_updates);
            if !previous_response.shallow_updates().is_empty() && shallow_lock.is_none() {
                if reject_shallow_remote {
                    return Err(Error::RejectShallowRemote);
                }
                shallow_lock = acquire_shallow_lock(&shallow_file).map(Some)?;
            }

            let (mut reader, may_read_to_end) =
                consume_received_pack(reader, consume_pack, &mut progress, should_interrupt)?;

            if may_read_to_end {
                // Assure the final flush packet is consumed.
                let has_read_to_end = reader.stopped_at().is_some();
                if !has_read_to_end {
                    read_remaining(&mut reader).await.map_err(Error::ReadRemainingBytes)?;
                }
            }
            drop(reader);

            if let Some(shallow_lock) = shallow_lock {
                if !previous_response.shallow_updates().is_empty() {
                    gix_shallow::write(shallow_lock, shallow_commits, previous_response.shallow_updates())?;
                }
            }
            Ok(Some(Outcome {
                last_response: previous_response,
                negotiate: NegotiateOutcome { action, rounds },
            }))
        }
    }
}

#[crate::bisync::only_async]
fn consume_received_pack<R, E>(
    reader: R,
    consume: impl FnOnce(&mut dyn std::io::BufRead, &mut dyn DynNestedProgress, &AtomicBool) -> Result<bool, E>,
    progress: &mut dyn DynNestedProgress,
    should_interrupt: &AtomicBool,
) -> Result<(R, bool), Error>
where
    R: crate::futures_io::AsyncBufRead + Unpin,
    E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
{
    let mut reader = crate::futures_lite::io::BlockOn::new(reader);
    let may_read_to_end =
        consume(&mut reader, progress, should_interrupt).map_err(|err| Error::ConsumePack(err.into()))?;
    Ok((reader.into_inner(), may_read_to_end))
}

#[crate::bisync::only_sync]
fn consume_received_pack<R, E>(
    mut reader: R,
    consume: impl FnOnce(&mut dyn std::io::BufRead, &mut dyn DynNestedProgress, &AtomicBool) -> Result<bool, E>,
    progress: &mut dyn DynNestedProgress,
    should_interrupt: &AtomicBool,
) -> Result<(R, bool), Error>
where
    R: std::io::BufRead,
    E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
{
    let may_read_to_end =
        consume(&mut reader, progress, should_interrupt).map_err(|err| Error::ConsumePack(err.into()))?;
    Ok((reader, may_read_to_end))
}

#[crate::bisync::only_async]
async fn read_remaining(reader: &mut (impl crate::futures_io::AsyncRead + Unpin)) -> std::io::Result<()> {
    crate::futures_lite::io::copy(reader, &mut crate::futures_lite::io::sink())
        .await
        .map(|_| ())
}

#[crate::bisync::only_sync]
fn read_remaining(reader: &mut impl std::io::Read) -> std::io::Result<()> {
    std::io::copy(reader, &mut std::io::sink()).map(|_| ())
}

fn acquire_shallow_lock(shallow_file: &Path) -> Result<gix_lock::File, Error> {
    gix_lock::File::acquire_to_update_resource(shallow_file, gix_lock::acquire::Fail::Immediately, None)
        .map_err(Into::into)
}

fn add_shallow_args(
    args: &mut Arguments,
    shallow: &Shallow,
    shallow_file: &std::path::Path,
) -> Result<(Option<nonempty::NonEmpty<gix_hash::ObjectId>>, Option<gix_lock::File>), Error> {
    let expect_change = *shallow != Shallow::NoChange;
    let shallow_lock = expect_change.then(|| acquire_shallow_lock(shallow_file)).transpose()?;

    let shallow_commits = gix_shallow::read(shallow_file)?;
    if (shallow_commits.is_some() || expect_change) && !args.can_use_shallow() {
        // NOTE: if this is an issue, we can always unshallow the repo ourselves.
        return Err(Error::MissingServerFeature {
            feature: "shallow",
            description: "shallow clones need server support to remain shallow, otherwise bigger than expected packs are sent effectively unshallowing the repository",
        });
    }
    if let Some(shallow_commits) = &shallow_commits {
        for commit in shallow_commits.iter() {
            args.shallow(commit);
        }
    }
    match shallow {
        Shallow::NoChange => {}
        Shallow::DepthAtRemote(commits) => args.deepen(commits.get() as usize),
        Shallow::Deepen(commits) => {
            args.deepen(*commits as usize);
            args.deepen_relative();
        }
        Shallow::Since { cutoff } => {
            args.deepen_since(cutoff.seconds);
        }
        Shallow::Exclude {
            remote_refs,
            since_cutoff,
        } => {
            if let Some(cutoff) = since_cutoff {
                args.deepen_since(cutoff.seconds);
            }
            for ref_ in remote_refs {
                args.deepen_not(ref_.as_ref().as_bstr());
            }
        }
    }
    Ok((shallow_commits, shallow_lock))
}

fn setup_remote_progress<'a>(
    progress: &mut dyn gix_features::progress::DynNestedProgress,
    reader: &mut Box<dyn ExtendedBufRead<'a> + Unpin + 'a>,
    should_interrupt: &'a AtomicBool,
) {
    reader.set_progress_handler(Some(Box::new({
        let mut remote_progress = progress.add_child_with_id("remote".to_string(), ProgressId::RemoteProgress.into());
        move |is_err: bool, data: &[u8]| {
            crate::RemoteProgress::translate_to_progress(is_err, data, &mut remote_progress);
            if should_interrupt.load(Ordering::Relaxed) {
                std::ops::ControlFlow::Break(())
            } else {
                std::ops::ControlFlow::Continue(())
            }
        }
    }) as HandleProgress<'a>));
}