summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-27 02:02:15 +0100
committertv <tv@krebsco.de>2026-10-27 02:02:15 +0100
commit06179a9df971636dd6c5e9f438baa9f133854830 (patch)
tree17544f5a8a9bff042cd17cd27860d574e303d523
parent2593edca179e86883e74c7039629564f4ea6ec55 (diff)
add support for several transports
-rw-r--r--src/main.rs1
-rw-r--r--src/simtest/identity.rs46
-rw-r--r--src/simtest/mod.rs1
-rw-r--r--src/swarm/topology.rs8
-rw-r--r--src/transport/firewall.rs4
-rw-r--r--src/transport/identity.rs478
-rw-r--r--src/transport/mod.rs8
-rw-r--r--src/transport/traffic.rs4
8 files changed, 378 insertions, 172 deletions
diff --git a/src/main.rs b/src/main.rs
index b67ac1d..b5e7fc6 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -177,6 +177,7 @@ async fn main() -> Result<()> {
identity: Some(transport::identity::IdentityConfig {
key: Arc::new(node_key),
scheme: "mycelium",
+ more: Vec::new(),
}),
initial_epoch,
initial_doc,
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index a33783f..9859d0e 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -2,6 +2,7 @@
//! through addresses (seeds, HELLO_REPLYs, the member list), and believed only once a
//! node proved itself at an address.
+use std::sync::Arc;
use std::time::Duration;
use serde_json::json;
@@ -12,7 +13,8 @@ use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::swarm::members;
use crate::transport::identity::{WHO_LABEL, signed};
-use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage};
+use crate::transport::sim::SimNet;
+use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage, Transport};
const SOON: Duration = Duration::from_secs(20);
@@ -126,3 +128,45 @@ async fn a_proof_holds_only_at_its_address() {
// Known by its node ID, not by its address.
assert!(a.get_peer_state("ghost").await.is_none());
}
+
+/// A swarm across two transports: a only on the first, c only on the second, b on both.
+/// a and c never reach each other, and still share everything through b.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn nodes_on_both_transports_carry_the_others() {
+ let sim = Sim::with_identity(54);
+ let second = SimNet::new(55);
+ let a = sim.start("a", &[], Start::Genesis);
+ let b = sim.start_with("b", &["a"], Start::Join, |config| {
+ let identity = config.identity.as_mut().unwrap();
+ let transport: Arc<dyn Transport> = second.transport("b");
+ identity.more.push(("other", transport));
+ });
+ let c = sim.start_with("c", &["other:b"], Start::Join, |config| {
+ config.transport = second.transport("c");
+ config.identity.as_mut().unwrap().scheme = "other";
+ });
+ sim.wait_ready(&[&a, &b, &c], SOON).await;
+ assert_eq!(b.transport.local_addresses(), ["sim:b", "other:b"]);
+
+ write(&c, "from/c", json!("c")).await;
+ write(&a, "from/a", json!("a")).await;
+ sim.wait_converged(&[&a, &b, &c], SOON).await;
+ assert_eq!(read(&a, "from/c").await, Some(json!("c")));
+ assert_eq!(read(&c, "from/a").await, Some(json!("a")));
+
+ // The member list says where c is, on a transport a doesn't have.
+ let c_id = c.local_id().to_owned();
+ let a_ref = &a;
+ sim.wait_until("a knows c's entry", SOON, move || {
+ let c_id = c_id.clone();
+ async move {
+ let replica = a_ref.doc.read().await;
+ members::addresses(&replica.doc, &c_id) == ["other:c"]
+ }
+ })
+ .await;
+ assert!(!a.transport.can_reach(c.local_id()));
+ assert!(!c.transport.can_reach(a.local_id()));
+ assert!(a.transport.can_reach(b.local_id()));
+ assert!(c.transport.can_reach(b.local_id()));
+}
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 4d7374a..be2dc7b 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -94,6 +94,7 @@ impl Sim {
identity: self.identity.then(|| IdentityConfig {
key: Arc::new(sim_key(id)),
scheme: "sim",
+ more: Vec::new(),
}),
initial_epoch: Epoch::GENESIS,
initial_doc: AutoCommit::new(),
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index bdfeb53..5150e38 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -144,9 +144,15 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
(current, stable)
};
let k = state.neighbours_k;
+ // Those that don't answer, and those on none of our transports.
let mut unreachable = BTreeSet::new();
for member in &current {
- if let Some(peer_state) = state.get_peer_state(member).await
+ if member == state.local_id() {
+ continue;
+ }
+ if !state.transport.can_reach(member) {
+ unreachable.insert(member.clone());
+ } else if let Some(peer_state) = state.get_peer_state(member).await
&& peer_state.lock().await.is_unreachable()
{
unreachable.insert(member.clone());
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index 3036b11..926acdd 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -84,6 +84,10 @@ impl Transport for FirewallTransport {
self.inner.learn_addresses(peer, addresses)
}
+ fn can_reach(&self, peer: &str) -> bool {
+ self.inner.can_reach(peer)
+ }
+
async fn resolve_seed(&self, seed: &str) -> Result<String> {
self.inner.resolve_seed(seed).await
}
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 6f000d0..2b9bf5a 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -1,22 +1,26 @@
-//! The identity layer: between the node and a transport whose identities are addresses
-//! (Mycelium's public keys), so that the rest of the node deals in node IDs
-//! (`crate::identity`) only, whatever transport reaches a peer.
+//! The identity layer: between the node and its transports, whose identities are
+//! addresses (Mycelium's public keys), so that the rest of the node deals in node IDs
+//! (`crate::identity`) only, whatever transport reaches a peer. A node may have several
+//! transports; a peer is reached on one both have, so nodes on several transports
+//! carry the traffic between the others.
//!
//! A node proves who it is to an address with WHO (which asks back) or ME (the answer):
//! its public node key, and a signature over its own address and the recipient's
//! (`signed`). The transport authenticates the sender's address, so a proof is good for
-//! that pair of addresses only: replayed from another address, or to another recipient,
-//! it fails. Once a node proved itself at an address, messages from there are its.
+//! that pair of addresses only: replayed from another address, to another recipient or
+//! on another transport, it fails. Once a node proved itself at an address, messages
+//! from there are its.
//!
//! * A message from an address no node proved itself at is dropped, and the address is
//! asked WHO (at most every `ASK_INTERVAL`): after a restart, the first message from
//! a peer brings its proof back, and the peer's node retries what was lost.
-//! * A seed (`resolve_seed`) is an address, or something the transport resolves to one;
-//! it is asked WHO, and its ME says which node is there.
+//! * 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 the address it
-//! proved itself at; 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 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).
//!
//! Addresses are `<scheme>:<transport identity>`, e.g. `mycelium:<public key>`.
@@ -26,7 +30,7 @@ use std::time::Duration;
use anyhow::{Result, anyhow, bail};
use async_trait::async_trait;
-use tokio::sync::oneshot;
+use tokio::sync::{mpsc, oneshot};
use tokio::time::Instant;
use tracing::{debug, info};
@@ -64,8 +68,11 @@ pub(crate) fn is_not_proven(error: &anyhow::Error) -> bool {
#[derive(Clone)]
pub(crate) struct IdentityConfig {
pub(crate) key: Arc<NodeKey>,
- /// The transport's address scheme, e.g. `mycelium`.
+ /// The address scheme of the node's transport (`NodeConfig::transport`), e.g.
+ /// `mycelium`.
pub(crate) scheme: &'static str,
+ /// Further transports, by scheme, after the first in preference.
+ pub(crate) more: Vec<(&'static str, Arc<dyn Transport>)>,
}
/// The bytes a WHO (`label`: `WHO_LABEL`) or ME proof signs: the label, then the
@@ -83,9 +90,25 @@ pub(crate) struct IdentityTransport {
shared: Arc<Shared>,
}
-struct Shared {
- inner: Arc<dyn Transport>,
+/// 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>,
key: Arc<NodeKey>,
node_id: String,
/// WHO and ME are counted here, under the address (`traffic`).
@@ -94,60 +117,78 @@ struct Shared {
book: Mutex<Book>,
}
-/// Who is where. Keys are transport identities (an address without its scheme).
+/// Who is where, by address.
#[derive(Default)]
struct Book {
- /// The node that proved itself at a transport identity, and when.
+ /// The node that proved itself at an address, and when.
at: HashMap<String, (String, Instant)>,
- /// The transport identity each node last proved itself at.
- proven: HashMap<String, String>,
+ /// The addresses each node last proved itself at, one per scheme.
+ proven: HashMap<String, Vec<String>>,
/// Where others say a node is.
hints: HashMap<String, Vec<String>>,
- /// When each transport identity was last asked WHO.
+ /// When each address was last asked WHO.
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)
+}
+
impl Book {
- /// `node` proved itself at `target`. Returns whether that is news.
- fn bind(&mut self, target: &str, node: &str, now: Instant) -> bool {
- let news = self.at.get(target).is_none_or(|(known, _)| known != node);
- // One address per node and one node per address: a node that moved, or an
- // address taken over by another node, leaves no stale entry behind.
- if let Some(previous) = self.proven.insert(node.to_owned(), target.to_owned())
- && previous != target
- {
- self.at.remove(&previous);
+ /// `node` proved itself at `address`. Returns whether that is news.
+ fn bind(&mut self, address: &str, node: &str, now: Instant) -> bool {
+ let news = self.at.get(address).is_none_or(|(known, _)| known != node);
+ // One address per node and scheme, and one node per address: a node that
+ // moved, or an address taken over by another node, leaves no stale entry.
+ let scheme = scheme_of(address);
+ let addresses = self.proven.entry(node.to_owned()).or_default();
+ if let Some(slot) = addresses.iter_mut().find(|at| scheme_of(at) == scheme) {
+ if *slot != address {
+ let previous = std::mem::replace(slot, address.to_owned());
+ self.at.remove(&previous);
+ }
+ } else {
+ addresses.push(address.to_owned());
}
- if let Some((previous, _)) = self.at.insert(target.to_owned(), (node.to_owned(), now))
+ if let Some((previous, _)) = self.at.insert(address.to_owned(), (node.to_owned(), now))
&& previous != node
- && self.proven.get(&previous).is_some_and(|at| at == target)
{
- self.proven.remove(&previous);
+ self.forget_proof(&previous, address);
}
if self.at.len() > MAX_ENTRIES
&& let Some(oldest) = self
.at
.iter()
.min_by_key(|(_, (_, at))| *at)
- .map(|(target, _)| target.clone())
+ .map(|(address, _)| address.clone())
&& let Some((node, _)) = self.at.remove(&oldest)
{
- self.proven.remove(&node);
+ self.forget_proof(&node, &oldest);
}
- for waiter in self.waiting.remove(target).into_iter().flatten() {
+ for waiter in self.waiting.remove(address).into_iter().flatten() {
let _ = waiter.send(node.to_owned());
}
news
}
- /// Whether to ask `target` WHO now (always, if `force`), noting that it was asked.
- fn ask(&mut self, target: &str, now: Instant, force: bool) -> bool {
+ fn forget_proof(&mut self, node: &str, address: &str) {
+ if let Some(addresses) = self.proven.get_mut(node) {
+ addresses.retain(|at| at != address);
+ if addresses.is_empty() {
+ self.proven.remove(node);
+ }
+ }
+ }
+
+ /// 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(target)
+ .get(address)
.is_some_and(|at| now.duration_since(*at) < ASK_INTERVAL)
{
return false;
@@ -156,22 +197,26 @@ impl Book {
self.asked
.retain(|_, at| now.duration_since(*at) < ASK_INTERVAL);
}
- self.asked.insert(target.to_owned(), now);
+ self.asked.insert(address.to_owned(), now);
true
}
}
impl IdentityTransport {
+ /// Over `transport`, whose addresses have `config.scheme`, and `config.more`.
pub(crate) fn new(
- inner: Arc<dyn Transport>,
+ transport: Arc<dyn Transport>,
config: IdentityConfig,
traffic: Arc<Traffic>,
) -> 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 {
- inner,
- scheme: config.scheme,
+ links,
key: config.key,
node_id,
traffic,
@@ -186,129 +231,172 @@ impl Shared {
self.book.lock_or_recover()
}
- fn address(&self, target: &str) -> String {
- format!("{}:{target}", self.scheme)
+ /// The transport `address` is on, and the transport identity in it; None if no
+ /// transport of ours has its scheme.
+ fn split<'a>(&self, address: &'a str) -> Option<(&Link, &'a str)> {
+ let (scheme, target) = address.split_once(':')?;
+ let link = self.links.iter().find(|link| link.scheme == scheme)?;
+ link.inner.is_peer_id(target).then_some((link, target))
}
- /// The transport identity in `address`, if it is one of this transport's.
- fn target_of<'a>(&self, address: &'a str) -> Option<&'a str> {
- address
- .strip_prefix(self.scheme)
- .and_then(|rest| rest.strip_prefix(':'))
- .filter(|target| self.inner.is_peer_id(target))
+ /// 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)
}
- fn proof(&self, label: &[u8], target: &str) -> IdentityProof {
- let message = signed(
- label,
- &self.address(self.inner.local_id()),
- &self.address(target),
- );
+ fn is_own(&self, address: &str) -> bool {
+ self.links.iter().any(|link| link.own_address() == address)
+ }
+
+ fn proof(&self, label: &[u8], link: &Link, target: &str) -> IdentityProof {
+ let message = signed(label, &link.own_address(), &link.address(target));
IdentityProof {
key: self.key.public(),
signature: self.key.sign(&message),
}
}
- /// Notes the node `proof` proves to be at `sender`; None if it proves nothing.
- fn prove(&self, sender: &str, label: &[u8], proof: &IdentityProof) -> Option<String> {
- let message = signed(
- label,
- &self.address(sender),
- &self.address(self.inner.local_id()),
- );
+ /// 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());
if !identity::verify(&proof.key, &message, &proof.signature) {
- debug!(from = %self.address(sender), "Ignoring an identity proof that doesn't hold");
+ debug!(from = %address, "Ignoring an identity proof that doesn't hold");
return None;
}
let node = identity::node_id_of(&proof.key);
if node == self.node_id {
- debug!(from = %self.address(sender), "Ignoring a proof of our own node key");
+ debug!(from = %address, "Ignoring a proof of our own node key");
return None;
}
- if self.lock().bind(sender, &node, Instant::now()) {
- debug!(node = %node, address = %self.address(sender), "Node proved its address");
+ if self.lock().bind(address, &node, Instant::now()) {
+ debug!(node = %node, address = %address, "Node proved its address");
}
Some(node)
}
- /// Asks `target` WHO, unless it was asked recently (`force`: anyway).
- fn ask(self: &Arc<Self>, target: &str, force: bool) {
- if !self.lock().ask(target, Instant::now(), force) {
+ /// Asks `address` WHO, unless it was asked recently (`force`: anyway).
+ fn ask(self: &Arc<Self>, address: &str, force: bool) {
+ if !self.lock().ask(address, Instant::now(), force) {
return;
}
let shared = Arc::clone(self);
- let target = target.to_owned();
- tokio::spawn(async move { shared.send_proof(MessageKind::Who, &target).await });
+ let address = address.to_owned();
+ tokio::spawn(async move { shared.send_proof(MessageKind::Who, &address).await });
}
- /// Sends a WHO or ME (`kind`) to `target`.
- async fn send_proof(&self, kind: MessageKind, target: &str) {
+ /// 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 {
+ return;
+ };
let message = if kind == MessageKind::Who {
- ParsedMessage::Who(self.proof(WHO_LABEL, target))
+ ParsedMessage::Who(self.proof(WHO_LABEL, link, target))
} else {
- ParsedMessage::Me(self.proof(ME_LABEL, target))
+ ParsedMessage::Me(self.proof(ME_LABEL, link, target))
};
- let encoded = match self.inner.encode(target, &message) {
+ let encoded = match link.inner.encode(target, &message) {
Ok(encoded) => encoded,
Err(error) => {
- debug!(to = %target, error = %format!("{error:#}"), "Couldn't encode an identity proof");
+ debug!(to = %address, error = %format!("{error:#}"), "Couldn't encode an identity proof");
return;
}
};
- let address = self.address(target);
- match self.inner.send_encoded(&encoded).await {
+ match link.inner.send_encoded(&encoded).await {
Ok(()) => self
.traffic
- .sent(&address, encoded.kind, encoded.payload_len),
+ .sent(address, encoded.kind, encoded.payload_len),
Err(error) => {
self.traffic
- .refused(&address, encoded.kind, encoded.payload_len);
+ .refused(address, encoded.kind, encoded.payload_len);
debug!(to = %address, error = %format!("{error:#}"), "Couldn't send an identity proof");
}
}
}
+ /// A message from `sender` on `links[index]`.
async fn receive(
self: Arc<Self>,
+ index: usize,
sender: 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(&self.address(&sender), MessageKind::Who, payload_len);
- if self.prove(&sender, WHO_LABEL, &proof).is_some() {
+ .received(&address, MessageKind::Who, payload_len);
+ if self.prove(link, &address, WHO_LABEL, &proof).is_some() {
let shared = Arc::clone(&self);
- tokio::spawn(async move { shared.send_proof(MessageKind::Me, &sender).await });
+ tokio::spawn(async move { shared.send_proof(MessageKind::Me, &address).await });
}
Ok(())
}
ParsedMessage::Me(proof) => {
self.traffic
- .received(&self.address(&sender), MessageKind::Me, payload_len);
- self.prove(&sender, ME_LABEL, &proof);
+ .received(&address, MessageKind::Me, payload_len);
+ self.prove(link, &address, ME_LABEL, &proof);
Ok(())
}
message => {
- let node = self.lock().at.get(&sender).map(|(node, _)| node.clone());
+ let node = self.lock().at.get(&address).map(|(node, _)| node.clone());
match node {
Some(node) => handler(node, message, payload_len).await,
None => {
debug!(
- from = %self.address(&sender),
+ from = %address,
kind = ?message.kind(),
"Dropping a message from an address no node proved itself at; asking who is there"
);
- self.ask(&sender, false);
+ self.ask(&address, false);
Ok(())
}
}
}
}
}
+
+ /// 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)))
+ }
+}
+
+/// 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;
+ });
+ }
+ drop(sender);
+ receiver.recv().await
}
#[async_trait]
@@ -321,148 +409,195 @@ 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.inner.delivery_limit()
+ self.shared
+ .links
+ .iter()
+ .map(|link| link.inner.delivery_limit())
+ .max()
+ .unwrap_or_default()
}
fn local_addresses(&self) -> Vec<String> {
- vec![self.shared.address(self.shared.inner.local_id())]
+ self.shared.links.iter().map(Link::own_address).collect()
}
fn addresses_of(&self, peer: &str) -> Vec<String> {
let book = self.shared.lock();
- match book.proven.get(peer) {
- Some(target) => vec![self.shared.address(target)],
- None => book
- .hints
- .get(peer)
- .into_iter()
- .flatten()
- .map(|target| self.shared.address(target))
- .collect(),
- }
+ book.proven
+ .get(peer)
+ .or_else(|| book.hints.get(peer))
+ .cloned()
+ .unwrap_or_default()
}
fn learn_addresses(&self, peer: &str, addresses: &[String]) {
if peer == self.shared.node_id || !identity::is_node_id(peer) {
return;
}
- let local = self.shared.inner.local_id();
- let mut targets: Vec<String> = addresses
- .iter()
- .filter_map(|address| self.shared.target_of(address))
- .filter(|target| *target != local)
- .map(str::to_owned)
- .collect();
- targets.dedup();
- targets.truncate(MAX_HINTS);
- if targets.is_empty() {
+ let mut known: Vec<String> = Vec::new();
+ for address in addresses {
+ if self.shared.split(address).is_some()
+ && !self.shared.is_own(address)
+ && !known.contains(address)
+ {
+ known.push(address.clone());
+ }
+ }
+ known.truncate(MAX_HINTS);
+ if known.is_empty() {
return;
}
let mut book = self.shared.lock();
if book.hints.len() < MAX_ENTRIES || book.hints.contains_key(peer) {
- book.hints.insert(peer.to_owned(), targets);
+ book.hints.insert(peer.to_owned(), known);
}
}
+ fn can_reach(&self, peer: &str) -> bool {
+ self.shared.destination(peer).is_some()
+ }
+
/// A node ID resolves to itself if this node knows where it is; anything else is
- /// an address (`<scheme>:...`) or what the transport resolves to one, asked WHO.
+ /// an address (`<scheme>:...`) or what one of the transports resolves to one (the
+ /// first that does), asked WHO.
async fn resolve_seed(&self, seed: &str) -> Result<String> {
if identity::is_node_id(seed) {
- let book = self.shared.lock();
- if book.proven.contains_key(seed) || book.hints.contains_key(seed) {
+ if self.shared.destination(seed).is_some() {
return Ok(seed.to_owned());
}
bail!("a node ID alone doesn't say where that node is; give its address");
}
- let target = match self.shared.target_of(seed) {
- Some(target) => target.to_owned(),
- None => self.shared.inner.resolve_seed(seed).await?,
+ let address = match self.shared.split(seed) {
+ Some(_) => seed.to_owned(),
+ 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 target == self.shared.inner.local_id() {
+ if self.shared.is_own(&address) {
bail!("{seed} is this node");
}
let receiver = {
let mut book = self.shared.lock();
- if let Some((node, _)) = book.at.get(&target) {
+ if let Some((node, _)) = book.at.get(&address) {
return Ok(node.clone());
}
let (sender, receiver) = oneshot::channel();
- let waiting = book.waiting.entry(target.clone()).or_default();
+ let waiting = book.waiting.entry(address.clone()).or_default();
waiting.retain(|waiter| !waiter.is_closed());
waiting.push(sender);
receiver
};
- self.shared.ask(&target, true);
+ self.shared.ask(&address, true);
match tokio::time::timeout(RESOLVE_TIMEOUT, receiver).await {
Ok(Ok(node)) => {
info!(seed = %seed, node = %node, "Seed is node");
Ok(node)
}
_ => Err(anyhow!(
- "no node answered at {} within {RESOLVE_TIMEOUT:?}",
- self.shared.address(&target)
+ "no node answered at {address} within {RESOLVE_TIMEOUT:?}"
)),
}
}
- /// For the address `peer` proved itself at, or else one it is said to be at (a send
- /// there is refused until it proves itself, `send_encoded`).
+ /// For the address `destination` picks (a send there is refused until `peer`
+ /// proved itself there, `send_encoded`).
fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
- let target = {
- let book = self.shared.lock();
- book.proven
- .get(peer)
- .or_else(|| book.hints.get(peer).and_then(|hints| hints.first()))
- .cloned()
+ let Some(address) = self.shared.destination(peer) else {
+ bail!("no address known for {peer} on this node's transports");
};
- let Some(target) = target else {
- bail!("no address known for {peer}");
+ let Some((link, target)) = self.shared.split(&address) else {
+ bail!("no transport for {address}");
};
- let mut encoded = self.shared.inner.encode(&target, message)?;
+ let mut encoded = link.inner.encode(target, message)?;
encoded.peer = peer.to_owned();
+ encoded.address = address;
Ok(encoded)
}
async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
let hints = {
let book = self.shared.lock();
- if book.proven.get(&message.peer) == Some(&message.address) {
+ 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 {
- for target in hints {
- self.shared.ask(&target, false);
+ 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 = message.address.clone();
- self.shared.inner.send_encoded(&inner).await
+ inner.peer = target.to_owned();
+ inner.address = target.to_owned();
+ link.inner.send_encoded(&inner).await
}
+ /// Every transport's consumer; returns when the first one does.
async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
let handler: Arc<HandlerFn> = Arc::from(handler);
- let shared = Arc::clone(&self.shared);
- self.shared
- .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(sender, message, payload_len, &*handler)
- .await
- })
- }))
+ 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 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
+ })
+ }))
+ .await
+ }));
+ }
+ first(consumers)
.await
+ .unwrap_or_else(|| Err(anyhow!("no transports")))
}
+ /// When the first transport exits.
async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
- self.shared.inner.wait_for_exit().await
+ 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")))
}
}
@@ -482,37 +617,40 @@ mod tests {
}
#[test]
- fn one_address_per_node_and_one_node_per_address() {
+ fn one_address_per_node_and_scheme_and_one_node_per_address() {
let now = Instant::now();
let mut book = Book::default();
- assert!(book.bind("x", "n1", now));
- assert!(!book.bind("x", "n1", now));
- // n1 moved to y: x is no longer n1's.
- assert!(book.bind("y", "n1", now));
- assert!(!book.at.contains_key("x"));
- assert_eq!(book.proven["n1"], "y");
- // n2 took over y: n1 is nowhere proven.
- assert!(book.bind("y", "n2", now));
- assert!(!book.proven.contains_key("n1"));
- assert_eq!(book.at["y"].0, "n2");
+ assert!(book.bind("s:x", "n1", now));
+ assert!(!book.bind("s:x", "n1", now));
+ // n1 moved to s:y: s:x is no longer n1's.
+ assert!(book.bind("s:y", "n1", now));
+ assert!(!book.at.contains_key("s:x"));
+ assert_eq!(book.proven["n1"], ["s:y"]);
+ // On another transport too.
+ assert!(book.bind("t:x", "n1", now));
+ assert_eq!(book.proven["n1"], ["s:y", "t:x"]);
+ // n2 took over s:y: n1 is proven on t only.
+ assert!(book.bind("s:y", "n2", now));
+ assert_eq!(book.proven["n1"], ["t:x"]);
+ assert_eq!(book.at["s:y"].0, "n2");
}
#[test]
fn asking_is_rate_limited() {
let now = Instant::now();
let mut book = Book::default();
- assert!(book.ask("x", now, false));
- assert!(!book.ask("x", now, false));
- assert!(book.ask("x", now, true));
- assert!(book.ask("x", now + ASK_INTERVAL, false));
+ assert!(book.ask("s:x", now, false));
+ assert!(!book.ask("s:x", now, false));
+ assert!(book.ask("s:x", now, true));
+ assert!(book.ask("s:x", now + ASK_INTERVAL, false));
}
#[tokio::test]
async fn a_waiting_seed_learns_the_node() {
let mut book = Book::default();
let (sender, receiver) = oneshot::channel();
- book.waiting.insert("x".into(), vec![sender]);
- book.bind("x", "n1", Instant::now());
+ book.waiting.insert("s:x".into(), vec![sender]);
+ book.bind("s:x", "n1", Instant::now());
assert_eq!(receiver.await.unwrap(), "n1");
}
}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 52db406..cbbef45 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -537,6 +537,14 @@ pub(crate) trait Transport: Send + Sync {
vec![peer.to_owned()]
}
+ /// Whether this node has a way to reach `peer` at all: an address of it on one of
+ /// its transports. Members it can't reach aren't its neighbours (`swarm::topology`).
+ /// By default yes, as for `local_addresses`.
+ fn can_reach(&self, peer: &str) -> bool {
+ let _ = peer;
+ true
+ }
+
/// Where others say `peer` can be reached (the member list, a HELLO_REPLY): hints,
/// to be checked before they are relied on. By default nothing to learn, as for
/// `local_addresses`.
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 59acaf2..7e4dac7 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -161,6 +161,10 @@ impl Transport for CountingTransport {
self.inner.learn_addresses(peer, addresses)
}
+ fn can_reach(&self, peer: &str) -> bool {
+ self.inner.can_reach(peer)
+ }
+
async fn resolve_seed(&self, seed: &str) -> Result<String> {
self.inner.resolve_seed(seed).await
}