introduce paths layer (split off identity layer)

1 files changed, 364 insertions, 0 deletions

diff --git a/src/transport/paths.rs b/src/transport/paths.rs
new file mode 100644
index 0000000..d1a4dc9
--- /dev/null
+++ b/src/transport/paths.rs
@@ -0,0 +1,364 @@
+//! 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`).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{Context, Result, anyhow};
+
+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)
+}
+
+pub(crate) struct Paths {
+    /// In order of preference.
+    links: Vec<Link>,
+}
+
+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(),
+        }
+    }
+
+    /// 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)
+    }
+
+    /// 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`.
+    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();
+        Ok(encoded)
+    }
+
+    /// 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")))
+    }
+}
+
+/// 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 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 = ParsedMessage::Status(Status {
+            heads: HeadsSummary::of(Vec::new()),
+            epoch: Epoch::GENESIS,
+            epoch_seq: 0,
+            ready: false,
+            syncs_with_you: false,
+        });
+        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());
+    }
+}