introduce addresses

8 files changed, 173 insertions, 25 deletions

diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index cb8a558..c2edd01 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -10,8 +10,10 @@
 //!
 //! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name
 //! for people reading /peers and metrics (not unique and never an identity: the public
-//! key stays that), and an `alive` stamp the node renews now and then: the members
-//! renewed within `--electorate-window` are the electorate of a remake.
+//! key stays that), where it can be reached (`addresses`, as its transport gives them:
+//! hints for the others' transports), and an `alive` stamp the node renews now and
+//! then: the members renewed within `--electorate-window` are the electorate of a
+//! remake.
 
 use std::collections::HashMap;
 use std::time::Duration;
@@ -108,6 +110,42 @@ fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(),
     }
 }
 
+fn addresses_path(pk: &str) -> Vec<String> {
+    let mut path = member_path(pk);
+    path.push("addresses".to_owned());
+    path
+}
+
+/// Addresses a member most likely has, at most: one per transport.
+const MAX_ADDRESSES: usize = 8;
+
+/// Where `pk` says it can be reached (at most `MAX_ADDRESSES`); none if its entry
+/// doesn't say.
+pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec<String> {
+    match store::read_path(doc, &addresses_path(pk)) {
+        Ok(Some(JsonValue::Array(items))) => items
+            .into_iter()
+            .filter_map(|item| match item {
+                JsonValue::String(address) => Some(address),
+                _ => None,
+            })
+            .take(MAX_ADDRESSES)
+            .collect(),
+        _ => Vec::new(),
+    }
+}
+
+/// Hands every other member's addresses to the transport (`Transport::learn_addresses`):
+/// the member list is how a node finds where members it never met are.
+pub(crate) fn learn_addresses(state: &AppState, doc: &AutoCommit) {
+    let me = state.local_id();
+    for pk in members(doc) {
+        if pk != me {
+            state.transport.learn_addresses(&pk, &addresses(doc, &pk));
+        }
+    }
+}
+
 fn alive_path(pk: &str) -> Vec<String> {
     let mut path = member_path(pk);
     path.push("alive".to_owned());
@@ -146,15 +184,16 @@ fn alive_renewal(window: Duration) -> Duration {
 }
 
 /// Registers this node if it is ready and its entry is missing, keeps its alias as
-/// configured (set, changed or removed after a restart with another `--alias`), and
-/// renews its `alive` stamp now and then (see `electorate`).
+/// configured (set, changed or removed after a restart with another `--alias`) and its
+/// addresses as the transport gives them, and renews its `alive` stamp now and then
+/// (see `electorate`).
 pub(crate) async fn ensure_registered(state: &AppState) {
     if !state.is_ready().await {
         return;
     }
     let local = state.local_id().to_owned();
     let own = member_path(&local);
-    let (registered, recorded_alias, alive) = {
+    let (registered, recorded_alias, recorded_addresses, alive) = {
         let replica = state.doc.read().await;
         let registered = store::read_path(&replica.doc, &own)
             .ok()
@@ -163,14 +202,17 @@ pub(crate) async fn ensure_registered(state: &AppState) {
         (
             registered,
             alias(&replica.doc, &local),
+            addresses(&replica.doc, &local),
             millis_at(&replica.doc, &alive_path(&local)),
         )
     };
     let wanted = state.alias.clone();
+    let mut own_addresses = state.transport.local_addresses();
+    own_addresses.truncate(MAX_ADDRESSES);
     let now = state.clock.now_millis().as_u64();
 
     if !registered {
-        let mut entry = json!({ "since": now, "alive": now });
+        let mut entry = json!({ "since": now, "alive": now, "addresses": own_addresses });
         if let Some(alias) = &wanted {
             entry["alias"] = json!(alias);
         }
@@ -199,6 +241,18 @@ pub(crate) async fn ensure_registered(state: &AppState) {
         }
     }
 
+    if recorded_addresses != own_addresses {
+        let path = addresses_path(&local);
+        let value = json!(own_addresses);
+        match state
+            .mutate_doc(move |doc| store::write_path(doc, &path, &value))
+            .await
+        {
+            Ok(()) => info!(addresses = ?own_addresses, "Addresses updated in the member list"),
+            Err(error) => warn!(error = %error, "Failed to record our addresses"),
+        }
+    }
+
     if recorded_alias == wanted {
         return;
     }
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 4c657ad..4a7f451 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -19,6 +19,9 @@ use crate::transport::{self, ParsedMessage};
 /// costs the recipient as much as it is long. A joining node needs only a few to find
 /// the swarm; the member list gives it the rest once it is ready.
 const MAX_LISTED_PEERS: usize = 64;
+/// Addresses of one peer listed in a HELLO_REPLY, at most (a node has one per transport,
+/// so a few).
+const MAX_LISTED_ADDRESSES: usize = 8;
 
 pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
     if peer.is_empty() {
@@ -202,6 +205,17 @@ pub(super) async fn send_hello_reply(
         peers.sort_by_cached_key(|known| topology::pair_score(peer, known));
         peers.truncate(MAX_LISTED_PEERS);
     }
+    let peers = peers
+        .into_iter()
+        .map(|known| {
+            let mut addresses = state.transport.addresses_of(&known);
+            addresses.truncate(MAX_LISTED_ADDRESSES);
+            transport::ListedPeer {
+                id: known,
+                addresses,
+            }
+        })
+        .collect();
 
     let auth = state
         .swarm_key
@@ -237,11 +251,17 @@ pub(super) async fn send_hello_reply(
 /// Greets the peers a HELLO_REPLY listed: at most `MAX_LISTED_PEERS`, only valid
 /// identities, and, once the member list is known, only those that are our neighbours
 /// (it says whom we greet; a listed peer that isn't one would get nothing from us, so
-/// it isn't remembered either).
-pub(super) async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[String]) {
+/// it isn't remembered either). Where the sender reaches them is a hint to the
+/// transport (at most `MAX_LISTED_ADDRESSES` each).
+pub(super) async fn discover_peers(
+    state: &AppState,
+    from_peer: &str,
+    advertised_peers: &[transport::ListedPeer],
+) {
     let view_known = topology::view_known(state);
     let mut taken = 0;
-    for discovered_peer in advertised_peers {
+    for listed in advertised_peers {
+        let discovered_peer = &listed.id;
         if discovered_peer == from_peer || !state.transport.is_peer_id(discovered_peer) {
             continue;
         }
@@ -258,6 +278,8 @@ pub(super) async fn discover_peers(state: &AppState, from_peer: &str, advertised
         }
         taken += 1;
 
+        let addresses = &listed.addresses[..listed.addresses.len().min(MAX_LISTED_ADDRESSES)];
+        state.transport.learn_addresses(discovered_peer, addresses);
         remember_peer(state, discovered_peer).await;
         if let Err(error) = send_hello_once(state, discovered_peer).await {
             warn!(
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index 73b80a2..3693d89 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -31,7 +31,8 @@ const MIN_WRITE_INTERVAL: Duration = Duration::from_secs(3600);
 #[serde(rename_all = "camelCase")]
 pub(crate) struct Remembered {
     pub(crate) swarm_id: String,
-    /// Public keys of members to join through (sorted).
+    /// Addresses of members to join through (sorted), as their member list entries give
+    /// them (`swarm::members::addresses`).
     #[serde(default)]
     pub(crate) seeds: Vec<String>,
 }
@@ -173,16 +174,22 @@ pub(crate) async fn update(state: &AppState, now: bool) {
     }
 }
 
-/// Members to rejoin through: those alive within the electorate window, stable ones
-/// first, at most `MAX_SEEDS`, sorted. Changes only when members join, leave or change
-/// tier, so the file rarely does.
+/// Where to rejoin through: the addresses of members alive within the electorate
+/// window, stable ones first, of at most `MAX_SEEDS` members, sorted. Changes only when
+/// members join, leave, move or change tier, so the file rarely does.
 fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec<String> {
     let (mut stable, transient): (Vec<String>, Vec<String>) = members::electorate(doc, now, window)
         .into_iter()
         .filter(|pk| pk != me)
         .partition(|pk| members::tier(doc, pk) == members::Tier::Stable);
     stable.extend(transient);
-    stable.truncate(MAX_SEEDS);
-    stable.sort();
-    stable
+    let mut seeds: Vec<String> = stable
+        .iter()
+        .filter(|pk| !members::addresses(doc, pk).is_empty())
+        .take(MAX_SEEDS)
+        .flat_map(|pk| members::addresses(doc, pk))
+        .collect();
+    seeds.sort();
+    seeds.dedup();
+    seeds
 }
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index f8a1c30..bdfeb53 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -134,6 +134,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
     }
     let (current, stable) = {
         let replica = state.doc.read().await;
+        members::learn_addresses(state, &replica.doc);
         let current = members::members(&replica.doc);
         let stable: BTreeSet<String> = current
             .iter()
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index df82aa7..3036b11 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -72,6 +72,18 @@ impl Transport for FirewallTransport {
         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)
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 61b1f4b..29c48d6 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -297,10 +297,18 @@ pub(crate) struct HelloRequest {
     pub(crate) handshake: u32,
 }
 
+/// A peer a HELLO_REPLY lists: who it is, and where the sender reaches it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct ListedPeer {
+    pub(crate) id: String,
+    /// As `Transport::addresses_of` gives them; hints for the recipient.
+    pub(crate) addresses: Vec<String>,
+}
+
 #[derive(Clone, Serialize, Deserialize)]
 pub(crate) struct HelloReply {
     pub(crate) swarm_id: String,
-    pub(crate) peers: Vec<String>,
+    pub(crate) peers: Vec<ListedPeer>,
     #[serde(with = "crate::transport::wire::codec::epoch")]
     pub(crate) epoch: Epoch,
     /// See `SyncEnvelope::epoch_seq`.
@@ -490,6 +498,26 @@ pub(crate) trait Transport: Send + Sync {
     /// a message sent is delivered or lost for good.
     fn delivery_limit(&self) -> std::time::Duration;
 
+    /// Where this node can be reached, as addresses other nodes can use: published in
+    /// the member list (`swarm::members`). By default its identity, for a transport
+    /// whose identities are addresses (Mycelium: a public key).
+    fn local_addresses(&self) -> Vec<String> {
+        vec![self.local_id().to_owned()]
+    }
+
+    /// Where `peer` can be reached, as far as this node knows: passed on to others (in
+    /// a HELLO_REPLY). By default its identity, as for `local_addresses`.
+    fn addresses_of(&self, peer: &str) -> Vec<String> {
+        vec![peer.to_owned()]
+    }
+
+    /// 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`.
+    fn learn_addresses(&self, peer: &str, addresses: &[String]) {
+        let _ = (peer, addresses);
+    }
+
     /// 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/traffic.rs b/src/transport/traffic.rs
index f9b8289..c27f0d7 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -145,6 +145,18 @@ impl Transport for CountingTransport {
         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)
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 009380f..ceeaf87 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 = 3;
+pub(crate) const VERSION: u8 = 4;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -300,7 +300,7 @@ mod tests {
     use crate::document::replica::Epoch;
     use crate::transport::{
         BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, Keepalive,
-        MAX_LISTED_HEADS,
+        ListedPeer, MAX_LISTED_HEADS,
     };
     use automerge::ChangeHash;
 
@@ -330,7 +330,10 @@ mod tests {
 
         let reply = ParsedMessage::HelloReply(HelloReply {
             swarm_id: "swarm".into(),
-            peers: vec!["p".into()],
+            peers: vec![ListedPeer {
+                id: "p".into(),
+                addresses: vec!["a".into()],
+            }],
             epoch,
             epoch_seq: 1,
             auth: Vec::new(),
@@ -342,7 +345,13 @@ mod tests {
         let ParsedMessage::HelloReply(decoded) = decode(&encode(&reply).unwrap()).unwrap() else {
             panic!("not a HELLO_REPLY");
         };
-        assert_eq!(decoded.peers, vec!["p".to_owned()]);
+        assert_eq!(
+            decoded.peers,
+            vec![ListedPeer {
+                id: "p".into(),
+                addresses: vec!["a".into()],
+            }]
+        );
         assert_eq!(decoded.away_ms, Some(5));
         assert_eq!(decoded.sender_run_id, [6; 16]);
     }
@@ -438,12 +447,15 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&hello).unwrap()),
-            format!("030097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+            format!("040097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
         );
 
         let reply = ParsedMessage::HelloReply(HelloReply {
             swarm_id: "swarm".into(),
-            peers: vec!["p".into()],
+            peers: vec![ListedPeer {
+                id: "p".into(),
+                addresses: vec!["a".into()],
+            }],
             epoch,
             epoch_seq: 1,
             auth: Vec::new(),
@@ -455,7 +467,7 @@ mod tests {
         assert_eq!(
             hex(&encode(&reply).unwrap()),
             format!(
-                "030199a5737761726d91a170{epoch_hex}01c400c2{}05{}",
+                "040199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
                 run("04"),
                 run("06")
             )
@@ -470,7 +482,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&ping).unwrap()),
-            format!("0303959290c400{epoch_hex}01c3c2")
+            format!("0403959290c400{epoch_hex}01c3c2")
         );
     }
 }