path: root/src/transport/paths.rs

//! Paths: one address space over any number of transports, below the identity layer
//! (`transport::identity`). An address is `<scheme>:<target>`: the transport by its
//! scheme (`mycelium`, `libp2p`, …), and on it a transport identity (Mycelium's public
//! key, libp2p's peer ID) or a location (`Transport::locate`), e.g. a multiaddr.
//! Addresses as paths keeps them are `<scheme>:<transport identity>` (`canonical`).
//!
//! Every transport carries the same messages (`wire`); paths routes each send to the
//! transport of its address, hands every transport's messages to one handler under the
//! 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::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};

/// Called with the sender's address, the message and its `payload_len`.
pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>
    + Send
    + Sync
    + 'static;

/// One of the node's transports.
struct Link {
    scheme: &'static str,
    inner: Arc<dyn Transport>,
}

impl Link {
    fn address(&self, target: &str) -> String {
        format!("{}:{target}", self.scheme)
    }

    fn own_address(&self) -> String {
        self.address(self.inner.local_id())
    }
}

/// The scheme of `address`.
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 {
    /// Over `transports`, by scheme, in order of preference.
    pub(crate) fn new(transports: Vec<(&'static str, Arc<dyn Transport>)>) -> Self {
        Self {
            links: transports
                .into_iter()
                .map(|(scheme, inner)| Link { scheme, inner })
                .collect(),
            down: Mutex::default(),
        }
    }

    /// The transport `address` is on, and the transport identity at it
    /// (`Transport::locate`, which notes where that is if the address says); None if no
    /// transport of ours has its scheme, or it isn't one of that transport's.
    fn split(&self, address: &str) -> Option<(&Link, String)> {
        let (scheme, target) = address.split_once(':')?;
        let link = self.links.iter().find(|link| link.scheme == scheme)?;
        let target = link.inner.locate(target)?;
        Some((link, target))
    }

    /// Whether one of our transports can use `address`.
    pub(crate) fn usable(&self, address: &str) -> bool {
        self.split(address).is_some()
    }

    /// `address` as paths keep it: `<scheme>:<transport identity>`.
    pub(crate) fn canonical(&self, address: &str) -> Option<String> {
        self.split(address)
            .map(|(link, target)| link.address(&target))
    }

    /// Where the node at `address` (as paths keep it) is, as others can use it
    /// (`Transport::locations`).
    pub(crate) fn locations(&self, address: &str) -> Vec<String> {
        let Some((link, target)) = self.split(address) else {
            return Vec::new();
        };
        link.inner
            .locations(&target)
            .iter()
            .map(|location| link.address(location))
            .collect()
    }

    /// Where `address`'s transport is in the order of preference (none of ours: last).
    pub(crate) fn preference(&self, address: &str) -> usize {
        let scheme = scheme_of(address);
        self.links
            .iter()
            .position(|link| link.scheme == scheme)
            .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)
    }

    /// Our address on the transport of `address`, as `address` names it there.
    pub(crate) fn own_address(&self, address: &str) -> Option<String> {
        let scheme = scheme_of(address);
        self.links
            .iter()
            .find(|link| link.scheme == scheme)
            .map(Link::own_address)
    }

    /// The longest of the transports': until then, something may still be on its way.
    pub(crate) fn delivery_limit(&self) -> Duration {
        self.links
            .iter()
            .map(|link| link.inner.delivery_limit())
            .max()
            .unwrap_or_default()
    }

    /// Each transport's (`Transport::local_addresses`), with its scheme.
    pub(crate) fn local_addresses(&self) -> Vec<String> {
        self.links
            .iter()
            .flat_map(|link| {
                link.inner
                    .local_addresses()
                    .into_iter()
                    .map(|target| link.address(&target))
            })
            .collect()
    }

    /// The schemes of the transports, in order of preference.
    pub(crate) fn transports(&self) -> Vec<String> {
        self.links
            .iter()
            .map(|link| link.scheme.to_owned())
            .collect()
    }

    /// Each transport's, with its scheme.
    pub(crate) fn discovered_seeds(&self) -> Vec<String> {
        self.links
            .iter()
            .flat_map(|link| {
                link.inner
                    .discovered_seeds()
                    .into_iter()
                    .map(|seed| link.address(&seed))
            })
            .collect()
    }

    /// The address of the node a seed leads to, as paths keep it: an address of ours
    /// (`<scheme>:...`) whose transport reads it at once; with a scheme of ours, what
    /// that transport resolves the rest to (an address the transport can't read as an
    /// identity, e.g. a libp2p multiaddr without its peer ID, still resolves); without
    /// one, what the first transport that can resolves it to.
    pub(crate) async fn resolve(&self, seed: &str) -> Result<String> {
        if let Some(address) = self.canonical(seed) {
            return Ok(address);
        }
        let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| {
            let link = self.links.iter().find(|link| link.scheme == scheme)?;
            Some((link, rest))
        });
        if let Some((link, rest)) = scheme_and_rest {
            let target = link
                .inner
                .resolve_seed(rest)
                .await
                .with_context(|| format!("{}: {seed}", link.scheme))?;
            return Ok(link.address(&target));
        }
        let mut failures = Vec::new();
        for link in &self.links {
            match link.inner.resolve_seed(seed).await {
                Ok(target) => return Ok(link.address(&target)),
                Err(error) => failures.push(format!("{}: {error:#}", link.scheme)),
            }
        }
        Err(anyhow!("{}", failures.join("; ")))
    }

    /// `message` encoded for `address`: `peer` and `address` of the result are
    /// `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)
            .with_context(|| format!("no transport for {address}"))?;
        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
            .split(&message.address)
            .with_context(|| format!("no transport for {}", message.address))?;
        let mut inner = message.clone();
        inner.peer = target.clone();
        inner.address = target;
        link.inner.send_encoded(&inner).await
    }

    /// How the transport of `address` is connected there (`Transport::connection`).
    pub(crate) fn connection(&self, address: &str) -> Option<Connection> {
        let (link, target) = self.split(address)?;
        link.inner.connection(&target)
    }

    /// Every transport's figures together.
    pub(crate) fn stats(&self) -> TransportStats {
        let mut stats = TransportStats::default();
        for link in &self.links {
            stats.merge(link.inner.stats());
        }
        stats
    }

    /// Every transport's consumer, each message handed to `handler` under the sender's
    /// address; returns when the first one does.
    pub(crate) async fn start_consumer(&self, handler: Arc<PathHandler>) -> Result<()> {
        let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
            Vec::new();
        for link in &self.links {
            let inner = Arc::clone(&link.inner);
            let scheme = link.scheme;
            let handler = Arc::clone(&handler);
            consumers.push(Box::pin(async move {
                inner
                    .start_consumer(Box::new(move |sender, message, payload_len| {
                        handler(format!("{scheme}:{sender}"), message, payload_len)
                    }))
                    .await
            }));
        }
        first(consumers)
            .await
            .unwrap_or_else(|| Err(anyhow!("no transports")))
    }

    /// When the first transport exits.
    pub(crate) async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
        let waits = self
            .links
            .iter()
            .map(|link| {
                let inner = Arc::clone(&link.inner);
                Box::pin(async move { inner.wait_for_exit().await })
                    as std::pin::Pin<Box<dyn Future<Output = _> + Send>>
            })
            .collect();
        first(waits)
            .await
            .unwrap_or_else(|| Err(std::io::Error::other("no transports")))
    }
}

