summaryrefslogtreecommitdiffstats
path: root/src/transport/paths.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/paths.rs')
-rw-r--r--src/transport/paths.rs188
1 files changed, 165 insertions, 23 deletions
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 30aab8d..dda2d87 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -17,16 +17,24 @@
//!
//! Paths measures the addresses of such nodes itself: every `PROBE_EVERY`, each address
//! it had a choice of within `ACTIVE_FOR` gets a PING with a random nonce, which the
-//! other side answers with a PONG at once, over the same path. Answers give each
+//! other side's paths answers at once with a PONG, over the same path. Answers give each
//! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a
//! send that failed within `DOWN_FOR`, make a path down. `rank` puts paths that are up
//! first, the one in use first among them, and the rest by round-trip time; it changes
//! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least
//! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the
-//! configured order decides. PING and PONG end here: the identity layer never sees them.
+//! configured order decides.
+//!
+//! PINGs and PONGs then go up as other messages do: they tell the node the peer is
+//! alive. The node's keepalive is a PING too (`swarm::peer::keepalive`), which paths
+//! sends like any message, and notes as the probe of the path it went over.
+//!
+//! The debug firewall (`transport::firewall`) works here, below everything else, through
+//! a `Gate`: messages from an address it blocks are dropped as they arrive, and sends to
+//! one are dropped silently, as by a lossy network.
use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;
use anyhow::{Context, Result, anyhow};
@@ -36,7 +44,14 @@ use tracing::{debug, info};
use crate::locks::MutexExt;
use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats};
use crate::transport::traffic::Traffic;
-use crate::transport::{Connection, EncodedMessage, MessageKind, ParsedMessage, Transport};
+use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport};
+
+/// Whether to drop what comes from or goes to an address (the debug firewall,
+/// `transport::firewall`; the identity layer knows which node is where).
+pub(crate) trait Gate: Send + Sync {
+ fn blocks_inbound(&self, address: &str) -> bool;
+ fn blocks_outbound(&self, address: &str) -> bool;
+}
/// 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>>
@@ -250,6 +265,16 @@ impl Table {
due
}
+ /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that
+ /// path, if paths measures it.
+ fn pinged(&mut self, address: &str, nonce: u64, now: Instant) {
+ if let Some(path) = self.paths.get_mut(address) {
+ path.probe = Some((nonce, now));
+ path.last_probe = Some(now);
+ self.counts.probes_sent += 1;
+ }
+ }
+
/// A PONG with `nonce` from `address`: the round trip, if it answers our PING out.
fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option<Duration> {
let path = self.paths.get_mut(address)?;
@@ -298,18 +323,21 @@ impl Table {
pub(crate) struct Paths {
/// In order of preference.
links: Vec<Link>,
- /// PINGs and PONGs are counted here, under the address (`traffic`).
+ /// Its own PINGs and PONGs are counted here, under the address (`traffic`).
traffic: Option<Arc<Traffic>>,
+ /// The debug firewall, if any.
+ gate: Option<Weak<dyn Gate>>,
/// A std mutex: never held across an await.
table: Mutex<Table>,
}
impl Paths {
- /// Over `transports`, by scheme, in order of preference; PINGs and PONGs counted in
- /// `traffic`.
+ /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs
+ /// counted in `traffic`, the firewall through `gate`.
pub(crate) fn new(
transports: Vec<(&'static str, Arc<dyn Transport>)>,
traffic: Option<Arc<Traffic>>,
+ gate: Option<Weak<dyn Gate>>,
) -> Self {
Self {
links: transports
@@ -317,10 +345,15 @@ impl Paths {
.map(|(scheme, inner)| Link { scheme, inner })
.collect(),
traffic,
+ gate,
table: Mutex::default(),
}
}
+ fn gate(&self) -> Option<Arc<dyn Gate>> {
+ self.gate.as_ref()?.upgrade()
+ }
+
/// 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.
@@ -519,6 +552,9 @@ impl Paths {
Ok(()) => {
let mut table = self.table.lock_or_recover();
table.sent(address);
+ if let Some(ParsedMessage::Ping(nonce)) = message.message.as_deref() {
+ table.pinged(address, *nonce, Instant::now());
+ }
if !failures.is_empty() {
table.counts.failovers += 1;
drop(table);
@@ -545,8 +581,15 @@ impl Paths {
}
}
- /// Sends `message` to its `address`.
+ /// Sends `message` to its `address` (dropped silently if the firewall blocks it).
pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> {
+ if self
+ .gate()
+ .is_some_and(|gate| gate.blocks_outbound(&message.address))
+ {
+ tracing::trace!(to = %message.address, "Firewall dropped outbound message");
+ return Ok(());
+ }
let (link, target) = self
.split(&message.address)
.with_context(|| format!("no transport for {}", message.address))?;
@@ -583,7 +626,8 @@ impl Paths {
stats
}
- /// Sends a PING or PONG with `nonce` to `address`, counted (`traffic`).
+ /// Sends a PING (or with `pong`, a PONG) with `nonce` to `address`, counted
+ /// (`traffic`).
async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> {
let message = if pong {
ParsedMessage::Pong(nonce)
@@ -622,8 +666,8 @@ impl Paths {
}
}
- /// A message from `address`: a PING is answered, a PONG noted; anything else goes to
- /// `handler`.
+ /// A message from `address`, handed on to `handler`: a PING answered and a PONG
+ /// noted first; dropped if the firewall blocks it.
async fn receive(
self: Arc<Self>,
address: String,
@@ -631,29 +675,31 @@ impl Paths {
payload_len: usize,
handler: &PathHandler,
) -> Result<()> {
+ if self
+ .gate()
+ .is_some_and(|gate| gate.blocks_inbound(&address))
+ {
+ tracing::trace!(from = %address, "Firewall dropped inbound message");
+ return Ok(());
+ }
match message {
ParsedMessage::Ping(nonce) => {
- if let Some(traffic) = &self.traffic {
- traffic.received(&address, MessageKind::Ping, payload_len);
- }
+ let paths = Arc::clone(&self);
+ let to = address.clone();
tokio::spawn(async move {
- if let Err(error) = self.send_probe(&address, true, nonce).await {
- debug!(to = %address, error = %format!("{error:#}"), "Couldn't answer a PING");
+ if let Err(error) = paths.send_probe(&to, true, nonce).await {
+ debug!(to = %to, error = %format!("{error:#}"), "Couldn't answer a PING");
}
});
- Ok(())
}
ParsedMessage::Pong(nonce) => {
- if let Some(traffic) = &self.traffic {
- traffic.received(&address, MessageKind::Pong, payload_len);
- }
self.table
.lock_or_recover()
.pong(&address, nonce, Instant::now());
- Ok(())
}
- message => handler(address, message, payload_len).await,
+ _ => {}
}
+ handler(address, message, payload_len).await
}
/// Every transport's consumer, each message handed to `handler` under the sender's
@@ -712,7 +758,7 @@ impl Paths {
}
/// A PING's nonce: random, so a PONG can't be made up for a PING not seen.
-fn random_nonce() -> u64 {
+pub(crate) fn random_nonce() -> u64 {
let mut bytes = [0u8; 8];
if getrandom::fill(&mut bytes).is_err() {
// No randomness: a nonce still tells answers apart, if not unguessably.
@@ -764,6 +810,7 @@ mod tests {
("other", second.transport(id) as Arc<dyn Transport>),
],
None,
+ None,
))
};
(on_both("a"), on_both("b"))
@@ -896,6 +943,99 @@ mod tests {
assert!(table.paths.is_empty() && table.chosen.is_empty());
}
+ /// The node's keepalive PING on a path is its probe: paths doesn't ping it too, and
+ /// the PONG measures it.
+ #[test]
+ fn keepalive_pings_are_probes() {
+ let mut table = Table::default();
+ let now = Instant::now();
+ let mut nonce = 100;
+ let mut next = || {
+ nonce += 1;
+ nonce
+ };
+ table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
+ table.pinged("libp2p:x", 7, now);
+ let due = table.due(now, &mut next);
+ assert_eq!(due.len(), 1);
+ assert_eq!(due[0].0, "mycelium:x");
+ assert!(
+ table
+ .pong("libp2p:x", 7, now + Duration::from_millis(20))
+ .is_some()
+ );
+ assert_eq!(table.counts.probes_sent, 2);
+ // A peer with one path isn't measured.
+ table.pinged("libp2p:z", 8, now);
+ assert!(!table.paths.contains_key("libp2p:z"));
+ }
+
+ /// Blocks one address, both ways.
+ struct Blocking(&'static str);
+
+ impl Gate for Blocking {
+ fn blocks_inbound(&self, address: &str) -> bool {
+ address == self.0
+ }
+ fn blocks_outbound(&self, address: &str) -> bool {
+ address == self.0
+ }
+ }
+
+ /// The firewall drops what comes from a blocked address before anything sees it
+ /// (a PING isn't answered), and sends to one silently.
+ #[tokio::test]
+ async fn the_firewall_drops_both_ways() {
+ let net = SimNet::new(4);
+ let gate: Arc<dyn Gate> = Arc::new(Blocking("sim:a"));
+ let b = Arc::new(Paths::new(
+ vec![("sim", net.transport("b") as Arc<dyn Transport>)],
+ None,
+ Some(Arc::downgrade(&gate)),
+ ));
+ let a = Arc::new(Paths::new(
+ vec![("sim", net.transport("a") as Arc<dyn Transport>)],
+ None,
+ None,
+ ));
+ let received = |paths: &Arc<Paths>| {
+ let (sender, 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(())
+ })
+ });
+ let paths = Arc::clone(paths);
+ tokio::spawn(async move { paths.start_consumer(handler).await });
+ received
+ };
+ let (mut at_a, mut at_b) = (received(&a), received(&b));
+
+ // a's PING never reaches b, so nothing answers it.
+ let ping = a.encode("sim:b", &ParsedMessage::Ping(5)).unwrap();
+ a.send(&ping).await.unwrap();
+ // b's send to a "works", and goes nowhere.
+ let status = b.encode("sim:a", &some_status()).unwrap();
+ b.send(&status).await.unwrap();
+ tokio::time::sleep(Duration::from_millis(200)).await;
+ assert!(at_b.try_recv().is_err());
+ assert!(at_a.try_recv().is_err());
+
+ // Without the gate: a PING is answered by b's paths, and goes on up there.
+ drop(gate);
+ a.send(&ping).await.unwrap();
+ assert_eq!(
+ at_b.recv().await.unwrap(),
+ ("sim:a".to_owned(), crate::transport::MessageKind::Ping)
+ );
+ assert_eq!(
+ at_a.recv().await.unwrap(),
+ ("sim:b".to_owned(), crate::transport::MessageKind::Pong)
+ );
+ }
+
/// A transport whose sends all fail.
struct Broken;
@@ -948,10 +1088,12 @@ mod tests {
("sim", net.transport("a") as Arc<dyn Transport>),
],
None,
+ None,
);
let b = Arc::new(Paths::new(
vec![("sim", net.transport("b") as Arc<dyn Transport>)],
None,
+ None,
));
let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 08:42:05 +0000