summaryrefslogtreecommitdiffstats
path: root/src/transport/identity.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/identity.rs')
-rw-r--r--src/transport/identity.rs569
1 files changed, 296 insertions, 273 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index af0fe74..c597d77 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -17,36 +17,45 @@
//! * A seed (`resolve_seed`) is an address, or something one of the transports resolves
//! to one; it is asked WHO, and its ME says which node is there.
//! * Where other nodes are comes from the member list and HELLO_REPLYs
-//! (`learn_addresses`): hints, not proofs. A message to a node goes to an address it
-//! proved itself at (on the first transport in the configured order that has one);
-//! with only hints, those are asked WHO, and the send is refused until one answers
-//! (the node retries refused sends).
+//! (`learn_addresses`): hints, not proofs. A message to a node goes to the addresses
+//! it proved itself at, best first (`Paths::rank`, `Paths::send_first`: the next
+//! when a send fails); with only hints, those are asked WHO, and the send is refused
+//! until one answers (the node retries refused sends). A node proven on some
+//! transports is asked WHO at its hints on the others too, now and then
+//! (`OTHER_PATHS_ASK_INTERVAL`), so that each transport both have becomes a path.
//!
-//! 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::firewall::Firewall;
+use crate::transport::paths::{Gate, PathHandler, Paths, scheme_of};
+use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats};
use crate::transport::traffic::Traffic;
use crate::transport::{
- EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind, ParsedMessage, Transport,
+ Destination, EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind,
+ ParsedMessage, Transport,
};
pub(crate) const WHO_LABEL: &[u8] = b"continuum/who/1";
pub(crate) const ME_LABEL: &[u8] = b"continuum/me/1";
/// An address is asked WHO at most this often, unless a seed is being resolved.
const ASK_INTERVAL: Duration = Duration::from_secs(5);
+/// A node's addresses on transports it hasn't proven itself on are asked WHO at most
+/// this often (`Book::unproven_hints`): over Tor, a WHO is a dial that takes seconds,
+/// and an address that doesn't answer shouldn't be pressed.
+const OTHER_PATHS_ASK_INTERVAL: Duration = Duration::from_secs(60);
/// How long resolving a seed waits for its ME.
const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
/// Addresses kept per node from hints: about one per transport.
@@ -92,25 +101,10 @@ 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: Arc<Paths>,
+ /// The debug firewall, by node ID; paths applies it (`Gate`).
+ firewall: Option<Arc<Firewall>>,
key: Arc<NodeKey>,
node_id: String,
/// WHO and ME are counted here, under the address (`traffic`).
@@ -132,11 +126,8 @@ struct Book {
asked: HashMap<String, Instant>,
/// Seeds being resolved: who waits for a ME from where.
waiting: HashMap<String, Vec<oneshot::Sender<String>>>,
-}
-
-/// The scheme of `address`.
-fn scheme_of(address: &str) -> &str {
- address.split_once(':').map_or("", |(scheme, _)| scheme)
+ /// Since start (`stats`).
+ counts: IdentityCounts,
}
impl Book {
@@ -151,6 +142,7 @@ impl Book {
if *slot != address {
let previous = std::mem::replace(slot, address.to_owned());
self.at.remove(&previous);
+ self.counts.address_changes += 1;
}
} else {
addresses.push(address.to_owned());
@@ -185,19 +177,63 @@ impl Book {
}
}
+ /// What the books hold now, and the counts.
+ fn stats(&self) -> IdentityStats {
+ let mut proven = std::collections::BTreeMap::new();
+ for addresses in self.proven.values() {
+ for address in addresses {
+ *proven.entry(scheme_of(address).to_owned()).or_insert(0) += 1;
+ }
+ }
+ IdentityStats {
+ proven,
+ hinted: self
+ .hints
+ .keys()
+ .filter(|node| !self.proven.contains_key(*node))
+ .count(),
+ counts: self.counts,
+ }
+ }
+
+ /// Where `node` is said to be on transports it hasn't proven itself on yet.
+ fn unproven_hints(&self, node: &str) -> Vec<String> {
+ let proven = self.proven.get(node);
+ self.hints
+ .get(node)
+ .into_iter()
+ .flatten()
+ .filter(|hint| {
+ proven.is_none_or(|proven| {
+ !proven
+ .iter()
+ .any(|address| scheme_of(address) == scheme_of(hint))
+ })
+ })
+ .cloned()
+ .collect()
+ }
+
/// Whether to ask `address` WHO now (always, if `force`), noting that it was asked.
fn ask(&mut self, address: &str, now: Instant, force: bool) -> bool {
- if !force
- && self
- .asked
- .get(address)
- .is_some_and(|at| now.duration_since(*at) < ASK_INTERVAL)
+ let interval = if force { Duration::ZERO } else { ASK_INTERVAL };
+ self.ask_every(address, now, interval)
+ }
+
+ /// Whether to ask `address` WHO now, if it wasn't within `interval`, noting that it
+ /// was asked.
+ fn ask_every(&mut self, address: &str, now: Instant, interval: Duration) -> bool {
+ if self
+ .asked
+ .get(address)
+ .is_some_and(|at| now.duration_since(*at) < interval)
{
return false;
}
if self.asked.len() >= MAX_ENTRIES {
+ // The longest interval: other paths' asks are kept until they are due again.
self.asked
- .retain(|_, at| now.duration_since(*at) < ASK_INTERVAL);
+ .retain(|_, at| now.duration_since(*at) < OTHER_PATHS_ASK_INTERVAL);
}
self.asked.insert(address.to_owned(), now);
true
@@ -206,94 +242,77 @@ impl Book {
impl IdentityTransport {
/// Over `transport`, whose addresses have `config.scheme`, and `config.more`.
+ /// `firewall`: the debug firewall, if any, which paths applies by address.
pub(crate) fn new(
transport: Arc<dyn Transport>,
config: IdentityConfig,
traffic: Arc<Traffic>,
+ firewall: Option<Arc<Firewall>>,
) -> 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();
- Self {
- shared: Arc::new(Shared {
- links,
+ let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| {
+ let gate: std::sync::Weak<dyn Gate> = shared.clone();
+ Shared {
+ paths: Arc::new(Paths::new(
+ std::iter::once((config.scheme, transport))
+ .chain(config.more)
+ .collect(),
+ Some(Arc::clone(&traffic)),
+ firewall.is_some().then_some(gate),
+ )),
+ firewall,
key: config.key,
node_id,
traffic,
book: Mutex::default(),
- }),
- }
+ }
+ });
+ Self { shared }
}
}
-impl Shared {
- fn lock(&self) -> MutexGuard<'_, Book> {
- 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))
+/// The firewall's rules are by node ID: an address is blocked if the node proven at it
+/// is. Addresses no node proved itself at aren't (nothing could name them).
+impl Gate for Shared {
+ fn blocks_inbound(&self, address: &str) -> bool {
+ self.blocks(address, Firewall::blocks_inbound)
}
- /// `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))
+ fn blocks_outbound(&self, address: &str) -> bool {
+ self.blocks(address, Firewall::blocks_outbound)
}
+}
- /// 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();
+impl Shared {
+ fn blocks(&self, address: &str, rule: fn(&Firewall, &str) -> bool) -> bool {
+ let Some(firewall) = &self.firewall else {
+ return false;
};
- 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)
+ let node = self.lock().at.get(address).map(|(node, _)| node.clone());
+ node.is_some_and(|node| rule(firewall, &node))
}
+}
- fn is_own(&self, address: &str) -> bool {
- self.links.iter().any(|link| link.own_address() == address)
+impl Shared {
+ fn lock(&self) -> MutexGuard<'_, Book> {
+ self.book.lock_or_recover()
}
- 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");
return None;
}
@@ -313,6 +332,27 @@ impl Shared {
if !self.lock().ask(address, Instant::now(), force) {
return;
}
+ self.send_who(address);
+ }
+
+ /// Asks `node`'s addresses on transports it hasn't proven itself on WHO, now and then
+ /// (`OTHER_PATHS_ASK_INTERVAL`): its ME there proves another path to it, which the
+ /// paths layer then measures, and fails over to.
+ fn ask_other_paths(self: &Arc<Self>, node: &str) {
+ let due: Vec<String> = {
+ let mut book = self.lock();
+ let now = Instant::now();
+ book.unproven_hints(node)
+ .into_iter()
+ .filter(|address| book.ask_every(address, now, OTHER_PATHS_ASK_INTERVAL))
+ .collect()
+ };
+ for address in due {
+ self.send_who(&address);
+ }
+ }
+
+ fn send_who(self: &Arc<Self>, address: &str) {
let shared = Arc::clone(self);
let address = address.to_owned();
tokio::spawn(async move { shared.send_proof(MessageKind::Who, &address).await });
@@ -320,22 +360,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 Some(message) = message else {
+ return;
};
- let encoded = match link.inner.encode(&target, &message) {
+ 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),
@@ -347,22 +391,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 });
}
@@ -371,7 +412,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 => {
@@ -379,6 +420,7 @@ impl Shared {
match node {
Some(node) => handler(node, message, payload_len).await,
None => {
+ self.lock().counts.dropped_unproven += 1;
debug!(
from = %address,
kind = ?message.kind(),
@@ -395,31 +437,23 @@ impl Shared {
/// Where to send to `peer`: an address it proved itself at, else one it is said to
/// be at; the most preferred transport's first.
fn destination(&self, peer: &str) -> Option<String> {
- let book = self.lock();
- let pick = |addresses: Option<&Vec<String>>| {
- addresses?
- .iter()
- .filter(|address| self.split(address).is_some())
- .min_by_key(|address| self.preference(address))
- .cloned()
- };
- pick(book.proven.get(peer)).or_else(|| pick(book.hints.get(peer)))
+ self.proven_at(peer)
+ .into_iter()
+ .next()
+ .or_else(|| self.hinted_at(peer).into_iter().next())
}
-}
-/// 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;
- });
+ /// The addresses `peer` proved itself at, best first (`Paths::rank`).
+ fn proven_at(&self, peer: &str) -> Vec<String> {
+ let proven = self.lock().proven.get(peer).cloned().unwrap_or_default();
+ self.paths.rank(&proven)
+ }
+
+ /// Where `peer` is said to be, best first.
+ fn hinted_at(&self, peer: &str) -> Vec<String> {
+ let hints = self.lock().hints.get(peer).cloned().unwrap_or_default();
+ self.paths.rank(&hints)
}
- drop(sender);
- receiver.recv().await
}
#[async_trait]
@@ -432,28 +466,26 @@ 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()
+ }
+
+ fn identities_of(&self, peer: &str) -> Vec<String> {
+ if peer == self.shared.node_id {
+ return self.shared.paths.own_addresses();
+ }
+ let book = self.shared.lock();
+ let mut addresses = book.proven.get(peer).cloned().unwrap_or_default();
+ for hint in book.hints.get(peer).into_iter().flatten() {
+ if !addresses.contains(hint) {
+ addresses.push(hint.clone());
+ }
+ }
+ addresses
}
/// Where `peer` proved itself, else where it is said to be, as others can use it.
@@ -468,7 +500,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);
}
@@ -483,8 +515,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);
@@ -505,25 +537,54 @@ impl Transport for IdentityTransport {
}
fn transports(&self) -> Vec<String> {
- self.shared
- .links
- .iter()
- .map(|link| link.scheme.to_owned())
- .collect()
+ self.shared.paths.transports()
+ }
+
+ /// Each transport those with its scheme, without it (`Paths::learn_relays`).
+ fn learn_relays(&self, addresses: &[String]) {
+ self.shared.paths.learn_relays(addresses);
+ }
+
+ /// The address `encode` picks, the transport's connection there, and every path
+ /// to `peer` as paths measures it, best first.
+ fn destination(&self, peer: &str) -> Option<Destination> {
+ let proven = self.shared.proven_at(peer);
+ let (address, is_proven) = match proven.first() {
+ Some(address) => (address.clone(), true),
+ None => (self.shared.hinted_at(peer).into_iter().next()?, false),
+ };
+ let connection = self.shared.paths.connection(&address);
+ Some(Destination {
+ address: self.shared.paths.location(&address),
+ proven: is_proven,
+ connection,
+ paths: self.shared.paths.info(&proven),
+ })
+ }
+
+ /// The paths layer answers them (`Paths::receive`).
+ fn answers_pings(&self) -> bool {
+ true
+ }
+
+ /// The best path's, where messages to `peer` go.
+ fn rtt(&self, peer: &str) -> Option<Duration> {
+ let address = self.shared.proven_at(peer).into_iter().next()?;
+ self.shared.paths.rtt(&address)
+ }
+
+ /// The identity layer's, and each transport's.
+ fn stats(&self) -> TransportStats {
+ let mut stats = self.shared.paths.stats();
+ stats.merge(TransportStats {
+ identity: Some(self.shared.lock().stats()),
+ ..TransportStats::default()
+ });
+ 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
@@ -536,45 +597,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 = {
@@ -594,9 +618,12 @@ impl Transport for IdentityTransport {
info!(seed = %seed, node = %node, "Seed is node");
Ok(node)
}
- _ => Err(anyhow!(
- "no node answered at {address} within {RESOLVE_TIMEOUT:?}"
- )),
+ _ => {
+ self.shared.lock().counts.seeds_unanswered += 1;
+ Err(anyhow!(
+ "no node answered at {address} within {RESOLVE_TIMEOUT:?}"
+ ))
+ }
}
}
@@ -606,86 +633,56 @@ 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)
}
+ /// To the addresses `peer` proved itself at, best first, the next when a send
+ /// fails (`Paths::send_first`); refused if it proved itself nowhere yet.
async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
- let hints = {
- let book = self.shared.lock();
- let proven = book
- .proven
- .get(&message.peer)
- .is_some_and(|addresses| addresses.contains(&message.address));
- if proven {
- None
- } else {
- Some(book.hints.get(&message.peer).cloned().unwrap_or_default())
- }
- };
- if let Some(hints) = hints {
+ let proven = self.shared.proven_at(&message.peer);
+ if !proven.is_empty() {
+ self.shared.ask_other_paths(&message.peer);
+ }
+ if proven.is_empty() {
+ let hints = {
+ let mut book = self.shared.lock();
+ book.counts.refused_unproven += 1;
+ book.hints.get(&message.peer).cloned().unwrap_or_default()
+ };
for address in hints {
self.shared.ask(&address, false);
}
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_first(message, &proven)
+ .await
+ .map(|_| ())
}
- /// 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
}
}
@@ -721,6 +718,32 @@ mod tests {
assert!(book.bind("s:y", "n2", now));
assert_eq!(book.proven["n1"], ["t:x"]);
assert_eq!(book.at["s:y"].0, "n2");
+ // Only n1's move from s:x to s:y changed an address.
+ assert_eq!(book.counts.address_changes, 1);
+ let stats = book.stats();
+ assert_eq!(stats.proven["s"], 1);
+ assert_eq!(stats.proven["t"], 1);
+ }
+
+ #[test]
+ fn other_paths_are_the_hints_on_transports_not_proven() {
+ let now = Instant::now();
+ let mut book = Book::default();
+ book.hints
+ .insert("n1".into(), vec!["s:x".into(), "t:x".into(), "u:x".into()]);
+ // Proven nowhere: every hint.
+ assert_eq!(book.unproven_hints("n1"), ["s:x", "t:x", "u:x"]);
+ // Proven on s (even at another address): the others.
+ book.bind("s:y", "n1", now);
+ assert_eq!(book.unproven_hints("n1"), ["t:x", "u:x"]);
+ assert!(book.ask_every("t:x", now, OTHER_PATHS_ASK_INTERVAL));
+ assert!(!book.ask_every("t:x", now + ASK_INTERVAL, OTHER_PATHS_ASK_INTERVAL));
+ assert!(book.ask_every(
+ "t:x",
+ now + OTHER_PATHS_ASK_INTERVAL,
+ OTHER_PATHS_ASK_INTERVAL
+ ));
+ assert!(book.unproven_hints("n2").is_empty());
}
#[test]
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-29 02:25:46 +0000