diff options
| -rw-r--r-- | src/transport/identity.rs | 60 | ||||
| -rw-r--r-- | src/transport/mod.rs | 6 | ||||
| -rw-r--r-- | src/transport/mycelium/mod.rs | 1 | ||||
| -rw-r--r-- | src/transport/p2p.rs | 1 | ||||
| -rw-r--r-- | src/transport/paths.rs | 201 |
5 files changed, 232 insertions, 37 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 7633ac1..dd3950b 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -17,10 +17,10 @@ //! * 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). //! //! Addresses and the transports behind them are the paths layer's //! (`transport::paths`): the books keep addresses as paths keep them. @@ -354,15 +354,22 @@ 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.paths.usable(address)) - .min_by_key(|address| self.paths.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 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) } } @@ -518,27 +525,26 @@ impl Transport for IdentityTransport { 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 mut book = self.shared.lock(); - let proven = book - .proven - .get(&message.peer) - .is_some_and(|addresses| addresses.contains(&message.address)); - if proven { - None - } else { + let proven = self.shared.proven_at(&message.peer); + if proven.is_empty() { + let hints = { + let mut book = self.shared.lock(); book.counts.refused_unproven += 1; - Some(book.hints.get(&message.peer).cloned().unwrap_or_default()) - } - }; - if let Some(hints) = hints { + 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()); } - self.shared.paths.send(message).await + self.shared + .paths + .send_first(message, &proven) + .await + .map(|_| ()) } /// Every transport's consumer (`Paths::start_consumer`); returns when the first diff --git a/src/transport/mod.rs b/src/transport/mod.rs index e46e82e..f20110a 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -134,7 +134,8 @@ pub(crate) struct Status { pub(crate) syncs_with_you: bool, } -#[derive(Debug)] +/// Cheap to clone: large parts are shared (`bytes::Bytes`). +#[derive(Debug, Clone)] pub(crate) enum ParsedMessage { Hello(HelloRequest), HelloReply(HelloReply), @@ -537,6 +538,9 @@ pub(crate) struct EncodedMessage { pub(crate) kind: MessageKind, /// Bytes the message takes on the wire (without the transport's own framing). pub(crate) payload_len: usize, + /// The message itself, under the paths layer (`transport::paths`): to encode it + /// again for another transport when a send there fails. None from a transport. + pub(crate) message: Option<std::sync::Arc<ParsedMessage>>, } /// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`). diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index cf979e2..43d14a8 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -450,6 +450,7 @@ pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<Enco body: serde_json::to_vec(&outbound)?.into(), kind: message.kind(), payload_len, + message: None, }) } diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs index 7a92407..30fdb23 100644 --- a/src/transport/p2p.rs +++ b/src/transport/p2p.rs @@ -1360,6 +1360,7 @@ impl Transport for Libp2pTransport { payload_len: body.len(), body: body.into(), kind: message.kind(), + message: None, }) } diff --git a/src/transport/paths.rs b/src/transport/paths.rs index d1a4dc9..3330b2d 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -9,11 +9,22 @@ //! 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`). +//! +//! A message to a node with several addresses goes to the first of them that takes it +//! (`send_first`), in the order `rank` gives: addresses whose last send failed, for +//! `DOWN_FOR`, after the others, and among each the configured order. A send that fails +//! is tried at once at the next address, the message encoded for its transport: when +//! one transport fails, the next carries on within the same send. -use std::sync::Arc; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; use std::time::Duration; use anyhow::{Context, Result, anyhow}; +use tokio::time::Instant; +use tracing::debug; + +use crate::locks::MutexExt; use crate::transport::stats::TransportStats; use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport}; @@ -45,9 +56,16 @@ pub(crate) fn scheme_of(address: &str) -> &str { address.split_once(':').map_or("", |(scheme, _)| scheme) } +/// An address whose send failed is ranked after the others for this long (unless a +/// send there works again). +const DOWN_FOR: Duration = Duration::from_secs(60); + pub(crate) struct Paths { /// In order of preference. links: Vec<Link>, + /// Addresses whose last send failed, and when (`DOWN_FOR`). A std mutex: never held + /// across an await. + down: Mutex<HashMap<String, Instant>>, } impl Paths { @@ -58,6 +76,7 @@ impl Paths { .into_iter() .map(|(scheme, inner)| Link { scheme, inner }) .collect(), + down: Mutex::default(), } } @@ -104,6 +123,29 @@ impl Paths { .unwrap_or(usize::MAX) } + /// The usable ones of `addresses` (as paths keep them), best first: those whose + /// last send failed (within `DOWN_FOR`) last, and by the configured order. + pub(crate) fn rank(&self, addresses: &[String]) -> Vec<String> { + let mut usable: Vec<(bool, usize, &String)> = addresses + .iter() + .filter(|address| self.usable(address)) + .map(|address| (false, self.preference(address), address)) + .collect(); + { + let now = Instant::now(); + let mut down = self.down.lock_or_recover(); + down.retain(|_, at| now.duration_since(*at) < DOWN_FOR); + for (failed, _, address) in &mut usable { + *failed = down.contains_key(*address); + } + } + usable.sort_by_key(|(failed, preference, _)| (*failed, *preference)); + usable + .into_iter() + .map(|(_, _, address)| address.clone()) + .collect() + } + /// Whether `address` is one of ours. pub(crate) fn is_own(&self, address: &str) -> bool { self.links.iter().any(|link| link.own_address() == address) @@ -193,7 +235,7 @@ impl Paths { } /// `message` encoded for `address`: `peer` and `address` of the result are - /// `address`. + /// `address`, and it keeps the message (to encode it for another address). pub(crate) fn encode(&self, address: &str, message: &ParsedMessage) -> Result<EncodedMessage> { let (link, target) = self .split(address) @@ -201,9 +243,67 @@ impl Paths { let mut encoded = link.inner.encode(&target, message)?; encoded.peer = address.to_owned(); encoded.address = address.to_owned(); + encoded.message = Some(Arc::new(message.clone())); Ok(encoded) } + /// Sends `message` to the first of `addresses` (`rank`ed) that takes it: to its own + /// `address` as encoded, to others encoded again for their transport (if it keeps + /// the message). Returns the address it went to; an error with every address's if + /// none took it. + pub(crate) async fn send_first( + &self, + message: &EncodedMessage, + addresses: &[String], + ) -> Result<String> { + let mut failures: Vec<(String, anyhow::Error)> = Vec::new(); + for address in addresses { + let encoded = if *address == message.address { + message.clone() + } else { + let Some(parsed) = &message.message else { + continue; + }; + match self.encode(address, parsed) { + Ok(mut encoded) => { + encoded.peer = message.peer.clone(); + encoded + } + Err(error) => { + failures.push((address.clone(), error)); + continue; + } + } + }; + match self.send(&encoded).await { + Ok(()) => { + self.down.lock_or_recover().remove(address); + if !failures.is_empty() { + debug!( + peer = %message.peer, + address = %address, + failed = %describe(&failures), + "Sent at another address after a send failed" + ); + } + return Ok(address.clone()); + } + Err(error) => { + self.down + .lock_or_recover() + .insert(address.clone(), Instant::now()); + failures.push((address.clone(), error)); + } + } + } + match failures.len() { + 0 => Err(anyhow!("no address to send to")), + // The transport's error as it is. + 1 => Err(failures.pop().expect("one").1), + _ => Err(anyhow!("{}", describe(&failures))), + } + } + /// Sends `message` to its `address`. pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> { let (link, target) = self @@ -269,6 +369,15 @@ impl Paths { } } +/// `address: error; …` for each failed send. +fn describe(failures: &[(String, anyhow::Error)]) -> String { + failures + .iter() + .map(|(address, error)| format!("{address}: {error:#}")) + .collect::<Vec<_>>() + .join("; ") +} + /// 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>>>, @@ -327,6 +436,86 @@ mod tests { assert!(a.resolve("nobody").await.is_err()); } + /// A transport whose sends all fail. + struct Broken; + + #[async_trait::async_trait] + impl Transport for Broken { + fn local_id(&self) -> &str { + "us" + } + fn is_peer_id(&self, id: &str) -> bool { + !id.is_empty() + } + fn delivery_limit(&self) -> Duration { + Duration::from_secs(1) + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { + Ok(seed.to_owned()) + } + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + crate::transport::mycelium::encode_message(peer, message) + } + async fn send_encoded(&self, _message: &EncodedMessage) -> Result<()> { + Err(anyhow!("broken")) + } + async fn start_consumer(&self, _handler: crate::transport::MessageHandler) -> Result<()> { + std::future::pending().await + } + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + std::future::pending().await + } + } + + fn some_status() -> ParsedMessage { + ParsedMessage::Status(Status { + heads: HeadsSummary::of(Vec::new()), + epoch: Epoch::GENESIS, + epoch_seq: 0, + ready: false, + syncs_with_you: false, + }) + } + + /// When the preferred transport fails, the same send goes over the next one, and + /// the failed address is ranked last for a while. + #[tokio::test] + async fn a_failed_send_goes_to_the_next_address() { + let net = SimNet::new(3); + let a = Paths::new(vec![ + ("broken", Arc::new(Broken) as Arc<dyn Transport>), + ("sim", net.transport("a") as Arc<dyn Transport>), + ]); + let b = Paths::new(vec![("sim", net.transport("b") as Arc<dyn Transport>)]); + let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); + let handler: Arc<PathHandler> = Arc::new(move |from, _, _| { + let sender = sender.clone(); + Box::pin(async move { + let _ = sender.send(from); + Ok(()) + }) + }); + tokio::spawn(async move { b.start_consumer(handler).await }); + + let addresses = vec!["sim:b".to_owned(), "broken:b".to_owned()]; + let ranked = a.rank(&addresses); + assert_eq!(ranked, ["broken:b", "sim:b"]); // The configured order. + let mut encoded = a.encode(&ranked[0], &some_status()).unwrap(); + encoded.peer = "node-b".to_owned(); + assert_eq!(a.send_first(&encoded, &ranked).await.unwrap(), "sim:b"); + assert_eq!(received.recv().await.unwrap(), "sim:a"); + assert_eq!(a.rank(&addresses), ["sim:b", "broken:b"]); + + // Without the message kept, nothing else can be tried: the transport's error. + encoded.message = None; + let error = a + .send_first(&encoded, &["broken:b".to_owned()]) + .await + .unwrap_err(); + assert_eq!(error.to_string(), "broken"); + assert!(a.send_first(&encoded, &[]).await.is_err()); + } + /// A message sent to an address arrives under the sender's address on that /// transport. #[tokio::test] @@ -341,13 +530,7 @@ mod tests { }) }); 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, - }); + let status = some_status(); for to in ["other:b", "sim:b"] { let encoded = a.encode(to, &status).unwrap(); assert_eq!(encoded.address, to); |