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]] = {}