/// `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>>>,
) -> Option<T> {
    let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
    for task in tasks {
        let sender = sender.clone();
        tokio::spawn(async move {
            let _ = sender.send(task.await).await;
        });
    }
    drop(sender);
    receiver.recv().await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::document::replica::Epoch;
    use crate::transport::sim::SimNet;
    use crate::transport::{HeadsSummary, Status};

    /// "a" on two sim nets, as `sim` (preferred) and `other`; "b" on both too.
    fn two_nets() -> (Paths, Paths) {
        let (first, second) = (SimNet::new(1), SimNet::new(2));
        let on_both = |id: &str| {
            Paths::new(vec![
                ("sim", first.transport(id) as Arc<dyn Transport>),
                ("other", second.transport(id) as Arc<dyn Transport>),
            ])
        };
        (on_both("a"), on_both("b"))
    }

    #[test]
    fn addresses_by_scheme() {
        let (a, _b) = two_nets();
        assert_eq!(a.local_addresses(), ["sim:a", "other:a"]);
        assert_eq!(a.transports(), ["sim", "other"]);
        assert_eq!(a.canonical("other:b").as_deref(), Some("other:b"));
        assert_eq!(a.canonical("pigeon:b"), None);
        assert_eq!(a.canonical("b"), None); // No scheme.
        assert!(a.preference("sim:b") < a.preference("other:b"));
        assert_eq!(a.preference("pigeon:b"), usize::MAX);
        assert!(a.is_own("other:a") && !a.is_own("other:b"));
        assert_eq!(a.own_address("other:b").as_deref(), Some("other:a"));
        assert_eq!(a.own_address("pigeon:b"), None);
    }

    #[tokio::test]
    async fn seeds_resolve_to_addresses() {
        let (a, _b) = two_nets();
        assert_eq!(a.resolve("other:b").await.unwrap(), "other:b");
        // Without a scheme: the first transport that resolves it.
        assert_eq!(a.resolve("b").await.unwrap(), "sim:b");
        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]
    async fn messages_arrive_under_the_senders_address() {
        let (a, b) = two_nets();
        let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
        let handler: Arc<PathHandler> = Arc::new(move |from, message, _| {
            let sender = sender.clone();
            Box::pin(async move {
                let _ = sender.send((from, message.kind()));
                Ok(())
            })
        });
        tokio::spawn(async move { b.start_consumer(handler).await });
        let status = some_status();
        for to in ["other:b", "sim:b"] {
            let encoded = a.encode(to, &status).unwrap();
            assert_eq!(encoded.address, to);
            a.send(&encoded).await.unwrap();
        }
        let mut from: Vec<String> = Vec::new();
        for _ in 0..2 {
            from.push(received.recv().await.unwrap().0);
        }
        from.sort();
        assert_eq!(from, ["other:a", "sim:a"]);
        assert!(a.encode("pigeon:b", &status).is_err());
    }
}