diff options
Diffstat (limited to 'src/transport/paths.rs')
| -rw-r--r-- | src/transport/paths.rs | 364 |
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()); + } +} |