[installinator] speed up HTTP downloads (#9900)

* Set the connect timeout to 5s per peer.
* Remember the last peer used to fetch an artifact.

Fixes #9617.

Author: sunshowers

installinator/src/artifact.rs

@@ -154,19 +154,23 @@ impl ArtifactClient {
             slog::o!("component" => "ArtifactClient", "peer" => addr.to_string()),
         );
 
-        // Set a connect timeout of 15 seconds (the progenitor default), and a
-        // total timeout of 5 minutes. The progenitor default for the total
-        // timeout is 15 seconds, which can easily be exceeded for large
-        // downloads. (Don't set the total timeout to be too long, though,
-        // because we're fetching ~2GiB artifacts over a LAN which really should
-        // take less than 5 minutes.)
+        // Set a connect timeout of 5 seconds and a total timeout of 5 minutes.
+        // The progenitor default for the total timeout is 15 seconds, which can
+        // easily be exceeded for large downloads. (Don't set the total timeout
+        // to be too long, though, because we're fetching ~2GiB artifacts over a
+        // LAN which really should take less than 5 minutes.)
+        //
+        // The connect timeout is deliberately short: these are LAN connections
+        // on the bootstrap network, and peers are tried serially. A long
+        // connect timeout (the progenitor default of 15s) means unresponsive
+        // peers block progress for a long time before the next peer is tried.
         //
         // Do not set a read timeout here -- instead, read timeouts are handled
         // by the fetch loop. (Why is the read timeout handled by the fetch
         // loop? So that it can also apply to the mock peer backend, and logic
         // shared across both.)
         let client = reqwest::ClientBuilder::new()
-            .connect_timeout(Duration::from_secs(15))
+            .connect_timeout(Duration::from_secs(5))
             .timeout(Duration::from_secs(5 * 60))
             .build()
             .expect("installinator artifact client created");

installinator/src/dispatch.rs

@@ -27,7 +27,7 @@ use crate::{
     ArtifactWriter, MeasurementToWrite, WriteDestination,
     artifact::{ArtifactIdOpts, ArtifactsToDownload, MeasurementArtifact},
     fetch::{FetchArtifactBackend, FetchedArtifact, HttpFetchBackend},
-    peers::DiscoveryMechanism,
+    peers::{DiscoveryMechanism, LastKnownPeer},
     reporter::{HttpProgressBackend, ProgressReporter, ReportProgressBackend},
 };
 
@@ -99,6 +99,7 @@ impl DebugDiscoverOpts {
                 self.opts.mechanism.discover_peers(&log).await?,
             )),
             Duration::from_secs(10),
+            None,
         );
         println!("discovered peers: {}", backend.peers().display());
         Ok(())
@@ -206,6 +207,8 @@ impl InstallOpts {
         );
         let progress_handle = progress_reporter.start();
         let discovery = &self.discover_opts.mechanism;
+        let last_known_peer = LastKnownPeer::new();
+        let last_known_peer = &last_known_peer;
 
         let engine = UpdateEngine::new(log, event_sender);
 
@@ -223,6 +226,7 @@ impl InstallOpts {
                         &cx,
                         &installinator_doc_id,
                         discovery,
+                        &last_known_peer,
                         log,
                     )
                     .await?;
