diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/document/meta.rs | 3 | ||||
| -rw-r--r-- | src/swarm/members.rs | 32 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 1 | ||||
| -rw-r--r-- | src/transport/identity.rs | 5 | ||||
| -rw-r--r-- | src/transport/libp2p.rs | 264 | ||||
| -rw-r--r-- | src/transport/mod.rs | 7 | ||||
| -rw-r--r-- | src/transport/paths.rs | 13 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 4 |
8 files changed, 288 insertions, 41 deletions
diff --git a/src/document/meta.rs b/src/document/meta.rs index 482b421..24ed7c7 100644 --- a/src/document/meta.rs +++ b/src/document/meta.rs @@ -7,6 +7,8 @@ //! * `votes/<pk>`: the votes register (`remake::records`). //! * `epochs/<generation>`: epoch metadata, written by snapshots (`remake::snapshot`). //! * `swarm`: the swarm ID, written once by the genesis node (`swarm::bootstrap`). +//! * `relays`: relays for the swarm's nodes, listed by an operator +//! (`swarm::members::relays`). /// The key every protocol path starts with. pub(crate) const ROOT: &str = "_continuum"; @@ -16,6 +18,7 @@ pub(crate) const REMAKES: &str = "remakes"; pub(crate) const VOTES: &str = "votes"; pub(crate) const EPOCHS: &str = "epochs"; pub(crate) const SWARM: &str = "swarm"; +pub(crate) const RELAYS: &str = "relays"; /// The store path `_continuum/<parts…>`. pub(crate) fn path(parts: &[&str]) -> Vec<String> { diff --git a/src/swarm/members.rs b/src/swarm/members.rs index c2edd01..3bfb9cf 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -146,6 +146,27 @@ pub(crate) fn learn_addresses(state: &AppState, doc: &AutoCommit) { } } +/// Relays listed for the swarm, at most this many. +const MAX_RELAYS: usize = 32; + +/// The relays listed for the swarm under `_continuum/relays` (by `--protocol-writers`): +/// addresses (`libp2p:<multiaddr>`, with or without the relay's peer ID) of relays that +/// aren't members, e.g. nodes that only relay. Every member's transport may +/// pick them (`Transport::learn_relays`), after the relays it is configured with. +pub(crate) fn relays(doc: &AutoCommit) -> Vec<String> { + match store::read_path(doc, &meta::path(&[meta::RELAYS])) { + Ok(Some(JsonValue::Array(items))) => items + .into_iter() + .filter_map(|item| match item { + JsonValue::String(address) => Some(address), + _ => None, + }) + .take(MAX_RELAYS) + .collect(), + _ => Vec::new(), + } +} + fn alive_path(pk: &str) -> Vec<String> { let mut path = member_path(pk); path.push("alive".to_owned()); @@ -303,6 +324,17 @@ mod tests { } #[test] + fn listed_relays() { + let mut doc = AutoCommit::new(); + assert!(relays(&doc).is_empty()); + let relay = "libp2p:/ip6/fd00::1/tcp/9661"; + store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!([relay, 7])).unwrap(); + assert_eq!(relays(&doc), [relay]); // Anything but text is ignored. + store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!(relay)).unwrap(); + assert!(relays(&doc).is_empty()); // Not a list. + } + + #[test] fn renewal_interval() { let day = Duration::from_secs(24 * 3600); assert_eq!(alive_renewal(7 * day), day); diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 1e4f5b9..1d98637 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -163,6 +163,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { let replica = state.doc.read().await; members::learn_addresses(state, &replica.doc); set_aliases(state, members::aliases(&replica.doc)); + state.transport.learn_relays(&members::relays(&replica.doc)); let current = members::members(&replica.doc); // Those that said they are alive within the electorate window. let alive: BTreeSet<String> = members::electorate( diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 432348e..c597d77 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -540,6 +540,11 @@ impl Transport for IdentityTransport { self.shared.paths.transports() } + /// Each transport those with its scheme, without it (`Paths::learn_relays`). + fn learn_relays(&self, addresses: &[String]) { + self.shared.paths.learn_relays(addresses); + } + /// The address `encode` picks, the transport's connection there, and every path /// to `peer` as paths measures it, best first. fn destination(&self, peer: &str) -> Option<Destination> { diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index 63f6335..a6778c6 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -30,7 +30,8 @@ //! 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 picks up to `AUTO_RELAYS` relays -//! (`relay_check`): those configured first, then members that relay. It needs one when +//! (`relay_check`): those configured first, then those listed in the document +//! (`learn_relays`), then members that relay. It needs one when //! it has no address others could dial, or when AutoNAT says it can't be dialled: //! AutoNAT asks connected peers (nodes of the swarm) to dial it back. Once AutoNAT says //! it can be dialled, it lets the relays it picked go. @@ -190,10 +191,40 @@ enum Command { }, /// See whether relays are needed now (one was just learned of). CheckRelays, - /// Find out which peer relays at `address`, configured without the relay's peer ID. + /// Find out which peer relays at `address`, named without the relay's peer ID + /// (`Book::finding`). FindRelay { address: Multiaddr }, } +/// Who named a relay to pick, in the order they are picked. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Named { + /// `Libp2pConfig::relays`. + Configured, + /// The relays listed in the document (`Transport::learn_relays`). + Listed, +} + +impl Named { + fn as_str(self) -> &'static str { + match self { + Self::Configured => "configured", + Self::Listed => "listed", + } + } +} + +/// A relay named to pick. +#[derive(Clone, Debug, PartialEq, Eq)] +struct NamedRelay { + /// Where, with the relay's peer ID. + at: Multiaddr, + /// As named: without the peer ID, it is found out again when the relay doesn't + /// grant a slot (it may have a new one). + given: Multiaddr, + named: Named, +} + /// Where peers are, and where we listen. #[derive(Default)] struct Book { @@ -203,12 +234,12 @@ struct Book { announced: Vec<Multiaddr>, /// `Libp2pConfig::relay`: this node relays, and says so. serving: bool, - /// Relays configured (`Libp2pConfig::relays`), and where (with the relay's peer ID): - /// picked first. - configured_relays: HashMap<PeerId, Multiaddr>, - /// Of those, the ones configured without the relay's peer ID, as configured: found - /// out again when the relay doesn't grant a slot (it may have a new peer ID). - unnamed_relays: HashMap<PeerId, Multiaddr>, + /// Relays named to pick (configured, listed), by peer: picked first. + named_relays: HashMap<PeerId, NamedRelay>, + /// Relays named without their peer ID, being found out (`Command::FindRelay`). + finding: HashMap<Multiaddr, Named>, + /// The relays listed in the document, as last learned (`learn_listed`). + listed: Vec<Multiaddr>, /// Members that say they relay, and where (with the relay's peer ID). relays: HashMap<PeerId, Multiaddr>, /// With `Libp2pConfig::public`: other peers that relay (identify says so), and @@ -481,6 +512,107 @@ fn published(all: &[Multiaddr]) -> Vec<Multiaddr> { } impl Book { + /// `named` names the relay at `given`; returns it if its peer is to be found out. + /// `local`: this node, never its own relay. + fn name_relay(&mut self, given: Multiaddr, named: Named, local: PeerId) -> Option<Multiaddr> { + match split_peer(&given) { + Some(peer) if peer == local => None, + Some(peer) => { + // Configured wins over listed. + if self + .named_relays + .get(&peer) + .is_none_or(|known| known.named == Named::Listed || named == Named::Configured) + { + self.named_relays.insert( + peer, + NamedRelay { + at: given.clone(), + given, + named, + }, + ); + } + None + } + None => { + let known = self.finding.contains_key(&given) + || self.named_relays.values().any(|relay| relay.given == given); + if known { + return None; + } + self.finding.insert(given.clone(), named); + Some(given) + } + } + } + + /// The relay named at `given` (without a peer ID) is `peer`: unless it is no longer + /// named (or is this node). Returns whether that's news. + fn relay_found(&mut self, given: Multiaddr, peer: PeerId, local: PeerId) -> bool { + let Some(named) = self.finding.remove(&given) else { + return false; + }; + let Ok(at) = given.clone().with_p2p(peer) else { + return false; + }; + if peer == local + || (named == Named::Listed + && self + .named_relays + .get(&peer) + .is_some_and(|known| known.named == Named::Configured)) + { + return false; + } + self.named_relays + .insert(peer, NamedRelay { at, given, named }); + true + } + + /// `relay` didn't grant a slot: if it was named without its peer ID, its address is + /// to be found out again, returned. + fn find_again(&mut self, relay: PeerId) -> Option<Multiaddr> { + if self + .named_relays + .get(&relay) + .is_none_or(|known| split_peer(&known.given).is_some()) + { + return None; + } + let known = self.named_relays.remove(&relay)?; + self.finding.insert(known.given.clone(), known.named); + Some(known.given) + } + + /// The relays listed in the document are `listed` now: returns those to find out, + /// None if nothing changed. + fn learn_listed(&mut self, listed: Vec<Multiaddr>, local: PeerId) -> Option<Vec<Multiaddr>> { + if listed == self.listed { + return None; + } + self.named_relays + .retain(|_, relay| relay.named != Named::Listed || listed.contains(&relay.given)); + self.finding + .retain(|given, named| *named != Named::Listed || listed.contains(given)); + self.listed = listed.clone(); + Some( + listed + .into_iter() + .filter_map(|given| self.name_relay(given, Named::Listed, local)) + .collect(), + ) + } + + /// The named relays `named` names, by peer, where. + fn named(&self, named: Named) -> HashMap<PeerId, Multiaddr> { + self.named_relays + .iter() + .filter(|(_, relay)| relay.named == named) + .map(|(peer, relay)| (*peer, relay.at.clone())) + .collect() + } + /// 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 { @@ -588,24 +720,17 @@ impl Libp2pTransport { .accept(PROTOCOL) .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?; let (commands, receiver) = mpsc::unbounded_channel(); - let mut configured_relays = HashMap::new(); - for relay in config.relays { - match split_peer(&relay) { - Some(peer) if peer != peer_id => { - configured_relays.insert(peer, relay); - } - Some(_) => {} // This node. - None => { - let _ = commands.send(Command::FindRelay { address: relay }); - } - } - } - let book = Arc::new(Mutex::new(Book { + let mut book = Book { announced: config.announce, serving: config.relay, - configured_relays, ..Book::default() - })); + }; + for relay in config.relays { + if let Some(address) = book.name_relay(relay, Named::Configured, peer_id) { + let _ = commands.send(Command::FindRelay { address }); + } + } + let book = Arc::new(Mutex::new(book)); let (stopping, stopped) = watch::channel(false); tokio::spawn(run( swarm, @@ -875,8 +1000,8 @@ fn relay_check( candidates } -/// Reserves slots on the relays `relay_check` picks (configured ones first, then -/// members, then public ones), and lets those picked go once AutoNAT says this node can +/// Reserves slots on the relays `relay_check` picks (configured ones first, then listed +/// ones, members, public ones), and lets those picked go once AutoNAT says this node can /// be dialled. fn pick_relays( swarm: &mut Swarm<Behaviour>, @@ -894,7 +1019,7 @@ fn pick_relays( } return; } - let (direct, configured, members, public) = { + let (direct, configured, listed, members, public) = { let mut book = lock(book); book.failed_relays .retain(|_, at| at.elapsed() < RELAY_PAUSE); @@ -913,7 +1038,8 @@ fn pick_relays( }; ( !book.direct().is_empty(), - usable(&book.configured_relays), + usable(&book.named(Named::Configured)), + usable(&book.named(Named::Listed)), usable(&book.relays), public, ) @@ -921,7 +1047,8 @@ fn pick_relays( let mut using: Vec<PeerId> = picked.values().copied().collect(); // Each kind for what the ones before left. for (kind, offers) in [ - ("configured", configured), + (Named::Configured.as_str(), configured), + (Named::Listed.as_str(), listed), ("member", members), ("public", public), ] { @@ -1059,19 +1186,16 @@ async fn run( } match tokio::time::timeout(CONNECT_TIMEOUT, identified).await { Ok(Ok(Ok(peer))) => { - if peer == local { - return; // This node. - } - if let Ok(relay) = address.clone().with_p2p(peer) { - debug!(relay = %Addressed(SCHEME, &relay), "Found a configured relay"); - let mut known = lock(&book); - known.configured_relays.insert(peer, relay); - known.unnamed_relays.insert(peer, address); + if lock(&book).relay_found(address.clone(), peer, local) { + debug!(relay = %Addressed(SCHEME, &address), peer = %Addressed(SCHEME, &peer), "Found a named relay"); + let _ = again.send(Command::CheckRelays); } - let _ = again.send(Command::CheckRelays); } _ => { - warn!(relay = %Addressed(SCHEME, &address), "Couldn't reach a configured relay; trying again later"); + if !lock(&book).finding.contains_key(&address) { + return; // No longer named. + } + warn!(relay = %Addressed(SCHEME, &address), "Couldn't reach a named relay; trying again later"); tokio::time::sleep(RELAY_RETRY).await; let _ = again.send(Command::FindRelay { address }); } @@ -1187,9 +1311,8 @@ async fn run( } else { book.counts.slots_not_granted += 1; book.failed_relays.insert(relay, tokio::time::Instant::now()); - // Configured without its peer ID: maybe another one now. - if let Some(address) = book.unnamed_relays.remove(&relay) { - book.configured_relays.remove(&relay); + // Named without its peer ID: maybe another one now. + if let Some(address) = book.find_again(relay) { let _ = again.send(Command::FindRelay { address }); } } @@ -1430,6 +1553,21 @@ impl Transport for Libp2pTransport { Some(peer.to_string()) } + /// Multiaddrs, with or without the relay's peer ID; others are ignored. + fn learn_relays(&self, targets: &[String]) { + let listed = targets + .iter() + .filter_map(|target| Multiaddr::from_str(target).ok()) + .collect(); + let Some(found) = lock(&self.book).learn_listed(listed, self.peer_id) else { + return; + }; + for address in found { + let _ = self.commands.send(Command::FindRelay { address }); + } + let _ = self.commands.send(Command::CheckRelays); + } + fn connection(&self, target: &str) -> Option<Connection> { let peer = PeerId::from_str(target).ok()?; lock(&self.book).connection(&peer) @@ -1936,6 +2074,50 @@ mod tests { } #[test] + fn named_relays() { + let (local, r1, r2) = (peer(), peer(), peer()); + let at = |port: u16| -> Multiaddr { format!("/ip6/fd00::1/tcp/{port}").parse().unwrap() }; + let mut book = Book::default(); + // With a peer ID: known at once; without: to be found out, once. + assert_eq!( + book.name_relay(at(1).with_p2p(r1).unwrap(), Named::Configured, local), + None + ); + assert_eq!( + book.name_relay(at(2), Named::Configured, local), + Some(at(2)) + ); + assert_eq!(book.name_relay(at(2), Named::Configured, local), None); + assert_eq!( + book.name_relay(at(3).with_p2p(local).unwrap(), Named::Configured, local), + None + ); + assert!(book.relay_found(at(2), r2, local)); + assert_eq!(book.named(Named::Configured).len(), 2); + // A relay named with its peer ID isn't found out again; one without is. + assert_eq!(book.find_again(r1), None); + assert_eq!(book.find_again(r2), Some(at(2))); + assert!(!book.named(Named::Configured).contains_key(&r2)); + + // Listed: configured wins; unlisted ones go, unless also configured. + let r3 = peer(); + let listed = vec![ + at(1).with_p2p(r1).unwrap(), + at(4).with_p2p(r3).unwrap(), + at(5), + ]; + assert_eq!(book.learn_listed(listed.clone(), local), Some(vec![at(5)])); + assert_eq!(book.learn_listed(listed, local), None); + assert_eq!(book.named_relays[&r1].named, Named::Configured); + assert_eq!(book.named(Named::Listed).keys().collect::<Vec<_>>(), [&r3]); + assert_eq!(book.learn_listed(Vec::new(), local), Some(Vec::new())); + assert!(book.named(Named::Listed).is_empty()); + assert!(!book.finding.contains_key(&at(5))); + assert!(!book.relay_found(at(5), peer(), local)); // No longer listed. + assert_eq!(book.named(Named::Configured).len(), 1); + } + + #[test] fn known_addresses() { let mut book = Book::default(); let (relay, us) = (peer(), peer()); diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 7cd976b..fb13ad9 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -683,6 +683,13 @@ pub(crate) trait Transport: Send + Sync { let _ = (peer, addresses); } + /// The relays listed for the swarm (`swarm::members::relays`), as addresses: under + /// the identity layer, each transport is given those with its scheme, without it. + /// By default nothing to learn. + fn learn_relays(&self, addresses: &[String]) { + let _ = 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> { diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 2fe534a..3baa092 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -480,6 +480,19 @@ impl Paths { .map(Link::own_address) } + /// Relays listed for the swarm (`Transport::learn_relays`): to each transport the + /// addresses with its scheme, without it. + pub(crate) fn learn_relays(&self, addresses: &[String]) { + for link in &self.links { + let targets: Vec<String> = addresses + .iter() + .filter_map(|address| address.strip_prefix(link.scheme)?.strip_prefix(':')) + .map(str::to_owned) + .collect(); + link.inner.learn_relays(&targets); + } + } + /// This node's own addresses, one per transport: `<scheme>:<its identity there>`. pub(crate) fn own_addresses(&self) -> Vec<String> { self.links.iter().map(Link::own_address).collect() diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 32f7257..b237db2 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -180,6 +180,10 @@ impl Transport for CountingTransport { self.inner.transports() } + fn learn_relays(&self, addresses: &[String]) { + self.inner.learn_relays(addresses) + } + fn destination(&self, peer: &str) -> Option<crate::transport::Destination> { self.inner.destination(peer) } |