summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/members.rs66
-rw-r--r--src/swarm/peer/handshake.rs28
-rw-r--r--src/swarm/state_file.rs21
-rw-r--r--src/swarm/topology.rs1
4 files changed, 100 insertions, 16 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()