failover between transports

1 files changed, 192 insertions, 9 deletions

diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index d1a4dc9..3330b2d 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -9,11 +9,22 @@
 //! sender's address, and puts together what the transports say about themselves
 //! (addresses, seeds they found, figures). The transports are in order of preference,
 //! as configured (`--transports`).
+//!
+//! A message to a node with several addresses goes to the first of them that takes it
+//! (`send_first`), in the order `rank` gives: addresses whose last send failed, for
+//! `DOWN_FOR`, after the others, and among each the configured order. A send that fails
+//! is tried at once at the next address, the message encoded for its transport: when
+//! one transport fails, the next carries on within the same send.
 
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow};
+use tokio::time::Instant;
+use tracing::debug;
+
+use crate::locks::MutexExt;
 
 use crate::transport::stats::TransportStats;
 use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport};
@@ -45,9 +56,16 @@ pub(crate) fn scheme_of(address: &str) -> &str {
     address.split_once(':').map_or("", |(scheme, _)| scheme)
 }
 
+/// An address whose send failed is ranked after the others for this long (unless a
+/// send there works again).
+const DOWN_FOR: Duration = Duration::from_secs(60);
+
 pub(crate) struct Paths {
     /// In order of preference.
     links: Vec<Link>,
+    /// Addresses whose last send failed, and when (`DOWN_FOR`). A std mutex: never held
+    /// across an await.
+    down: Mutex<HashMap<String, Instant>>,
 }
 
 impl Paths {
@@ -58,6 +76,7 @@ impl Paths {
                 .into_iter()
                 .map(|(scheme, inner)| Link { scheme, inner })
                 .collect(),
+            down: Mutex::default(),
         }
     }
 
@@ -104,6 +123,29 @@ impl Paths {
             .unwrap_or(usize::MAX)
     }
 
