diff options
Diffstat (limited to 'src')
38 files changed, 4550 insertions, 809 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs index 9e907ac..cea3644 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -17,6 +17,7 @@ use crate::document::store; use crate::locks::MutexExt; use crate::state::{AppState, HelloState, Incidents, Liveness}; use crate::swarm::peer::forgotten_path; +use crate::transport::Destination; use crate::transport::traffic::PeerTraffic; #[derive(Debug, Serialize)] @@ -50,6 +51,9 @@ pub(crate) struct PeerStatus { alias: Option<String>, /// Where this node reaches it, or would (`Transport::addresses_of`). addresses: Vec<String>, + /// Where messages to it go now: the address, whether the peer proved itself there, + /// and the connection there (libp2p: direct or relayed). Absent if nowhere. + destination: Option<Destination>, validated_swarm: Option<String>, status: PeerLivenessStatus, handshake_complete: bool, @@ -65,9 +69,6 @@ pub(crate) struct PeerStatus { heads: Option<usize>, /// We keep it alive and sync with it on our own; other peers we only answer. neighbour: bool, - /// Effective round-trip time in milliseconds (smoothed, raised by requests still - /// unanswered); absent without a recent sample. - rtt_ms: Option<f64>, /// Messages and payload bytes exchanged with it, by kind, since we started (or /// since it was last listed). traffic: PeerTraffic, @@ -120,10 +121,6 @@ pub(crate) async fn list_peers( for (peer, peer_state_mutex) in snapshot { let neighbour = crate::swarm::topology::is_neighbour(&state, &peer); - let rtt_ms = state - .lock_rtt() - .effective(&peer, now, state.timing.rtt_fresh_for()) - .map(|rtt| rtt.as_secs_f64() * 1000.0); let traffic = state.traffic.peer(&peer).unwrap_or_default(); let sync_throughput = state .throughput @@ -146,11 +143,13 @@ pub(crate) async fn list_peers( let alias = aliases.remove(&peer); let addresses = state.transport.addresses_of(&peer); + let destination = state.transport.destination(&peer); peers.insert( peer, PeerStatus { alias, addresses, + destination, validated_swarm: ps.standing.swarm_id().map(str::to_owned), status, handshake_complete: ps.hello_state == HelloState::Established, @@ -163,7 +162,6 @@ pub(crate) async fn list_peers( same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), heads: ps.reported_heads, neighbour, - rtt_ms, traffic, incidents: ps.incidents, sync_throughput, diff --git a/src/api/stats.rs b/src/api/stats.rs index fa64600..07521ab 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -16,6 +16,7 @@ use serde::Serialize; use super::peers::EpochInfo; use crate::state::{AppState, Incidents}; use crate::swarm::{members, stability, topology}; +use crate::transport::stats::TransportStats; use crate::transport::traffic::PeerTraffic; #[derive(Debug, Serialize)] @@ -44,6 +45,8 @@ pub(crate) struct StatsResponse { propagation: PropagationStats, /// This node's own files (`files`), by kind. disk: BTreeMap<&'static str, DiskStats>, + /// How the transports find and reach peers (`Transport::stats`). + transports: TransportStats, } /// Reads and writes of one kind of file, since the node started. @@ -93,6 +96,9 @@ struct DocumentStats { saved_bytes_secs_ago: Option<u64>, /// How long the last measurement took (copy and save), in milliseconds. saved_bytes_took_ms: Option<u64>, + /// How long takers of the document's lock waited for it, since start: the node's + /// work on the one document queues there. + lock_wait: crate::document::lock::LockWaits, } #[derive(Debug, Serialize)] @@ -242,6 +248,7 @@ async fn collect(state: &AppState) -> StatsResponse { saved_bytes: saved.map(|s| s.bytes), saved_bytes_secs_ago: saved.map(|s| s.at.elapsed().as_secs()), saved_bytes_took_ms: saved.map(|s| s.took.as_millis() as u64), + lock_wait: state.doc.waits(), }, catch_up: CatchUpStats { sources: node.catch_up_sources, @@ -316,5 +323,6 @@ async fn collect(state: &AppState) -> StatsResponse { (kind.as_str(), stats) }) .collect(), + transports: state.transport.stats(), } } diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 01cd271..9cbeff4 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -9,6 +9,8 @@ use tokio::time::Instant; use super::StatsResponse; use crate::locks::MutexExt; use crate::state::{AppState, Incidents, Liveness}; +use crate::transport::Destination; +use crate::transport::stats::{Reachability, TransportStats}; use crate::transport::traffic::{ByKind, Counter, PeerTraffic}; pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8"; @@ -101,6 +103,11 @@ fn escape(value: &str) -> String { .replace('\n', "\\n") } +/// The transport of an address. +fn scheme(address: &str) -> &str { + address.split_once(':').map_or("", |(scheme, _)| scheme) +} + fn flag(value: bool) -> f64 { if value { 1.0 } else { 0.0 } } @@ -116,8 +123,8 @@ struct Peer { diverged_secs: Option<f64>, traffic: PeerTraffic, incidents: Incidents, - rtt_secs: Option<f64>, throughput: Option<f64>, + destination: Option<Destination>, } async fn peers(state: &AppState) -> Vec<Peer> { @@ -127,10 +134,6 @@ async fn peers(state: &AppState) -> Vec<Peer> { let mut peers = Vec::with_capacity(snapshot.len()); for (key, peer_state) in snapshot { let neighbour = crate::swarm::topology::is_neighbour(state, &key); - let rtt_secs = state - .lock_rtt() - .effective(&key, now, state.timing.rtt_fresh_for()) - .map(|rtt| rtt.as_secs_f64()); let throughput = state .throughput .lock_or_recover() @@ -152,9 +155,9 @@ async fn peers(state: &AppState) -> Vec<Peer> { .map(|t| now.saturating_duration_since(t).as_secs_f64()), traffic, incidents: peer_state.incidents, - rtt_secs, throughput, alias: aliases.remove(&key), + destination: state.transport.destination(&key), key, }); } @@ -169,10 +172,11 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou (direction, "hello", by.hello), (direction, "hello_reply", by.hello_reply), (direction, "sync", by.sync), - (direction, "ping", by.ping), - (direction, "pong", by.pong), + (direction, "status", by.status), + (direction, "status_reply", by.status_reply), (direction, "blob", by.blob), (direction, "identity", by.identity), + (direction, "paths", by.paths), ] }; let mut series = Vec::new(); @@ -182,6 +186,181 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou series } +/// The transports' figures (`TransportStats`). +fn transports(out: &mut Out, stats: &TransportStats) { + if let Some(identity) = &stats.identity { + out.family( + "identity_proven_nodes", + "gauge", + "Nodes proven at an address, by transport.", + ); + for (transport, nodes) in &identity.proven { + out.sample( + "identity_proven_nodes", + &[("transport", transport)], + *nodes as f64, + ); + } + out.single( + "identity_hinted_nodes", + "gauge", + "Nodes known only from hints, proven nowhere.", + identity.hinted as f64, + ); + let counts = &identity.counts; + for (name, help, value) in [ + ( + "identity_proofs_rejected_total", + "Identity proofs that didn't hold.", + counts.proofs_rejected, + ), + ( + "identity_address_changes_total", + "Nodes that proved themselves at another address on a transport than before.", + counts.address_changes, + ), + ( + "identity_dropped_unproven_total", + "Messages dropped because no node proved itself at their address.", + counts.dropped_unproven, + ), + ( + "identity_refused_unproven_total", + "Sends refused because the peer hadn't proven where it is yet.", + counts.refused_unproven, + ), + ( + "identity_seeds_unanswered_total", + "Seeds no node answered at in time.", + counts.seeds_unanswered, + ), + ] { + out.single(name, "counter", help, value as f64); + } + } + if let Some(paths) = &stats.paths { + let counts = &paths.counts; + out.family( + "paths_probes_total", + "counter", + "PINGs to paths of peers with several, by result.", + ); + for (result, count) in [ + ("sent", counts.probes_sent), + ("answered", counts.probes_answered), + ("unanswered", counts.probes_unanswered), + ] { + out.sample("paths_probes_total", &[("result", result)], count as f64); + } + for (name, help, value) in [ + ( + "paths_send_failures_total", + "Sends that failed on a path, each tried at the next.", + counts.send_failures, + ), + ( + "paths_failovers_total", + "Sends that went over another path after one failed.", + counts.failovers, + ), + ( + "paths_switches_total", + "Changes of the path in use for a faster one.", + counts.switches, + ), + ] { + out.single(name, "counter", help, value as f64); + } + } + if let Some(libp2p) = &stats.libp2p { + out.family( + "libp2p_reachability", + "gauge", + "What AutoNAT says: 1 for the current status.", + ); + for status in Reachability::ALL { + out.sample( + "libp2p_reachability", + &[("status", status.as_str())], + flag(libp2p.reachability == status), + ); + } + out.family( + "libp2p_connections", + "gauge", + "Open libp2p connections, direct or through a relay.", + ); + for (kind, count) in [ + ("direct", libp2p.connections.direct), + ("relayed", libp2p.connections.relayed), + ] { + out.sample("libp2p_connections", &[("kind", kind)], count as f64); + } + let slots = &libp2p.relay_slots; + out.single( + "libp2p_relay_slots", + "gauge", + "Slots held on relays, to be reachable through them.", + slots.held as f64, + ); + out.single( + "libp2p_relays_paused", + "gauge", + "Relays picked that never granted a slot, not picked again for a while.", + slots.paused as f64, + ); + out.family( + "libp2p_relay_slot_events_total", + "counter", + "Relay slots granted, lost after being granted, and never granted.", + ); + for (event, count) in [ + ("granted", slots.granted), + ("lost", slots.lost), + ("not_granted", slots.not_granted), + ] { + out.sample( + "libp2p_relay_slot_events_total", + &[("event", event)], + count as f64, + ); + } + out.family( + "libp2p_hole_punches_total", + "counter", + "Direct connections tried over a relayed one (DCUtR), by result.", + ); + for (result, count) in [ + ("succeeded", libp2p.hole_punches.succeeded), + ("failed", libp2p.hole_punches.failed), + ] { + out.sample( + "libp2p_hole_punches_total", + &[("result", result)], + count as f64, + ); + } + let failures = &libp2p.dial_failures; + out.family( + "libp2p_dial_failures_total", + "counter", + "libp2p connections that couldn't be made, by cause.", + ); + for (cause, count) in [ + ("unreachable", failures.unreachable), + ("wrong_peer_id", failures.wrong_peer_id), + ("no_addresses", failures.no_addresses), + ("other", failures.other), + ] { + out.sample( + "libp2p_dial_failures_total", + &[("cause", cause)], + count as f64, + ); + } + } +} + /// A per-peer incident counter: its name, its help and how to read it. type IncidentSeries = (&'static str, &'static str, fn(&Incidents) -> f64); @@ -241,6 +420,94 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } + out.family( + "document_lock_wait_seconds", + "histogram", + "Time waited for the document's lock, by mode (read, write).", + ); + let lock_wait = &document.lock_wait; + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + let mut cumulative = 0; + for (le_us, count) in waits.buckets() { + cumulative += count; + let le = match le_us { + Some(us) => (us as f64 / 1e6).to_string(), + None => "+Inf".to_owned(), + }; + out.sample( + "document_lock_wait_seconds_bucket", + &[("mode", mode), ("le", &le)], + cumulative as f64, + ); + } + out.sample( + "document_lock_wait_seconds_sum", + &[("mode", mode)], + waits.sum_us as f64 / 1e6, + ); + out.sample( + "document_lock_wait_seconds_count", + &[("mode", mode)], + waits.count as f64, + ); + } + out.family( + "document_lock_wait_max_seconds", + "gauge", + "Longest wait for the document's lock since start, by mode.", + ); + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + out.sample( + "document_lock_wait_max_seconds", + &[("mode", mode)], + waits.max_us as f64 / 1e6, + ); + } + + out.family( + "document_lock_wait_seconds", + "histogram", + "Time waited for the document's lock, by mode (read, write).", + ); + let lock_wait = &document.lock_wait; + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + let mut cumulative = 0; + for (le_us, count) in waits.buckets() { + cumulative += count; + let le = match le_us { + Some(us) => (us as f64 / 1e6).to_string(), + None => "+Inf".to_owned(), + }; + out.sample( + "document_lock_wait_seconds_bucket", + &[("mode", mode), ("le", &le)], + cumulative as f64, + ); + } + out.sample( + "document_lock_wait_seconds_sum", + &[("mode", mode)], + waits.sum_us as f64 / 1e6, + ); + out.sample( + "document_lock_wait_seconds_count", + &[("mode", mode)], + waits.count as f64, + ); + } + out.family( + "document_lock_wait_max_seconds", + "gauge", + "Longest wait for the document's lock since start, by mode.", + ); + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + out.sample( + "document_lock_wait_max_seconds", + &[("mode", mode)], + waits.max_us as f64 / 1e6, + ); + } + let catch_up = &stats.catch_up; out.single( "catch_up_sources_total", @@ -503,6 +770,8 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } + transports(&mut out, &stats.transports); + let peers = peers(state).await; out.family( @@ -521,6 +790,61 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } + out.family( + "peer_destination_info", + "gauge", + "Where messages to the peer go: its transport, whether it proved itself there, and the connection (always 1).", + ); + for peer in &peers { + if let Some(destination) = &peer.destination { + let transport = scheme(&destination.address); + out.sample( + "peer_destination_info", + &[ + ("peer", &peer.key), + ("transport", transport), + ("proven", if destination.proven { "true" } else { "false" }), + ( + "connection", + destination.connection.map_or("", |c| c.as_str()), + ), + ], + 1.0, + ); + } + } + + out.family( + "peer_path_rtt_seconds", + "gauge", + "Smoothed round-trip time of the PINGs on each path to a peer with several.", + ); + for peer in &peers { + for path in peer.destination.iter().flat_map(|d| &d.paths) { + if let Some(ms) = path.rtt_ms { + out.sample( + "peer_path_rtt_seconds", + &[("peer", &peer.key), ("transport", scheme(&path.address))], + ms / 1000.0, + ); + } + } + } + out.family( + "peer_path_down", + "gauge", + "1 while a path to the peer is down (PINGs unanswered, or a send failed).", + ); + for peer in &peers { + for path in peer.destination.iter().flat_map(|d| &d.paths) { + out.sample( + "peer_path_down", + &[("peer", &peer.key), ("transport", scheme(&path.address))], + flag(path.down), + ); + } + } + out.family("peer_neighbour", "gauge", "1 if the peer is a neighbour."); for peer in &peers { out.sample( @@ -564,16 +888,6 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { } } out.family( - "peer_rtt_seconds", - "gauge", - "Effective round-trip time (smoothed, raised by unanswered requests).", - ); - for peer in &peers { - if let Some(rtt) = peer.rtt_secs { - out.sample("peer_rtt_seconds", &[("peer", &peer.key)], rtt); - } - } - out.family( "peer_sync_throughput_bytes_per_second", "gauge", "Smoothed sync throughput in large exchanges.", diff --git a/src/config/mod.rs b/src/config/mod.rs index 2de44ac..d4a7c9c 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -240,7 +240,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::List, "NAME,[NAME...]", Some("mycelium"), - "The transports this node uses, in order of preference: mycelium, libp2p (in builds with the libp2p feature). A peer is reached on the first one both have; nodes on several carry the traffic between nodes that share none.", + "The transports this node uses, in order of preference: mycelium, libp2p and tor (in builds with the libp2p and tor features). A peer is reached on the first one both have, until the paths are measured (then on the fastest); nodes on several carry the traffic between nodes that share none.", ), option( "libp2p-listen", @@ -285,6 +285,20 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "With the libp2p transport: use public libp2p infrastructure: join the IPFS DHT (through the IPFS bootstrap nodes) to find peers' addresses and public relays, picked after members that relay. Public relays only keep circuits for minutes and some KiB: enough to punch holes through NATs. Makes this node's peer ID and addresses known to the DHT.", ), option( + "tor-state-dir", + Kind::Text, + "PATH", + None, + "With the tor transport: arti's state and cache (in `state` and `cache` below), among them the onion service's key, which makes its address: keep it with the node. Required with tor.", + ), + option( + "tor-port", + Kind::Integer, + "PORT", + Some("9662"), + "With the tor transport: the port of this node's onion service (published with its onion address).", + ), + option( "libp2p-key-file", Kind::Text, "PATH", @@ -420,6 +434,11 @@ pub(crate) struct Settings { pub(crate) libp2p_public: bool, #[cfg_attr(not(feature = "libp2p"), allow(dead_code))] pub(crate) libp2p_key_file: Option<PathBuf>, + #[cfg_attr(not(feature = "tor"), allow(dead_code))] + pub(crate) tor_state_dir: Option<PathBuf>, + #[serde(default = "default_tor_port")] + #[cfg_attr(not(feature = "tor"), allow(dead_code))] + pub(crate) tor_port: u16, pub(crate) mycelium_rest_url: Option<String>, pub(crate) mycelium_key_file: Option<PathBuf>, pub(crate) mycelium_network_key_file: Option<PathBuf>, @@ -455,8 +474,14 @@ fn default_transports() -> Vec<String> { vec!["mycelium".to_owned()] } /// The transports `transports` may name, and whether this build has each. -pub(crate) const TRANSPORTS: &[(&str, bool)] = - &[("mycelium", true), ("libp2p", cfg!(feature = "libp2p"))]; +pub(crate) const TRANSPORTS: &[(&str, bool)] = &[ + ("mycelium", true), + ("libp2p", cfg!(feature = "libp2p")), + ("tor", cfg!(feature = "tor")), +]; +fn default_tor_port() -> u16 { + 9662 +} fn default_mycelium_peers() -> Vec<String> { vec!["tls://127.0.0.1:9651".to_owned()] } @@ -521,6 +546,11 @@ impl Settings { if self.transports.is_empty() { bail!("transports: a node needs at least one"); } + if self.transports.iter().any(|name| name == "tor") && self.tor_state_dir.is_none() { + bail!( + "tor needs tor-state-dir: where arti keeps its state, the onion service's key among it" + ); + } for (index, name) in self.transports.iter().enumerate() { match TRANSPORTS.iter().find(|(known, _)| known == name) { None => bail!( diff --git a/src/document/lock.rs b/src/document/lock.rs new file mode 100644 index 0000000..a394b52 --- /dev/null +++ b/src/document/lock.rs @@ -0,0 +1,175 @@ +//! The lock around the replica (`AppState::doc`), counting how long its takers wait +//! (/admin/stats `document.lockWait`): whether the one document is what the node's +//! work queues behind. + +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use serde::Serialize; +use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; + +use super::replica::Replica; +use crate::locks::MutexExt; + +/// Upper bounds (inclusive, microseconds) of the wait buckets; a last bucket takes the +/// rest. +const WAIT_BUCKETS_US: [u64; 7] = [10, 100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000]; + +/// `tokio::sync::RwLock<Replica>`, with its waits counted. +pub(crate) struct DocumentLock { + replica: RwLock<Replica>, + waits: Mutex<LockWaits>, +} + +/// How long takers of the lock waited for it, by mode, since start. +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct LockWaits { + /// Readers: wait for a writer only. + pub(crate) read: Waits, + /// Writers (applying sync messages, generating them, writes, saves): wait for + /// everyone. + pub(crate) write: Waits, +} + +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct Waits { + /// Per bucket of `WAIT_BUCKETS_US`, then the rest; not cumulative. + buckets: [u64; WAIT_BUCKETS_US.len() + 1], + pub(crate) count: u64, + pub(crate) sum_us: u64, + pub(crate) max_us: u64, +} + +impl Waits { + fn record(&mut self, waited: Duration) { + let us = u64::try_from(waited.as_micros()).unwrap_or(u64::MAX); + let bucket = WAIT_BUCKETS_US + .iter() + .position(|&bound| us <= bound) + .unwrap_or(WAIT_BUCKETS_US.len()); + self.buckets[bucket] += 1; + self.count += 1; + self.sum_us = self.sum_us.saturating_add(us); + self.max_us = self.max_us.max(us); + } + + /// Each bucket's upper bound in microseconds (None for the last) and count. + pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ { + WAIT_BUCKETS_US + .iter() + .copied() + .map(Some) + .chain([None]) + .zip(self.buckets.iter().copied()) + } +} + +impl Serialize for Waits { + fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> { + #[derive(Serialize)] + #[serde(rename_all = "camelCase")] + struct Bucket { + le_us: Option<u64>, + count: u64, + } + #[derive(Serialize)] + #[serde(rename_all = "camelCase")] + struct Out { + count: u64, + sum_us: u64, + max_us: u64, + /// Not cumulative; the last bucket has no bound. + buckets: Vec<Bucket>, + } + Out { + count: self.count, + sum_us: self.sum_us, + max_us: self.max_us, + buckets: self + .buckets() + .map(|(le_us, count)| Bucket { le_us, count }) + .collect(), + } + .serialize(serializer) + } +} + +#[derive(Clone, Copy)] +enum Mode { + Read, + Write, +} + +impl DocumentLock { + pub(crate) fn new(replica: Replica) -> Self { + Self { + replica: RwLock::new(replica), + waits: Mutex::default(), + } + } + + fn waited(&self, mode: Mode, since: Instant) { + let waited = since.elapsed(); + let mut waits = self.waits.lock_or_recover(); + match mode { + Mode::Read => waits.read.record(waited), + Mode::Write => waits.write.record(waited), + } + } + + pub(crate) async fn read(&self) -> RwLockReadGuard<'_, Replica> { + let since = Instant::now(); + let guard = self.replica.read().await; + self.waited(Mode::Read, since); + guard + } + + pub(crate) async fn write(&self) -> RwLockWriteGuard<'_, Replica> { + let since = Instant::now(); + let guard = self.replica.write().await; + self.waited(Mode::Write, since); + guard + } + + /// Outside the runtime only (`spawn_blocking`), like `RwLock::blocking_read`. + pub(crate) fn blocking_read(&self) -> RwLockReadGuard<'_, Replica> { + let since = Instant::now(); + let guard = self.replica.blocking_read(); + self.waited(Mode::Read, since); + guard + } + + /// Outside the runtime only (`spawn_blocking`), like `RwLock::blocking_write`. + pub(crate) fn blocking_write(&self) -> RwLockWriteGuard<'_, Replica> { + let since = Instant::now(); + let guard = self.replica.blocking_write(); + self.waited(Mode::Write, since); + guard + } + + pub(crate) fn waits(&self) -> LockWaits { + *self.waits.lock_or_recover() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn waits_fall_in_their_buckets() { + let mut waits = Waits::default(); + waits.record(Duration::from_micros(5)); + waits.record(Duration::from_micros(10)); + waits.record(Duration::from_millis(2)); + waits.record(Duration::from_secs(60)); + let buckets: Vec<_> = waits.buckets().collect(); + assert_eq!(buckets[0], (Some(10), 2)); + assert_eq!(buckets[3], (Some(10_000), 1)); + assert_eq!(buckets[7], (None, 1)); + assert_eq!(waits.count, 4); + assert_eq!(waits.sum_us, 5 + 10 + 2_000 + 60_000_000); + assert_eq!(waits.max_us, 60_000_000); + } +} diff --git a/src/document/mod.rs b/src/document/mod.rs index 3f489aa..b86d1cc 100644 --- a/src/document/mod.rs +++ b/src/document/mod.rs @@ -2,6 +2,7 @@ //! (merged-view paths over JSON) the HTTP API and remakes work with. pub(crate) mod file; +pub(crate) mod lock; pub(crate) mod meta; pub(crate) mod replica; pub(crate) mod store; diff --git a/src/main.rs b/src/main.rs index f2494d8..f174c4e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,6 +5,7 @@ use std::future::IntoFuture; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, Result}; use automerge::AutoCommit; @@ -53,7 +54,10 @@ async fn main() -> Result<()> { let args = &config.settings; // Read before anything starts, so a bad key file fails right away. - let node_key = identity::NodeKey::load(&args.node_key_file, args.create_node_key)?; + let node_key = Arc::new(identity::NodeKey::load( + &args.node_key_file, + args.create_node_key, + )?); info!(node_id = %node_key.id(), "Node identity"); let swarm_key = match &args.swarm_key_file { Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)), @@ -205,11 +209,12 @@ async fn main() -> Result<()> { .with_context(|| format!("libp2p-relays {address:?}")) }) .collect::<Result<Vec<_>>>()?; - let key = transport::p2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?; + let key = + transport::libp2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?; ( "libp2p", - Arc::new(transport::p2p::Libp2pTransport::start( - transport::p2p::Libp2pConfig { + Arc::new(transport::libp2p::Libp2pTransport::start( + transport::libp2p::Libp2pConfig { key, listen, announce, @@ -222,9 +227,25 @@ async fn main() -> Result<()> { )?), ) } + #[cfg(feature = "tor")] + "tor" => { + let state_dir = args + .tor_state_dir + .clone() + .context("tor needs tor-state-dir")?; + let transport = tokio::select! { + transport = transport::tor::TorTransport::start(transport::tor::TorConfig { + key: Arc::clone(&node_key), + state_dir, + port: args.tor_port, + }) => transport?, + _ = stopped(stop.clone()) => return Ok(()), + }; + ("tor", Arc::new(transport) as Arc<dyn Transport>) + } other => anyhow::bail!("transports: no {other} in this build"), }; - debug!( + info!( transport = transport.0, address = transport.1.local_id(), "Transport ready" @@ -250,7 +271,7 @@ async fn main() -> Result<()> { let node = node::Node::new(node::NodeConfig { transport, identity: Some(transport::identity::IdentityConfig { - key: Arc::new(node_key), + key: Arc::clone(&node_key), scheme, more: transports.collect(), }), @@ -294,24 +315,9 @@ async fn main() -> Result<()> { node.spawn_workers(); - let joining = bootstrap::bootstrap(&state, &seeds, start); - tokio::select! { - joined = joining => joined.context("Joining swarm")?, - _ = stopped(stop.clone()) => { - info!("Stopped while joining the swarm"); - return Ok(()); - } - } - - node::spawn_ready_workers(&state); - - info!( - swarm_id = ?state.swarm_id().await, - ready = state.is_ready().await, - "Bootstrap completed" - ); - swarm::state_file::update(&state, true).await; - + // The API is up while the node joins: what it serves then is the handlers' business + // (`docs/lifecycle.md`), and /peers and /admin/stats show how joining goes. + let joined = AtomicBool::new(false); let app = api::router(state.clone(), auth_state, args.debug_api); let listener = tokio::net::TcpListener::bind(args.bind_addr) @@ -331,9 +337,32 @@ async fn main() -> Result<()> { let stopping = state.clone(); tokio::pin!(server_fut); + // Joins, then runs until stopped; returns only if joining fails. + let joining = async { + bootstrap::bootstrap(&state, &seeds, start) + .await + .context("Joining swarm")?; + node::spawn_ready_workers(&state); + info!( + swarm_id = ?state.swarm_id().await, + ready = state.is_ready().await, + "Bootstrap completed" + ); + swarm::state_file::update(&state, true).await; + joined.store(true, Ordering::Release); + std::future::pending::<Result<()>>().await + }; + tokio::pin!(joining); + tokio::select! { + failed = joining.as_mut() => { + let _ = shutdown_tx.send(()); + let _ = server_fut.as_mut().await; + return failed; + } + exit_status = state.transport.wait_for_exit() => { - error!("Mycelium runtime process exited. Status: {:?}", exit_status); + error!(status = ?exit_status, "A transport stopped"); let _ = shutdown_tx.send(()); let _ = server_fut.as_mut().await; } @@ -344,7 +373,6 @@ async fn main() -> Result<()> { } else { info!("Axum HTTP engine shutdown cleanly."); } - std::mem::drop(state); } _ = stopped(stop.clone()) => { @@ -354,6 +382,10 @@ async fn main() -> Result<()> { } info!("HTTP API server stopped"); + if !joined.load(Ordering::Acquire) { + info!("Stopped while joining the swarm"); + return Ok(()); + } swarm::peer::farewell(&stopping).await; swarm::state_file::update(&stopping, true).await; document::file::flush(&stopping).await; diff --git a/src/node.rs b/src/node.rs index 8e4ace7..8aab601 100644 --- a/src/node.rs +++ b/src/node.rs @@ -23,7 +23,7 @@ use crate::swarm::state_file::StateFile; use crate::swarm::sync::SyncEvent; use crate::swarm::{self, peer, sync}; use crate::transport::Transport; -use crate::transport::firewall::{Firewall, FirewallTransport}; +use crate::transport::firewall::Firewall; use crate::transport::identity::{IdentityConfig, IdentityTransport}; use crate::transport::traffic::{CountingTransport, Traffic}; @@ -77,6 +77,7 @@ impl Node { config.transport, identity, Arc::clone(&traffic), + Some(Arc::clone(&firewall)), )) as Arc<dyn Transport>, None => config.transport, }; @@ -84,20 +85,18 @@ impl Node { let replica = Replica::new(config.initial_epoch, config.initial_doc); let stamp = replica.stamps(); let state = AppState { - doc: Arc::new(RwLock::new(replica)), + doc: Arc::new(crate::document::lock::DocumentLock::new(replica)), epoch_tx, stamp, lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining { discovered_swarm_id: None, })), peer_states: Arc::new(RwLock::new(HashMap::new())), - // Counted outside the firewall: what it drops outbound counts as sent (it - // plays a lossy network), what it drops inbound never arrived. Both above - // the identity layer, so by node ID. - transport: Arc::new(CountingTransport::new( - Arc::new(FirewallTransport::new(transport, Arc::clone(&firewall))), - Arc::clone(&traffic), - )), + // Counted above the firewall (in the paths layer, under the identity layer): + // what it drops outbound counts as sent (it plays a lossy network), what it + // drops inbound never arrived. Without the identity layer (tests) there is + // no firewall. + transport: Arc::new(CountingTransport::new(transport, Arc::clone(&traffic))), firewall, traffic, stats: Default::default(), diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index ed35389..3d235b0 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -14,7 +14,7 @@ use crate::swarm::bootstrap::Start; use crate::swarm::members; use crate::transport::identity::{WHO_LABEL, signed}; use crate::transport::sim::SimNet; -use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage, Transport}; +use crate::transport::{HeadsSummary, IdentityProof, ParsedMessage, Status, Transport}; const SOON: Duration = Duration::from_secs(20); @@ -41,6 +41,22 @@ async fn nodes_are_their_node_ids() { for node in [&a, &b, &c] { peers_are_node_ids(node).await; } + // Where a sends to b: where b proved itself, on the sim's transport. + let destination = a.transport.destination(b.local_id()).unwrap(); + assert_eq!( + ( + destination.address.as_str(), + destination.proven, + destination.connection + ), + ("sim:b", true, None) + ); + // Its only path, measured by keepalives once there were any. + assert_eq!(destination.paths.len(), 1); + assert_eq!(destination.paths[0].address, "sim:b"); + let stats = a.transport.stats(); + assert_eq!(stats.identity.unwrap().proven["sim"], 2); + assert!(stats.libp2p.is_none()); let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned()); let a = &a; sim.wait_until( @@ -83,8 +99,8 @@ async fn members_find_each_other_through_the_member_list() { } } -fn ping(node: &AppState) -> ParsedMessage { - ParsedMessage::Ping(Keepalive { +fn status(node: &AppState) -> ParsedMessage { + ParsedMessage::Status(Status { heads: HeadsSummary::of(Vec::new()), epoch: node.local_epoch(), epoch_seq: 0, @@ -115,15 +131,18 @@ async fn a_proof_holds_only_at_its_address() { who(&ghost, "sim:ghost", "sim:b"), ] { sim.net.inject("ghost", "a", &forged).await.unwrap(); - sim.net.inject("ghost", "a", &ping(&a)).await.unwrap(); + sim.net.inject("ghost", "a", &status(&a)).await.unwrap(); assert!(a.get_peer_state(&ghost.id()).await.is_none()); } + let counts = a.transport.stats().identity.unwrap().counts; + assert_eq!(counts.proofs_rejected, 2); + assert!(counts.dropped_unproven >= 2); sim.net .inject("ghost", "a", &who(&ghost, "sim:ghost", "sim:a")) .await .unwrap(); - sim.net.inject("ghost", "a", &ping(&a)).await.unwrap(); + sim.net.inject("ghost", "a", &status(&a)).await.unwrap(); assert!(a.get_peer_state(&ghost.id()).await.is_some()); // Known by its node ID, not by its address. assert!(a.get_peer_state("ghost").await.is_none()); @@ -176,7 +195,7 @@ async fn nodes_on_both_transports_carry_the_others() { #[cfg(feature = "libp2p")] #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn nodes_over_libp2p() { - use crate::transport::p2p::{Libp2pConfig, Libp2pTransport}; + use crate::transport::libp2p::{Libp2pConfig, Libp2pTransport}; use ::libp2p::identity::Keypair; use sha2::Digest; diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs index ce1dfd5..20ad15c 100644 --- a/src/simtest/timing.rs +++ b/src/simtest/timing.rs @@ -10,7 +10,7 @@ use crate::document::replica::Epoch; use crate::state::{AppState, Liveness}; use crate::swarm::bootstrap::Start; use crate::transport::sim::LinkParams; -use crate::transport::{HeadsSummary, HelloRequest, Keepalive, MessageKind, ParsedMessage, RunId}; +use crate::transport::{HeadsSummary, HelloRequest, MessageKind, ParsedMessage, RunId, Status}; /// Long enough for anything that should happen within a few keepalives. const SOON: Duration = Duration::from_secs(10); @@ -311,8 +311,8 @@ async fn late_epoch_does_not_flip_a_peers_epoch() { handshake: 1, }) }; - let ping = |epoch: Epoch, epoch_seq: u32| { - ParsedMessage::Ping(Keepalive { + let status = |epoch: Epoch, epoch_seq: u32| { + ParsedMessage::Status(Status { heads: HeadsSummary::of(Vec::new()), epoch, epoch_seq, @@ -336,20 +336,20 @@ async fn late_epoch_does_not_flip_a_peers_epoch() { // The ghost switched twice; we hear of the second switch first. sim.net - .inject("ghost", "a", &ping(epoch(2), 2)) + .inject("ghost", "a", &status(epoch(2), 2)) .await .unwrap(); assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); - // The PING from between the switches arrives late: ignored. + // The STATUS from between the switches arrives late: ignored. sim.net - .inject("ghost", "a", &ping(epoch(1), 1)) + .inject("ghost", "a", &status(epoch(1), 1)) .await .unwrap(); assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); // So is one from before both. sim.net - .inject("ghost", "a", &ping(Epoch::GENESIS, 0)) + .inject("ghost", "a", &status(Epoch::GENESIS, 0)) .await .unwrap(); assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); @@ -366,7 +366,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() { Some(second_run) ); sim.net - .inject("ghost", "a", &ping(epoch(3), 1)) + .inject("ghost", "a", &status(epoch(3), 1)) .await .unwrap(); assert_eq!(recorded().await, (Some(epoch(3)), Some(1))); diff --git a/src/state/host.rs b/src/state/host.rs index f6c460d..6eab803 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -29,7 +29,7 @@ impl AppState { // Tell peers about the new epoch now instead of with the next keepalive (up to a // keepalive interval later). Ready peers already on it learn it from our sync - // opening, which goes out right away; everyone else gets a PING, so peers that + // opening, which goes out right away; everyone else gets a STATUS, so peers that // never saw the commit start catching up (we don't sync with them, so nothing // else would tell them). let local = self.local_epoch(); @@ -52,7 +52,7 @@ impl AppState { } for id in announce { - if let Err(error) = crate::swarm::peer::send_ping(self, id).await { + if let Err(error) = crate::swarm::peer::send_status(self, id).await { debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed"); } } @@ -161,14 +161,24 @@ impl Host for AppState { fn resume(&self) -> impl Future<Output = ()> + Send { let state = self.clone(); async move { - let mut lifecycle = state.lifecycle.write().await; - if let NodeLifecycle::Joining { - discovered_swarm_id: Some(swarm_id), - } = &*lifecycle - { - let swarm_id = swarm_id.clone(); - debug!(swarm_id = %swarm_id, "Node resumed"); - *lifecycle = NodeLifecycle::Ready { swarm_id }; + let resumed = { + let mut lifecycle = state.lifecycle.write().await; + let swarm_id = match &*lifecycle { + NodeLifecycle::Joining { + discovered_swarm_id: Some(swarm_id), + } => Some(swarm_id.clone()), + _ => None, + }; + if let Some(swarm_id) = &swarm_id { + debug!(swarm_id = %swarm_id, "Node resumed"); + *lifecycle = NodeLifecycle::Ready { + swarm_id: swarm_id.clone(), + }; + } + swarm_id.is_some() + }; + if resumed { + crate::swarm::peer::announce(&state); } } } diff --git a/src/state/mod.rs b/src/state/mod.rs index c8243a2..fec164d 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -24,7 +24,7 @@ mod read; mod session; mod timing; -pub(crate) use peer::{HelloState, Incidents, Liveness, PeerState, Standing}; +pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing}; pub(crate) use read::{ReadSource, StaleView}; pub(crate) use session::SyncSession; pub(crate) use timing::Timing; @@ -40,7 +40,8 @@ use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch}; use tokio::time::Instant; use tracing::{debug, error, warn}; -use crate::document::replica::{Epoch, Replica, Stamp}; +use crate::document::lock::DocumentLock; +use crate::document::replica::{Epoch, Stamp}; use crate::locks::MutexExt; use crate::remake::{Host, RemakeState}; use crate::swarm::sync::{ChangeSet, SyncEvent}; @@ -58,7 +59,7 @@ pub(crate) enum NodeLifecycle { #[derive(Clone)] pub(crate) struct AppState { /// The document and its epoch. Derefs to `AutoCommit`. - pub(crate) doc: Arc<RwLock<Replica>>, + pub(crate) doc: Arc<DocumentLock>, /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers). /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`. @@ -167,7 +168,7 @@ pub(crate) struct CatchUpSource { pub(crate) reply_timeouts: u32, /// When we last warned that catching up from it stalls. pub(crate) last_warned: Option<Instant>, - /// Our last probe PING to it. + /// Our last probe STATUS to it. pub(crate) last_probe: Option<Instant>, /// Its keepalives in a row saying it wouldn't sync with us. pub(crate) refusals: u32, diff --git a/src/state/peer.rs b/src/state/peer.rs index 8086f63..b723ccd 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -67,7 +67,7 @@ const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); #[derive(Clone)] pub(crate) struct PeerState { - /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for + /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. Fixed for /// the pair, so computed once. pub(crate) leads: bool, @@ -85,7 +85,12 @@ pub(crate) struct PeerState { pub(crate) liveness: Liveness, pub(crate) added_at: Instant, pub(crate) last_seen: Option<Instant>, - pub(crate) last_ping_at: Option<Instant>, + /// Our last keepalive to it (a PING or a STATUS). + pub(crate) last_keepalive_at: Option<Instant>, + /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours. + pub(crate) keepalive_received_at: Option<Instant>, + /// The last STATUS between us, either way (`status_refresh`). + pub(crate) last_status_at: Option<Instant>, /// Last time a sync message from this peer was applied (survives resets: the /// keepalive heal looks at it). @@ -133,7 +138,8 @@ pub(crate) struct PeerState { } /// Trouble with a peer, counted since it was first listed. Lost keepalives and refused -/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING. +/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING, +/// every STATUS_REPLY a STATUS. #[derive(Debug, Default, Clone, Copy, serde::Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct Incidents { @@ -157,6 +163,15 @@ impl Incidents { } } +/// What a keepalive due is (`PeerState::take_keepalive_due`). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Keepalive { + /// A PING: the peer's transport answers it with a PONG. + Ping, + /// A STATUS: our state, answered with the peer's. + Status, +} + /// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). const UNREACHABLE_AFTER_HELLOS: u32 = 3; @@ -172,7 +187,9 @@ impl PeerState { liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, - last_ping_at: None, + last_keepalive_at: None, + keepalive_received_at: None, + last_status_at: None, last_sync_rx: None, last_resync_at: None, diverged_since: None, @@ -287,17 +304,38 @@ impl PeerState { true } - /// True (and records the ping) if an established, quiet peer needs a keepalive. + /// The keepalive an established, quiet peer needs, if any (and records it): a PING, + /// or a STATUS if none went between us for `status_refresh`. + /// + /// A PING and its PONG (or a STATUS and its STATUS_REPLY) tell both sides that the + /// other is alive, so one per interval does for the pair. The two sides take turns: + /// after a keepalive from the peer, the next is ours (the one leading the pair, + /// `leads`, sends the first), so both measure the round trip and each sends every + /// other interval. On our turn we send once the pair was quiet for an interval; on its + /// turn only after half an interval more, in case it doesn't (a keepalive or its + /// answer was lost). Without turns, both sides would get due at about the same time + /// (each one's quiet time starts with the same exchange) and their keepalives would + /// often cross. /// - /// One PING and its PONG tell both sides that the other is alive, so only one side of - /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs - /// makes the other side's peer look recently heard from, so the other side never - /// pings while it does. If the pings stop, the other side pings once it has heard - /// nothing for half an interval more. Without that, both sides would get due at about - /// the same time (each one's quiet time starts with the same exchange) and their - /// PINGs would often cross. - pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { - let quiet_enough = if self.leads { + /// When they do cross anyway (both sent before seeing the other's: the peer's came + /// within half an interval after ours), each side would take the next turn for its + /// own and they would keep crossing; then the next turn is the leader's. + pub(crate) fn take_keepalive_due( + &mut self, + timing: &Timing, + now: Instant, + ) -> Option<Keepalive> { + let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) { + (None, None) => self.leads, + (None, Some(_)) => true, + (Some(_), None) => false, + (Some(sent), Some(received)) if received > sent => { + let crossed = received.saturating_duration_since(sent) < timing.keepalive / 2; + !crossed || self.leads + } + (Some(_), Some(_)) => false, + }; + let quiet_enough = if our_turn { timing.keepalive } else { timing.keepalive + timing.keepalive / 2 @@ -307,16 +345,24 @@ impl PeerState { && now.saturating_duration_since(self.added_at) >= timing.keepalive && self.quiet_for(now) >= quiet_enough && self - .last_ping_at + .last_keepalive_at .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive); - - if due { - self.last_ping_at = Some(now); + if !due { + return None; } - due + self.last_keepalive_at = Some(now); + let status = self + .last_status_at + .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh); + Some(if status { + Keepalive::Status + } else { + Keepalive::Ping + }) } - /// Self-healing check on PING/PONG. Returns true if a re-handshake should start. + /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake + /// should start. pub(crate) fn needs_heal( &mut self, local: &transport::HeadsSummary, @@ -378,6 +424,88 @@ impl PeerState { mod tests { use super::*; + /// The two sides of a pair take turns: each sends every other interval. + #[test] + fn keepalives_take_turns() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut a = PeerState::new(true, &timing); + let mut b = PeerState::new(false, &timing); + a.hello_state = HelloState::Established; + b.hello_state = HelloState::Established; + b.added_at = a.added_at; + let mut senders = Vec::new(); + let mut now = a.added_at; + for _ in 0..40 { + now += Duration::from_secs(5); + // Whoever is due sends; the other one gets it at once, and both hear from + // each other (the keepalive, its answer). + if a.take_keepalive_due(&timing, now).is_some() { + senders.push("a"); + b.keepalive_received_at = Some(now); + } else if b.take_keepalive_due(&timing, now).is_some() { + senders.push("b"); + a.keepalive_received_at = Some(now); + } else { + continue; + } + a.mark_seen(now); + b.mark_seen(now); + } + // 200 s: one keepalive a minute, a first, then taking turns. + assert_eq!(senders, ["a", "b", "a"]); + } + + /// Keepalives that crossed (both sides sent before seeing the other's) leave the next + /// turn to the leader, rather than both taking it and crossing again. + #[test] + fn crossed_keepalives_go_back_to_taking_turns() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut a = PeerState::new(true, &timing); + let mut b = PeerState::new(false, &timing); + let start = a.added_at; + b.added_at = start; + for peer in [&mut a, &mut b] { + peer.hello_state = HelloState::Established; + } + // Both sent at 100 s and got the other's a moment later. + let sent = start + Duration::from_secs(100); + for peer in [&mut a, &mut b] { + peer.last_keepalive_at = Some(sent); + peer.keepalive_received_at = Some(sent + Duration::from_millis(50)); + peer.mark_seen(sent + Duration::from_millis(50)); + } + let next = sent + timing.keepalive + Duration::from_secs(1); + assert!(a.take_keepalive_due(&timing, next).is_some()); // The leader's turn. + assert!(b.take_keepalive_due(&timing, next).is_none()); + } + + /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet. + #[test] + fn keepalives_are_pings_and_now_and_then_a_status() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut peer = PeerState::new(true, &timing); + peer.hello_state = HelloState::Established; + let start = peer.added_at; + assert_eq!(peer.take_keepalive_due(&timing, start), None); // Too new. + let mut kinds = Vec::new(); + for interval in 1..=6 { + let now = start + timing.keepalive * interval; + let due = peer.take_keepalive_due(&timing, now); + if due == Some(Keepalive::Status) { + peer.last_status_at = Some(now); + } + kinds.extend(due); + // Not again within the interval. + assert_eq!(peer.take_keepalive_due(&timing, now), None); + } + use Keepalive::{Ping, Status}; + assert_eq!(kinds, [Status, Ping, Ping, Status, Ping, Ping]); + // Heard from just now: nothing due. + let now = start + timing.keepalive * 7; + peer.mark_seen(now); + assert_eq!(peer.take_keepalive_due(&timing, now), None); + } + /// R41: unreachability lasts until the peer is heard from, not for a set time, and /// probes back off meanwhile. #[test] diff --git a/src/state/timing.rs b/src/state/timing.rs index 0dab3da..9055ed6 100644 --- a/src/state/timing.rs +++ b/src/state/timing.rs @@ -18,10 +18,14 @@ const PROBE_MAX_FACTOR: u32 = 6; #[derive(Clone, Copy, Debug)] pub(crate) struct Timing { pub(crate) keepalive: Duration, + /// How often a quiet pair's keepalive is a STATUS rather than a PING + /// (`PeerState::take_keepalive_due`): its state (heads, readiness, epoch) for the heal + /// and against lost announcements. Changes go out as they happen anyway. + pub(crate) status_refresh: Duration, pub(crate) dormant_after: Duration, pub(crate) resync_after: Duration, pub(crate) dormant_probe: Duration, - /// While catching up: PING the source after this much silence (`swarm::source`). + /// While catching up: send the source a STATUS after this much silence (`swarm::source`). pub(crate) catch_up_probe: Duration, /// First wait for an answer to a sync message before sending again; doubles while /// answers stay out (`SyncSession::reply_backoff`). @@ -42,9 +46,10 @@ impl Timing { /// The timings for `--keepalive`. /// /// Each node keeps about `k` neighbours (`--neighbours`) whatever the swarm's size, - /// and only one side of a pair pings, so idle, a node sends about `k` messages and - /// receives about `k` (PINGs and PONGs) per keepalive interval (more on a stable - /// member others prefer). The interval trades that against how soon a silent peer is + /// and only one side of a pair sends them, so idle, a node sends about `k` messages and + /// receives about `k` per keepalive interval (more on a stable member others prefer): + /// PINGs and PONGs (a few bytes), every third a STATUS and STATUS_REPLY (heads, + /// readiness, epoch) instead. The interval trades that against how soon a silent peer is /// noticed (`dormant_after`, 4 intervals) and a gap closes (within an interval). /// Recommended: /// @@ -68,7 +73,10 @@ impl Timing { ) -> Self { Self { keepalive, - dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row + // Every third keepalive; with `resync_after` at two keepalives, a heal needs + // two STATUSes that differ: within about two refreshes. + status_refresh: keepalive * 3, + dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost keepalives in a row // >= 2 keepalives, so a second exchange confirms the difference; and >= twice // the first sync reply timeout, so a lost message's sender resends it first. resync_after: (keepalive * 2).max(sync_reply_timeout * 2), diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 7c4c444..a9ab1b2 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -119,6 +119,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() { state.mark_ready(swarm_id.clone()).await?; info!(swarm_id = %swarm_id, "Node is ready"); + crate::swarm::peer::announce(state); return Ok(()); } } diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs index c1a196b..0d45b32 100644 --- a/src/swarm/pair.rs +++ b/src/swarm/pair.rs @@ -1,5 +1,5 @@ -//! Which side of a pair of nodes leads, for duties only one side needs to do (sending -//! keepalive PINGs). +//! Which side of a pair of nodes leads, for what one side needs to start (the first +//! keepalive; then the two take turns, `PeerState::take_keepalive_due`). //! //! Both sides decide on their own, from the two keys alone, and agree. The leader is //! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index b04aad9..2e89739 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -11,7 +11,7 @@ use super::epoch::observe_peer_epoch; use super::handshake::{ discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply, }; -use super::keepalive::{heal_if_diverged, local_keepalive, note_keepalive}; +use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status, status_received}; use super::lanes::SyncLanes; use crate::state::AppState; use crate::swarm::sync::SyncEvent; @@ -30,6 +30,11 @@ pub(crate) async fn handle_message( debug!("Ignoring message from self"); return Ok(()); } + // PING and PONG only say the peer is alive (`keepalive::note_alive`). + if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) { + note_alive(state, &peer, &message).await; + return Ok(()); + } // WHO and ME end at the identity layer (`transport::identity`). let Some((epoch, epoch_seq)) = message.epoch_stamp() else { return Ok(()); @@ -143,40 +148,42 @@ pub(crate) async fn handle_message( transport::ParsedMessage::Blob(blob) => { crate::swarm::blob::handle(state, &peer, blob).await; } - transport::ParsedMessage::Ping(remote) => { - let local = on_keepalive(state, &peer, &remote, reactivated).await; + transport::ParsedMessage::Status(remote) => { + status_received(state, &peer).await; + let local = on_status(state, &peer, &remote, reactivated).await; if let Err(error) = state .transport - .send_message(&peer, &ParsedMessage::Pong(local)) + .send_message(&peer, &ParsedMessage::StatusReply(local)) .await { - debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); + debug!(peer = %peer, error = %format!("{error:#}"), "STATUS_REPLY failed"); } } - transport::ParsedMessage::Pong(remote) => { - // It answers our PING: a round-trip sample. - state.lock_rtt().pong_received(&peer, Instant::now()); - on_keepalive(state, &peer, &remote, reactivated).await; + transport::ParsedMessage::StatusReply(remote) => { + on_status(state, &peer, &remote, reactivated).await; } - transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) => {} + transport::ParsedMessage::Who(_) + | transport::ParsedMessage::Me(_) + | transport::ParsedMessage::Ping(_) + | transport::ParsedMessage::Pong(_) => {} } Ok(()) } -/// What a PING and a PONG have in common: a handshake if needed, the peer's keepalive -/// noted, and the heal if our documents differ. Returns our keepalive for the peer. -async fn on_keepalive( +/// What a STATUS and a STATUS_REPLY have in common: a handshake if needed, the peer's status +/// noted, and the heal if our documents differ. Returns our status for the peer. +async fn on_status( state: &AppState, peer: &str, - remote: &transport::Keepalive, + remote: &transport::Status, reactivated: bool, -) -> transport::Keepalive { +) -> transport::Status { if reactivated || peer_swarm(state, peer).await.is_none() { ensure_handshake(state, peer).await; } - note_keepalive(state, peer, remote).await; - let local = local_keepalive(state, peer).await; + note_status(state, peer, remote).await; + let local = local_status(state, peer).await; heal_if_diverged(state, peer, &local, remote).await; local } diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index ce2951b..c8a93eb 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -1,12 +1,19 @@ -//! Keepalives (PING / PONG) and the self-healing re-handshake when they show our -//! documents differ while sync is idle. +//! Keepalives and the self-healing re-handshake. +//! +//! A quiet pair's keepalive (`PeerState::take_keepalive_due`) is a PING, answered with a +//! PONG by the peer's paths layer (`transport::paths`), over the same path: it tells both +//! sides the other is alive, and measures the path (`swarm::rtt`). Every `Timing::status_refresh` it is a +//! STATUS instead, answered with a STATUS_REPLY: each side's state (heads, readiness, +//! epoch), which also go out as they change (relay announcements, the epoch after a +//! switch, readiness: `announce`). When a STATUS shows our documents differ while sync +//! is idle, the pair re-handshakes (`heal_if_diverged`). use std::sync::atomic::Ordering; use std::time::Duration; use anyhow::Result; use tokio::time::Instant; -use tracing::warn; +use tracing::{debug, info, warn}; use super::handshake::ensure_handshake; use crate::state::{AppState, Liveness, Standing}; @@ -14,37 +21,111 @@ use crate::swarm::sync::SyncEvent; use crate::swarm::topology; use crate::transport::{self, ParsedMessage}; -/// Sends `peer` a PING with our keepalive, noting the time for its round trip. +/// Sends `peer` a STATUS with our status. +pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> { + let status = local_status(state, peer).await; + if let Some(peer_state) = state.get_peer_state(peer).await { + peer_state.lock().await.last_status_at = Some(Instant::now()); + } + state + .transport + .send_message(peer, &ParsedMessage::Status(status)) + .await +} + +/// Sends `peer` a PING (the paths layer measures the path with it). pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { - let keepalive = local_keepalive(state, peer).await; - state.lock_rtt().ping_sent(peer, Instant::now()); + let nonce = crate::transport::paths::random_nonce(); state .transport - .send_message(peer, &ParsedMessage::Ping(keepalive)) + .send_message(peer, &ParsedMessage::Ping(nonce)) .await } -/// How long shutting down waits for the last PINGs (`farewell`). +/// A PING or PONG from `peer`: it is alive, and after a PONG the paths layer has a +/// new round-trip time to it (`swarm::rtt`). A transport without the paths layer (`Transport::answers_pings`: in tests) +/// leaves answering a PING to us. Only peers we know already: a PING makes none. +pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMessage) { + let now = Instant::now(); + match message { + ParsedMessage::Ping(nonce) if !state.transport.answers_pings() => { + let pong = ParsedMessage::Pong(*nonce); + if let Err(error) = state.transport.send_message(peer, &pong).await { + debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); + } + } + ParsedMessage::Pong(_) => { + if let Some(rtt) = state.transport.rtt(peer) { + state.lock_rtt().measured(peer, rtt, now); + } + } + _ => {} + } + let Some(peer_state) = state.get_peer_state(peer).await else { + return; + }; + let reactivated = { + let mut peer_state = peer_state.lock().await; + if matches!(message, ParsedMessage::Ping(_)) { + // Its keepalive (or a path probe): the next is ours. + peer_state.keepalive_received_at = Some(now); + } + if peer_state.unreachable_since.is_some() { + // Reachable again (`mark_seen` clears it). + state.topology_changed.notify_one(); + } + peer_state.mark_seen(now) + }; + crate::swarm::source::heard_from(state, peer); + if reactivated { + info!(peer = %peer, "Dormant peer is active again; re-handshaking"); + ensure_handshake(state, peer).await; + } +} + +/// How long shutting down waits for the last STATUSes (`farewell`). const FAREWELL_TIMEOUT: Duration = Duration::from_secs(2); -/// At shutdown: one last PING to each neighbour, saying we aren't ready. A neighbour -/// catching up from us picks another source now, instead of once we go dormant. Bounded -/// in time: our transport may be gone already. -pub(crate) async fn farewell(state: &AppState) { - state.stopping.store(true, Ordering::Release); - let mut pings = tokio::task::JoinSet::new(); +/// A STATUS to each neighbour, at most `within` (all sent: none). +async fn status_to_neighbours(state: &AppState, within: Option<Duration>) { + let mut statuses = tokio::task::JoinSet::new(); for (peer, _) in state.peers_snapshot().await { if topology::is_neighbour(state, &peer) { let state = state.clone(); - pings.spawn(async move { send_ping(&state, &peer).await }); + statuses.spawn(async move { + if let Err(error) = send_status(&state, &peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "STATUS failed"); + } + }); } } - let all_sent = async { while pings.join_next().await.is_some() {} }; - let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await; + let all_sent = async { while statuses.join_next().await.is_some() {} }; + match within { + Some(within) => { + let _ = tokio::time::timeout(within, all_sent).await; + } + None => all_sent.await, + } +} + +/// Our readiness changed (we became ready, or ready again after a rejoin): a STATUS +/// to each neighbour now, rather than with the next refresh. A peer opens sync towards +/// us only once it knows we are ready. +pub(crate) fn announce(state: &AppState) { + let state = state.clone(); + tokio::spawn(async move { status_to_neighbours(&state, None).await }); +} + +/// At shutdown: one last STATUS to each neighbour, saying we aren't ready. A neighbour +/// catching up from us picks another source now, instead of once we go dormant. Bounded +/// in time: our transport may be gone already. +pub(crate) async fn farewell(state: &AppState) { + state.stopping.store(true, Ordering::Release); + status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await; } -/// Our keepalive for `peer`. -pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { +/// Our status for `peer`. +pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Status { // Heads and epoch are published together (`Replica::stamps`), so they always match, // and reading them doesn't wait for a sync message being applied. let (heads, epoch, epoch_seq) = { @@ -64,7 +145,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: } None => false, }; - transport::Keepalive { + transport::Status { heads, epoch, epoch_seq, @@ -73,8 +154,19 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: } } -/// What a peer's keepalive says about it. -pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) { +/// A STATUS (not a reply) from `peer`: its keepalive, and a STATUS between us +/// (`Timing::status_refresh`). +pub(super) async fn status_received(state: &AppState, peer: &str) { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + let now = Instant::now(); + peer_state.keepalive_received_at = Some(now); + peer_state.last_status_at = Some(now); + } +} + +/// What a peer's status says about it. +pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) { if let Some(peer_state) = state.get_peer_state(peer).await { let mut peer_state = peer_state.lock().await; let became_ready = remote.ready && peer_state.reports_ready != Some(true); @@ -103,8 +195,8 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp pub(super) async fn heal_if_diverged( state: &AppState, peer: &str, - local: &transport::Keepalive, - remote: &transport::Keepalive, + local: &transport::Status, + remote: &transport::Status, ) { // Heads of documents from different epochs are never equal and never converge. // And a peer that isn't our neighbour isn't ours to keep in step. diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs index 0249f3b..326f3a9 100644 --- a/src/swarm/peer/mod.rs +++ b/src/swarm/peer/mod.rs @@ -2,7 +2,8 @@ //! swarm-wide forget records, and dispatch of inbound messages. //! //! * `handshake`: discovering peers, HELLO / HELLO_REPLY, membership. -//! * `keepalive`: PING / PONG and the self-healing re-handshake. +//! * `keepalive`: PING / PONG and STATUS / STATUS_REPLY, and the self-healing +//! re-handshake. //! * `epoch`: the epochs peers advertise. //! * `admit`: admission with a swarm key. //! * `forget`: swarm-wide forget records. @@ -24,6 +25,6 @@ pub(crate) use handshake::{ apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once, }; pub(crate) use inbound::handle_message; -pub(crate) use keepalive::{farewell, send_ping}; +pub(crate) use keepalive::{announce, farewell, send_status}; pub(crate) use lanes::SyncLanes; pub(crate) use worker::run_peer_worker; diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index 057e4bd..f3d0e84 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -10,9 +10,9 @@ use tracing::{debug, info, warn}; use super::forget::{AppliedForgets, apply_forget_records, forget_strangers}; use super::handshake::retry_hello_if_due; -use super::keepalive::send_ping; +use super::keepalive::{send_ping, send_status}; use crate::locks::MutexExt; -use crate::state::AppState; +use crate::state::{AppState, Keepalive}; use crate::swarm::sync::SyncEvent; use crate::swarm::topology; @@ -50,16 +50,20 @@ pub(crate) async fn run_peer_worker(state: AppState) { // Only neighbours are kept alive and judged; others we only answer. let neighbour = topology::is_neighbour(&state, peer); - let (became_dormant, ping_due, unanswered) = { + let (became_dormant, keepalive, unanswered) = { let mut peer_state = peer_state_mutex.lock().await; let became_dormant = ready && neighbour && peer_state.check_dormancy(&state.timing, now); - let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now); + let keepalive = if neighbour { + peer_state.take_keepalive_due(&state.timing, now) + } else { + None + }; if !neighbour { peer_state.stop_hello_retries(); } let unanswered = peer_state.take_overdue_sync_reply(now); - (became_dormant, ping_due, unanswered) + (became_dormant, keepalive, unanswered) }; if let Some(changes) = unanswered { @@ -80,12 +84,21 @@ pub(crate) async fn run_peer_worker(state: AppState) { ); } - let probe_due = - !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now()); - if (ping_due || probe_due) - && let Err(error) = send_ping(&state, peer).await - { - debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed"); + // The catch-up source gets STATUSes: we need its readiness, and whether it + // syncs with us. + let keepalive = match keepalive { + None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => { + Some(Keepalive::Status) + } + keepalive => keepalive, + }; + let sent = match keepalive { + Some(Keepalive::Ping) => send_ping(&state, peer).await, + Some(Keepalive::Status) => send_status(&state, peer).await, + None => Ok(()), + }; + if let Err(error) = sent { + debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed"); } if let Err(error) = retry_hello_if_due(&state, peer).await { diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 202ae40..6172f67 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -2,7 +2,7 @@ //! //! A node pushes its own changes to its neighbours, but nothing it receives goes any //! further by itself. So after applying changes from one neighbour, a node *announces* -//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100 +//! its new heads to its other neighbours with a STATUS (a keepalive: heads, about 100 //! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the //! announced heads as the peer's heads in its sync state, Automerge asks for what's //! missing (`need`), and the announcer answers with exactly those changes. Each node @@ -11,7 +11,7 @@ //! //! Announcements are coalesced: the first applied change schedules one announcement //! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A -//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the +//! round costs one STATUS and one STATUS_REPLY per neighbour (plus the pulls), whatever the //! write rate; the delay is the price per hop. Without coalescing, every applied sync //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). @@ -27,7 +27,7 @@ use tracing::debug; use crate::locks::MutexExt; use crate::state::{AppState, Standing}; -use crate::swarm::peer::send_ping; +use crate::swarm::peer::send_status; use crate::swarm::sync::SyncEvent; use crate::swarm::topology; use crate::transport::{self, HeadsSummary}; @@ -97,7 +97,7 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) { .cloned() .collect(); for peer in neighbours { - if let Err(error) = send_ping(state, &peer).await { + if let Err(error) = send_status(state, &peer).await { debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); } } @@ -111,7 +111,7 @@ pub(crate) struct Pulling { /// The neighbour we are pulling from, and since when. current: Option<(String, Instant)>, /// The latest keepalive of each neighbour waiting to be pulled from. - waiting: HashMap<String, transport::Keepalive>, + waiting: HashMap<String, transport::Status>, } /// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a @@ -119,7 +119,7 @@ pub(crate) struct Pulling { pub(crate) async fn pull_if_behind( state: &AppState, peer: &str, - remote: &transport::Keepalive, + remote: &transport::Status, ) -> bool { if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) { return false; diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs index ed6cc73..cf91ac2 100644 --- a/src/swarm/rtt.rs +++ b/src/swarm/rtt.rs @@ -1,15 +1,16 @@ -//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and -//! HELLO → HELLO_REPLY. No probes of our own. +//! Round-trip times to peers. Mostly the paths layer's (`transport::paths`): the +//! smoothed round-trip time of the path a peer's messages go over, measured by PINGs +//! (keepalives, and the paths layer's own), taken whenever a PONG from the peer arrives +//! (`measured`). Before a peer's path is measured (while joining: no keepalives yet), +//! HELLO → HELLO_REPLY gives one, smoothed (exponential moving average, weight 1/4); +//! a HELLO_REPLY only counts when exactly one HELLO was outstanding: with two, we can't +//! tell which one the answer belongs to (an answer to the older one would look too +//! fast). //! -//! A sample only counts when exactly one request was outstanding: with two, we can't tell -//! which one the answer belongs to (an answer to the older one would look too fast). -//! Samples are smoothed (exponential moving average, weight 1/4) to ride out jitter. -//! -//! What we use is the *effective* round-trip time: the smoothed value, raised to the age -//! of the oldest request still unanswered (the round trip takes at least that long), so -//! a peer that stops answering drifts away by itself. Without a recent sample (nodes -//! move; an old value says little) it is unknown. Silence alone doesn't count: we don't -//! ask non-neighbours anything, so their quiet is ours. +//! Without a recent value (nodes move; an old value says little) it is unknown. A lost +//! PING or HELLO is no sample: a peer that stops answering goes dormant, and is given up +//! as a catch-up source, by itself. Silence alone doesn't count: we don't ask +//! non-neighbours anything, so their quiet is ours. use std::collections::HashMap; use std::time::Duration; @@ -18,9 +19,11 @@ use tokio::time::Instant; #[derive(Debug, Default)] pub(crate) struct RttTable { - pings: HashMap<String, Outstanding>, hellos: HashMap<String, Outstanding>, - smoothed: HashMap<String, Sample>, + /// From the paths layer, as last measured. + from_paths: HashMap<String, Sample>, + /// From HELLOs, smoothed. + from_hellos: HashMap<String, Sample>, } #[derive(Debug, Clone, Copy)] @@ -36,85 +39,73 @@ struct Outstanding { } impl RttTable { - pub(crate) fn ping_sent(&mut self, peer: &str, now: Instant) { - Self::sent(&mut self.pings, peer, now); - } - - pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) { - if let Some(sample) = Self::answered(&mut self.pings, peer, now) { - self.record(peer, sample, now); - } + /// The paths layer's round-trip time to `peer` (`Transport::rtt`), as a PONG from + /// it arrived. + pub(crate) fn measured(&mut self, peer: &str, rtt: Duration, now: Instant) { + self.from_paths + .insert(peer.to_owned(), Sample { rtt, at: now }); } pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) { - Self::sent(&mut self.hellos, peer, now); + self.hellos + .entry(peer.to_owned()) + .and_modify(|outstanding| outstanding.count += 1) + .or_insert(Outstanding { + sent: now, + count: 1, + }); } pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) { - if let Some(sample) = Self::answered(&mut self.hellos, peer, now) { - self.record(peer, sample, now); + let Some(outstanding) = self.hellos.remove(peer) else { + return; + }; + if outstanding.count != 1 { + return; } + let sample = now.saturating_duration_since(outstanding.sent); + let rtt = match self.from_hellos.get(peer) { + Some(old) => (old.rtt * 3 + sample) / 4, + None => sample, + }; + self.from_hellos + .insert(peer.to_owned(), Sample { rtt, at: now }); } - /// The effective round-trip time: the smoothed value if sampled within `fresh_for`, - /// raised to the age of the oldest unanswered request; None without a recent sample. + /// The round-trip time: the paths layer's if measured within `fresh_for`, else + /// HELLOs' if sampled within it; None without either. pub(crate) fn effective( &self, peer: &str, now: Instant, fresh_for: Duration, ) -> Option<Duration> { - let sample = self - .smoothed + let fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for; + self.from_paths .get(peer) - .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?; - let waiting = [self.pings.get(peer), self.hellos.get(peer)] - .into_iter() - .flatten() - .map(|outstanding| now.saturating_duration_since(outstanding.sent)) - .max() - .unwrap_or_default(); - Some(sample.rtt.max(waiting)) + .filter(fresh) + .or_else(|| self.from_hellos.get(peer).filter(fresh)) + .map(|sample| sample.rtt) } - /// The effective round-trip time of every peer that has one. + /// The round-trip time of every peer that has one. pub(crate) fn all_effective( &self, now: Instant, fresh_for: Duration, ) -> impl Iterator<Item = (&str, Duration)> { - self.smoothed.keys().filter_map(move |peer| { + let mut peers: Vec<&String> = self + .from_paths + .keys() + .chain(self.from_hellos.keys()) + .collect(); + peers.sort_unstable(); + peers.dedup(); + peers.into_iter().filter_map(move |peer| { self.effective(peer, now, fresh_for) .map(|rtt| (peer.as_str(), rtt)) }) } - - fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) { - map.entry(peer.to_owned()) - .and_modify(|outstanding| outstanding.count += 1) - .or_insert(Outstanding { - sent: now, - count: 1, - }); - } - - fn answered( - map: &mut HashMap<String, Outstanding>, - peer: &str, - now: Instant, - ) -> Option<Duration> { - let outstanding = map.remove(peer)?; - (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent)) - } - - fn record(&mut self, peer: &str, sample: Duration, now: Instant) { - let rtt = match self.smoothed.get(peer) { - Some(old) => (old.rtt * 3 + sample) / 4, - None => sample, - }; - self.smoothed - .insert(peer.to_owned(), Sample { rtt, at: now }); - } } #[cfg(test)] @@ -124,55 +115,57 @@ mod tests { const FRESH: Duration = Duration::from_secs(300); #[test] - fn samples_only_unambiguous_answers_and_smooths() { + fn hellos_sample_only_unambiguous_answers_and_smooth() { let mut table = RttTable::default(); let start = Instant::now(); let at = |ms| start + Duration::from_millis(ms); - table.ping_sent("a", start); - table.pong_received("a", at(100)); + table.hello_sent("a", start); + table.hello_reply_received("a", at(100)); assert_eq!( table.effective("a", at(100), FRESH), Some(Duration::from_millis(100)) ); - - table.ping_sent("a", at(100)); - table.pong_received("a", at(300)); + table.hello_sent("a", at(100)); + table.hello_reply_received("a", at(300)); assert_eq!( table.effective("a", at(300), FRESH), Some(Duration::from_millis(125)) ); - - // Two PINGs out: the PONG can't be attributed. - table.ping_sent("a", at(300)); - table.ping_sent("a", at(400)); - table.pong_received("a", at(450)); + // Two HELLOs out: the HELLO_REPLY can't be attributed. + table.hello_sent("a", at(300)); + table.hello_sent("a", at(400)); + table.hello_reply_received("a", at(410)); assert_eq!( - table.effective("a", at(450), FRESH), + table.effective("a", at(410), FRESH), Some(Duration::from_millis(125)) ); - - // An unsolicited PONG changes nothing. - table.pong_received("b", start); + // An unsolicited HELLO_REPLY changes nothing. + table.hello_reply_received("b", start); assert_eq!(table.effective("b", start, FRESH), None); } #[test] - fn unanswered_requests_raise_it_and_old_samples_expire() { + fn the_path_wins_nothing_waits_and_old_values_expire() { let mut table = RttTable::default(); let start = Instant::now(); - table.ping_sent("a", start); - table.pong_received("a", start + Duration::from_millis(50)); - - // A PING unanswered for 40 s: at least 40 s. - let sent = start + Duration::from_secs(60); - table.ping_sent("a", sent); - let later = sent + Duration::from_secs(40); + table.hello_sent("a", start); + table.hello_reply_received("a", start + Duration::from_millis(80)); + table.measured("a", Duration::from_millis(20), start); assert_eq!( - table.effective("a", later, FRESH), - Some(Duration::from_secs(40)) + table.effective("a", start, FRESH), + Some(Duration::from_millis(20)) ); - - // No sample within the freshness window: unknown. + // A HELLO unanswered for long raises nothing. + table.hello_sent("a", start); + assert_eq!( + table.effective("a", start + Duration::from_secs(40), FRESH), + Some(Duration::from_millis(20)) + ); + assert_eq!( + table.all_effective(start, FRESH).collect::<Vec<_>>(), + [("a", Duration::from_millis(20))] + ); + // No value within the freshness window: unknown. let stale = start + FRESH + Duration::from_secs(1); assert_eq!(table.effective("a", stale, FRESH), None); } diff --git a/src/swarm/source.rs b/src/swarm/source.rs index 699b03b..d80e4d7 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -12,7 +12,7 @@ //! A source must be ready and willing to sync with us, as far as we know (its sync //! messages and keepalives say so). It counts as alive while any message from it //! arrives, so a slow transfer of a large document never looks like a stall. When it -//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or +//! is quiet for `Timing::catch_up_probe`, we send it a STATUS; after three unanswered probes, or //! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the //! first time may be a handshake still under way), the next peer takes over. //! @@ -268,7 +268,7 @@ pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: } } -/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval. +/// Whether to send `peer` a STATUS now: it is the source and has been quiet for a probe interval. pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { if !catching_up(state) { return false; diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index 308c2a5..b5548d5 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -96,7 +96,7 @@ pub(crate) struct SavedSize { /// Passing changes on (`swarm::relay`), since start. #[derive(Debug, Clone, Copy, Default)] pub(crate) struct Relay { - /// Announcement rounds (each a PING to the neighbours). + /// Announcement rounds (each a STATUS to the neighbours). pub(crate) announce_rounds: u64, /// Pulls started. pub(crate) pulls: u64, diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 011f049..df4bb2d 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -39,7 +39,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { return Ok(()); } let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - trace!("Peer was forgotten; skipping synchronization"); + trace!("Peer was forgotten; skipping sync"); return Ok(()); }; @@ -50,7 +50,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { if matches!(peer_state.liveness, Liveness::Dormant { .. }) { // Reactivation re-handshakes, and PeerValidated then restarts sync. - trace!("Skipping synchronization with dormant peer"); + trace!("Skipping sync with dormant peer"); return Ok(()); } @@ -59,7 +59,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { .is_some_and(|epoch| epoch != state.local_epoch()) { // Unrelated document histories must never be synced. - trace!("Skipping synchronization with peer on another epoch"); + trace!("Skipping sync with peer on another epoch"); return Ok(()); } @@ -222,7 +222,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { tokio::spawn(async move { if let Err(error) = send_generated(&state, &peer, &peer_state_mutex, session, encoded).await { - warn!(peer = %peer, error = %format!("{error:#}"), "Synchronization processing failed"); + warn!(peer = %peer, error = %format!("{error:#}"), "Sync processing failed"); } }); Ok(()) @@ -393,7 +393,7 @@ pub(crate) async fn handle_sync( peer_state .sync .receive(&mut replica.doc, message) - .context("Automerge synchronization failed")?; + .context("Automerge sync failed")?; peer_state.last_sync_rx = Some(Instant::now()); if let Some(wait) = &peer_state.sync.reply { throughput @@ -551,7 +551,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece let mut dirty_peers = HashSet::<String>::new(); let mut document_dirty = false; - info!("Starting synchronization worker"); + info!("Starting the sync worker"); loop { if dirty_peers.is_empty() && !document_dirty { @@ -567,7 +567,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece } }, None => { - warn!("Synchronization event channel closed"); + warn!("Sync event channel closed"); return; } } @@ -617,7 +617,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece warn!( peer = %peer, error = %format!("{error:#}"), - "Synchronization processing failed" + "Sync processing failed" ); } } diff --git a/src/transport/challenge.rs b/src/transport/challenge.rs new file mode 100644 index 0000000..f06d2fe --- /dev/null +++ b/src/transport/challenge.rs @@ -0,0 +1,277 @@ +//! A challenge handshake that proves, at the start of a connection, which key is at the +//! other end: for transports that don't authenticate their peers themselves (an onion +//! service learns nothing about who connects to it, `transport::tor`). +//! +//! Both sides send a HELLO: their public key and a fresh random nonce. Then each sends +//! a PROOF: its signature over a label, its role (dialer or listener), the connection's +//! context, and both HELLOs (its own first). Each checks the other's signature against +//! the key in the other's HELLO, and from then on knows that key is at the other end. +//! +//! * A recorded PROOF is useless elsewhere: it signs the other side's nonce, fresh for +//! each connection. +//! * The roles keep a side's own PROOF from being reflected back to it. +//! * The context binds the handshake to the connection as the dialer meant it: the +//! listener's address, as the dialer dialed it and as the listener knows itself. A +//! party in the middle that passes one side's handshake on to someone else fails +//! there: the context it was signed for is a different one. +//! +//! What it doesn't do: protect what follows. That is the connection's job, so the +//! handshake is only enough over one that is encrypted and tamper-proof end to end, and +//! authenticates the listener to the dialer (a Tor stream to an onion service). Over a +//! bare network (UDP, say), it would need key agreement and protected messages as well. +//! +//! The signing key is the node key (`crate::identity`); the label keeps these +//! signatures apart from the node key's others (WHO and ME). + +use std::time::Duration; + +use anyhow::{Result, anyhow, bail}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +use crate::identity::{self, NodeKey}; + +const LABEL: &[u8] = b"continuum/challenge/1"; +/// The first byte of a HELLO: this handshake's version. +const VERSION: u8 = 1; +const HELLO_BYTES: usize = 1 + 32 + 32; +const PROOF_BYTES: usize = 64; + +/// Which end of the connection a side is. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Role { + Dialer, + Listener, +} + +impl Role { + fn byte(self) -> u8 { + match self { + Self::Dialer => 0, + Self::Listener => 1, + } + } + + fn other(self) -> Self { + match self { + Self::Dialer => Self::Listener, + Self::Listener => Self::Dialer, + } + } +} + +/// One side's HELLO. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct Hello { + pub(crate) key: [u8; 32], + pub(crate) nonce: [u8; 32], +} + +impl Hello { + /// Ours, with a fresh nonce. + pub(crate) fn new(key: &NodeKey) -> Result<Self> { + let mut nonce = [0u8; 32]; + getrandom::fill(&mut nonce).map_err(|error| anyhow!("No randomness: {error}"))?; + Ok(Self { + key: key.public(), + nonce, + }) + } + + pub(crate) fn to_bytes(self) -> [u8; HELLO_BYTES] { + let mut bytes = [0u8; HELLO_BYTES]; + bytes[0] = VERSION; + bytes[1..33].copy_from_slice(&self.key); + bytes[33..].copy_from_slice(&self.nonce); + bytes + } + + pub(crate) fn from_bytes(bytes: &[u8; HELLO_BYTES]) -> Result<Self> { + if bytes[0] != VERSION { + bail!( + "challenge handshake version {} (we speak {VERSION})", + bytes[0] + ); + } + let mut hello = Self { + key: [0; 32], + nonce: [0; 32], + }; + hello.key.copy_from_slice(&bytes[1..33]); + hello.nonce.copy_from_slice(&bytes[33..]); + Ok(hello) + } +} + +/// What the side in `role` signs: the label, its role, the context, then its own HELLO +/// and the other side's. +fn signed(role: Role, context: &str, own: &Hello, other: &Hello) -> Vec<u8> { + let mut bytes = LABEL.to_vec(); + bytes.push(role.byte()); + bytes.extend_from_slice(&(context.len() as u32).to_be_bytes()); + bytes.extend_from_slice(context.as_bytes()); + for hello in [own, other] { + bytes.extend_from_slice(&hello.key); + bytes.extend_from_slice(&hello.nonce); + } + bytes +} + +/// Our PROOF, as the side in `role`. +pub(crate) fn prove( + key: &NodeKey, + role: Role, + context: &str, + own: &Hello, + other: &Hello, +) -> [u8; PROOF_BYTES] { + key.sign(&signed(role, context, own, other)) +} + +/// Whether `proof` is the other side's (the one not in `role`) for this handshake. +pub(crate) fn verify( + role: Role, + context: &str, + own: &Hello, + other: &Hello, + proof: &[u8; PROOF_BYTES], +) -> bool { + identity::verify( + &other.key, + &signed(role.other(), context, other, own), + proof, + ) +} + +/// Runs the handshake on `stream` as the side in `role`, within `timeout`: returns the +/// public key at the other end. `context`: the listener's address, as dialed (module +/// docs). +pub(crate) async fn run<S>( + stream: &mut S, + key: &NodeKey, + role: Role, + context: &str, + timeout: Duration, +) -> Result<[u8; 32]> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let exchange = async { + let own = Hello::new(key)?; + stream.write_all(&own.to_bytes()).await?; + stream.flush().await?; + let mut hello = [0u8; HELLO_BYTES]; + stream.read_exact(&mut hello).await?; + let other = Hello::from_bytes(&hello)?; + if other.key == own.key { + bail!("the other end has our own key"); + } + stream + .write_all(&prove(key, role, context, &own, &other)) + .await?; + stream.flush().await?; + let mut proof = [0u8; PROOF_BYTES]; + stream.read_exact(&mut proof).await?; + if !verify(role, context, &own, &other, &proof) { + bail!("the other end's proof doesn't hold"); + } + Ok(other.key) + }; + tokio::time::timeout(timeout, exchange) + .await + .map_err(|_| anyhow!("no handshake within {timeout:?}"))? +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(seed: u8) -> NodeKey { + NodeKey::from_seed([seed; 32]) + } + + #[test] + fn a_proof_holds_for_its_handshake_only() { + let (a, b) = (key(1), key(2)); + let (ha, hb) = (Hello::new(&a).unwrap(), Hello::new(&b).unwrap()); + let context = "example.onion:9663"; + let proof = prove(&a, Role::Dialer, context, &ha, &hb); + // b, the listener, checks a's proof. + assert!(verify(Role::Listener, context, &hb, &ha, &proof)); + // Another context (passed on to another listener). + assert!(!verify( + Role::Listener, + "other.onion:9663", + &hb, + &ha, + &proof + )); + // Another connection: b's nonce is a new one. + let hb2 = Hello::new(&b).unwrap(); + assert!(!verify(Role::Listener, context, &hb2, &ha, &proof)); + // Reflected: a can't be fooled by its own proof as if b had sent it. + assert!(!verify(Role::Dialer, context, &ha, &hb, &proof)); + // Someone else's key in the HELLO. + let hc = Hello { + key: key(3).public(), + nonce: ha.nonce, + }; + assert!(!verify(Role::Listener, context, &hb, &hc, &proof)); + } + + #[test] + fn hellos_round_trip_and_versions_are_checked() { + let hello = Hello::new(&key(1)).unwrap(); + assert_eq!(Hello::from_bytes(&hello.to_bytes()).unwrap(), hello); + let mut bytes = hello.to_bytes(); + bytes[0] = VERSION + 1; + assert!(Hello::from_bytes(&bytes).is_err()); + } + + #[tokio::test] + async fn both_ends_learn_each_others_key() { + let (mut dialer, mut listener) = tokio::io::duplex(1024); + let (a, b) = (key(1), key(2)); + let context = "example.onion:9663"; + let (at_a, at_b) = tokio::join!( + run( + &mut dialer, + &a, + Role::Dialer, + context, + Duration::from_secs(5) + ), + run( + &mut listener, + &b, + Role::Listener, + context, + Duration::from_secs(5) + ), + ); + assert_eq!(at_a.unwrap(), b.public()); + assert_eq!(at_b.unwrap(), a.public()); + } + + #[tokio::test] + async fn a_different_context_fails_on_both_ends() { + let (mut dialer, mut listener) = tokio::io::duplex(1024); + let (a, b) = (key(1), key(2)); + let (at_a, at_b) = tokio::join!( + run( + &mut dialer, + &a, + Role::Dialer, + "meant.onion:1", + Duration::from_secs(5) + ), + run( + &mut listener, + &b, + Role::Listener, + "actual.onion:1", + Duration::from_secs(5) + ), + ); + assert!(at_a.is_err() && at_b.is_err()); + } +} diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs index 78d00fc..7b0ec0d 100644 --- a/src/transport/firewall.rs +++ b/src/transport/firewall.rs @@ -1,13 +1,14 @@ +//! The debug firewall (`/debug/firewall`, with `--debug-api`): drops messages from and +//! to peers, by node ID, to test partitions on a real swarm. It works at the bottom of +//! the node, in the paths layer (`transport::paths::Gate`), so it drops everything a +//! partition would: WHO and ME, PINGs and their PONGs, as well as the node's messages. + use std::collections::HashSet; -use std::sync::{Arc, RwLock, RwLockReadGuard}; +use std::sync::{RwLock, RwLockReadGuard}; -use anyhow::Result; -use async_trait::async_trait; use serde::{Deserialize, Serialize}; use crate::locks::RwLockExt; -use crate::transport::EncodedMessage; -use crate::transport::{MessageHandler, ParsedMessage, Transport}; #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub(crate) struct FirewallRules { @@ -46,87 +47,3 @@ impl Firewall { self.read().outbound.contains(peer) } } - -pub(crate) struct FirewallTransport { - inner: Arc<dyn Transport>, - firewall: Arc<Firewall>, -} - -impl FirewallTransport { - pub(crate) fn new(inner: Arc<dyn Transport>, firewall: Arc<Firewall>) -> Self { - Self { inner, firewall } - } -} - -#[async_trait] -impl Transport for FirewallTransport { - fn local_id(&self) -> &str { - self.inner.local_id() - } - - fn is_peer_id(&self, id: &str) -> bool { - self.inner.is_peer_id(id) - } - - fn delivery_limit(&self) -> std::time::Duration { - self.inner.delivery_limit() - } - - fn local_addresses(&self) -> Vec<String> { - self.inner.local_addresses() - } - - fn addresses_of(&self, peer: &str) -> Vec<String> { - self.inner.addresses_of(peer) - } - - fn learn_addresses(&self, peer: &str, addresses: &[String]) { - self.inner.learn_addresses(peer, addresses) - } - - fn can_reach(&self, peer: &str) -> bool { - self.inner.can_reach(peer) - } - - fn discovered_seeds(&self) -> Vec<String> { - self.inner.discovered_seeds() - } - - fn transports(&self) -> Vec<String> { - self.inner.transports() - } - - async fn resolve_seed(&self, seed: &str) -> Result<String> { - self.inner.resolve_seed(seed).await - } - - fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - self.inner.encode(peer, message) - } - - async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - if self.firewall.blocks_outbound(&message.peer) { - tracing::trace!(peer = %message.peer, "Firewall dropped outbound message"); - return Ok(()); - } - self.inner.send_encoded(message).await - } - - async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { - let firewall = Arc::clone(&self.firewall); - - self.inner - .start_consumer(Box::new(move |sender, message, payload_len| { - if firewall.blocks_inbound(&sender) { - tracing::trace!(peer = %sender, "Firewall dropped inbound message"); - return Box::pin(async { Ok(()) }); - } - handler(sender, message, payload_len) - })) - .await - } - - async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { - self.inner.wait_for_exit().await - } -} diff --git a/src/transport/identity.rs b/src/transport/identity.rs index af0fe74..94faded 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 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) } - /// 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)) + 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,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. @@ -468,7 +486,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 +501,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 +523,49 @@ 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, 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, + 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 +578,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 +599,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 +614,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 +699,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] diff --git a/src/transport/p2p.rs b/src/transport/libp2p.rs index 955c0f7..54bfc76 100644 --- a/src/transport/p2p.rs +++ b/src/transport/libp2p.rs @@ -6,7 +6,9 @@ //! without the peer ID, which a connection finds out. Messages go one per stream of the //! `/continuum/1` protocol: their length (4 bytes, big endian), then the message as //! `wire` encodes it. A send is done once the stream is written and closed; a connection -//! that can't be made, or a stream that can't be written, refuses it. +//! that can't be made, or a stream that can't be written, refuses it. A stream's protocol +//! goes with its first data, without waiting for the other side to agree +//! (`swarm_config`): opening a stream costs no round trip. //! //! Connected peers tell each other where they listen (libp2p's identify protocol), so a //! node learns the addresses of peers that dialled it. A node publishes where it can be @@ -21,6 +23,10 @@ //! that works, or for good if it doesn't, within its limits (`RELAY_CIRCUIT_DURATION`, //! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). //! +//! Of several connections to one peer (both dialled at once, or AutoNAT dialled back), +//! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`), +//! by the side with the smaller peer ID. +//! //! A relay says so where it publishes its addresses: next to each, the same address //! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the //! member list. A node that needs a relay and has none configured picks up to @@ -55,12 +61,20 @@ use anyhow::{Context, Result, anyhow, bail}; use async_trait::async_trait; use futures::{AsyncReadExt, AsyncWriteExt, StreamExt}; use tokio::sync::{Semaphore, mpsc, oneshot, watch}; -use tracing::{debug, info, warn}; +use tracing::{debug, info, trace, warn}; use crate::locks::MutexExt; -use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport, wire}; +use crate::transport::stats::{ + Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats, +}; +use crate::transport::{ + Connection, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, + is_probe, wire, +}; const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1"); +/// Sends taking this long are logged with where the time went (at debug). +const SLOW_SEND: Duration = Duration::from_millis(5); /// The largest message read: anyone who can connect can send anything, so this bounds /// what one stream costs. Sync messages above 16 chunks travel as blobs. const MAX_MESSAGE: usize = 128 << 20; @@ -99,6 +113,17 @@ const RELAY_CHECK: Duration = Duration::from_secs(30); /// answers at its address: a relay restarted without its libp2p key) isn't picked again /// for this long. const RELAY_PAUSE: Duration = Duration::from_secs(600); +/// A peer nobody mentioned for this long (the member list, identify, mDNS, a +/// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted +/// without their libp2p key, whose new ones the member list has since. Members are +/// mentioned every few seconds (`swarm::topology`). Checked every `PRUNE_EVERY`. +const STALE_AFTER: Duration = Duration::from_secs(600); +const PRUNE_EVERY: Duration = Duration::from_secs(60); +/// Several connections to one peer (both dialled at once, or AutoNAT dialled back to +/// see whether it can) are cut to one (`Book::duplicates`) this long after the last one +/// opened: long enough for what it was dialled for. Looked for every `DEDUPE_EVERY`. +const DUPLICATE_GRACE: Duration = Duration::from_secs(10); +const DEDUPE_EVERY: Duration = Duration::from_secs(5); /// How long a message of `bytes` may take on a stream. fn stream_timeout(bytes: usize) -> Duration { @@ -187,6 +212,17 @@ struct Book { public_relays: HashMap<PeerId, Multiaddr>, /// Relays picked that never granted a slot, and when (`RELAY_PAUSE`). failed_relays: HashMap<PeerId, tokio::time::Instant>, + /// Relays that granted us a slot we hold. + slots: std::collections::HashSet<PeerId>, + /// Open connections: to whom, whether through a relay, and the remote address. + connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>, + /// What AutoNAT says, and the address it confirmed while public. + reachability: Reachability, + public_address: Option<Multiaddr>, + /// Since start (`stats`). + counts: Counts, + /// When each peer was last mentioned (`STALE_AFTER`). + seen: HashMap<PeerId, tokio::time::Instant>, /// Peers found on the LAN (mDNS) not known yet to speak continuum. lan: std::collections::HashSet<PeerId>, /// Peers that speak continuum (identify says so). @@ -196,7 +232,138 @@ struct Book { discovered: Vec<Multiaddr>, } +/// What `Libp2pStats` counts since start. +#[derive(Default)] +struct Counts { + slots_granted: u64, + slots_lost: u64, + slots_not_granted: u64, + hole_punches: HolePunches, + dial_failures: DialFailures, +} + +/// Counts a dial that failed with `error` (not one that wasn't needed). +fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::DialError) { + use ::libp2p::swarm::DialError; + match error { + DialError::DialPeerConditionFalse(_) => {} + DialError::Transport(_) => failures.unreachable += 1, + DialError::WrongPeerId { .. } => failures.wrong_peer_id += 1, + DialError::NoAddresses => failures.no_addresses += 1, + _ => failures.other += 1, + } +} + impl Book { + /// How we are connected to `peer`: directly if any connection is direct. + fn connection(&self, peer: &PeerId) -> Option<Connection> { + let mut found = None; + for (connected, relayed, _) in self.connections.values() { + if connected == peer { + if !relayed { + return Some(Connection::Direct); + } + found = Some(Connection::Relayed); + } + } + found + } + + /// The connections to `peer` to close, of several: all but one, the oldest direct + /// one (the oldest relayed one if none is direct). + fn duplicates(&self, peer: &PeerId) -> Vec<ConnectionId> { + let mut connections: Vec<(bool, ConnectionId)> = self + .connections + .iter() + .filter(|(_, (connected, _, _))| connected == peer) + .map(|(id, (_, relayed, _))| (*relayed, *id)) + .collect(); + // Direct (false) first, then the oldest. + connections.sort(); + connections.into_iter().skip(1).map(|(_, id)| id).collect() + } + + /// The remote address of a connection to `peer`: a direct one if there is one. + fn location(&self, peer: &PeerId) -> Option<&Multiaddr> { + let mut found = None; + for (connected, relayed, address) in self.connections.values() { + if connected == peer { + if !relayed { + return Some(address); + } + found = Some(address); + } + } + found + } + + fn stats(&self) -> Libp2pStats { + let mut connections = Connections::default(); + for (_, relayed, _) in self.connections.values() { + if *relayed { + connections.relayed += 1; + } else { + connections.direct += 1; + } + } + Libp2pStats { + reachability: self.reachability, + public_address: self.public_address.as_ref().map(Multiaddr::to_string), + connections, + relay_slots: RelaySlots { + held: self.slots.len(), + paused: self + .failed_relays + .values() + .filter(|at| at.elapsed() < RELAY_PAUSE) + .count(), + granted: self.counts.slots_granted, + lost: self.counts.slots_lost, + not_granted: self.counts.slots_not_granted, + }, + hole_punches: self.counts.hole_punches, + dial_failures: self.counts.dial_failures, + } + } + + /// `peer` was mentioned just now. + fn touch(&mut self, peer: PeerId) { + self.seen.insert(peer, tokio::time::Instant::now()); + } + + /// Forgets the peers not mentioned within `STALE_AFTER` before `now`, but those + /// `connected`. Returns how many. + fn prune(&mut self, now: tokio::time::Instant, connected: impl Fn(&PeerId) -> bool) -> usize { + let stale: Vec<PeerId> = self + .addresses + .keys() + .chain(self.relays.keys()) + .chain(self.public_relays.keys()) + .chain(self.continuum.iter()) + .chain(self.lan.iter()) + .filter(|peer| { + !connected(peer) + && self + .seen + .get(peer) + .is_none_or(|at| now.saturating_duration_since(*at) >= STALE_AFTER) + }) + .copied() + .collect::<std::collections::HashSet<_>>() + .into_iter() + .collect(); + for peer in &stale { + self.addresses.remove(peer); + self.relays.remove(peer); + self.public_relays.remove(peer); + self.continuum.remove(peer); + self.lan.remove(peer); + self.failed_relays.remove(peer); + self.seen.remove(peer); + } + stale.len() + } + /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first. fn discover(&mut self, peer: PeerId) { let Some(address) = self.addresses.get(&peer).and_then(|known| known.first()) else { @@ -254,6 +421,20 @@ fn published(all: &[Multiaddr]) -> Vec<Multiaddr> { } impl Book { + /// Whether all of `addresses` are known of `peer` (`remember`). + fn knows(&self, peer: PeerId, addresses: &[Multiaddr]) -> bool { + let Some(known) = self.addresses.get(&peer) else { + return false; + }; + addresses.iter().all(|address| { + let mut address = address.clone(); + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + known.contains(&address) + }) + } + /// `peer` is at `address` (without its `/p2p` part): kept first. fn remember(&mut self, peer: PeerId, mut address: Multiaddr) { if let Some(Protocol::P2p(_)) = address.iter().last() { @@ -262,6 +443,7 @@ impl Book { if !self.addresses.contains_key(&peer) && self.addresses.len() >= MAX_PEERS { return; } + self.touch(peer); let addresses = self.addresses.entry(peer).or_default(); addresses.retain(|known| *known != address); addresses.insert(0, address); @@ -372,7 +554,7 @@ impl Libp2pTransport { }, stopping, )); - info!(peer_id = %peer_id, "libp2p transport started"); + info!(peer_id = %peer_id, "Transport started"); Ok(Self { local_id: peer_id.to_string(), peer_id, @@ -542,10 +724,25 @@ fn network_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> { Behaviour::new(key, relay_client, features) .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>) })? - .with_swarm_config(|config| config.with_idle_connection_timeout(IDLE_TIMEOUT)) + .with_swarm_config(|config| swarm_config(config, features)) .build()) } +/// The swarm's settings. Streams are opened without waiting for the other side to agree +/// on the protocol (`V1Lazy`): the protocol and the first data go together, saving a +/// round trip per message (one stream each, module docs). That is safe only towards +/// peers sure to speak the protocol, so not with public libp2p infrastructure +/// (`Libp2pConfig::public`), whose nodes speak only some: there a protocol they don't +/// would fail late, and unclearly. +fn swarm_config(config: ::libp2p::swarm::Config, features: Features) -> ::libp2p::swarm::Config { + let config = config.with_idle_connection_timeout(IDLE_TIMEOUT); + if features.public { + config + } else { + config.with_substream_upgrade_protocol_override(::libp2p::core::upgrade::Version::V1Lazy) + } +} + fn memory_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> { use ::libp2p::Transport as _; use ::libp2p::core::transport::{MemoryTransport, OrTransport}; @@ -564,7 +761,7 @@ fn memory_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> { transport, Behaviour::new(&key, relay_client, features)?, peer_id, - ::libp2p::swarm::Config::with_tokio_executor().with_idle_connection_timeout(IDLE_TIMEOUT), + swarm_config(::libp2p::swarm::Config::with_tokio_executor(), features), )) } @@ -628,8 +825,10 @@ fn pick_relays( ) { if nat.is_public() && !picked.is_empty() { info!("Reachable directly (AutoNAT); letting the relays picked go"); - for listener in std::mem::take(picked).into_keys() { + let mut book = lock(book); + for (listener, relay) in std::mem::take(picked) { swarm.remove_listener(listener); + book.slots.remove(&relay); } return; } @@ -660,11 +859,11 @@ fn pick_relays( for (relay, address) in wanted { match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) { Ok(listener) => { - info!(relay = %relay, address = %address, "Picked a member as libp2p relay"); + info!(relay = %relay, address = %address, "Picked a member as relay"); picked.insert(listener, relay); } Err(error) => { - debug!(relay = %relay, error = %error, "Couldn't use a member as libp2p relay"); + debug!(relay = %relay, error = %error, "Couldn't use a member as relay"); } } } @@ -688,15 +887,14 @@ async fn run( ) { let serving = relaying.serving; let mut nat = ::libp2p::autonat::NatStatus::Unknown; - // Relays picked among the members, by listener, and those that granted a slot. + // Relays picked among the members, by listener. let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new(); - let mut granted: std::collections::HashSet<PeerId> = std::collections::HashSet::new(); let mut check = tokio::time::interval(RELAY_CHECK); check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP); dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - // Relays we listen through, by listener, as configured. - let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> = + // Relays we listen through, by listener: as configured, and their peer. + let mut reservations: HashMap<::libp2p::core::transport::ListenerId, (Multiaddr, PeerId)> = HashMap::new(); let reserve_later = |configured: Multiaddr| { let again = again.clone(); @@ -713,8 +911,37 @@ async fn run( ConnectionId, (Multiaddr, oneshot::Sender<Result<PeerId, String>>), > = HashMap::new(); + let mut prune = tokio::time::interval(PRUNE_EVERY); + prune.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // Peers that may have several connections, and when to cut them to one. Only the + // side with the smaller peer ID does that, so the two never close different ones. + let mut dedupe_due: HashMap<PeerId, tokio::time::Instant> = HashMap::new(); + let mut dedupe = tokio::time::interval(DEDUPE_EVERY); + dedupe.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); loop { tokio::select! { + _ = dedupe.tick(), if !dedupe_due.is_empty() => { + let now = tokio::time::Instant::now(); + let due: Vec<PeerId> = dedupe_due + .iter() + .filter(|(_, at)| **at <= now) + .map(|(peer, _)| *peer) + .collect(); + for peer in due { + dedupe_due.remove(&peer); + let duplicates = lock(&book).duplicates(&peer); + for connection in duplicates { + debug!(peer = %peer, connection = %connection, "Closing a duplicate connection"); + swarm.close_connection(connection); + } + } + } + _ = prune.tick() => { + let forgotten = lock(&book).prune(tokio::time::Instant::now(), |peer| swarm.is_connected(peer)); + if forgotten > 0 { + debug!(forgotten, "Forgot peers nobody mentions any more"); + } + } _ = dht_bootstrap.tick(), if relaying.public => { if let Some(dht) = swarm.behaviour_mut().kad.as_mut() && let Err(error) = dht.bootstrap() @@ -750,6 +977,7 @@ async fn run( opts = opts.extend_addresses_through_behaviour(); } if let Err(error) = swarm.dial(opts.build()) { + count_dial_failure(&mut lock(&book).counts.dial_failures, &error); for done in connecting.remove(&peer).into_iter().flatten() { let _ = done.send(Err(error.to_string())); } @@ -774,7 +1002,7 @@ async fn run( } } _ => { - warn!(relay = %relay, "Couldn't reach a libp2p relay; trying again later"); + warn!(relay = %relay, "Couldn't reach a relay; trying again later"); tokio::time::sleep(RELAY_RETRY).await; let _ = again.send(Command::Reserve { relay, configured }); } @@ -782,13 +1010,16 @@ async fn run( }); } Some(Command::Reserve { relay, configured }) => { + let Some(relay_peer) = split_peer(&relay) else { + continue; // The arm above takes those. + }; match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) { Ok(listener) => { - debug!(relay = %relay, "Reserving a slot on a libp2p relay"); - reservations.insert(listener, configured); + debug!(relay = %relay, "Reserving a slot on a relay"); + reservations.insert(listener, (configured, relay_peer)); } Err(error) => { - warn!(relay = %relay, error = %error, "Couldn't use a libp2p relay; trying again later"); + warn!(relay = %relay, error = %error, "Couldn't use a relay; trying again later"); reserve_later(configured); } } @@ -807,7 +1038,29 @@ async fn run( } }, event = swarm.select_next_some() => match event { - SwarmEvent::ConnectionEstablished { peer_id, connection_id, .. } => { + SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, .. } => { + { + let mut book = lock(&book); + book.touch(peer_id); + let relayed = endpoint.is_relayed(); + book.connections.insert( + connection_id, + (peer_id, relayed, endpoint.get_remote_address().clone()), + ); + if relaying.local < peer_id + && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 1 + { + dedupe_due.insert(peer_id, tokio::time::Instant::now() + DUPLICATE_GRACE); + } + debug!( + peer = %peer_id, + connection = %connection_id, + dialled = endpoint.is_dialer(), + address = %endpoint.get_remote_address(), + relayed, + "Connection opened" + ); + } for done in connecting.remove(&peer_id).into_iter().flatten() { let _ = done.send(Ok(())); } @@ -816,7 +1069,17 @@ async fn run( let _ = done.send(Ok(peer_id)); } } + SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => { + lock(&book).connections.remove(&connection_id); + debug!( + peer = %peer_id, + connection = %connection_id, + cause = ?cause, + "Connection closed" + ); + } SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => { + count_dial_failure(&mut lock(&book).counts.dial_failures, &error); if let Some((_, done)) = identifying.remove(&connection_id) { let _ = done.send(Err(error.to_string())); } @@ -832,7 +1095,7 @@ async fn run( } } SwarmEvent::NewListenAddr { address, .. } => { - info!(address = %address, "libp2p listening"); + info!(address = %address, "Listening"); // A relay tells those reserving where it is: its external addresses. if serving && !is_local(&address) && !is_relayed(&address) { swarm.add_external_address(address.clone()); @@ -843,7 +1106,7 @@ async fn run( } } SwarmEvent::ExpiredListenAddr { address, .. } => { - info!(address = %address, "libp2p no longer listening"); + info!(address = %address, "No longer listening"); if serving { swarm.remove_external_address(&address); } @@ -853,14 +1116,33 @@ async fn run( lock(&book) .listening .retain(|known| !addresses.contains(known)); - if let Some(relay) = reservations.remove(&listener_id) { - warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later"); + if let Some((relay, relay_peer)) = reservations.remove(&listener_id) { + { + let mut book = lock(&book); + if book.slots.remove(&relay_peer) { + book.counts.slots_lost += 1; + } else { + book.counts.slots_not_granted += 1; + } + } + warn!(relay = %relay, reason = ?reason, "Lost the slot on a relay; reserving again later"); reserve_later(relay); } // A relay picked among the members: the next check picks again, another // one if this one never granted the slot. if let Some(relay) = picked.remove(&listener_id) { - if granted.remove(&relay) { + let lost = { + let mut book = lock(&book); + let lost = book.slots.remove(&relay); + if lost { + book.counts.slots_lost += 1; + } else { + book.counts.slots_not_granted += 1; + book.failed_relays.insert(relay, tokio::time::Instant::now()); + } + lost + }; + if lost { debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying"); } else { info!( @@ -868,50 +1150,77 @@ async fn run( pause = ?RELAY_PAUSE, "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while" ); - lock(&book) - .failed_relays - .insert(relay, tokio::time::Instant::now()); } } } SwarmEvent::ListenerError { error, .. } => { - warn!(error = %error, "libp2p listener failed"); + warn!(error = %error, "Listener failed"); } SwarmEvent::Behaviour(BehaviourEvent::RelayClient( ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. }, )) => { - granted.insert(relay_peer_id); - lock(&book).failed_relays.remove(&relay_peer_id); + { + let mut book = lock(&book); + book.slots.insert(relay_peer_id); + book.failed_relays.remove(&relay_peer_id); + if !renewal { + book.counts.slots_granted += 1; + } + } if !renewal { - info!(relay = %relay_peer_id, "Reachable through a libp2p relay"); + info!(relay = %relay_peer_id, "Reachable through a relay"); } } SwarmEvent::Behaviour(BehaviourEvent::Autonat( ::libp2p::autonat::Event::StatusChanged { old, new }, )) => { - info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)"); + info!(old = ?old, new = ?new, "Reachability (AutoNAT)"); + { + use ::libp2p::autonat::NatStatus; + let mut book = lock(&book); + (book.reachability, book.public_address) = match &new { + NatStatus::Unknown => (Reachability::Unknown, None), + NatStatus::Private => (Reachability::Private, None), + NatStatus::Public(address) => { + (Reachability::Public, Some(address.clone())) + } + }; + } nat = new; } SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => { - debug!(event = ?event, "libp2p relay"); + debug!(event = ?event, "Relay event"); } SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result { - Ok(_) => info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"), - Err(error) => debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"), + Ok(_) => { + lock(&book).counts.hole_punches.succeeded += 1; + info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"); + } + Err(error) => { + lock(&book).counts.hole_punches.failed += 1; + debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"); + } }, SwarmEvent::Behaviour(BehaviourEvent::Identify( - ::libp2p::identify::Event::Received { peer_id, info, .. }, + ::libp2p::identify::Event::Received { connection_id, peer_id, info }, )) => { - debug!( - peer = %peer_id, - listen = ?info.listen_addrs, - observed = %info.observed_addr, - agent = %info.agent_version, - "libp2p peer identified" - ); let mut book = lock(&book); + let listen = published(&info.listen_addrs); + // Every new connection identifies again: logged when it says something new. + if book.knows(peer_id, &listen) { + trace!(peer = %peer_id, connection = %connection_id, "Peer identified again"); + } else { + debug!( + peer = %peer_id, + connection = %connection_id, + listen = ?info.listen_addrs, + observed = %info.observed_addr, + agent = %info.agent_version, + "Peer identified" + ); + } // Last first: `remember` puts each in front. - for address in published(&info.listen_addrs).into_iter().rev() { + for address in listen.into_iter().rev() { book.remember(peer_id, address); } if relaying.public @@ -940,7 +1249,7 @@ async fn run( if peer == relaying.local { continue; } - debug!(peer = %peer, address = %address, "libp2p peer on the LAN"); + debug!(peer = %peer, address = %address, "Peer on the LAN"); let continuum = { let mut book = lock(&book); book.remember(peer, address.clone()); @@ -1068,6 +1377,29 @@ impl Transport for Libp2pTransport { Some(peer.to_string()) } + fn connection(&self, target: &str) -> Option<Connection> { + let peer = PeerId::from_str(target).ok()?; + lock(&self.book).connection(&peer) + } + + fn location(&self, target: &str) -> Option<String> { + let peer = PeerId::from_str(target).ok()?; + let address = lock(&self.book).location(&peer)?.clone(); + Some( + address + .with_p2p(peer) + .unwrap_or_else(|address| address) + .to_string(), + ) + } + + fn stats(&self) -> TransportStats { + TransportStats { + libp2p: Some(lock(&self.book).stats()), + ..TransportStats::default() + } + } + /// Nodes found on the LAN (mDNS) that speak continuum. fn discovered_seeds(&self) -> Vec<String> { std::mem::take(&mut lock(&self.book).discovered) @@ -1125,66 +1457,89 @@ impl Transport for Libp2pTransport { payload_len: body.len(), body: body.into(), kind: message.kind(), + message: None, }) } async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { let peer = PeerId::from_str(&message.peer) .with_context(|| format!("{} is no libp2p peer ID", message.peer))?; + let started = std::time::Instant::now(); self.connect(peer).await?; + let connected = started.elapsed(); let mut control = self.control.clone(); let body = message.body.clone(); let write = async move { let mut stream = control.open_stream(peer, PROTOCOL).await?; + let opened = started.elapsed(); stream.write_all(&(body.len() as u32).to_be_bytes()).await?; stream.write_all(&body).await?; stream.close().await?; - Ok::<(), anyhow::Error>(()) + Ok::<_, anyhow::Error>(opened) }; let timeout = stream_timeout(message.body.len()); - tokio::time::timeout(timeout, write) + let opened = tokio::time::timeout(timeout, write) .await .map_err(|_| anyhow!("sending to {peer} took longer than {timeout:?}"))? - .with_context(|| format!("Sending to {peer}")) + .with_context(|| format!("Sending to {peer}"))?; + let took = started.elapsed(); + if took >= SLOW_SEND { + // Where a send's time goes: the swarm task (connecting), the stream's protocol + // negotiation (opening), or the connection (writing). + debug!( + peer = %peer, + kind = ?message.kind, + bytes = message.body.len(), + connect_ms = connected.as_secs_f64() * 1000.0, + open_ms = (opened - connected).as_secs_f64() * 1000.0, + write_ms = (took - opened).as_secs_f64() * 1000.0, + "Slow send" + ); + } + Ok(()) } /// Reads inbound streams (`READS` at once) and handles their messages one at a - /// time, in the order they were read. + /// time, in the order they were read; PINGs and PONGs at once (`is_probe`). async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { let Some(mut incoming) = self.incoming.lock().await.take() else { bail!("libp2p consumer already started"); }; - let (sender, mut received) = mpsc::channel::<(PeerId, Vec<u8>)>(READS); + let handler: Arc<HandlerFn> = Arc::from(handler); + let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS); let reads = Arc::new(Semaphore::new(READS)); + let probes = Arc::clone(&handler); tokio::spawn(async move { while let Some((peer, stream)) = incoming.next().await { let Ok(permit) = Arc::clone(&reads).acquire_owned().await else { break; }; let sender = sender.clone(); + let probes = Arc::clone(&probes); tokio::spawn(async move { match read_message(stream).await { - Ok(body) => { - let _ = sender.send((peer, body)).await; - } + Ok(body) => match wire::decode(&body) { + Ok(message) if is_probe(&message) => { + hand_on_probe(&probes, peer.to_string(), message, body.len()); + } + Ok(message) => { + let _ = sender.send((peer, message, body.len())).await; + } + Err(error) => { + debug!(peer = %peer, error = %error, "Dropping a message that doesn't decode"); + } + }, Err(error) => { - debug!(peer = %peer, error = %format!("{error:#}"), "Dropping an unreadable libp2p stream"); + debug!(peer = %peer, error = %format!("{error:#}"), "Dropping an unreadable stream"); } } drop(permit); }); } }); - while let Some((peer, body)) = received.recv().await { - match wire::decode(&body) { - Ok(message) => { - if let Err(error) = handler(peer.to_string(), message, body.len()).await { - warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process inbound message"); - } - } - Err(error) => { - debug!(peer = %peer, error = %error, "Dropping a libp2p message that doesn't decode"); - } + while let Some((peer, message, payload_len)) = received.recv().await { + if let Err(error) = handler(peer.to_string(), message, payload_len).await { + warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process inbound message"); } } Ok(()) @@ -1203,7 +1558,7 @@ mod tests { use super::*; use crate::document::replica::Epoch; - use crate::transport::{HeadsSummary, Keepalive}; + use crate::transport::{HeadsSummary, Status}; /// Memory ports are process-wide. static NEXT_PORT: AtomicU64 = AtomicU64::new(40_000); @@ -1246,8 +1601,8 @@ mod tests { memory_node_with(true, Vec::new(), false) } - fn ping() -> ParsedMessage { - ParsedMessage::Ping(Keepalive { + fn status() -> ParsedMessage { + ParsedMessage::Status(Status { heads: HeadsSummary::of(Vec::new()), epoch: Epoch::GENESIS, epoch_seq: 3, @@ -1287,15 +1642,21 @@ mod tests { b.resolve_seed(&a_address.to_string()).await.unwrap(), a.local_id() ); - b.send_message(a.local_id(), &ping()).await.unwrap(); + b.send_message(a.local_id(), &status()).await.unwrap(); let (from, message) = at_a.recv().await.unwrap(); assert_eq!(from, b.local_id()); - assert!(matches!(message, ParsedMessage::Ping(keepalive) if keepalive.epoch_seq == 3)); + assert!(matches!(message, ParsedMessage::Status(status) if status.epoch_seq == 3)); // a knows no address of b, but b's connection is there. - a.send_message(b.local_id(), &ping()).await.unwrap(); + a.send_message(b.local_id(), &status()).await.unwrap(); let (from, _) = at_b.recv().await.unwrap(); assert_eq!(from, a.local_id()); + + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); + assert_eq!(b.connection(a.local_id()), Some(Connection::Direct)); + let stats = a.stats().libp2p.unwrap(); + assert!(stats.connections.direct >= 1); + assert_eq!(stats.connections.relayed, 0); } #[tokio::test] @@ -1347,7 +1708,7 @@ mod tests { let (a, b) = (Arc::new(a), Arc::new(b)); let _at_a = received(Arc::clone(&a)); b.resolve_seed(&a_address.to_string()).await.unwrap(); - b.send_message(a.local_id(), &ping()).await.unwrap(); + b.send_message(a.local_id(), &status()).await.unwrap(); let b_id = b.local_id().to_owned(); assert!( eventually(|| a.locations(&b_id) == [b_address.to_string()]).await, @@ -1418,11 +1779,16 @@ mod tests { "{relayed}" ); + assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await); + let slots = b.stats().libp2p.unwrap().relay_slots; + assert_eq!((slots.granted, slots.not_granted), (1, 0)); + let (a, _) = memory_node(); assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); - a.send_message(b.local_id(), &ping()).await.unwrap(); + a.send_message(b.local_id(), &status()).await.unwrap(); let (from, _) = at_b.recv().await.unwrap(); assert_eq!(from, a.local_id()); + assert!(a.connection(b.local_id()).is_some()); } fn peer() -> PeerId { @@ -1496,6 +1862,46 @@ mod tests { assert_eq!(book.discovered, [address.with_p2p(lan).unwrap()]); } + #[test] + fn known_addresses() { + let mut book = Book::default(); + let (relay, us) = (peer(), peer()); + let direct: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap(); + let relayed: Multiaddr = format!("/ip6/fd00::2/tcp/9661/p2p/{relay}/p2p-circuit/p2p/{us}") + .parse() + .unwrap(); + assert!(!book.knows(us, &[direct.clone()])); + book.remember(us, relayed.clone()); + book.remember(us, direct.clone()); + // In any order, with or without the peer ID. + assert!(book.knows(us, &[relayed.clone(), direct.clone()])); + assert!(book.knows(us, &[direct.clone().with_p2p(us).unwrap()])); + assert!(!book.knows(us, &["/ip6/fd00::3/tcp/9661".parse().unwrap()])); + } + + /// Peers nobody mentions any more are forgotten, but connected ones. + #[test] + fn stale_peers_are_forgotten() { + let start = tokio::time::Instant::now(); + let (old, current, connected) = (peer(), peer(), peer()); + let address: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap(); + let mut book = Book::default(); + for peer in [old, current, connected] { + book.remember(peer, address.clone()); + } + book.relays.insert(old, address.clone()); + book.continuum.insert(old); + let later = start + STALE_AFTER; + // `current` was mentioned again since. + book.seen.insert(current, later); + assert_eq!(book.prune(later, |peer| *peer == connected), 1); + assert!(!book.addresses.contains_key(&old)); + assert!(!book.relays.contains_key(&old) && !book.continuum.contains(&old)); + assert!(book.addresses.contains_key(¤t)); + assert!(book.addresses.contains_key(&connected)); + assert_eq!(book.prune(later, |_| false), 1); // `connected`, once it isn't. + } + /// A relay doesn't dial the nodes it relays for through itself. #[test] fn no_circuits_through_ourselves() { @@ -1522,7 +1928,43 @@ mod tests { .public() .to_peer_id() .to_string(); - assert!(b.send_message(&stranger, &ping()).await.is_err()); + assert!(b.send_message(&stranger, &status()).await.is_err()); assert!(b.resolve_seed(&stranger).await.is_err()); + assert_eq!(b.connection(&stranger), None); + } + + #[test] + fn a_direct_connection_wins_over_a_relayed_one() { + let mut book = Book::default(); + let (relayed_only, both) = (peer(), peer()); + let mut id = 0; + let mut connect = |book: &mut Book, peer: PeerId, relayed: bool| { + id += 1; + let address: Multiaddr = format!("/ip4/10.0.0.{id}/tcp/1").parse().unwrap(); + book.connections + .insert(ConnectionId::new_unchecked(id), (peer, relayed, address)); + }; + connect(&mut book, relayed_only, true); + connect(&mut book, both, true); + connect(&mut book, both, false); + assert_eq!(book.connection(&relayed_only), Some(Connection::Relayed)); + assert_eq!(book.connection(&both), Some(Connection::Direct)); + assert_eq!(book.connection(&peer()), None); + assert_eq!( + book.location(&both).map(|address| address.to_string()), + Some("/ip4/10.0.0.3/tcp/1".to_owned()) + ); + // Of several, the oldest direct one stays; a relayed one only without a direct. + connect(&mut book, both, false); + assert_eq!( + book.duplicates(&both), + [ + ConnectionId::new_unchecked(4), + ConnectionId::new_unchecked(2) + ] + ); + assert!(book.duplicates(&relayed_only).is_empty()); + let connections = book.stats().connections; + assert_eq!((connections.direct, connections.relayed), (1, 2)); } } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 3f77306..25ad56d 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -11,13 +11,20 @@ use std::sync::atomic::{AtomicUsize, Ordering}; pub(crate) use crate::document::replica::Epoch; +// Used by the transports that don't authenticate their peers themselves (`tor`). +#[cfg_attr(not(feature = "tor"), allow(dead_code))] +pub(crate) mod challenge; pub(crate) mod firewall; pub(crate) mod identity; -pub(crate) mod mycelium; #[cfg(feature = "libp2p")] -pub(crate) mod p2p; +pub(crate) mod libp2p; +pub(crate) mod mycelium; +pub(crate) mod paths; #[cfg(test)] pub(crate) mod sim; +pub(crate) mod stats; +#[cfg(feature = "tor")] +pub(crate) mod tor; pub(crate) mod traffic; pub(crate) mod wire; @@ -115,9 +122,9 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] { hasher.finalize().into() } -/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient. +/// STATUS and STATUS_REPLY. Addressed to one peer: `syncs_with_you` is about the recipient. #[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct Keepalive { +pub(crate) struct Status { /// Only comparable between nodes on the same epoch. #[serde(with = "crate::transport::wire::codec::heads")] pub(crate) heads: HeadsSummary, @@ -132,18 +139,23 @@ pub(crate) struct Keepalive { 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), Sync(SyncEnvelope), - Ping(Keepalive), - Pong(Keepalive), + Status(Status), + StatusReply(Status), Blob(BlobMessage), /// Asks the recipient who it is, saying who the sender is (`transport::identity`). Who(IdentityProof), /// Answers a WHO. Me(IdentityProof), + /// Asks for a PONG with this nonce, over the same path (`transport::paths`). + Ping(u64), + /// Answers a PING. + Pong(u64), } /// A node's proof that it holds the key of its node ID (`identity`), sending from its @@ -262,11 +274,13 @@ pub(crate) enum MessageKind { Hello, HelloReply, Sync, - Ping, - Pong, + Status, + StatusReply, Blob, Who, Me, + Ping, + Pong, } impl ParsedMessage { @@ -275,26 +289,31 @@ impl ParsedMessage { ParsedMessage::Hello(_) => MessageKind::Hello, ParsedMessage::HelloReply(_) => MessageKind::HelloReply, ParsedMessage::Sync(_) => MessageKind::Sync, - ParsedMessage::Ping(_) => MessageKind::Ping, - ParsedMessage::Pong(_) => MessageKind::Pong, + ParsedMessage::Status(_) => MessageKind::Status, + ParsedMessage::StatusReply(_) => MessageKind::StatusReply, ParsedMessage::Blob(_) => MessageKind::Blob, ParsedMessage::Who(_) => MessageKind::Who, ParsedMessage::Me(_) => MessageKind::Me, + ParsedMessage::Ping(_) => MessageKind::Ping, + ParsedMessage::Pong(_) => MessageKind::Pong, } } /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them - /// but WHO and ME (which the node never sees). + /// but WHO, ME, PING and PONG (which the node never sees). pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> { match self { ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)), ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)), ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)), - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => { - Some((keepalive.epoch, keepalive.epoch_seq)) + ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => { + Some((status.epoch, status.epoch_seq)) } ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)), - ParsedMessage::Who(_) | ParsedMessage::Me(_) => None, + ParsedMessage::Who(_) + | ParsedMessage::Me(_) + | ParsedMessage::Ping(_) + | ParsedMessage::Pong(_) => None, } } } @@ -330,6 +349,40 @@ pub(crate) struct ListedPeer { pub(crate) addresses: Vec<String>, } +/// Where messages to a peer go now (`Transport::destination`). +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Destination { + /// As the identity layer keeps it: `<scheme>:<transport identity>`. + pub(crate) address: String, + /// The peer proved itself there; else it is only said to be there, and sends wait + /// for its proof. + pub(crate) proven: bool, + /// The transport's connection to it, if the transport has connections. + pub(crate) connection: Option<Connection>, + /// Every address it proved itself at, best first, as the paths layer measures them + /// (`transport::paths`). + pub(crate) paths: Vec<stats::PathInfo>, +} + +/// How a transport with connections (libp2p) is connected to a peer. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Connection { + Direct, + /// Only through a relay. + Relayed, +} + +impl Connection { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Direct => "direct", + Self::Relayed => "relayed", + } + } +} + #[derive(Clone, Serialize, Deserialize)] pub(crate) struct HelloReply { pub(crate) swarm_id: String, @@ -504,6 +557,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`). @@ -513,6 +569,28 @@ pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn + 'static; pub(crate) type MessageHandler = Box<HandlerFn>; +/// Whether `message` is a PING or PONG. The paths layer times them, so a transport's +/// consumer hands them on as soon as they are read (`hand_on_probe`), not after the +/// messages read before them are handled (which can take a while: sync). +pub(crate) fn is_probe(message: &ParsedMessage) -> bool { + matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) +} + +/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own. +pub(crate) fn hand_on_probe( + handler: &std::sync::Arc<HandlerFn>, + sender: String, + message: ParsedMessage, + payload_len: usize, +) { + let handler = std::sync::Arc::clone(handler); + tokio::spawn(async move { + if let Err(error) = handler(sender.clone(), message, payload_len).await { + tracing::warn!(peer = %sender, error = %format!("{error:#}"), "Failed to process inbound message"); + } + }); +} + #[async_trait] pub(crate) trait Transport: Send + Sync { /// This node's identity on the transport (for Mycelium: its public key). @@ -584,6 +662,47 @@ pub(crate) trait Transport: Send + Sync { let _ = (peer, addresses); } + /// Where messages to `peer` go now. None if nowhere, and by default, for a + /// transport whose identities are addresses (`peer` is where they go). + fn destination(&self, peer: &str) -> Option<Destination> { + let _ = peer; + None + } + + /// Where messages to `target` (a transport identity) go now, as an address that + /// names `target` too: the remote address of the connection in use, say. By default + /// nowhere but `target` itself (Mycelium). + fn location(&self, target: &str) -> Option<String> { + let _ = target; + None + } + + /// How this node is connected to `target` (a transport identity), if it is. By + /// default not at all, for a transport without connections (Mycelium). + fn connection(&self, target: &str) -> Option<Connection> { + let _ = target; + None + } + + /// The smoothed round-trip time of the path messages to `peer` go over, if + /// measured (`transport::paths`). By default none. + fn rtt(&self, peer: &str) -> Option<std::time::Duration> { + let _ = peer; + None + } + + /// Whether the transport answers PINGs with PONGs itself (the paths layer, + /// `transport::paths`); if not (a bare transport, in tests), the node does. By + /// default not. + fn answers_pings(&self) -> bool { + false + } + + /// Figures about the transport itself (`stats`). By default none. + fn stats(&self) -> stats::TransportStats { + stats::TransportStats::default() + } + /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's /// identity, as `local_id` gives it on that node. async fn resolve_seed(&self, seed: &str) -> Result<String>; diff --git a/src/transport/mycelium/daemon.rs b/src/transport/mycelium/daemon.rs index f986d52..58ebe0c 100644 --- a/src/transport/mycelium/daemon.rs +++ b/src/transport/mycelium/daemon.rs @@ -66,19 +66,14 @@ impl MyceliumSetup { pub(crate) async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> { // TODO warn for all ignored options if rest_url_opt is set if let Some(url) = config.rest_url_opt { - tracing::info!( - "Configured to interface with an external standalone Mycelium instance at: {}", - url - ); + tracing::info!(url = %url, "Using a daemon running elsewhere"); return Ok(Self { runtime: MyceliumRuntime::External, rest_url: url, }); } - tracing::info!( - "Configured to initialize an embedded user-space Mycelium background worker" - ); + tracing::info!("Starting the supervised daemon"); let key = match config.key_file_opt { Some(key_file) => KeySource::Path(key_file), @@ -101,7 +96,7 @@ impl MyceliumSetup { let supervised_node = SupervisedMycelium::spawn(key, private_net_config, config.tcp_port, &config.peers) .await - .context("Failed to securely bootstrap embedded Mycelium background worker")?; + .context("Starting the supervised Mycelium daemon")?; let resolved_url = Url::parse(&format!( "http://127.0.0.1:{}", @@ -204,7 +199,7 @@ impl SupervisedMycelium { } // Own process group: a Ctrl-C in the terminal goes to continuum only, // which then stops Mycelium itself, instead of both exiting at once (which - // shows up as "Mycelium runtime process exited"). + // shows up as "A transport stopped"). if libc::setpgid(0, 0) != 0 { return Err(std::io::Error::last_os_error()); } @@ -252,9 +247,7 @@ impl SupervisedMycelium { cmd.stdout(Stdio::piped()); cmd.stderr(Stdio::piped()); - let mut child = cmd - .spawn() - .context("Failed to execute mycelium process daemon")?; + let mut child = cmd.spawn().context("Running `mycelium`")?; // Its output goes where ours does. if let Some(mut stdout) = child.stdout.take() { diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 354aebc..2bb399c 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -8,7 +8,9 @@ use thiserror::Error; use url::Url; use crate::transport::wire; -use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; +use crate::transport::{ + EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, is_probe, +}; pub(crate) mod daemon; @@ -28,7 +30,7 @@ const INBOX_POLLS: usize = 2; /// (`post_timeout`). const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); /// The slowest our Mycelium is expected to take a message body, in bytes per second: -/// a large sync message gets more time than a PING, instead of failing (and being +/// a large sync message gets more time than a STATUS, instead of failing (and being /// generated and sent whole again) at a fixed limit. const POST_MIN_RATE: u64 = 1 << 20; @@ -144,11 +146,11 @@ impl MyceliumTransport { Ok(response) if response.status() == StatusCode::NO_CONTENT => break, Ok(response) if response.status().is_success() => discarded += 1, Ok(response) => { - tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox"); + tracing::warn!(status = %response.status(), "Couldn't drain the inbox"); break; } Err(error) => { - tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox"); + tracing::warn!(error = %error, "Couldn't drain the inbox"); break; } } @@ -274,7 +276,7 @@ impl Transport for MyceliumTransport { let mut other_versions = std::collections::HashSet::new(); tracing::info!( endpoint = self.messages_endpoint.as_str(), - "Starting Mycelium message consumer" + "Starting the consumer" ); // Several polls at once, so one message's HTTP round trip overlaps the next one's @@ -289,6 +291,8 @@ impl Transport for MyceliumTransport { } drop(tx); + // PINGs and PONGs aren't handled after the message before them (`is_probe`). + let handler: std::sync::Arc<HandlerFn> = std::sync::Arc::from(handler); while let Some(inbound_message) = rx.recv().await { match parse_message(&inbound_message) { Ok((sender, parsed)) => { @@ -300,7 +304,9 @@ impl Transport for MyceliumTransport { "Received message" ); let payload_len = base64_decoded_len(&inbound_message.payload); - if let Err(error) = handler(sender, parsed, payload_len).await { + if is_probe(&parsed) { + hand_on_probe(&handler, sender, parsed, payload_len); + } else if let Err(error) = handler(sender, parsed, payload_len).await { tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); } } @@ -351,7 +357,7 @@ async fn poll_inbox( tx: tokio::sync::mpsc::Sender<InboundMessage>, ) { loop { - tracing::trace!("Polling Mycelium inbox"); + tracing::trace!("Polling the inbox"); let response = match client .get(endpoint.clone()) @@ -362,7 +368,7 @@ async fn poll_inbox( { Ok(resp) => resp, Err(error) => { - tracing::warn!(error = %error, "Mycelium inbox poll failed"); + tracing::warn!(error = %error, "Polling the inbox failed"); tokio::time::sleep(Duration::from_millis(250)).await; continue; } @@ -371,13 +377,13 @@ async fn poll_inbox( let status = response.status(); if status == StatusCode::NO_CONTENT { - tracing::trace!("Mycelium inbox is empty"); + tracing::trace!("The inbox is empty"); tokio::time::sleep(EMPTY_QUEUE_DELAY).await; continue; } if !status.is_success() { - tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status"); + tracing::warn!(status = %status, "Polling the inbox failed"); tokio::time::sleep(Duration::from_millis(500)).await; continue; } @@ -392,7 +398,7 @@ async fn poll_inbox( continue; } Err(error) => { - tracing::warn!(error = %error, "Failed to read raw Mycelium response stream"); + tracing::warn!(error = %error, "Couldn't read a message from the inbox"); tokio::time::sleep(Duration::from_millis(100)).await; continue; } @@ -404,7 +410,7 @@ async fn poll_inbox( tracing::error!( error = %error, raw_payload = %String::from_utf8_lossy(&raw_bytes), - "Failed to decode inbound Mycelium message structure" + "Dropping a message that doesn't decode" ); tokio::time::sleep(Duration::from_millis(500)).await; continue; @@ -450,6 +456,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, }) } @@ -552,7 +559,7 @@ pub(crate) enum TransportError { mod tests { use super::*; use crate::transport::{ - BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, Keepalive, MAX_LISTED_HEADS, + BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, MAX_LISTED_HEADS, Status, }; use automerge::ChangeHash; @@ -571,7 +578,7 @@ mod tests { #[test] fn largest_keepalive_fits_in_one_chunk() { // A digest with a full sample: the most a keepalive carries. - let keepalive = Keepalive { + let status = Status { heads: HeadsSummary::of( (0..=MAX_LISTED_HEADS as u8) .map(|n| ChangeHash([n; 32])) @@ -585,17 +592,17 @@ mod tests { ready: true, syncs_with_you: true, }; - let payload = wire::encode(&ParsedMessage::Pong(keepalive.clone())).unwrap(); + let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap(); // Mycelium's message data: a length byte, the topic, then our payload. let size = 1 + TOPIC.len() + payload.len(); assert!( size + 50 <= MYCELIUM_CHUNK, - "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes" + "STATUS with {MAX_LISTED_HEADS} heads takes {size} bytes" ); - let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else { - panic!("not a PONG"); + let ParsedMessage::StatusReply(decoded) = wire::decode(&payload).unwrap() else { + panic!("not a STATUS_REPLY"); }; - assert_eq!(decoded.heads, keepalive.heads); + assert_eq!(decoded.heads, status.heads); } #[test] @@ -648,17 +655,17 @@ mod tests { #[test] fn messages_round_trip_through_the_api_body() { - let keepalive = ParsedMessage::Ping(Keepalive { + let status = ParsedMessage::Status(Status { heads: HeadsSummary::of(vec![ChangeHash([1; 32])]), epoch: Epoch::GENESIS, epoch_seq: 0, ready: true, syncs_with_you: true, }); - let encoded = encode_message("peer", &keepalive).unwrap(); + let encoded = encode_message("peer", &status).unwrap(); assert!(matches!( decode_message(&encoded.body).unwrap(), - ParsedMessage::Ping(_) + ParsedMessage::Status(_) )); } diff --git a/src/transport/paths.rs b/src/transport/paths.rs new file mode 100644 index 0000000..ba16794 --- /dev/null +++ b/src/transport/paths.rs @@ -0,0 +1,1247 @@ +//! 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`). +//! +//! A message to a node with several addresses goes to the first of them that takes it +//! (`send_first`), in the order `rank` gives. 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. +//! +//! Paths measures the addresses of such nodes itself (and every path the node's +//! keepalive PINGs go over, below): every `PROBE_EVERY`, each address it had a choice of +//! within `ACTIVE_FOR` gets a PING with a random nonce, which the +//! 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. +//! +//! 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, Weak}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use tokio::time::Instant; +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, 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>> + + 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) +} + +/// A path whose send failed is down for this long (unless it answers a PING, or a +/// send there works again). +const DOWN_FOR: Duration = Duration::from_secs(60); +/// How often a path in use is pinged, how long an answer may take, and how many PINGs +/// in a row unanswered make it down. +const PROBE_EVERY: Duration = Duration::from_secs(30); +const PROBE_TIMEOUT: Duration = Duration::from_secs(15); +const UNANSWERED_DOWN: u32 = 3; +/// How often the probe task looks for paths due. +const PROBE_TICK: Duration = Duration::from_secs(5); +/// Paths this node had a choice of within this long are probed; ones it didn't use for +/// twice as long are forgotten. +const ACTIVE_FOR: Duration = Duration::from_secs(300); +/// The path in use is changed for one this much faster (a fraction of its round-trip +/// time), and at least `SWITCH_MIN`. +const SWITCH_GAIN: f64 = 0.25; +const SWITCH_MIN: Duration = Duration::from_millis(5); +/// Paths kept at most: they are of peers this node chose among, but bounded anyway. +const MAX_PATHS: usize = 4096; + +/// What paths knows about one path (an address as paths keep it). +#[derive(Debug, Default)] +struct PathState { + /// When it was last among several to choose from (`rank`). + used: Option<Instant>, + /// Smoothed round-trip time of its PINGs. + rtt: Option<Duration>, + /// The PING out, by nonce, and when it went. + probe: Option<(u64, Instant)>, + last_probe: Option<Instant>, + /// PINGs in a row unanswered. + unanswered: u32, + /// When a send there last failed. + failed_at: Option<Instant>, +} + +impl PathState { + fn is_down(&self, now: Instant) -> bool { + self.unanswered >= UNANSWERED_DOWN + || self + .failed_at + .is_some_and(|at| now.saturating_duration_since(at) < DOWN_FOR) + } +} + +/// Every path's state, and which one is in use for each set to choose from. +#[derive(Debug, Default)] +struct Table { + paths: HashMap<String, PathState>, + /// By the set's addresses (sorted, joined): the one in use. + chosen: HashMap<String, String>, + counts: PathCounts, +} + +impl Table { + /// `usable`, with each one's place in the configured order, best first (as + /// `Paths::rank` says), noting the choice. + fn rank(&mut self, mut usable: Vec<(String, usize)>, now: Instant) -> Vec<String> { + usable.sort_by_key(|(_, preference)| *preference); + usable.dedup_by(|a, b| a.0 == b.0); + if usable.len() < 2 { + return usable.into_iter().map(|(address, _)| address).collect(); + } + if self.paths.len() >= MAX_PATHS { + self.forget_unused(now); + } + for (address, _) in &usable { + if self.paths.len() < MAX_PATHS || self.paths.contains_key(address) { + self.paths.entry(address.clone()).or_default().used = Some(now); + } + } + let rtt = |table: &Self, address: &str| table.paths.get(address).and_then(|path| path.rtt); + let (up, down): (Vec<_>, Vec<_>) = usable.into_iter().partition(|(address, _)| { + !self + .paths + .get(address) + .is_some_and(|path| path.is_down(now)) + }); + let mut set: Vec<&str> = up + .iter() + .chain(down.iter()) + .map(|(address, _)| address.as_str()) + .collect(); + set.sort_unstable(); + let key = set.join(" "); + let mut ranked: Vec<String> = Vec::new(); + if !up.is_empty() { + let mut current = self + .chosen + .get(&key) + .filter(|chosen| up.iter().any(|(address, _)| address == *chosen)) + .cloned() + .unwrap_or_else(|| up[0].0.clone()); + let fastest = up + .iter() + .filter_map(|(address, _)| Some((address, rtt(self, address)?))) + .min_by_key(|(_, rtt)| *rtt); + if let (Some((fastest, fastest_rtt)), Some(current_rtt)) = + (fastest, rtt(self, ¤t)) + && *fastest != current + && current_rtt.saturating_sub(fastest_rtt) + >= SWITCH_MIN.max(current_rtt.mul_f64(SWITCH_GAIN)) + { + info!( + from = %current, + to = %fastest, + from_rtt = ?current_rtt, + to_rtt = ?fastest_rtt, + "Switching to a faster path" + ); + self.counts.switches += 1; + current = fastest.clone(); + } + if self.chosen.len() < MAX_PATHS || self.chosen.contains_key(&key) { + self.chosen.insert(key, current.clone()); + } + // The rest: measured ones by round-trip time, then unmeasured ones in the + // configured order. + let mut rest: Vec<&(String, usize)> = up + .iter() + .filter(|(address, _)| *address != current) + .collect(); + rest.sort_by_key(|(address, preference)| { + ( + rtt(self, address).is_none(), + rtt(self, address), + *preference, + ) + }); + ranked.push(current); + ranked.extend(rest.into_iter().map(|(address, _)| address.clone())); + } + ranked.extend(down.into_iter().map(|(address, _)| address)); + ranked + } + + /// A send to `address` failed. + fn failed(&mut self, address: &str, now: Instant) { + self.counts.send_failures += 1; + if let Some(path) = self.paths.get_mut(address) { + path.failed_at = Some(now); + // A PING that didn't go out gets no answer; it is no unanswered one. + path.probe = None; + } + } + + /// A send to `address` worked. + fn sent(&mut self, address: &str) { + if let Some(path) = self.paths.get_mut(address) { + path.failed_at = None; + } + } + + /// The PINGs due now, with their nonces (from `nonce`): to every path used within + /// `ACTIVE_FOR` that wasn't pinged within `PROBE_EVERY` and has no PING out. A PING + /// out for `PROBE_TIMEOUT` counts as unanswered. + fn due(&mut self, now: Instant, mut nonce: impl FnMut() -> u64) -> Vec<(String, u64)> { + self.forget_unused(now); + let mut due = Vec::new(); + for (address, path) in &mut self.paths { + if let Some((_, sent)) = path.probe + && now.saturating_duration_since(sent) >= PROBE_TIMEOUT + { + path.probe = None; + path.unanswered += 1; + self.counts.probes_unanswered += 1; + if path.unanswered == UNANSWERED_DOWN { + info!(address = %address, "A path stopped answering PINGs; using others"); + } + } + let active = path + .used + .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR); + let rested = path + .last_probe + .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY); + if active && rested && path.probe.is_none() { + let nonce = nonce(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; + due.push((address.clone(), nonce)); + } + } + due + } + + /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that + /// path, which its PONG measures, even if it is a peer's only path (then it isn't + /// pinged otherwise: `due` pings only paths there was a choice of). + fn pinged(&mut self, address: &str, nonce: u64, now: Instant) { + if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) { + self.forget_unused(now); + if self.paths.len() >= MAX_PATHS { + return; + } + } + let path = self.paths.entry(address.to_owned()).or_default(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; + } + + /// The PING with `nonce` to `address` (from `due`) goes now: its round trip counts + /// from here, not from when it was found due. + fn sending(&mut self, address: &str, nonce: u64, now: Instant) { + if let Some(path) = self.paths.get_mut(address) + && let Some((expected, sent)) = &mut path.probe + && *expected == nonce + { + *sent = now; + } + } + + /// 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)?; + let (expected, sent) = path.probe?; + if expected != nonce { + return None; + } + let sample = now.saturating_duration_since(sent); + path.probe = None; + if path.unanswered >= UNANSWERED_DOWN { + info!(address = %address, "A path answers PINGs again"); + } + path.unanswered = 0; + path.failed_at = None; + path.rtt = Some(match path.rtt { + Some(rtt) => (rtt * 3 + sample) / 4, + None => sample, + }); + self.counts.probes_answered += 1; + path.rtt + } + + /// Forgets the paths neither used nor pinged for twice `ACTIVE_FOR`, and choices + /// among them. + fn forget_unused(&mut self, now: Instant) { + let recent = |at: Option<Instant>| { + at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR * 2) + }; + self.paths + .retain(|_, path| recent(path.used) || recent(path.last_probe)); + let paths = &self.paths; + self.chosen + .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address))); + } + + fn info(&self, address: &str, now: Instant) -> PathInfo { + let path = self.paths.get(address); + PathInfo { + address: address.to_owned(), + rtt_ms: path + .and_then(|path| path.rtt) + .map(|rtt| rtt.as_secs_f64() * 1000.0), + down: path.is_some_and(|path| path.is_down(now)), + } + } +} + +pub(crate) struct Paths { + /// In order of preference. + links: Vec<Link>, + /// 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; 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 + .into_iter() + .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. + 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) + } + + /// The usable ones of `addresses` (as paths keep them), best first: paths that are + /// up before those down, the one in use first, then by round-trip time, then in + /// the configured order (module docs). Several addresses are paths to measure. + pub(crate) fn rank(&self, addresses: &[String]) -> Vec<String> { + let usable: Vec<(String, usize)> = addresses + .iter() + .filter(|address| self.usable(address)) + .map(|address| (address.clone(), self.preference(address))) + .collect(); + self.table.lock_or_recover().rank(usable, Instant::now()) + } + + /// The smoothed round-trip time of the path to `address`, if measured. + pub(crate) fn rtt(&self, address: &str) -> Option<Duration> { + self.table.lock_or_recover().paths.get(address)?.rtt + } + + /// What paths knows about each of `addresses`. + pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> { + let now = Instant::now(); + let infos = { + let table = self.table.lock_or_recover(); + addresses + .iter() + .map(|address| table.info(address, now)) + .collect() + }; + self.located(infos) + } + + /// `infos` with where each path goes, where the transport knows + /// (`Transport::location`); asked outside the table's lock. + fn located(&self, mut infos: Vec<PathInfo>) -> Vec<PathInfo> { + for info in &mut infos { + if let Some((link, location)) = self + .split(&info.address) + .and_then(|(link, target)| Some((link, link.inner.location(&target)?))) + { + info.address = link.address(&location); + } + } + infos + } + + /// 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`, 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) + .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(); + 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; + } + } + }; + // A PING is noted before it goes: on a fast link, its PONG may be back + // before the send returns. + if let Some(ParsedMessage::Ping(nonce)) = message.message.as_deref() { + self.table + .lock_or_recover() + .pinged(address, *nonce, Instant::now()); + } + match self.send(&encoded).await { + Ok(()) => { + let mut table = self.table.lock_or_recover(); + table.sent(address); + if !failures.is_empty() { + table.counts.failovers += 1; + drop(table); + debug!( + peer = %message.peer, + address = %address, + failed = %describe(&failures), + "Sent at another address after a send failed" + ); + } + return Ok(address.clone()); + } + Err(error) => { + self.table.lock_or_recover().failed(address, 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` (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))?; + 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) + } + + /// Paths' own figures, and every transport's. + pub(crate) fn stats(&self) -> TransportStats { + let mut stats = TransportStats::default(); + for link in &self.links { + stats.merge(link.inner.stats()); + } + let now = Instant::now(); + let (paths, counts) = { + let table = self.table.lock_or_recover(); + let paths: Vec<PathInfo> = table + .paths + .keys() + .map(|address| table.info(address, now)) + .collect(); + (paths, table.counts) + }; + let mut paths = self.located(paths); + paths.sort_by(|a, b| a.address.cmp(&b.address)); + stats.paths = Some(PathsStats { paths, counts }); + stats + } + + /// 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) + } else { + ParsedMessage::Ping(nonce) + }; + let encoded = self.encode(address, &message)?; + let sent = self.send(&encoded).await; + if let Some(traffic) = &self.traffic { + match &sent { + Ok(()) => traffic.sent(address, encoded.kind, encoded.payload_len), + Err(_) => traffic.refused(address, encoded.kind, encoded.payload_len), + } + } + sent + } + + /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. Each PING + /// goes on its own, timed from when it goes: a send that takes long (a dial) neither + /// holds up the others nor counts in their round trips. + async fn probe(self: Arc<Self>) { + let mut tick = tokio::time::interval(PROBE_TICK); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let due = self + .table + .lock_or_recover() + .due(Instant::now(), random_nonce); + for (address, nonce) in due { + let paths = Arc::clone(&self); + tokio::spawn(async move { + paths + .table + .lock_or_recover() + .sending(&address, nonce, Instant::now()); + if let Err(error) = paths.send_probe(&address, false, nonce).await { + debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING"); + paths + .table + .lock_or_recover() + .failed(&address, Instant::now()); + } + }); + } + } + } + + /// 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, + message: ParsedMessage, + 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) => { + let paths = Arc::clone(&self); + let to = address.clone(); + tokio::spawn(async move { + if let Err(error) = paths.send_probe(&to, true, nonce).await { + debug!(to = %to, error = %format!("{error:#}"), "Couldn't answer a PING"); + } + }); + } + ParsedMessage::Pong(nonce) => { + self.table + .lock_or_recover() + .pong(&address, nonce, Instant::now()); + } + _ => {} + } + handler(address, message, payload_len).await + } + + /// Every transport's consumer, each message handed to `handler` under the sender's + /// address (PINGs and PONGs end here), and the probe task; returns when the first + /// consumer does. + pub(crate) async fn start_consumer(self: &Arc<Self>, handler: Arc<PathHandler>) -> Result<()> { + let probing = tokio::spawn(Arc::clone(self).probe()); + 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); + let paths = Arc::clone(self); + consumers.push(Box::pin(async move { + inner + .start_consumer(Box::new(move |sender, message, payload_len| { + let paths = Arc::clone(&paths); + let handler = Arc::clone(&handler); + Box::pin(async move { + paths + .receive( + format!("{scheme}:{sender}"), + message, + payload_len, + &*handler, + ) + .await + }) + })) + .await + })); + } + let result = first(consumers) + .await + .unwrap_or_else(|| Err(anyhow!("no transports"))); + probing.abort(); + result + } + + /// 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"))) + } +} + +/// A PING's nonce: random, so a PONG can't be made up for a PING not seen. +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. + return std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_nanos() as u64); + } + u64::from_le_bytes(bytes) +} + +/// `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>>>, +) -> 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() -> (Arc<Paths>, Arc<Paths>) { + let (first, second) = (SimNet::new(1), SimNet::new(2)); + let on_both = |id: &str| { + Arc::new(Paths::new( + vec![ + ("sim", first.transport(id) as Arc<dyn Transport>), + ("other", second.transport(id) as Arc<dyn Transport>), + ], + None, + None, + )) + }; + (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()); + } + + fn paths(addresses: &[&str]) -> Vec<(String, usize)> { + addresses + .iter() + .enumerate() + .map(|(preference, address)| (address.to_string(), preference)) + .collect() + } + + /// Answers every PING due `rtt` later, as from the path it went to (`rtts`). + fn answer(table: &mut Table, now: Instant, rtts: &[(&str, u64)]) -> Instant { + let mut nonces = 0; + let due = table.due(now, || { + nonces += 1; + nonces + }); + for (address, nonce) in due { + let rtt = rtts + .iter() + .find(|(to, _)| *to == address) + .map(|(_, ms)| *ms); + if let Some(ms) = rtt { + table.pong(&address, nonce, now + Duration::from_millis(ms)); + } + } + now + PROBE_EVERY + } + + #[test] + fn the_configured_order_until_measured_then_the_faster_path() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]); + // libp2p a bit slower: not enough to switch. + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:x", 110), ("mycelium:x", 100)]); + } + assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]); + assert_eq!(table.counts.switches, 0); + // Much slower: switched, and it stays switched while that holds. + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:x", 300), ("mycelium:x", 100)]); + } + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.counts.switches, 1); + // Small differences under SWITCH_MIN never switch. + let lan = paths(&["libp2p:y", "mycelium:y"]); + table.rank(lan.clone(), now); + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:y", 4), ("mycelium:y", 1)]); + } + assert_eq!(table.rank(lan, now), ["libp2p:y", "mycelium:y"]); + } + + #[test] + fn unanswered_or_failing_paths_are_down() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + table.rank(set.clone(), now); + // libp2p stops answering: down after UNANSWERED_DOWN PINGs, then last. + for _ in 0..=UNANSWERED_DOWN { + now = answer(&mut table, now, &[("mycelium:x", 100)]); + table.rank(set.clone(), now); + } + assert!(table.info("libp2p:x", now).down); + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + // It answers again: up, and measured. + now = answer(&mut table, now, &[("libp2p:x", 50), ("mycelium:x", 100)]); + assert!(!table.info("libp2p:x", now).down); + + // A failed send: down for DOWN_FOR, unless a send there works. + table.failed("mycelium:x", now); + assert!(table.info("mycelium:x", now).down); + assert!(!table.info("mycelium:x", now + DOWN_FOR).down); + table.sent("mycelium:x"); + assert!(!table.info("mycelium:x", now).down); + } + + #[test] + fn only_paths_in_use_are_pinged() { + let mut table = Table::default(); + let now = Instant::now(); + let mut nonce = 0; + let mut next = || { + nonce += 1; + nonce + }; + // One address: nothing to choose, nothing to measure. + table.rank(paths(&["libp2p:z"]), now); + assert!(table.due(now, &mut next).is_empty()); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + assert_eq!(table.due(now, &mut next).len(), 2); + // Not again within PROBE_EVERY; a stale answer is ignored. + assert!(table.due(now + PROBE_TIMEOUT, &mut next).is_empty()); + assert_eq!(table.pong("libp2p:x", 99, now), None); + // Unused for ACTIVE_FOR: no more PINGs; for twice as long: forgotten. + assert!(table.due(now + ACTIVE_FOR, &mut next).is_empty()); + table.due(now + ACTIVE_FOR * 2, &mut next); + assert!(table.paths.is_empty() && table.chosen.is_empty()); + } + + /// A PING's round trip counts from when it goes, not from when it was found due. + #[test] + fn round_trips_count_from_the_send() { + let mut table = Table::default(); + let now = Instant::now(); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + let due = table.due(now, || 7); + assert_eq!(due.len(), 2); + // Sent a second late (after another path's slow send). + let sent = now + Duration::from_secs(1); + table.sending("libp2p:x", 7, sent); + table.sending("libp2p:x", 8, sent + Duration::from_secs(1)); // not its PING + table.pong("libp2p:x", 7, sent + Duration::from_millis(100)); + let rtt = table.info("libp2p:x", now).rtt_ms.unwrap(); + assert!((rtt - 100.0).abs() < 1e-6, "{rtt}"); + } + + /// 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's only path is measured by its keepalives, and not pinged otherwise. + table.pinged("libp2p:z", 8, now); + assert!( + table + .pong("libp2p:z", 8, now + Duration::from_millis(30)) + .is_some() + ); + let rtt = table.info("libp2p:z", now).rtt_ms.unwrap(); + assert!((rtt - 30.0).abs() < 1e-6, "{rtt}"); + let later = now + PROBE_EVERY; + assert!( + table + .due(later, &mut next) + .iter() + .all(|(address, _)| address != "libp2p:z") + ); + // Forgotten once the keepalives stop. + table.due(now + ACTIVE_FOR * 2, &mut next); + 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; + + #[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>), + ], + 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, _, _| { + 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] + 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 = some_status(); + 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()); + } +} diff --git a/src/transport/sim.rs b/src/transport/sim.rs index 149798b..500050b 100644 --- a/src/transport/sim.rs +++ b/src/transport/sim.rs @@ -319,10 +319,10 @@ impl Transport for SimTransport { mod tests { use super::*; use crate::document::replica::Epoch; - use crate::transport::{HeadsSummary, Keepalive}; + use crate::transport::{HeadsSummary, Status}; - fn ping() -> ParsedMessage { - ParsedMessage::Ping(Keepalive { + fn status() -> ParsedMessage { + ParsedMessage::Status(Status { heads: HeadsSummary::of(Vec::new()), epoch: Epoch::GENESIS, epoch_seq: 0, @@ -353,13 +353,13 @@ mod tests { let received = record(net.transport("b")); let settle = || tokio::time::sleep(Duration::from_millis(30)); - a.send_message("b", &ping()).await.unwrap(); + a.send_message("b", &status()).await.unwrap(); settle().await; net.take_down("b"); - a.send_message("b", &ping()).await.unwrap(); + a.send_message("b", &status()).await.unwrap(); settle().await; net.bring_up("b"); - a.send_message("b", &ping()).await.unwrap(); + a.send_message("b", &status()).await.unwrap(); settle().await; assert_eq!(*received.lock().unwrap(), vec!["a", "a"]); @@ -381,7 +381,7 @@ mod tests { let a = net.transport("a"); let _b = net.transport("b"); for _ in 0..200 { - a.send_message("b", &ping()).await.unwrap(); + a.send_message("b", &status()).await.unwrap(); } tokio::time::sleep(Duration::from_millis(50)).await; net.counts() @@ -408,8 +408,8 @@ mod tests { let at_a = record(Arc::clone(&a)); let at_b = record(Arc::clone(&b)); - a.send_message("b", &ping()).await.unwrap(); - b.send_message("a", &ping()).await.unwrap(); + a.send_message("b", &status()).await.unwrap(); + b.send_message("a", &status()).await.unwrap(); tokio::time::sleep(Duration::from_millis(30)).await; assert!(at_b.lock().unwrap().is_empty()); @@ -420,8 +420,8 @@ mod tests { async fn inject_returns_once_handled() { let net = SimNet::new(1); let received = record(net.transport("b")); - net.inject("ghost", "b", &ping()).await.unwrap(); + net.inject("ghost", "b", &status()).await.unwrap(); assert_eq!(*received.lock().unwrap(), vec!["ghost"]); - assert!(net.inject("ghost", "nobody", &ping()).await.is_err()); + assert!(net.inject("ghost", "nobody", &status()).await.is_err()); } } diff --git a/src/transport/stats.rs b/src/transport/stats.rs new file mode 100644 index 0000000..398ffdf --- /dev/null +++ b/src/transport/stats.rs @@ -0,0 +1,189 @@ +//! Figures about the transports themselves (`Transport::stats`), for /admin/stats +//! (`transports`, `docs/measurements.md`): how they find and reach peers, besides the +//! messages counted per peer (`traffic`). + +use std::collections::BTreeMap; + +use serde::Serialize; + +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct TransportStats { + /// The identity layer's (`transport::identity`); absent without one. + pub(crate) identity: Option<IdentityStats>, + /// The paths layer's (`transport::paths`); absent without one. + pub(crate) paths: Option<PathsStats>, + /// The libp2p transport's; absent if it isn't one of this node's. + pub(crate) libp2p: Option<Libp2pStats>, +} + +impl TransportStats { + /// Takes what `other` has. + pub(crate) fn merge(&mut self, other: TransportStats) { + if other.identity.is_some() { + self.identity = other.identity; + } + if other.paths.is_some() { + self.paths = other.paths; + } + if other.libp2p.is_some() { + self.libp2p = other.libp2p; + } + } +} + +/// How the paths to peers with several addresses do (`transport::paths`). +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PathsStats { + /// Every path measured: those of peers this node chose among lately, and those its + /// keepalives went over. + pub(crate) paths: Vec<PathInfo>, + #[serde(flatten)] + pub(crate) counts: PathCounts, +} + +/// One path, as `transport::paths` measures it. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PathInfo { + /// `<scheme>:<where it goes>`: where the transport says the path goes now + /// (`Transport::location`; libp2p: the remote address of its connection, ending in + /// the peer ID), else whom it goes to, `<scheme>:<transport identity>` (a libp2p + /// peer ID without a connection, a Mycelium public key, an onion service). + pub(crate) address: String, + /// Smoothed round-trip time of its PINGs (ours, and our keepalives); absent before + /// an answer. + pub(crate) rtt_ms: Option<f64>, + /// Its PINGs went unanswered three times in a row, or a send there failed within a + /// minute: used only when no other path works. + pub(crate) down: bool, +} + +/// Counts since start. +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PathCounts { + pub(crate) probes_sent: u64, + pub(crate) probes_answered: u64, + pub(crate) probes_unanswered: u64, + /// Sends that failed on a path (each tried at the next). + pub(crate) send_failures: u64, + /// Sends that went over another path after one failed. + pub(crate) failovers: u64, + /// Changes of the path in use for one faster. + pub(crate) switches: u64, +} + +/// Who is where (`transport::identity`): the books now, and counts since start. +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct IdentityStats { + /// Nodes proven at an address, by transport (scheme); a node on several counts on + /// each. + pub(crate) proven: BTreeMap<String, usize>, + /// Nodes known only from hints (the member list, HELLO_REPLYs), proven nowhere. + pub(crate) hinted: usize, + #[serde(flatten)] + pub(crate) counts: IdentityCounts, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct IdentityCounts { + /// Proofs that didn't hold (a bad signature, or made for other addresses): normally + /// none. + pub(crate) proofs_rejected: u64, + /// Nodes that proved themselves at another address on a transport than before: a + /// node restarted with a new transport identity (a new libp2p key or Mycelium key). + pub(crate) address_changes: u64, + /// Messages dropped because no node proved itself at their address: some after a + /// restart are normal, a steady stream is not. + pub(crate) dropped_unproven: u64, + /// Sends refused because the peer hadn't proven where it is yet (`NotProven`). + pub(crate) refused_unproven: u64, + /// Seeds no node answered at in time. + pub(crate) seeds_unanswered: u64, +} + +/// Whether others can dial this node, as AutoNAT found out. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Reachability { + #[default] + Unknown, + /// Others can't dial it: it needs a relay. + Private, + Public, +} + +impl Reachability { + pub(crate) const ALL: [Reachability; 3] = [Self::Unknown, Self::Private, Self::Public]; + + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Unknown => "unknown", + Self::Private => "private", + Self::Public => "public", + } + } +} + +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Libp2pStats { + pub(crate) reachability: Reachability, + /// The address AutoNAT confirmed others can dial, while public. + pub(crate) public_address: Option<String>, + /// Open connections, now. + pub(crate) connections: Connections, + pub(crate) relay_slots: RelaySlots, + /// Direct connections tried over a relayed one (DCUtR), since start. + pub(crate) hole_punches: HolePunches, + /// Connections that couldn't be made, by cause, since start. + pub(crate) dial_failures: DialFailures, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Connections { + pub(crate) direct: usize, + /// Through a relay (a circuit). + pub(crate) relayed: usize, +} + +/// Slots this node holds on relays, to be reachable through them. +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RelaySlots { + /// Held now. + pub(crate) held: usize, + /// Relays picked that never granted a slot, not picked again for a while now. + pub(crate) paused: usize, + /// Since start: slots granted (not counting renewals), lost after being granted, and + /// asked for but never granted (a relay unreachable, or another peer at its + /// address: a relay restarted without its libp2p key). + pub(crate) granted: u64, + pub(crate) lost: u64, + pub(crate) not_granted: u64, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct HolePunches { + pub(crate) succeeded: u64, + /// The relay carries on. + pub(crate) failed: u64, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct DialFailures { + /// No address of the peer could be reached (refused, timed out, no route). + pub(crate) unreachable: u64, + /// Another peer answered at the address: one restarted without its libp2p key. + pub(crate) wrong_peer_id: u64, + /// No address known for the peer. + pub(crate) no_addresses: u64, + pub(crate) other: u64, +} diff --git a/src/transport/tor.rs b/src/transport/tor.rs new file mode 100644 index 0000000..b49cc67 --- /dev/null +++ b/src/transport/tor.rs @@ -0,0 +1,673 @@ +//! Tor as a transport (`--transports tor`, in builds with the `tor` feature), for +//! reachability: every node runs an onion service (arti), reachable from anywhere +//! without a forwarded port or a relay. Not for anonymity: a node's other addresses are +//! in the member list next to its onion address. +//! +//! A node's transport identity is its onion service, `<onion>.onion:<port>`. The onion +//! service's own key lives in arti's state (`--tor-state-dir`), so its address stays the +//! same across restarts. +//! +//! Peers talk over one long-lived connection per pair (opening a stream over Tor costs +//! a round trip through the rendezvous circuit): a Tor stream to the other's onion +//! service, either way. Tor authenticates the onion service to the one who dials it, +//! but nothing about the dialer: whoever dials arrives anonymously. So at the start of +//! every connection, the challenge handshake (`transport::challenge`) proves both node +//! keys, bound to the onion address dialed, and then the dialer says which onion +//! service is its own (`send_claim`): the connection's peer from then on. That claim is +//! checked before the connection is used (`Shared::check`): the onion service must be +//! where the dialer's node key is. A dial proves that (Tor proves the onion service, the +//! handshake the key there), so a claim holds if dialing the claimed onion service found +//! the same key, now or before; if it wasn't dialed yet, the listener dials it to check, +//! and closes that connection again. A claim whose check failed is refused without a +//! dial for a while (`FAILED_CHECK_FOR`). Then messages go as on libp2p: their length (4 +//! bytes, big endian), then the message as `wire` encodes it. A connection that breaks +//! is dialed again on the next send. Both ends may dial at once; each then sends on the +//! newest connection and reads from both until the older one closes. +//! +//! Onion circuits are slow (round trips of a second or more, seconds to set one up), +//! so the timeouts are generous and messages smaller than on libp2p; the paths layer +//! (`transport::paths`) prefers faster transports while they work. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result, anyhow, bail}; +use arti_client::config::TorClientConfigBuilder; +use arti_client::{DataStream, TorClient}; +use async_trait::async_trait; +use futures::StreamExt; +use safelog::DisplayRedacted as _; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf}; +use tokio::sync::{mpsc, watch}; +use tor_cell::relaycell::msg::{Connected, End, EndReason}; +use tor_hsservice::config::OnionServiceConfigBuilder; +use tor_hsservice::{RunningOnionService, StreamRequest, handle_rend_requests}; +use tor_proto::stream::IncomingStreamRequest; +use tor_rtcompat::PreferredRuntime; +use tracing::{debug, info, warn}; + +use crate::identity::NodeKey; +use crate::locks::MutexExt; +use crate::transport::challenge::{self, Role}; +use crate::transport::{ + EncodedMessage, HandlerFn, Hex, MessageHandler, ParsedMessage, Transport, hand_on_probe, + is_probe, wire, +}; + +/// A message read, with the sender's onion service and its payload length. +type Inbound = (String, ParsedMessage, usize); + +/// The onion service's nickname in arti's state (one per node). +const NICKNAME: &str = "continuum"; +/// The largest message read. Sync messages above 16 chunks travel as blobs, and a chunk +/// is at most 4 MiB. +const MAX_MESSAGE: usize = 8 << 20; +/// How long reaching an onion service may take (building circuits, the rendezvous). +const CONNECT_TIMEOUT: Duration = Duration::from_secs(60); +/// How long the challenge handshake may take, a few round trips. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); +/// How long a message may take, before what its size adds (`stream_timeout`). +const STREAM_TIMEOUT: Duration = Duration::from_secs(30); +/// The slowest a message is expected to go over Tor, in bytes per second. +const MIN_RATE: u64 = 64 << 10; +/// Messages read and not handled yet, at most. +const READS: usize = 32; +/// A claim whose check failed (`Shared::check`) is refused without dialing again for +/// this long: claims of onion services that aren't the claimer's don't make us dial +/// over and over. A dial of our own there that finds the claimer's key ends it early. +const FAILED_CHECK_FOR: Duration = Duration::from_secs(600); +/// Onion services remembered (with the node key found there, `Shared::proven`), at most. +const MAX_PEERS: usize = 4096; + +/// How long a message of `bytes` may take. +fn stream_timeout(bytes: usize) -> Duration { + STREAM_TIMEOUT + Duration::from_millis(bytes as u64 * 1000 / MIN_RATE) +} + +/// What a Tor transport is made with. +pub(crate) struct TorConfig { + /// The node key: it proves who we are in the challenge handshake. + pub(crate) key: Arc<NodeKey>, + /// Arti's state (the onion service's key, among others) and cache, in `state` and + /// `cache` below. + pub(crate) state_dir: PathBuf, + /// The onion service's port. + pub(crate) port: u16, +} + +/// `<onion>.onion:<port>`, from `target`, which may end in `/<node key>` (how nodes +/// published their onion service before it was their transport identity). +fn parse_location(target: &str) -> Option<String> { + let location = match target.split_once('/') { + Some((location, key)) if is_key(key) => location, + Some(_) => return None, + None => target, + }; + let (host, port) = location.rsplit_once(':')?; + let name = host.strip_suffix(".onion")?; + if name.is_empty() || !name.bytes().all(|byte| byte.is_ascii_alphanumeric()) { + return None; + } + port.parse::<u16>().ok()?; + Some(location.to_ascii_lowercase()) +} + +/// Whether `id` is a node key, 64 lowercase hex digits. +fn is_key(id: &str) -> bool { + id.len() == 64 + && id + .bytes() + .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f')) +} + +/// After the handshake, the dialer says what the connection is for (`read_claim`): +/// carrying messages from its onion service, `claim`, or (empty) only checking which +/// node key is at the listener's (`Shared::check`). +async fn send_claim<S: AsyncWrite + Unpin>(stream: &mut S, claim: &str) -> Result<()> { + let length = u8::try_from(claim.len()).map_err(|_| anyhow!("{claim} is too long"))?; + stream.write_all(&[length]).await?; + stream.write_all(claim.as_bytes()).await?; + stream.flush().await?; + Ok(()) +} + +/// The dialer's claim (`send_claim`): its onion service, or None for a check. +async fn read_claim<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Option<String>> { + let mut length = [0u8; 1]; + stream.read_exact(&mut length).await?; + if length[0] == 0 { + return Ok(None); + } + let mut claim = vec![0u8; length[0] as usize]; + stream.read_exact(&mut claim).await?; + let claim = String::from_utf8(claim).map_err(|_| anyhow!("a claim that isn't text"))?; + match parse_location(&claim) { + Some(location) if location == claim => Ok(Some(location)), + _ => bail!("{claim} is no onion address with a port"), + } +} + +/// One connection to a peer: where to write. A reader task reads the other half. +struct Connection { + id: u64, + writer: tokio::sync::Mutex<WriteHalf<DataStream>>, +} + +struct Shared { + key: Arc<NodeKey>, + client: Arc<TorClient<PreferredRuntime>>, + /// Our onion service, `<onion>.onion:<port>`: our transport identity. + local_id: String, + /// The node key found at each onion service dialed: proven there (`check`). + proven: Mutex<HashMap<String, [u8; 32]>>, + /// Claims whose check failed lately (`FAILED_CHECK_FOR`): an onion service and the + /// node key claiming it, and when. + failed_checks: Mutex<HashMap<(String, [u8; 32]), Instant>>, + /// The connection to write to each peer on, by onion service. + connections: Mutex<HashMap<String, Arc<Connection>>>, + /// One dial at a time per onion service. + dialing: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>, + next_connection: AtomicU64, + /// Messages read, for the consumer: in order, and PINGs and PONGs apart, to be + /// handed on at once (`is_probe`). + inbound: mpsc::Sender<Inbound>, + probes: mpsc::Sender<Inbound>, +} + +fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> { + mutex.lock_or_recover() +} + +pub(crate) struct TorTransport { + shared: Arc<Shared>, + /// Kept: dropping it stops the onion service. + _service: Arc<RunningOnionService>, + /// Messages, and PINGs and PONGs (`Shared::inbound`, `Shared::probes`). + incoming: tokio::sync::Mutex<Option<(mpsc::Receiver<Inbound>, mpsc::Receiver<Inbound>)>>, + stopped: watch::Receiver<bool>, +} + +impl TorTransport { + /// Launches our onion service, and bootstraps arti in the background (that takes a + /// while: it fetches what it needs to know of the Tor network). Until it is done, + /// dials wait for it (arti bootstraps on demand), and the onion service is published + /// once it is: the node starts without waiting for Tor. + pub(crate) async fn start(config: TorConfig) -> Result<Self> { + let tor_config = TorClientConfigBuilder::from_directories( + config.state_dir.join("state"), + config.state_dir.join("cache"), + ) + .build() + .context("Configuring Tor")?; + let client = TorClient::builder() + .config(tor_config) + .create_unbootstrapped_async() + .await + .context("Starting Tor")?; + { + let client = client.clone(); + let state_dir = config.state_dir.display().to_string(); + info!(state_dir = %state_dir, "Bootstrapping Tor"); + // Launching the onion service starts it too; this waits for it to be done. + tokio::spawn(async move { + match client.bootstrap().await { + Ok(()) => info!("Tor bootstrapped"), + // Dials try again (on demand). + Err(error) => warn!(error = %error, "Bootstrapping Tor failed"), + } + }); + } + let service_config = OnionServiceConfigBuilder::default() + .nickname( + NICKNAME + .parse() + .map_err(|error| anyhow!("Onion service nickname: {error}"))?, + ) + .build() + .context("Configuring the onion service")?; + let (service, requests) = client + .launch_onion_service(service_config) + .context("Launching the onion service")? + .ok_or_else(|| anyhow!("The onion service is disabled in arti's configuration"))?; + let name = service + .onion_address() + .ok_or_else(|| anyhow!("The onion service has no address"))?; + // Arti shows onion addresses redacted unless asked: we publish ours. + let onion = format!("{}:{}", name.display_unredacted(), config.port); + + let (inbound, incoming) = mpsc::channel(READS); + let (probes, incoming_probes) = mpsc::channel(READS); + let shared = Arc::new(Shared { + key: config.key, + client, + local_id: onion.clone(), + proven: Mutex::default(), + failed_checks: Mutex::default(), + connections: Mutex::default(), + dialing: Mutex::default(), + next_connection: AtomicU64::new(0), + inbound, + probes, + }); + let (stopping, stopped) = watch::channel(false); + { + let shared = Arc::clone(&shared); + let port = config.port; + tokio::spawn(async move { + let requests = handle_rend_requests(requests); + futures::pin_mut!(requests); + while let Some(request) = requests.next().await { + let shared = Arc::clone(&shared); + tokio::spawn(async move { shared.accept(request, port).await }); + } + warn!("The onion service stopped"); + let _ = stopping.send(true); + }); + } + info!(address = %onion, "Transport started (the onion service may take a minute to be reachable)"); + Ok(Self { + shared, + _service: service, + incoming: tokio::sync::Mutex::new(Some((incoming, incoming_probes))), + stopped, + }) + } +} + +impl Shared { + /// A connection to the onion service at `location`, made if there is none. + async fn connection(self: &Arc<Self>, location: &str) -> Result<Arc<Connection>> { + if let Some(connection) = lock(&self.connections).get(location) { + return Ok(Arc::clone(connection)); + } + let dialing = self.dialing(location); + let _one_at_a_time = dialing.lock().await; + // Someone else may have connected meanwhile. + if let Some(connection) = lock(&self.connections).get(location) { + return Ok(Arc::clone(connection)); + } + let (_, stream) = self.dial(location, &self.local_id).await?; + Ok(self.register(location.to_owned(), stream)) + } + + /// The lock that keeps dials to `location` one at a time. + fn dialing(&self, location: &str) -> Arc<tokio::sync::Mutex<()>> { + let mut dialing = lock(&self.dialing); + if dialing.len() >= MAX_PEERS { + // Only those in use. + dialing.retain(|_, lock| Arc::strong_count(lock) > 1); + } + Arc::clone(dialing.entry(location.to_owned()).or_default()) + } + + /// Dials the onion service at `location`, runs the handshake and says what for + /// (`send_claim`): returns the node key there, now proven there, and the stream. + async fn dial(&self, location: &str, claim: &str) -> Result<([u8; 32], DataStream)> { + let (host, port) = location + .rsplit_once(':') + .and_then(|(host, port)| Some((host, port.parse::<u16>().ok()?))) + .ok_or_else(|| anyhow!("{location} is no onion address with a port"))?; + let mut stream = tokio::time::timeout(CONNECT_TIMEOUT, self.client.connect((host, port))) + .await + .map_err(|_| anyhow!("reaching {location} took longer than {CONNECT_TIMEOUT:?}"))? + .with_context(|| format!("Reaching {location}"))?; + let key = challenge::run( + &mut stream, + &self.key, + Role::Dialer, + location, + HANDSHAKE_TIMEOUT, + ) + .await + .with_context(|| format!("Handshake with {location}"))?; + tokio::time::timeout(HANDSHAKE_TIMEOUT, send_claim(&mut stream, claim)) + .await + .map_err(|_| anyhow!("no handshake with {location} within {HANDSHAKE_TIMEOUT:?}"))? + .with_context(|| format!("Handshake with {location}"))?; + let mut proven = lock(&self.proven); + if proven.len() < MAX_PEERS || proven.contains_key(location) { + proven.insert(location.to_owned(), key); + } + Ok((key, stream)) + } + + /// Whether the node with `key` is at the onion service at `location` (a dialer + /// claims it is: module docs). Dialed to find out, unless that was done already, or + /// failed within `FAILED_CHECK_FOR`. + async fn check(&self, location: &str, key: [u8; 32]) -> Result<bool> { + if lock(&self.proven).get(location) == Some(&key) { + return Ok(true); + } + let claim = (location.to_owned(), key); + if let Some(failed) = self.failed_lately(&claim) { + bail!("its check failed {}s ago", failed.as_secs()); + } + let dialing = self.dialing(location); + let _one_at_a_time = dialing.lock().await; + // A dial or check meanwhile may have found out. + if lock(&self.proven).get(location) == Some(&key) { + return Ok(true); + } + if let Some(failed) = self.failed_lately(&claim) { + bail!("its check failed {}s ago", failed.as_secs()); + } + let checked = self + .dial(location, "") + .await + .map(|(found, _check_only)| found == key); + if !matches!(checked, Ok(true)) { + let mut failed = lock(&self.failed_checks); + if failed.len() >= MAX_PEERS { + failed.retain(|_, at| at.elapsed() < FAILED_CHECK_FOR); + } + if failed.len() < MAX_PEERS { + failed.insert(claim, Instant::now()); + } + } + checked + } + + /// How long ago the check of `claim` failed, if within `FAILED_CHECK_FOR`. + fn failed_lately(&self, claim: &(String, [u8; 32])) -> Option<Duration> { + let mut failed = lock(&self.failed_checks); + let since = failed.get(claim)?.elapsed(); + if since < FAILED_CHECK_FOR { + return Some(since); + } + failed.remove(claim); + None + } + + /// A stream to our onion service: accepted if for our port, then the handshake, and + /// the dialer's claim checked (module docs). + async fn accept(self: Arc<Self>, request: StreamRequest, port: u16) { + let ours = matches!(request.request(), IncomingStreamRequest::Begin(begin) if begin.port() == port); + if !ours { + let _ = request + .reject(End::new_with_reason(EndReason::CONNECTREFUSED)) + .await; + return; + } + let mut stream = match request.accept(Connected::new_empty()).await { + Ok(stream) => stream, + Err(error) => { + debug!(error = %error, "Couldn't accept a stream to our onion service"); + return; + } + }; + let handshake = async { + let key = challenge::run( + &mut stream, + &self.key, + Role::Listener, + &self.local_id, + HANDSHAKE_TIMEOUT, + ) + .await?; + let claim = tokio::time::timeout(HANDSHAKE_TIMEOUT, read_claim(&mut stream)) + .await + .map_err(|_| anyhow!("no claim within {HANDSHAKE_TIMEOUT:?}"))??; + Ok::<_, anyhow::Error>((key, claim)) + }; + let (key, claim) = match handshake.await { + Ok(done) => done, + Err(error) => { + debug!(error = %format!("{error:#}"), "Dropping a connection to our onion service: the handshake failed"); + return; + } + }; + // A check (`check`): the handshake was all. + let Some(location) = claim else { + return; + }; + if location == self.local_id { + debug!("Dropping a connection to our onion service: it claims to be ours"); + return; + } + match self.check(&location, key).await { + Ok(true) => { + self.register(location, stream); + } + Ok(false) => { + debug!(claimed = %location, key = %Hex(&key), "Dropping a connection to our onion service: another node is at the onion service it claims"); + } + Err(error) => { + debug!(claimed = %location, error = %format!("{error:#}"), "Dropping a connection to our onion service: couldn't check the onion service it claims"); + } + } + } + + /// A connection to the onion service at `location`, authenticated: the one to write + /// to it on from now, and read from until it closes. + fn register(self: &Arc<Self>, location: String, stream: DataStream) -> Arc<Connection> { + let (reader, writer) = tokio::io::split(stream); + let id = self.next_connection.fetch_add(1, Ordering::Relaxed); + let connection = Arc::new(Connection { + id, + writer: tokio::sync::Mutex::new(writer), + }); + lock(&self.connections).insert(location.clone(), Arc::clone(&connection)); + debug!(peer = %location, connection = id, "Connection opened"); + let shared = Arc::clone(self); + tokio::spawn(async move { + let error = shared.read(&location, reader).await; + debug!(peer = %location, connection = id, error = %format!("{error:#}"), "Connection closed"); + shared.forget(&location, id); + }); + connection + } + + /// Reads messages from `location` until the connection fails; returns why. + async fn read(&self, location: &str, mut reader: ReadHalf<DataStream>) -> anyhow::Error { + loop { + let mut length = [0u8; 4]; + if let Err(error) = reader.read_exact(&mut length).await { + return error.into(); + } + let length = u32::from_be_bytes(length) as usize; + if length > MAX_MESSAGE { + return anyhow!("a message of {length} bytes; at most {MAX_MESSAGE} are read"); + } + let mut body = vec![0u8; length]; + match tokio::time::timeout(stream_timeout(length), reader.read_exact(&mut body)).await { + Ok(Ok(_)) => {} + Ok(Err(error)) => return error.into(), + Err(_) => { + return anyhow!( + "{length} bytes took longer than {:?}", + stream_timeout(length) + ); + } + } + let message = match wire::decode(&body) { + Ok(message) => message, + Err(error) => { + debug!(peer = %location, error = %error, "Dropping a message that doesn't decode"); + continue; + } + }; + let channel = if is_probe(&message) { + &self.probes + } else { + &self.inbound + }; + if channel + .send((location.to_owned(), message, body.len())) + .await + .is_err() + { + return anyhow!("the transport stopped"); + } + } + } + + /// The connection `id` to `location` is gone: not to be written to any more. + fn forget(&self, location: &str, id: u64) { + let mut connections = lock(&self.connections); + if connections + .get(location) + .is_some_and(|connection| connection.id == id) + { + connections.remove(location); + } + } +} + +#[async_trait] +impl Transport for TorTransport { + fn local_id(&self) -> &str { + &self.shared.local_id + } + + fn is_peer_id(&self, id: &str) -> bool { + parse_location(id).is_some_and(|location| location == id) + } + + /// The longest a message may take. + fn delivery_limit(&self) -> Duration { + stream_timeout(MAX_MESSAGE) + } + + fn local_addresses(&self) -> Vec<String> { + vec![self.shared.local_id.clone()] + } + + /// An onion service (`<onion>.onion:<port>`, with a node key after it as nodes + /// published it before, which is left out). + fn locate(&self, target: &str) -> Option<String> { + parse_location(target) + } + + /// An onion service is its own transport identity: nothing to find out. + async fn resolve_seed(&self, seed: &str) -> Result<String> { + let location = parse_location(seed) + .ok_or_else(|| anyhow!("{seed} is no onion address with a port"))?; + if location == self.shared.local_id { + bail!("{seed} is this node"); + } + Ok(location) + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let body = wire::encode(message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + address: peer.to_owned(), + payload_len: body.len(), + body: body.into(), + kind: message.kind(), + message: None, + }) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + let location = &message.peer; + if !self.is_peer_id(location) { + bail!("{location} is no Tor transport identity"); + } + if message.body.len() > MAX_MESSAGE { + bail!( + "a message of {} bytes; at most {MAX_MESSAGE} go over Tor", + message.body.len() + ); + } + let connection = self.shared.connection(location).await?; + let timeout = stream_timeout(message.body.len()); + let written = tokio::time::timeout(timeout, async { + let mut writer = connection.writer.lock().await; + writer + .write_all(&(message.body.len() as u32).to_be_bytes()) + .await?; + writer.write_all(&message.body).await?; + writer.flush().await + }) + .await; + let result = match written { + Ok(Ok(())) => Ok(()), + Ok(Err(error)) => Err(anyhow::Error::from(error)), + Err(_) => Err(anyhow!( + "sending to {location} took longer than {timeout:?}" + )), + }; + if result.is_err() { + // Dialed again on the next send. + self.shared.forget(location, connection.id); + } + result.with_context(|| format!("Sending to {location}")) + } + + /// Handles the messages read from every connection, one at a time, in the order + /// they were read. + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let Some((mut incoming, mut probes)) = self.incoming.lock().await.take() else { + bail!("Tor consumer already started"); + }; + let handler: Arc<HandlerFn> = Arc::from(handler); + { + let handler = Arc::clone(&handler); + tokio::spawn(async move { + while let Some((peer, message, payload_len)) = probes.recv().await { + hand_on_probe(&handler, peer, message, payload_len); + } + }); + } + while let Some((peer, message, payload_len)) = incoming.recv().await { + if let Err(error) = handler(peer.clone(), message, payload_len).await { + warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Ok(()) + } + + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + let mut stopped = self.stopped.clone(); + let _ = stopped.wait_for(|stopped| *stopped).await; + Err(std::io::Error::other("the onion service stopped")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const ONION: &str = "vww6ybal4bd7szmgncyruucpgfkqahzddi37ktceo3ah7ngmcopnpyyd.onion"; + + #[test] + fn locations() { + let key = "ab".repeat(32); + assert!(is_key(&key)); + assert!(!is_key(&key.to_uppercase())); + assert!(!is_key(&key[1..])); + let location = format!("{ONION}:9662"); + assert_eq!(parse_location(&location), Some(location.clone())); + // As nodes published them before: the key is left out. + assert_eq!( + parse_location(&format!("{ONION}:9662/{key}")), + Some(location.clone()) + ); + assert_eq!( + parse_location(&location.to_uppercase().replace(".ONION", ".onion")), + Some(location) + ); + // A port is needed, a key must be one, and only onion addresses. + assert_eq!(parse_location(ONION), None); + assert_eq!(parse_location(&format!("{ONION}:9662/xyz")), None); + assert_eq!(parse_location("example.org:9662"), None); + assert_eq!(parse_location(&format!("{ONION}:99999")), None); + } + + #[tokio::test] + async fn claims_round_trip() { + let (mut dialer, mut listener) = tokio::io::duplex(1024); + let location = format!("{ONION}:9662"); + send_claim(&mut dialer, &location).await.unwrap(); + send_claim(&mut dialer, "").await.unwrap(); + send_claim(&mut dialer, "example.org:1").await.unwrap(); + assert_eq!(read_claim(&mut listener).await.unwrap(), Some(location)); + assert_eq!(read_claim(&mut listener).await.unwrap(), None); + assert!(read_claim(&mut listener).await.is_err()); + } +} diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index b332d3d..64595cb 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -5,8 +5,8 @@ //! (for Mycelium: the topic, or the base64 of its HTTP API). Counters only grow (from //! the start of the process); rates are for whoever reads them. //! -//! A message dropped by the debug firewall still counts as sent (the firewall plays a -//! lossy network) but not as received. +//! A message dropped by the debug firewall (in the paths layer, below) still counts as +//! sent (the firewall plays a lossy network) but not as received. use std::collections::HashMap; use std::sync::{Arc, Mutex, MutexGuard}; @@ -37,12 +37,14 @@ pub(crate) struct ByKind { pub(crate) hello: Counter, pub(crate) hello_reply: Counter, pub(crate) sync: Counter, - pub(crate) ping: Counter, - pub(crate) pong: Counter, + pub(crate) status: Counter, + pub(crate) status_reply: Counter, pub(crate) blob: Counter, /// WHO and ME (`transport::identity`), counted by the identity layer under the /// address they went to or came from. pub(crate) identity: Counter, + /// PING and PONG (`transport::paths`), counted by the paths layer under the address. + pub(crate) paths: Counter, } impl ByKind { @@ -51,10 +53,11 @@ impl ByKind { MessageKind::Hello => &mut self.hello, MessageKind::HelloReply => &mut self.hello_reply, MessageKind::Sync => &mut self.sync, - MessageKind::Ping => &mut self.ping, - MessageKind::Pong => &mut self.pong, + MessageKind::Status => &mut self.status, + MessageKind::StatusReply => &mut self.status_reply, MessageKind::Blob => &mut self.blob, MessageKind::Who | MessageKind::Me => &mut self.identity, + MessageKind::Ping | MessageKind::Pong => &mut self.paths, } } } @@ -173,6 +176,26 @@ impl Transport for CountingTransport { self.inner.transports() } + fn destination(&self, peer: &str) -> Option<crate::transport::Destination> { + self.inner.destination(peer) + } + + fn connection(&self, target: &str) -> Option<crate::transport::Connection> { + self.inner.connection(target) + } + + fn stats(&self) -> crate::transport::stats::TransportStats { + self.inner.stats() + } + + fn answers_pings(&self) -> bool { + self.inner.answers_pings() + } + + fn rtt(&self, peer: &str) -> Option<std::time::Duration> { + self.inner.rtt(peer) + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { self.inner.resolve_seed(seed).await } @@ -215,20 +238,20 @@ mod tests { #[test] fn totals_survive_retain() { let traffic = Traffic::default(); - traffic.sent("a", MessageKind::Ping, 100); + traffic.sent("a", MessageKind::Status, 100); traffic.sent("b", MessageKind::Sync, 1000); - traffic.received("b", MessageKind::Pong, 90); + traffic.received("b", MessageKind::StatusReply, 90); traffic.refused("a", MessageKind::Sync, 500); let b = traffic.peer("b").unwrap(); assert_eq!(b.sent.sync.messages, 1); - assert_eq!(b.received.pong.bytes, 90); + assert_eq!(b.received.status_reply.bytes, 90); traffic.retain(|peer| peer == "b"); assert!(traffic.peer("a").is_none()); let total = traffic.total(); - assert_eq!(total.sent.ping.bytes, 100); + assert_eq!(total.sent.status.bytes, 100); assert_eq!(total.sent.sync.bytes, 1000); assert_eq!(total.refused.sync.bytes, 500); } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 5577f42..fcbee85 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 5; +pub(crate) const VERSION: u8 = 6; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -22,11 +22,13 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::Hello => 0, MessageKind::HelloReply => 1, MessageKind::Sync => 2, - MessageKind::Ping => 3, - MessageKind::Pong => 4, + MessageKind::Status => 3, + MessageKind::StatusReply => 4, MessageKind::Blob => 5, MessageKind::Who => 6, MessageKind::Me => 7, + MessageKind::Ping => 8, + MessageKind::Pong => 9, } } @@ -55,13 +57,16 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> { ParsedMessage::Hello(hello) => rmp_serde::encode::write(&mut payload, hello)?, ParsedMessage::HelloReply(reply) => rmp_serde::encode::write(&mut payload, reply)?, ParsedMessage::Sync(envelope) => rmp_serde::encode::write(&mut payload, envelope)?, - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => { - rmp_serde::encode::write(&mut payload, keepalive)? + ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => { + rmp_serde::encode::write(&mut payload, status)? } ParsedMessage::Blob(blob) => rmp_serde::encode::write(&mut payload, blob)?, ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => { rmp_serde::encode::write(&mut payload, proof)? } + ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => { + rmp_serde::encode::write(&mut payload, nonce)? + } } Ok(payload) } @@ -80,16 +85,20 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> { ParsedMessage::HelloReply(rmp_serde::from_slice(body)?) } else if is(MessageKind::Sync) { ParsedMessage::Sync(rmp_serde::from_slice(body)?) - } else if is(MessageKind::Ping) { - ParsedMessage::Ping(rmp_serde::from_slice(body)?) - } else if is(MessageKind::Pong) { - ParsedMessage::Pong(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Status) { + ParsedMessage::Status(rmp_serde::from_slice(body)?) + } else if is(MessageKind::StatusReply) { + ParsedMessage::StatusReply(rmp_serde::from_slice(body)?) } else if is(MessageKind::Blob) { ParsedMessage::Blob(rmp_serde::from_slice(body)?) } else if is(MessageKind::Who) { ParsedMessage::Who(rmp_serde::from_slice(body)?) } else if is(MessageKind::Me) { ParsedMessage::Me(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Ping) { + ParsedMessage::Ping(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Pong) { + ParsedMessage::Pong(rmp_serde::from_slice(body)?) } else { return Err(WireError::UnknownKind(kind)); }) @@ -322,7 +331,7 @@ mod tests { use crate::document::replica::Epoch; use crate::transport::{ BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof, - Keepalive, ListedPeer, MAX_LISTED_HEADS, + ListedPeer, MAX_LISTED_HEADS, Status, }; use automerge::ChangeHash; @@ -380,7 +389,7 @@ mod tests { #[test] fn keepalives_keep_their_kind_and_heads() { - let keepalive = Keepalive { + let status = Status { heads: HeadsSummary::Digest { digest: [8; 32], sample: vec![ChangeHash([9; 32])], @@ -390,17 +399,18 @@ mod tests { ready: true, syncs_with_you: false, }; - let decoded = decode(&encode(&ParsedMessage::Pong(keepalive.clone())).unwrap()).unwrap(); - let ParsedMessage::Pong(decoded) = decoded else { - panic!("not a PONG"); + let decoded = + decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap(); + let ParsedMessage::StatusReply(decoded) = decoded else { + panic!("not a STATUS_REPLY"); }; - assert_eq!(decoded.heads, keepalive.heads); + assert_eq!(decoded.heads, status.heads); assert_eq!(decoded.heads.listed(), [ChangeHash([9; 32])]); } #[test] fn too_many_listed_heads_are_refused() { - let keepalive = Keepalive { + let status = Status { heads: HeadsSummary::Heads( (0..=MAX_LISTED_HEADS as u8) .map(|n| ChangeHash([n; 32])) @@ -411,7 +421,7 @@ mod tests { ready: true, syncs_with_you: true, }; - assert!(decode(&encode(&ParsedMessage::Ping(keepalive)).unwrap()).is_err()); + assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err()); } #[test] @@ -469,7 +479,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("050097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("060097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -489,13 +499,13 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "050199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "060199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) ); - let ping = ParsedMessage::Ping(Keepalive { + let status = ParsedMessage::Status(Status { heads: HeadsSummary::Heads(Vec::new()), epoch, epoch_seq: 1, @@ -503,8 +513,8 @@ mod tests { syncs_with_you: false, }); assert_eq!( - hex(&encode(&ping).unwrap()), - format!("0503959290c400{epoch_hex}01c3c2") + hex(&encode(&status).unwrap()), + format!("0603959290c400{epoch_hex}01c3c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -513,7 +523,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("050692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("060692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -526,5 +536,15 @@ mod tests { panic!("not a ME"); }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); + + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "060807"); + assert_eq!( + hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()), + "0609cfffffffffffffffff" + ); + let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else { + panic!("not a PONG"); + }; + assert_eq!(nonce, 42); } } |