squash me

Author: Byron

gix-protocol/src/fetch/error.rs

@@ -33,6 +33,8 @@ pub enum Error {
     },
     #[error("Failed to consume the pack sent by the remove")]
     ConsumePack(Box<dyn std::error::Error + Send + Sync + 'static>),
+    #[error("Failed to read remaining bytes in stream")]
+    ReadRemainingBytes(#[source] std::io::Error),
 }
 
 impl crate::transport::IsSpuriousError for Error {

gix-protocol/src/fetch/function.rs

@@ -1,10 +1,10 @@
-use crate::{fetch::Arguments, transport::packetline::read::ProgressAction};
-use std::path::Path;
-use std::sync::atomic::{AtomicBool, Ordering};
-
 use crate::fetch::{
     negotiate, Context, Error, Negotiate, NegotiateOutcome, Options, Outcome, ProgressId, RefMap, Shallow, Tags,
 };
+use crate::{fetch::Arguments, transport::packetline::read::ProgressAction};
+use gix_features::progress::DynNestedProgress;
+use std::path::Path;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 /// Perform one fetch operation, relying on a `transport`, right after a [`ref_map`](RefMap::new()) was created so
 /// it's clear what the remote has.
@@ -32,7 +32,7 @@ use crate::fetch::{
 pub async fn fetch<P, T, E>(
     ref_map: &RefMap,
     negotiate: &mut impl Negotiate,
-    consume_pack: impl FnOnce(&mut dyn std::io::BufRead, &mut P, &AtomicBool) -> Result<(), E>,
+    consume_pack: impl FnOnce(&mut dyn std::io::BufRead, &mut dyn DynNestedProgress, &AtomicBool) -> Result<(), E>,
     mut progress: P,
     should_interrupt: &AtomicBool,
     Context {
@@ -114,7 +114,7 @@ where
             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 mut 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));
@@ -157,11 +157,37 @@ where
             }
 
             #[cfg(feature = "async-client")]
-            let mut reader = crate::futures_lite::io::BlockOn::new(reader);
+            let mut rd = crate::futures_lite::io::BlockOn::new(reader);
+            #[cfg(not(feature = "async-client"))]
+            let mut rd = reader;
+            consume_pack(&mut rd, &mut progress, should_interrupt).map_err(|err| Error::ConsumePack(err.into()))?;
+            #[cfg(feature = "async-client")]
+            {
+                reader = rd.into_inner();
+            }
             #[cfg(not(feature = "async-client"))]
-            let mut reader = reader;
-            consume_pack(&mut reader, &mut progress, should_interrupt).map_err(|err| Error::ConsumePack(err.into()))?;
+            {
+                reader = rd;
+            }
+
+            // Assure the final flush packet is consumed.
+            let has_read_to_end = reader.stopped_at().is_some();
+            #[cfg(feature = "async-client")]
+            {
+                if !has_read_to_end {
+                    futures_lite::io::copy(&mut reader, &mut futures_lite::io::sink())
+                        .await
+                        .map_err(Error::ReadRemainingBytes)?;
+                }
+            }
+            #[cfg(not(feature = "async-client"))]
+            {
+                if !has_read_to_end {
+                    std::io::copy(&mut reader, &mut std::io::sink()).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())?;

gix/src/remote/connection/fetch/error.rs

@@ -1,9 +1,12 @@
 use crate::config;
 
 /// The error returned by [`receive()`](super::Prepare::receive()).
+// TODO: remove unused variants
 #[derive(Debug, thiserror::Error)]
 #[allow(missing_docs)]
 pub enum Error {
+    #[error(transparent)]
+    Fetch(#[from] gix_protocol::fetch::Error),
     #[error("The value to configure pack threads should be 0 to auto-configure or the amount of threads to use")]
     PackThreads(#[from] config::unsigned_integer::Error),
     #[error("The value to configure the pack index version should be 1 or 2")]
@@ -49,11 +52,6 @@ pub enum Error {
     NegotiationAlgorithmConfig(#[from] config::key::GenericErrorWithValue),
     #[error("Failed to read remaining bytes in stream")]
     ReadRemainingBytes(#[source] std::io::Error),
-    #[error("None of the refspec(s) {} matched any of the {num_remote_refs} refs on the remote", refspecs.iter().map(|r| r.to_ref().instruction().to_bstring().to_string()).collect::<Vec<_>>().join(", "))]
-    NoMapping {
-        refspecs: Vec<gix_refspec::RefSpec>,
-        num_remote_refs: usize,
-    },
 }
 
 impl gix_protocol::transport::IsSpuriousError for Error {