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