From 2da740c387c7c6b99ebeb61334d1af91ca772c9b Mon Sep 17 00:00:00 2001 From: DeepZima Date: Wed, 25 Mar 2026 11:55:12 +0100 Subject: [PATCH] Feat/static peer discovery (#1690) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Enabling peers to be discovered in environments where mDNS is unavailable (SSH sessions, headless servers, Docker).** ## Motivation Exo discovers peers exclusively via mDNS, which works great on a local network but breaks once you move beyond a single L2 broadcast domain: - SSH sessions on macOS — TCC blocks mDNS multicast from non-GUI sessions (#1488) - Headless servers/rack machines — #1682 ("DGX Spark does not find other nodes") - Docker Compose — mDNS is often unavailable across container networks; e.g. #1462 (E2E test framework) needs an alternative Related works: #1488 (working implementation made by @AlexCheema and closed because SSH had a GUI workaround), #1023 (Headscale WAN then closed due to merge conflicts), #1656 (discovery cleanup, open). This PR introduces an optional bootstrap mechanism for peer discovery while leaving the existing mDNS behavior unchanged. ## Changes Adds two new CLI flags: - `--bootstrap-peers` (env: `EXO_BOOTSTRAP_PEERS`) — comma-separated libp2p multiaddrs to dial on startup and retry periodically - `--libp2p-port` — fixed TCP port for libp2p to listen on (default: OS-assigned). Required when bootstrap peers, so other nodes know which port to dial. 8 files: - `rust/networking/src/discovery.rs`: Store bootstrap addrs, dial in existing retry loop - `rust/networking/src/swarm.rs`: Thread `bootstrap_peers` parameter to `Behaviour` - `rust/networking/examples/chatroom.rs`: Updated call site for new create_swarm signature - `rust/networking/tests/bootstrap_peers.rs`: Integration tests - `rust/exo_pyo3_bindings/src/networking.rs`: Accept optional `bootstrap_peers` in PyO3 constructor - `rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi` : Update type stub - `src/exo/routing/router.py`: Pass peers to `NetworkingHandle` - `src/exo/main.py` : `--bootstrap-peers` CLI arg + `EXO_BOOTSTRAP_PEERS` env var ## Why It Works Bootstrap peers are dialed in the existing retry loop — the same path taken by peers when mDNS-discovered. The swarm handles connection, Noise handshake, and gossipsub mesh joining from there. PeerId is intentionally not required in the multiaddr, the Noise handshake discovers it. Docker Compose example: ```yaml services: exo-1: environment: EXO_BOOTSTRAP_PEERS: "/ip4/exo-2/tcp/30000" exo-2: environment: EXO_BOOTSTRAP_PEERS: "/ip4/exo-1/tcp/30000" ``` ## Test Plan ### Manual Testing
Docker Compose config ``` services: exo-node1: build: context: . dockerfile: Dockerfile.bootstrap-test container_name: exo-bootstrap-node1 hostname: exo-node1 command: ["-q", "--libp2p-port", "30000", "--bootstrap-peers", "/ip4/172.30.20.3/tcp/30000"] environment: - EXO_LIBP2P_NAMESPACE=bootstrap-test ports: - "52415:52415" networks: bootstrap-net: ipv4_address: 172.30.20.2 deploy: resources: limits: memory: 4g exo-node2: build: context: . dockerfile: Dockerfile.bootstrap-test container_name: exo-bootstrap-node2 hostname: exo-node2 command: ["-q", "--libp2p-port", "30000", "--bootstrap-peers", "/ip4/172.30.20.2/tcp/30000"] environment: - EXO_LIBP2P_NAMESPACE=bootstrap-test ports: - "52416:52415" networks: bootstrap-net: ipv4_address: 172.30.20.3 deploy: resources: limits: memory: 4g networks: bootstrap-net: driver: bridge ipam: config: - subnet: 172.30.20.0/24 ```
Two containers on a bridge network (`172.30.20.0/24`), fixed IPs, `--libp2p-port 30000`, cross-referencing `--bootstrap-peers`. Both nodes found each other and established a connection then ran the election protocol. ### Automated Testing 4 Rust integration tests in `rust/networking/tests/bootstrap_peers.rs` (`cargo test -p networking`): | Test | What it verifies | Result | |------|-----------------|--------| | `two_nodes_connect_via_bootstrap_peers` | Node B discovers Node A via bootstrap addr (real TCP connection) | PASS | | `create_swarm_with_empty_bootstrap_peers` | Backward compatibility — no bootstrap peers works | PASS | | `create_swarm_ignores_invalid_bootstrap_addrs` | Invalid multiaddrs silently filtered | PASS | | `create_swarm_with_fixed_port` | `listen_port` parameter works | PASS | All 4 pass. The connection test takes ~6s --------- Signed-off-by: DeepZima Co-authored-by: Evan --- rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi | 2 +- rust/exo_pyo3_bindings/src/networking.rs | 11 +- rust/networking/examples/chatroom.rs | 11 +- rust/networking/src/discovery.rs | 10 +- rust/networking/src/swarm.rs | 25 +++-- rust/networking/tests/bootstrap_peers.rs | 107 +++++++++++++++++++ src/exo/main.py | 27 ++++- src/exo/routing/router.py | 12 ++- 8 files changed, 189 insertions(+), 16 deletions(-) create mode 100644 rust/networking/tests/bootstrap_peers.rs diff --git a/rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi b/rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi index 2af8239e..dc189ba4 100644 --- a/rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi +++ b/rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi @@ -42,7 +42,7 @@ class MessageTooLargeError(builtins.Exception): @typing.final class NetworkingHandle: - def __new__(cls, identity: Keypair) -> NetworkingHandle: ... + def __new__(cls, identity: Keypair, bootstrap_peers: list[builtins.str], listen_port: builtins.int) -> NetworkingHandle: ... async def gossipsub_subscribe(self, topic: builtins.str) -> builtins.bool: r""" Subscribe to a `GossipSub` topic. diff --git a/rust/exo_pyo3_bindings/src/networking.rs b/rust/exo_pyo3_bindings/src/networking.rs index 0e4f1063..9bcea5eb 100644 --- a/rust/exo_pyo3_bindings/src/networking.rs +++ b/rust/exo_pyo3_bindings/src/networking.rs @@ -180,7 +180,12 @@ impl PyNetworkingHandle { // ---- Lifecycle management methods ---- #[new] - fn py_new(identity: Bound<'_, PyKeypair>) -> PyResult { + #[pyo3(signature = (identity, bootstrap_peers, listen_port))] + fn py_new( + identity: Bound<'_, PyKeypair>, + bootstrap_peers: Vec, + listen_port: u16, + ) -> PyResult { // create communication channels let (to_swarm, from_client) = mpsc::channel(MPSC_CHANNEL_SIZE); @@ -189,7 +194,9 @@ impl PyNetworkingHandle { // create networking swarm (within tokio context!! or it crashes) let _guard = pyo3_async_runtimes::tokio::get_runtime().enter(); - let swarm = create_swarm(identity, from_client).pyerr()?.into_stream(); + let swarm = create_swarm(identity, from_client, bootstrap_peers, listen_port) + .pyerr()? + .into_stream(); Ok(Self { swarm: Arc::new(Mutex::new(swarm)), diff --git a/rust/networking/examples/chatroom.rs b/rust/networking/examples/chatroom.rs index cd25c99a..87fc87ee 100644 --- a/rust/networking/examples/chatroom.rs +++ b/rust/networking/examples/chatroom.rs @@ -16,9 +16,14 @@ async fn main() { let (to_swarm, from_client) = mpsc::channel(20); // Configure swarm - let mut swarm = swarm::create_swarm(identity::Keypair::generate_ed25519(), from_client) - .expect("Swarm creation failed") - .into_stream(); + let mut swarm = swarm::create_swarm( + identity::Keypair::generate_ed25519(), + from_client, + vec![], + 0, + ) + .expect("Swarm creation failed") + .into_stream(); // Create a Gossipsub topic & subscribe let (tx, rx) = oneshot::channel(); diff --git a/rust/networking/src/discovery.rs b/rust/networking/src/discovery.rs index a6cbd0e0..2ee2f970 100644 --- a/rust/networking/src/discovery.rs +++ b/rust/networking/src/discovery.rs @@ -104,6 +104,7 @@ pub struct Behaviour { // state-tracking for managed behaviors & mDNS-discovered peers managed: managed::Behaviour, mdns_discovered: HashMap>, + bootstrap_peers: Vec, retry_delay: Delay, // retry interval @@ -112,10 +113,11 @@ pub struct Behaviour { } impl Behaviour { - pub fn new(keypair: &identity::Keypair) -> io::Result { + pub fn new(keypair: &identity::Keypair, bootstrap_peers: Vec) -> io::Result { Ok(Self { managed: managed::Behaviour::new(keypair)?, mdns_discovered: HashMap::new(), + bootstrap_peers, retry_delay: Delay::new(RETRY_CONNECT_INTERVAL), pending_events: WakerDeque::new(), }) @@ -368,6 +370,12 @@ impl NetworkBehaviour for Behaviour { self.dial(p, ma) } } + // dial bootstrap peers (for environments where mDNS is unavailable) + for addr in &self.bootstrap_peers { + self.pending_events.push_back(ToSwarm::Dial { + opts: DialOpts::unknown_peer_id().address(addr.clone()).build(), + }) + } self.retry_delay.reset(RETRY_CONNECT_INTERVAL) // reset timeout } diff --git a/rust/networking/src/swarm.rs b/rust/networking/src/swarm.rs index 15babc43..a88863c6 100644 --- a/rust/networking/src/swarm.rs +++ b/rust/networking/src/swarm.rs @@ -142,19 +142,29 @@ fn filter_swarm_event(event: SwarmEvent) -> Option { } } -/// Create and configure a swarm which listens to all ports on OS +/// Create and configure a swarm. +/// +/// - `listen_port`: TCP port to listen on. `0` lets the OS assign one. +/// - `bootstrap_peers`: multiaddrs to dial for environments without mDNS. pub fn create_swarm( keypair: identity::Keypair, from_client: mpsc::Receiver, + bootstrap_peers: Vec, + listen_port: u16, ) -> alias::AnyResult { + let parsed_bootstrap_peers: Vec = bootstrap_peers + .iter() + .filter(|s| !s.is_empty()) + .filter_map(|s| s.parse().ok()) + .collect(); + let mut swarm = SwarmBuilder::with_existing_identity(keypair) .with_tokio() .with_other_transport(tcp_transport)? - .with_behaviour(Behaviour::new)? + .with_behaviour(|keypair| Behaviour::new(keypair, parsed_bootstrap_peers))? .build(); - // Listen on all interfaces and whatever port the OS assigns - swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?; + swarm.listen_on(format!("/ip4/0.0.0.0/tcp/{listen_port}").parse()?)?; Ok(Swarm { swarm, from_client }) } @@ -246,9 +256,12 @@ mod behaviour { } impl Behaviour { - pub fn new(keypair: &identity::Keypair) -> alias::AnyResult { + pub fn new( + keypair: &identity::Keypair, + bootstrap_peers: Vec, + ) -> alias::AnyResult { Ok(Self { - discovery: discovery::Behaviour::new(keypair)?, + discovery: discovery::Behaviour::new(keypair, bootstrap_peers)?, gossipsub: gossipsub_behaviour(keypair), }) } diff --git a/rust/networking/tests/bootstrap_peers.rs b/rust/networking/tests/bootstrap_peers.rs new file mode 100644 index 00000000..ef9eb294 --- /dev/null +++ b/rust/networking/tests/bootstrap_peers.rs @@ -0,0 +1,107 @@ +use futures_lite::StreamExt; +use networking::swarm::{FromSwarm, create_swarm}; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio::time::timeout; + +/// Helper: find a free TCP port. +fn free_port() -> u16 { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.local_addr().unwrap().port() +} + +/// Two nodes connect via bootstrap peers — no mDNS needed. +/// +/// Node A listens on a fixed port. Node B bootstraps to A's address. +/// We verify that B emits `FromSwarm::Discovered` for A's peer ID. +#[tokio::test] +async fn two_nodes_connect_via_bootstrap_peers() { + let port_a = free_port(); + + // Node A: listens on a known port, no bootstrap peers + let keypair_a = libp2p::identity::Keypair::generate_ed25519(); + let peer_id_a = keypair_a.public().to_peer_id(); + let (_tx_a, rx_a) = mpsc::channel(16); + let swarm_a = create_swarm(keypair_a, rx_a, vec![], port_a).expect("create swarm A"); + let mut stream_a = swarm_a.into_stream(); + + // Node B: bootstraps to A's address + let keypair_b = libp2p::identity::Keypair::generate_ed25519(); + let (_tx_b, rx_b) = mpsc::channel(16); + let swarm_b = create_swarm( + keypair_b, + rx_b, + vec![format!("/ip4/127.0.0.1/tcp/{port_a}")], + 0, + ) + .expect("create swarm B"); + let mut stream_b = swarm_b.into_stream(); + + // Wait for B to discover A (connection established) + let connected = timeout(Duration::from_secs(10), async { + loop { + tokio::select! { + Some(event) = stream_a.next() => { + // A will also see B connect, but we check from B's perspective + let _ = event; + } + Some(event) = stream_b.next() => { + if let FromSwarm::Discovered { peer_id } = event { + if peer_id == peer_id_a { + return true; + } + } + } + } + } + }) + .await; + + assert!( + connected.is_ok() && connected.unwrap(), + "Node B should discover Node A via bootstrap peer" + ); +} + +/// Empty bootstrap peers should work (backward compatible). +#[tokio::test] +async fn create_swarm_with_empty_bootstrap_peers() { + let keypair = libp2p::identity::Keypair::generate_ed25519(); + let (_tx, rx) = mpsc::channel(16); + let swarm = create_swarm(keypair, rx, vec![], 0); + assert!( + swarm.is_ok(), + "create_swarm with no bootstrap peers should succeed" + ); +} + +/// Invalid multiaddr strings are silently filtered out. +#[tokio::test] +async fn create_swarm_ignores_invalid_bootstrap_addrs() { + let keypair = libp2p::identity::Keypair::generate_ed25519(); + let (_tx, rx) = mpsc::channel(16); + let swarm = create_swarm( + keypair, + rx, + vec![ + "not-a-valid-multiaddr".to_string(), + "".to_string(), + "/ip4/10.0.0.1/tcp/30000".to_string(), // valid + ], + 0, + ); + assert!( + swarm.is_ok(), + "create_swarm should succeed even with invalid bootstrap addrs" + ); +} + +/// Fixed listen port works correctly. +#[tokio::test] +async fn create_swarm_with_fixed_port() { + let port = free_port(); + let keypair = libp2p::identity::Keypair::generate_ed25519(); + let (_tx, rx) = mpsc::channel(16); + let swarm = create_swarm(keypair, rx, vec![], port); + assert!(swarm.is_ok(), "create_swarm with fixed port should succeed"); +} diff --git a/src/exo/main.py b/src/exo/main.py index bf5ae688..93922cc2 100644 --- a/src/exo/main.py +++ b/src/exo/main.py @@ -47,7 +47,11 @@ class Node: keypair = get_node_id_keypair() node_id = NodeId(keypair.to_node_id()) session_id = SessionId(master_node_id=node_id, election_clock=0) - router = Router.create(keypair) + router = Router.create( + keypair, + bootstrap_peers=args.bootstrap_peers, + listen_port=args.libp2p_port, + ) await router.register_topic(topics.GLOBAL_EVENTS) await router.register_topic(topics.LOCAL_EVENTS) await router.register_topic(topics.COMMANDS) @@ -272,6 +276,9 @@ def main(): if args.offline: logger.info("Running in OFFLINE mode — no internet checks, local models only") + if args.bootstrap_peers: + logger.info(f"Bootstrap peers: {args.bootstrap_peers}") + if args.no_batch: os.environ["EXO_NO_BATCH"] = "1" logger.info("Continuous batching disabled (--no-batch)") @@ -308,6 +315,8 @@ class Args(CamelCaseModel): offline: bool = os.getenv("EXO_OFFLINE", "false").lower() == "true" no_batch: bool = False fast_synch: bool | None = None # None = auto, True = force on, False = force off + bootstrap_peers: list[str] = [] + libp2p_port: int @classmethod def parse(cls) -> Self: @@ -365,6 +374,22 @@ class Args(CamelCaseModel): action="store_true", help="Disable continuous batching, use sequential generation", ) + parser.add_argument( + "--bootstrap-peers", + type=lambda s: [p for p in s.split(",") if p], + default=os.getenv("EXO_BOOTSTRAP_PEERS", "").split(",") + if os.getenv("EXO_BOOTSTRAP_PEERS") + else [], + dest="bootstrap_peers", + help="Comma-separated libp2p multiaddrs to dial on startup (env: EXO_BOOTSTRAP_PEERS)", + ) + parser.add_argument( + "--libp2p-port", + type=int, + default=0, + dest="libp2p_port", + help="Fixed TCP port for libp2p to listen on (0 = OS-assigned).", + ) fast_synch_group = parser.add_mutually_exclusive_group() fast_synch_group.add_argument( "--fast-synch", diff --git a/src/exo/routing/router.py b/src/exo/routing/router.py index 0d4f4190..9447d6aa 100644 --- a/src/exo/routing/router.py +++ b/src/exo/routing/router.py @@ -1,3 +1,4 @@ +from collections.abc import Sequence from copy import copy from itertools import count from math import inf @@ -102,8 +103,15 @@ class TopicRouter[T: CamelCaseModel]: class Router: @classmethod - def create(cls, identity: Keypair) -> "Router": - return cls(handle=NetworkingHandle(identity)) + def create( + cls, + identity: Keypair, + bootstrap_peers: Sequence[str] = (), + listen_port: int = 0, + ) -> "Router": + return cls( + handle=NetworkingHandle(identity, list(bootstrap_peers), listen_port) + ) def __init__(self, handle: NetworkingHandle): self.topic_routers: dict[str, TopicRouter[CamelCaseModel]] = {}