+    /// The usable ones of `addresses` (as paths keep them), best first: those whose
+    /// last send failed (within `DOWN_FOR`) last, and by the configured order.
+    pub(crate) fn rank(&self, addresses: &[String]) -> Vec<String> {
+        let mut usable: Vec<(bool, usize, &String)> = addresses
+            .iter()
+            .filter(|address| self.usable(address))
+            .map(|address| (false, self.preference(address), address))
+            .collect();
+        {
+            let now = Instant::now();
+            let mut down = self.down.lock_or_recover();
+            down.retain(|_, at| now.duration_since(*at) < DOWN_FOR);
+            for (failed, _, address) in &mut usable {
+                *failed = down.contains_key(*address);
+            }
+        }
+        usable.sort_by_key(|(failed, preference, _)| (*failed, *preference));
+        usable
+            .into_iter()
+            .map(|(_, _, address)| address.clone())
+            .collect()
+    }
+
     /// Whether `address` is one of ours.
     pub(crate) fn is_own(&self, address: &str) -> bool {
         self.links.iter().any(|link| link.own_address() == address)
@@ -193,7 +235,7 @@ impl Paths {
     }
 
     /// `message` encoded for `address`: `peer` and `address` of the result are
-    /// `address`.
+    /// `address`, and it keeps the message (to encode it for another address).
     pub(crate) fn encode(&self, address: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
         let (link, target) = self
             .split(address)
@@ -201,9 +243,67 @@ impl Paths {
         let mut encoded = link.inner.encode(&target, message)?;
         encoded.peer = address.to_owned();
         encoded.address = address.to_owned();
+        encoded.message = Some(Arc::new(message.clone()));
         Ok(encoded)
     }
 
+    /// Sends `message` to the first of `addresses` (`rank`ed) that takes it: to its own
+    /// `address` as encoded, to others encoded again for their transport (if it keeps
+    /// the message). Returns the address it went to; an error with every address's if
+    /// none took it.
+    pub(crate) async fn send_first(
+        &self,
+        message: &EncodedMessage,
+        addresses: &[String],
+    ) -> Result<String> {
+        let mut failures: Vec<(String, anyhow::Error)> = Vec::new();
+        for address in addresses {
+            let encoded = if *address == message.address {
+                message.clone()
+            } else {
+                let Some(parsed) = &message.message else {
+                    continue;
+                };
+                match self.encode(address, parsed) {
+                    Ok(mut encoded) => {
+                        encoded.peer = message.peer.clone();
+                        encoded
+                    }
+                    Err(error) => {
+                        failures.push((address.clone(), error));
+                        continue;
+                    }
+                }
+            };
+            match self.send(&encoded).await {
+                Ok(()) => {
+                    self.down.lock_or_recover().remove(address);
+                    if !failures.is_empty() {
+                        debug!(
+                            peer = %message.peer,
+                            address = %address,
+                            failed = %describe(&failures),
+                            "Sent at another address after a send failed"
+                        );
+                    }
+                    return Ok(address.clone());
+                }
+                Err(error) => {
+                    self.down
+                        .lock_or_recover()
+                        .insert(address.clone(), Instant::now());
+                    failures.push((address.clone(), error));
+                }
+            }
+        }
+        match failures.len() {
+            0 => Err(anyhow!("no address to send to")),
+            // The transport's error as it is.
+            1 => Err(failures.pop().expect("one").1),
+            _ => Err(anyhow!("{}", describe(&failures))),
+        }
+    }
+
     /// Sends `message` to its `address`.
     pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> {
         let (link, target) = self
@@ -269,6 +369,15 @@ impl Paths {
     }
 }
 
+/// `address: error; …` for each failed send.
+fn describe(failures: &[(String, anyhow::Error)]) -> String {
+    failures
+        .iter()
+        .map(|(address, error)| format!("{address}: {error:#}"))
+        .collect::<Vec<_>>()
+        .join("; ")
+}
+
 /// The first of the `results` to arrive, from tasks that run until they end.
 async fn first<T: Send + 'static>(
     tasks: Vec<std::pin::Pin<Box<dyn Future<Output = T> + Send>>>,
@@ -327,6 +436,86 @@ mod tests {
         assert!(a.resolve("nobody").await.is_err());
     }
 
+    /// A transport whose sends all fail.
+    struct Broken;
+
+    #[async_trait::async_trait]
+    impl Transport for Broken {
+        fn local_id(&self) -> &str {
+            "us"
+        }
+        fn is_peer_id(&self, id: &str) -> bool {
+            !id.is_empty()
+        }
+        fn delivery_limit(&self) -> Duration {
+            Duration::from_secs(1)
+        }
+        async fn resolve_seed(&self, seed: &str) -> Result<String> {
+            Ok(seed.to_owned())
+        }
+        fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+            crate::transport::mycelium::encode_message(peer, message)
+        }
+        async fn send_encoded(&self, _message: &EncodedMessage) -> Result<()> {
+            Err(anyhow!("broken"))
+        }
+        async fn start_consumer(&self, _handler: crate::transport::MessageHandler) -> Result<()> {
+            std::future::pending().await
+        }
+        async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+            std::future::pending().await
+        }
+    }
+
+    fn some_status() -> ParsedMessage {
+        ParsedMessage::Status(Status {
+            heads: HeadsSummary::of(Vec::new()),
+            epoch: Epoch::GENESIS,
+            epoch_seq: 0,
+            ready: false,
+            syncs_with_you: false,
+        })
+    }
+
+    /// When the preferred transport fails, the same send goes over the next one, and
+    /// the failed address is ranked last for a while.
+    #[tokio::test]
+    async fn a_failed_send_goes_to_the_next_address() {
+        let net = SimNet::new(3);
+        let a = Paths::new(vec![
+            ("broken", Arc::new(Broken) as Arc<dyn Transport>),
+            ("sim", net.transport("a") as Arc<dyn Transport>),
+        ]);
+        let b = Paths::new(vec![("sim", net.transport("b") as Arc<dyn Transport>)]);
+        let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
+        let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {
+            let sender = sender.clone();
+            Box::pin(async move {
+                let _ = sender.send(from);
+                Ok(())
+            })
+        });
+        tokio::spawn(async move { b.start_consumer(handler).await });
+
+        let addresses = vec!["sim:b".to_owned(), "broken:b".to_owned()];
+        let ranked = a.rank(&addresses);
+        assert_eq!(ranked, ["broken:b", "sim:b"]); // The configured order.
+        let mut encoded = a.encode(&ranked[0], &some_status()).unwrap();
+        encoded.peer = "node-b".to_owned();
+        assert_eq!(a.send_first(&encoded, &ranked).await.unwrap(), "sim:b");
+        assert_eq!(received.recv().await.unwrap(), "sim:a");
+        assert_eq!(a.rank(&addresses), ["sim:b", "broken:b"]);
+
+        // Without the message kept, nothing else can be tried: the transport's error.
+        encoded.message = None;
+        let error = a
+            .send_first(&encoded, &["broken:b".to_owned()])
+            .await
+            .unwrap_err();
+        assert_eq!(error.to_string(), "broken");
+        assert!(a.send_first(&encoded, &[]).await.is_err());
+    }
+
     /// A message sent to an address arrives under the sender's address on that
     /// transport.
     #[tokio::test]
@@ -341,13 +530,7 @@ mod tests {
             })
         });
         tokio::spawn(async move { b.start_consumer(handler).await });
-        let status = ParsedMessage::Status(Status {
-            heads: HeadsSummary::of(Vec::new()),
-            epoch: Epoch::GENESIS,
-            epoch_seq: 0,
-            ready: false,
-            syncs_with_you: false,
-        });
+        let status = some_status();
         for to in ["other:b", "sim:b"] {
             let encoded = a.encode(to, &status).unwrap();
             assert_eq!(encoded.address, to);