@@ -326,9 +330,14 @@ impl InstallOpts {
                     let to_download = to_download.into_value(cx.token()).await;
                     let host_phase_2_id = to_download.host_phase_2_id();
 
-                    let host_phase_2_artifact =
-                        fetch_artifact(&cx, &host_phase_2_id, discovery, log)
-                            .await?;
+                    let host_phase_2_artifact = fetch_artifact(
+                        &cx,
+                        &host_phase_2_id,
+                        discovery,
+                        last_known_peer,
+                        log,
+                    )
+                    .await?;
 
                     // Check that the sha256 of the data we got from wicket
                     // matches the data we asked for. We do not retry this for
@@ -364,9 +373,14 @@ impl InstallOpts {
                         to_download_2.into_value(cx.token()).await;
                     let control_plane_id = to_download.control_plane_id();
 
-                    let control_plane_artifact =
-                        fetch_artifact(&cx, &control_plane_id, discovery, log)
-                            .await?;
+                    let control_plane_artifact = fetch_artifact(
+                        &cx,
+                        &control_plane_id,
+                        discovery,
+                        last_known_peer,
+                        log,
+                    )
+                    .await?;
 
                     // Check that the sha256 of the data we got from wicket
                     // matches the data we asked for. We do not retry this for
@@ -420,7 +434,7 @@ impl InstallOpts {
                                 async move |cx2| {
                                     let measurement_artifact =
                                         fetch_artifact(
-                                            &cx2, &c.hash, discovery, log,
+                                            &cx2, &c.hash, discovery, last_known_peer, log,
                                         )
                                         .await?;
 
@@ -675,6 +689,7 @@ async fn fetch_artifact(
     cx: &StepContext,
     id: &ArtifactHashId,
     discovery: &DiscoveryMechanism,
+    last_known_peer: &LastKnownPeer,
     log: &slog::Logger,
 ) -> Result<FetchedArtifact> {
     // TODO: Not sure why slog::o!("artifact" => ?id) isn't working, figure it
@@ -684,20 +699,24 @@ async fn fetch_artifact(
         cx,
         &log,
         || async {
+            let preferred = last_known_peer.get();
             Ok(FetchArtifactBackend::new(
                 &log,
                 Box::new(HttpFetchBackend::new(
                     &log,
                     discovery.discover_peers(&log).await?,
                 )),
                 Duration::from_secs(10),
+                preferred,
             ))
         },
         id,
     )
     .await
     .with_context(|| format!("error fetching image with id {id:?}"))?;
 
+    last_known_peer.set(artifact.peer);
+
     slog::info!(
         log,
         "fetched {} bytes from {}",

installinator/src/fetch.rs

@@ -127,16 +127,18 @@ pub(crate) struct FetchArtifactBackend {
     log: slog::Logger,
     imp: Box<dyn FetchArtifactImpl>,
     read_timeout: Duration,
+    preferred_peer: Option<PeerAddress>,
 }
 
 impl FetchArtifactBackend {
     pub(crate) fn new(
         log: &slog::Logger,
         imp: Box<dyn FetchArtifactImpl>,
         read_timeout: Duration,
+        preferred_peer: Option<PeerAddress>,
     ) -> Self {
         let log = log.new(slog::o!("component" => "Peers"));
-        Self { log, imp, read_timeout }
+        Self { log, imp, read_timeout, preferred_peer }
     }
 
     pub(crate) async fn fetch_artifact(
@@ -154,7 +156,7 @@ impl FetchArtifactBackend {
 
         slog::debug!(log, "start fetch from peers"; "remaining_peers" => remaining_peers);
 
-        for &peer in peers.peers() {
+        for &peer in peers.iter_with_preferred(self.preferred_peer) {
             remaining_peers -= 1;
 
             slog::debug!(

installinator/src/mock_peers.rs

@@ -837,6 +837,7 @@ mod tests {
                     &log,
                     Box::new(peers),
                     read_timeout,
+                    None,
                 )),
                 Some(Err(error)) => {
                     future::err(DiscoverPeersError::Retry(error))

installinator/src/peers.rs

@@ -7,6 +7,7 @@ use std::{
     fmt,
     net::{AddrParseError, IpAddr, SocketAddr},
     str::FromStr,
+    sync::Mutex,
 };
 
 use anyhow::{Result, bail};
@@ -118,6 +119,21 @@ impl PeerAddresses {
     pub(crate) fn display(&self) -> impl fmt::Display + use<> {
         self.peers().iter().join(", ")
     }
+
+    /// Returns an iterator that yields the preferred peer first (if present in
+    /// the set), then the remaining peers in BTreeSet order.
+    pub(crate) fn iter_with_preferred(
+        &self,
+        preferred: Option<PeerAddress>,
+    ) -> impl Iterator<Item = &PeerAddress> {
+        // Look up the preferred peer in the set to get a reference with the
+        // right lifetime. (If it's not in the set, the correct behavior is to
+        // skip it.)
+        let preferred_ref = preferred.and_then(|p| self.peers.get(&p));
+        preferred_ref
+            .into_iter()
+            .chain(self.peers.iter().filter(move |&p| Some(p) != preferred_ref))
+    }
 }
 
 impl FromIterator<PeerAddress> for PeerAddresses {
@@ -157,3 +173,98 @@ impl FromStr for PeerAddress {
         Ok(Self { address })
     }
 }
+
+/// Tracks the last peer that successfully delivered an artifact.
+///
+/// Created once per install session and shared across all artifact fetches, so
+/// that a peer discovered while fetching an earlier artifact is tried first for
+/// subsequent artifacts.
+#[derive(Debug)]
+pub(crate) struct LastKnownPeer {
+    inner: Mutex<Option<PeerAddress>>,
+}
+
+impl LastKnownPeer {
+    pub(crate) fn new() -> Self {
+        Self { inner: Mutex::new(None) }
+    }
+
+    pub(crate) fn get(&self) -> Option<PeerAddress> {
+        *self.inner.lock().expect("last known peer lock poisoned")
+    }
+
+    pub(crate) fn set(&self, peer: PeerAddress) {
+        *self.inner.lock().expect("last known peer lock poisoned") = Some(peer);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{Ipv6Addr, SocketAddrV6};
+
+    /// Build a peer address from a distinguishing segment. Returns
+    /// `[::n]:8000`.
+    fn peer(last_segment: u16) -> PeerAddress {
+        PeerAddress::new(SocketAddr::V6(SocketAddrV6::new(
+            Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, last_segment),
+            8000,
+            0,
+            0,
+        )))
+    }
+
+    fn addrs(ns: &[u16]) -> PeerAddresses {
+        ns.iter().map(|&n| peer(n)).collect()
+    }
+
+    /// Collect the last segment of each peer address for easy comparison.
+    fn collect_last_segment(
+        peers: &PeerAddresses,
+        preferred: Option<PeerAddress>,
+    ) -> Vec<u16> {
+        peers
+            .iter_with_preferred(preferred)
+            .map(|p| match p.address() {
+                SocketAddr::V6(a) => a.ip().segments()[7],
+                SocketAddr::V4(_) => {
+                    unreachable!("we only use IPv6 addresses in these tests")
+                }
+            })
+            .collect()
+    }
+
+    #[test]
+    fn iter_no_preferred_preserves_btree_order() {
+        let peers = addrs(&[3, 1, 2]);
+        assert_eq!(collect_last_segment(&peers, None), vec![1, 2, 3]);
+    }
+
+    #[test]
+    fn last_known_peer_preferred() {
+        let peers = addrs(&[3, 1, 2]);
+        let lkp = LastKnownPeer::new();
+
+        // Initially no preferred peer: normal BTreeSet order.
+        assert_eq!(collect_last_segment(&peers, lkp.get()), vec![1, 2, 3]);
+
+        // Simulate a successful fetch from peer ::3.
+        lkp.set(peer(3));
+        assert_eq!(collect_last_segment(&peers, lkp.get()), vec![3, 1, 2]);
+
+        // Subsequent fetch succeeds from peer ::1 instead.
+        lkp.set(peer(1));
+        assert_eq!(collect_last_segment(&peers, lkp.get()), vec![1, 2, 3]);
+    }
+
+    #[test]
+    fn last_known_peer_stale() {
+        // The remembered peer may not appear in a later discovery round.
+        let lkp = LastKnownPeer::new();
+        lkp.set(peer(1));
+
+        let new_peers = addrs(&[2, 3]);
+        // peer(1) is gone; iteration should fall back to normal order.
+        assert_eq!(collect_last_segment(&new_peers, lkp.get()), vec![2, 3]);
+    }
+}