introduce paths layer (split off identity layer)

1 files changed, 63 insertions, 250 deletions

diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 17c7795..7633ac1 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -22,22 +22,22 @@
 //!   with only hints, those are asked WHO, and the send is refused until one answers
 //!   (the node retries refused sends).
 //!
-//! Addresses are `<scheme>:<transport identity>`, e.g. `mycelium:<public key>`, or
-//! `<scheme>:<location>` where a transport needs more than an identity to reach a node
-//! (`Transport::locate`, `Transport::locations`); the books keep the identities.
+//! Addresses and the transports behind them are the paths layer's
+//! (`transport::paths`): the books keep addresses as paths keep them.
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
 
-use anyhow::{Context, Result, anyhow, bail};
+use anyhow::{Result, anyhow, bail};
 use async_trait::async_trait;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::oneshot;
 use tokio::time::Instant;
 use tracing::{debug, info};
 
 use crate::identity::{self, NodeKey};
 use crate::locks::MutexExt;
+use crate::transport::paths::{PathHandler, Paths, scheme_of};
 use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats};
 use crate::transport::traffic::Traffic;
 use crate::transport::{
@@ -94,25 +94,8 @@ pub(crate) struct IdentityTransport {
     shared: Arc<Shared>,
 }
 
-/// 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())
-    }
-}
-
 struct Shared {
-    /// In order of preference.
-    links: Vec<Link>,
+    paths: Paths,
     key: Arc<NodeKey>,
     node_id: String,
     /// WHO and ME are counted here, under the address (`traffic`).
@@ -138,11 +121,6 @@ struct Book {
     counts: IdentityCounts,
 }
 
-/// The scheme of `address`.
-fn scheme_of(address: &str) -> &str {
-    address.split_once(':').map_or("", |(scheme, _)| scheme)
-}
-
 impl Book {
     /// `node` proved itself at `address`. Returns whether that is news.
     fn bind(&mut self, address: &str, node: &str, now: Instant) -> bool {
@@ -236,13 +214,14 @@ impl IdentityTransport {
         traffic: Arc<Traffic>,
     ) -> Self {
         let node_id = config.key.id();
-        let links = std::iter::once((config.scheme, transport))
-            .chain(config.more)
-            .map(|(scheme, inner)| Link { scheme, inner })
-            .collect();
+        let paths = Paths::new(
+            std::iter::once((config.scheme, transport))
+                .chain(config.more)
+                .collect(),
+        );
         Self {
             shared: Arc::new(Shared {
-                links,
+                paths,
                 key: config.key,
                 node_id,
                 traffic,
@@ -257,66 +236,19 @@ impl Shared {
         self.book.lock_or_recover()
     }
 
-    /// 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))
-    }
-
-    /// `address` as the books keep it: `<scheme>:<transport identity>`.
-    fn canonical(&self, address: &str) -> Option<String> {
-        self.split(address)
-            .map(|(link, target)| link.address(&target))
-    }
-
-    /// Where the node at `address` (as the books keep it) is, as others can use it
-    /// (`Transport::locations`).
-    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).
-    fn preference(&self, address: &str) -> usize {
-        let scheme = scheme_of(address);
-        self.links
-            .iter()
-            .position(|link| link.scheme == scheme)
-            .unwrap_or(usize::MAX)
-    }
-
-    fn is_own(&self, address: &str) -> bool {
-        self.links.iter().any(|link| link.own_address() == address)
-    }
-
-    fn proof(&self, label: &[u8], link: &Link, target: &str) -> IdentityProof {
-        let message = signed(label, &link.own_address(), &link.address(target));
-        IdentityProof {
+    /// Our proof for `address` (as paths keep it), from our address on its transport.
+    fn proof(&self, label: &[u8], address: &str) -> Option<IdentityProof> {
+        let message = signed(label, &self.paths.own_address(address)?, address);
+        Some(IdentityProof {
             key: self.key.public(),
             signature: self.key.sign(&message),
-        }
+        })
     }
 
-    /// Notes the node `proof` proves to be at `address` (on `link`); None if it proves
-    /// nothing.
-    fn prove(
-        &self,
-        link: &Link,
-        address: &str,
-        label: &[u8],
-        proof: &IdentityProof,
-    ) -> Option<String> {
-        let message = signed(label, address, &link.own_address());
+    /// Notes the node `proof` proves to be at `address` (as paths keep it); None if it
+    /// proves nothing.
+    fn prove(&self, address: &str, label: &[u8], proof: &IdentityProof) -> Option<String> {
+        let message = signed(label, address, &self.paths.own_address(address)?);
         if !identity::verify(&proof.key, &message, &proof.signature) {
             self.lock().counts.proofs_rejected += 1;
             debug!(from = %address, "Ignoring an identity proof that doesn't hold");
@@ -345,22 +277,26 @@ impl Shared {
 
     /// Sends a WHO or ME (`kind`) to `address`.
     async fn send_proof(&self, kind: MessageKind, address: &str) {
-        let Some((link, target)) = self.split(address) else {
+        let Some(address) = self.paths.canonical(address) else {
             return;
         };
+        let address = address.as_str();
         let message = if kind == MessageKind::Who {
-            ParsedMessage::Who(self.proof(WHO_LABEL, link, &target))
+            self.proof(WHO_LABEL, address).map(ParsedMessage::Who)
         } else {
-            ParsedMessage::Me(self.proof(ME_LABEL, link, &target))
+            self.proof(ME_LABEL, address).map(ParsedMessage::Me)
         };
-        let encoded = match link.inner.encode(&target, &message) {
+        let Some(message) = message else {
+            return;
+        };
+        let encoded = match self.paths.encode(address, &message) {
             Ok(encoded) => encoded,
             Err(error) => {
                 debug!(to = %address, error = %format!("{error:#}"), "Couldn't encode an identity proof");
                 return;
             }
         };
-        match link.inner.send_encoded(&encoded).await {
+        match self.paths.send(&encoded).await {
             Ok(()) => self
                 .traffic
                 .sent(address, encoded.kind, encoded.payload_len),
@@ -372,22 +308,19 @@ impl Shared {
         }
     }
 
-    /// A message from `sender` on `links[index]`.
+    /// A message from `address`.
     async fn receive(
         self: Arc<Self>,
-        index: usize,
-        sender: String,
+        address: String,
         message: ParsedMessage,
         payload_len: usize,
         handler: &HandlerFn,
     ) -> Result<()> {
-        let link = &self.links[index];
-        let address = link.address(&sender);
         match message {
             ParsedMessage::Who(proof) => {
                 self.traffic
                     .received(&address, MessageKind::Who, payload_len);
-                if self.prove(link, &address, WHO_LABEL, &proof).is_some() {
+                if self.prove(&address, WHO_LABEL, &proof).is_some() {
                     let shared = Arc::clone(&self);
                     tokio::spawn(async move { shared.send_proof(MessageKind::Me, &address).await });
                 }
@@ -396,7 +329,7 @@ impl Shared {
             ParsedMessage::Me(proof) => {
                 self.traffic
                     .received(&address, MessageKind::Me, payload_len);
-                self.prove(link, &address, ME_LABEL, &proof);
+                self.prove(&address, ME_LABEL, &proof);
                 Ok(())
             }
             message => {
@@ -425,29 +358,14 @@ impl Shared {
         let pick = |addresses: Option<&Vec<String>>| {
             addresses?
                 .iter()
-                .filter(|address| self.split(address).is_some())
-                .min_by_key(|address| self.preference(address))
+                .filter(|address| self.paths.usable(address))
+                .min_by_key(|address| self.paths.preference(address))
                 .cloned()
         };
         pick(book.proven.get(peer)).or_else(|| pick(book.hints.get(peer)))
     }
 }
 
-/// 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) = 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
-}
-
 #[async_trait]
 impl Transport for IdentityTransport {
     fn local_id(&self) -> &str {
@@ -458,28 +376,12 @@ impl Transport for IdentityTransport {
         identity::is_node_id(id)
     }
 
-    /// The longest of the transports': until then, something may still be on its way.
     fn delivery_limit(&self) -> Duration {
-        self.shared
-            .links
-            .iter()
-            .map(|link| link.inner.delivery_limit())
-            .max()
-            .unwrap_or_default()
+        self.shared.paths.delivery_limit()
     }
 
-    /// Each transport's (`Transport::local_addresses`), with its scheme.
     fn local_addresses(&self) -> Vec<String> {
-        self.shared
-            .links
-            .iter()
-            .flat_map(|link| {
-                link.inner
-                    .local_addresses()
-                    .into_iter()
-                    .map(|target| link.address(&target))
-            })
-            .collect()
+        self.shared.paths.local_addresses()
     }
 
     /// Where `peer` proved itself, else where it is said to be, as others can use it.
@@ -494,7 +396,7 @@ impl Transport for IdentityTransport {
         };
         let mut locations = Vec::new();
         for address in addresses {
-            for location in self.shared.locations(&address) {
+            for location in self.shared.paths.locations(&address) {
                 if !locations.contains(&location) {
                     locations.push(location);
                 }
@@ -509,8 +411,8 @@ impl Transport for IdentityTransport {
         }
         let mut known: Vec<String> = Vec::new();
         for address in addresses {
-            if let Some(address) = self.shared.canonical(address)
-                && !self.shared.is_own(&address)
+            if let Some(address) = self.shared.paths.canonical(address)
+                && !self.shared.paths.is_own(&address)
                 && !known.contains(&address)
             {
                 known.push(address);
@@ -531,11 +433,7 @@ impl Transport for IdentityTransport {
     }
 
     fn transports(&self) -> Vec<String> {
-        self.shared
-            .links
-            .iter()
-            .map(|link| link.scheme.to_owned())
-            .collect()
+        self.shared.paths.transports()
     }
 
     /// The address `encode` picks, and the transport's connection there.
@@ -547,10 +445,7 @@ impl Transport for IdentityTransport {
             .proven
             .get(peer)
             .is_some_and(|addresses| addresses.contains(&address));
-        let connection = self
-            .shared
-            .split(&address)
-            .and_then(|(link, target)| link.inner.connection(&target));
+        let connection = self.shared.paths.connection(&address);
         Some(Destination {
             address,
             proven,
@@ -560,28 +455,16 @@ impl Transport for IdentityTransport {
 
     /// The identity layer's, and each transport's.
     fn stats(&self) -> TransportStats {
-        let mut stats = TransportStats {
+        let mut stats = self.shared.paths.stats();
+        stats.merge(TransportStats {
             identity: Some(self.shared.lock().stats()),
             ..TransportStats::default()
-        };
-        for link in &self.shared.links {
-            stats.merge(link.inner.stats());
-        }
+        });
         stats
     }
 
-    /// Each transport's, with its scheme.
     fn discovered_seeds(&self) -> Vec<String> {
-        self.shared
-            .links
-            .iter()
-            .flat_map(|link| {
-                link.inner
-                    .discovered_seeds()
-                    .into_iter()
-                    .map(|seed| link.address(&seed))
-            })
-            .collect()
+        self.shared.paths.discovered_seeds()
     }
 
     /// A node ID resolves to itself if this node knows where it is; anything else is
@@ -594,45 +477,8 @@ impl Transport for IdentityTransport {
             }
             bail!("a node ID alone doesn't say where that node is; give its address");
         }
-        // With a scheme of ours, only that transport resolves the rest (an address the
-        // transport can't read as an identity, e.g. a libp2p multiaddr without its peer
-        // ID, still resolves); without one, the first transport that can.
-        let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| {
-            let link = self
-                .shared
-                .links
-                .iter()
-                .find(|link| link.scheme == scheme)?;
-            Some((link, rest))
-        });
-        let address = match self.shared.canonical(seed) {
-            Some(address) => address,
-            None => match scheme_and_rest {
-                Some((link, rest)) => {
-                    let target = link
-                        .inner
-                        .resolve_seed(rest)
-                        .await
-                        .with_context(|| format!("{}: {seed}", link.scheme))?;
-                    link.address(&target)
-                }
-                None => {
-                    let mut failures = Vec::new();
-                    let mut resolved = None;
-                    for link in &self.shared.links {
-                        match link.inner.resolve_seed(seed).await {
-                            Ok(target) => {
-                                resolved = Some(link.address(&target));
-                                break;
-                            }
-                            Err(error) => failures.push(format!("{}: {error:#}", link.scheme)),
-                        }
-                    }
-                    resolved.ok_or_else(|| anyhow!("{}", failures.join("; ")))?
-                }
-            },
-        };
-        if self.shared.is_own(&address) {
+        let address = self.shared.paths.resolve(seed).await?;
+        if self.shared.paths.is_own(&address) {
             bail!("{seed} is this node");
         }
         let receiver = {
@@ -667,12 +513,8 @@ impl Transport for IdentityTransport {
         let Some(address) = self.shared.destination(peer) else {
             bail!("no address known for {peer} on this node's transports");
         };
-        let Some((link, target)) = self.shared.split(&address) else {
-            bail!("no transport for {address}");
-        };
-        let mut encoded = link.inner.encode(&target, message)?;
+        let mut encoded = self.shared.paths.encode(&address, message)?;
         encoded.peer = peer.to_owned();
-        encoded.address = address;
         Ok(encoded)
     }
 
@@ -696,58 +538,29 @@ impl Transport for IdentityTransport {
             }
             return Err(NotProven(message.peer.clone()).into());
         }
-        let Some((link, target)) = self.shared.split(&message.address) else {
-            bail!("no transport for {}", message.address);
-        };
-        let mut inner = message.clone();
-        inner.peer = target.clone();
-        inner.address = target;
-        link.inner.send_encoded(&inner).await
+        self.shared.paths.send(message).await
     }
 
-    /// Every transport's consumer; returns when the first one does.
+    /// Every transport's consumer (`Paths::start_consumer`); returns when the first
+    /// one does.
     async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
         let handler: Arc<HandlerFn> = Arc::from(handler);
-        let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
-            Vec::new();
-        for (index, link) in self.shared.links.iter().enumerate() {
-            let inner = Arc::clone(&link.inner);
-            let shared = Arc::clone(&self.shared);
+        let shared = Arc::clone(&self.shared);
+        let receive: Arc<PathHandler> = Arc::new(move |address, message, payload_len| {
+            let shared = Arc::clone(&shared);
             let handler = Arc::clone(&handler);
-            consumers.push(Box::pin(async move {
-                inner
-                    .start_consumer(Box::new(move |sender, message, payload_len| {
-                        let shared = Arc::clone(&shared);
-                        let handler = Arc::clone(&handler);
-                        Box::pin(async move {
-                            shared
-                                .receive(index, sender, message, payload_len, &*handler)
-                                .await
-                        })
-                    }))
+            Box::pin(async move {
+                shared
+                    .receive(address, message, payload_len, &*handler)
                     .await
-            }));
-        }
-        first(consumers)
-            .await
-            .unwrap_or_else(|| Err(anyhow!("no transports")))
+            })
+        });
+        self.shared.paths.start_consumer(receive).await
     }
 
     /// When the first transport exits.
     async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
-        let waits = self
-            .shared
-            .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")))
+        self.shared.paths.wait_for_exit().await
     }
 }