add support for several transports

4 files changed, 324 insertions, 170 deletions

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
     }