From c32d3ef7bda4353875f28c6ce09f3dce736c4ced Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Fri, 23 Oct 2026 13:21:12 +0200
Subject: pick pair leader by hash instead of key order

---
 src/swarm/mod.rs  |  1 +
 src/swarm/pair.rs | 78 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/swarm/peer.rs |  2 +-
 src/swarm/sync.rs |  2 +-
 4 files changed, 81 insertions(+), 2 deletions(-)
 create mode 100644 src/swarm/pair.rs

(limited to 'src/swarm')

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..7200a91
--- /dev/null
+++ b/src/swarm/pair.rs
@@ -0,0 +1,78 @@
+//! 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 b6c9700..cec8842 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 d82238e..2c2079a 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(),
-- 
cgit v1.3.1