diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/state.rs | 4 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/pair.rs | 71 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 2 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 2 |
5 files changed, 76 insertions, 4 deletions
diff --git a/src/state.rs b/src/state.rs index f991fb8..cc4a790 100644 --- a/src/state.rs +++ b/src/state.rs @@ -177,7 +177,7 @@ impl PeerState { /// openings cross, and each side answers the other's: harmless between peers that /// share history, but an empty peer answers with empty heads, which Automerge takes /// for "lost everything", so the whole document goes out a second time. Only the - /// side with the smaller key opens; the other waits for its opening, up to + /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening, up to /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all. pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> { if we_open || self.sync_state.their_heads.is_some() { @@ -276,7 +276,7 @@ impl PeerState { /// True (and records the ping) if an established, quiet peer needs a keepalive. /// /// One PING and its PONG tell both sides that the other is alive, so only one side of - /// a pair needs to ping: the one with the smaller key (`we_ping`). Each of its PINGs + /// a pair needs to ping: the one leading the pair (`we_ping`, `swarm::pair`). Each of its PINGs /// makes the other side's peer look recently heard from, so the other side never /// pings while it does. If the pings stop, the other side pings once it has heard /// nothing for half an interval more. Without that, both sides would get due at about diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index a4277c7..7b53afc 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -3,5 +3,6 @@ pub(crate) mod auth; pub(crate) mod bootstrap; +pub(crate) mod pair; pub(crate) mod peer; pub(crate) mod sync; diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs new file mode 100644 index 0000000..c5e46c7 --- /dev/null +++ b/src/swarm/pair.rs @@ -0,0 +1,71 @@ +//! Which side of a pair of nodes leads, for duties only one side needs to do (opening +//! the sync after a reset, sending keepalive PINGs). +//! +//! Both sides decide on their own, from the two keys alone, and agree. The leader is +//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a +//! global order would make the node with the smallest key lead every one of its pairs +//! and the one with the largest lead none, so every such duty would pile up on the same +//! few nodes. With the hash, each node leads about half of its pairs. + +use sha2::{Digest, Sha256}; + +const LABEL: &[u8] = b"continuum/pair-leader/1"; + +/// Whether `local` leads its pair with `peer`. Exactly one of `leads(a, b)` and +/// `leads(b, a)` is true for any two different keys. +pub(crate) fn leads(local: &str, peer: &str) -> bool { + let (low, high) = if local <= peer { + (local, peer) + } else { + (peer, local) + }; + let mut hash = Sha256::new(); + // Length-prefixed fields, so no two different pairs hash the same input. + for field in [LABEL, low.as_bytes(), high.as_bytes()] { + hash.update((field.len() as u32).to_be_bytes()); + hash.update(field); + } + let leader = if hash.finalize()[0] & 1 == 0 { low } else { high }; + leader == local +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(n: u32) -> String { + format!("{:064x}", u64::from(n).wrapping_mul(0x9e37_79b9_7f4a_7c15)) + } + + #[test] + fn exactly_one_side_leads() { + for a in 0..50 { + for b in 0..50 { + if a != b { + assert_ne!(leads(&key(a), &key(b)), leads(&key(b), &key(a))); + } + } + } + } + + #[test] + fn leading_is_spread_across_nodes() { + let nodes: Vec<String> = (0..64).map(key).collect(); + let mut smallest_key_leads = 0; + for (i, node) in nodes.iter().enumerate() { + let led = nodes + .iter() + .filter(|peer| *peer != node && leads(node, peer)) + .count(); + // 63 pairs per node: about half each; far from all or none. + assert!((10..=53).contains(&led), "node {i} leads {led} of 63 pairs"); + for peer in &nodes { + if node < peer && leads(node, peer) { + smallest_key_leads += 1; + } + } + } + // Not simply the smaller key of each pair (64 * 63 / 2 = 2016 pairs). + assert!((800..=1216).contains(&smallest_key_leads), "{smallest_key_leads}"); + } +} diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 5928ac2..8c6843a 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -410,7 +410,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { let (became_dormant, ping_due, unanswered) = { let mut peer_state = peer_state_mutex.lock().await; let became_dormant = ready && peer_state.check_dormancy(&state.timing, now); - let we_ping = state.local_id() < peer.as_str(); + let we_ping = crate::swarm::pair::leads(state.local_id(), peer); let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now); let unanswered = peer_state.take_overdue_sync_reply(now); (became_dormant, ping_due, unanswered) diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 15b4ced..57f42e7 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -63,7 +63,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool }; if !retry_delivery { - let we_open = state.local_id() < peer; + let we_open = crate::swarm::pair::leads(state.local_id(), peer); if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) { trace!( wait_ms = wait.as_millis(), |