From 747ed7a3e9596aa69d885a10fe16e2c77334e7b0 Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Thu, 16 Jul 2026 16:04:05 +0300 Subject: [PATCH] Added randezvous --- Cargo.lock | 89 +++++- Cargo.toml | 1 + README.md | 36 ++- apps/artist-dht-cli/src/main.rs | 27 +- crates/artist-dht/README.md | 51 +-- crates/artist-dht/src/config.rs | 14 +- crates/artist-dht/src/lib.rs | 2 +- crates/artist-dht/src/service.rs | 8 +- crates/federation-net/Cargo.toml | 3 + crates/federation-net/src/config.rs | 49 +++ crates/federation-net/src/engine.rs | 176 ++++++++--- crates/federation-net/src/lib.rs | 5 + crates/federation-net/src/rendezvous.rs | 344 +++++++++++++++++++++ crates/federation-net/tests/integration.rs | 64 +++- 14 files changed, 788 insertions(+), 81 deletions(-) create mode 100644 crates/federation-net/src/rendezvous.rs diff --git a/Cargo.lock b/Cargo.lock index 464d349..e96c6af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -529,6 +529,21 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" + [[package]] name = "critical-section" version = "1.2.0" @@ -810,6 +825,21 @@ dependencies = [ "winapi", ] +[[package]] +name = "document-features" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4b8a88685455ed29a21542a33abd9cb6510b6b129abadabdcef0f4c55bc8f61" +dependencies = [ + "litrs", +] + +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "ed25519" version = "3.0.0" @@ -919,9 +949,11 @@ dependencies = [ "anyhow", "blake3", "data-encoding", + "futures", "iroh", "iroh-base", "iroh-tickets", + "mainline", "postcard", "rand 0.9.4", "serde", @@ -955,6 +987,17 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "futures-core", + "futures-sink", + "spin 0.9.8", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1793,7 +1836,7 @@ dependencies = [ "iroh-base", "iroh-dns", "iroh-metrics", - "lru", + "lru 0.18.1", "n0-error", "n0-future", "noq", @@ -1965,6 +2008,12 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "litrs" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092" + [[package]] name = "lock_api" version = "0.4.14" @@ -1993,6 +2042,12 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "lru" +version = "0.16.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39" + [[package]] name = "lru" version = "0.18.1" @@ -2014,6 +2069,28 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3d25b0e0b648a86960ac23b7ad4abb9717601dec6f66c165f5b037f3f03065f" +[[package]] +name = "mainline" +version = "7.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dfad9601b9117218868271c05403853593d4817e7abeab5c95e11eee16382bb" +dependencies = [ + "crc", + "document-features", + "dyn-clone", + "ed25519-dalek", + "flume", + "futures-lite", + "getrandom 0.4.3", + "lru 0.16.4", + "serde", + "serde_bencode", + "serde_bytes", + "sha1_smol", + "thiserror 2.0.18", + "tracing", +] + [[package]] name = "matchers" version = "0.2.0" @@ -3161,6 +3238,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_bencode" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a70dfc7b7438b99896e7f8992363ab8e2c4ba26aa5ec675d32d1c3c2c33d413e" +dependencies = [ + "serde", + "serde_bytes", +] + [[package]] name = "serde_bytes" version = "0.11.19" diff --git a/Cargo.toml b/Cargo.toml index d50ac87..596c521 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ federation-net = { path = "crates/federation-net" } iroh = "1" iroh-base = "1" iroh-tickets = "1" +mainline = "7" tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time", "io-util", "io-std", "signal"] } serde = { version = "1", features = ["derive"] } postcard = { version = "1", features = ["alloc"] } diff --git a/README.md b/README.md index 4a686fe..e446bdc 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,9 @@ as an opaque serde-serializable payload. * Persistent peer identity (`/identity.key`, created on first start). * Connection establishment via a shareable string ticket (`fnet...`). +* Optional automatic peer discovery (**rendezvous**): peers of a network find + each other through the BitTorrent Mainline DHT knowing nothing but the + network id — no tickets and no bootstrap servers. * An application-level handshake that isolates networks and schemas. * Typed message exchange in both directions over one QUIC connection. * Network events (connect, disconnect, message, protocol error). @@ -20,10 +23,9 @@ as an opaque serde-serializable payload. ## What it deliberately does not do (yet) -No global peer discovery, gossip, broadcast overlay, DHT, content search, -file/chunk/streaming transfer, database sync, CRDTs, authorization, ACLs, -HTTP APIs or metrics. The architecture allows adding these later as separate -modules or ALPN protocols. +No gossip, broadcast overlay, content search, file/chunk/streaming transfer, +database sync, CRDTs, authorization, ACLs, HTTP APIs or metrics. The +architecture allows adding these later as separate modules or ALPN protocols. ## Usage @@ -126,6 +128,32 @@ reject each other, because they could not decode each other's messages. Any backwards-incompatible change to your message type requires a new schema name. +## Rendezvous (peer discovery by network id) + +Passing a `RendezvousConfig` to the config builder enables automatic peer +discovery: every peer periodically publishes its own Iroh address into a +shared BEP44 mutable record in the public BitTorrent Mainline DHT and dials +the addresses other peers published there. The record's signing key is +derived deterministically from the `NetworkId`, so knowing the network id is +enough to find and join the network; the first peer of a new network simply +publishes itself and waits. + +```rust +let config = NetworkConfig::builder() + .data_dir("./peer-a") + .network_id(NetworkId::from_name("example-network")) + .schema_id(SchemaId::from_name("demo-message-v1")) + .rendezvous(federation_net::RendezvousConfig::default()) + .build()?; +``` + +Because the rendezvous record is world-readable, the network id acts as a +public rendezvous token: anyone who knows it can discover and join the +network. Treat the id of a private network like a shared secret (e.g. +`myorg-prod-7f3a`); the handshake still rejects peers whose network id does +not match exactly. Rendezvous is off by default — without it, peers are +connected via tickets only. + ## Tickets A `PeerTicket` is a self-contained invitation string with the `fnet` prefix diff --git a/apps/artist-dht-cli/src/main.rs b/apps/artist-dht-cli/src/main.rs index b38f453..e329b96 100644 --- a/apps/artist-dht-cli/src/main.rs +++ b/apps/artist-dht-cli/src/main.rs @@ -8,7 +8,8 @@ use std::sync::mpsc as std_mpsc; use anyhow::Context; use artist_dht::{ - Artist, ArtistDhtConfig, ArtistDhtEvent, ArtistDhtService, NetworkId, PeerTicket, SearchOutcome, + Artist, ArtistDhtConfig, ArtistDhtEvent, ArtistDhtService, NetworkId, PeerTicket, + RendezvousConfig, SearchOutcome, }; use clap::Parser; use rustyline::ExternalPrinter; @@ -30,9 +31,16 @@ struct Args { name: String, /// Ticket(s) of peers to connect to on startup; may be repeated. + /// Optional: peers of the same network normally find each other + /// automatically through the mainline DHT. #[arg(long)] connect: Vec, + /// Disable automatic peer discovery through the mainline DHT; only the + /// tickets given via --connect are used. + #[arg(long)] + no_bootstrap: bool, + /// Enable verbose logging. #[arg(long)] verbose: bool, @@ -61,11 +69,13 @@ async fn main() -> anyhow::Result<()> { .with_writer(std::io::stderr) .init(); - let config = ArtistDhtConfig::builder() + let mut config_builder = ArtistDhtConfig::builder() .data_dir(&args.data_dir) - .network_id(NetworkId::from_name(&args.network_id)) - .build() - .context("invalid configuration")?; + .network_id(NetworkId::from_name(&args.network_id)); + if !args.no_bootstrap { + config_builder = config_builder.rendezvous(RendezvousConfig::default()); + } + let config = config_builder.build().context("invalid configuration")?; let (service, mut events) = ArtistDhtService::start(config) .await @@ -96,7 +106,12 @@ async fn main() -> anyhow::Result<()> { } } } - if args.connect.is_empty() { + if !args.no_bootstrap { + println!( + "Discovering '{}' peers via the mainline DHT... (may take up to a minute)", + args.network_id + ); + } else if args.connect.is_empty() { println!("Waiting for peers... (share the ticket above)"); } println!("Type /help for commands."); diff --git a/crates/artist-dht/README.md b/crates/artist-dht/README.md index 3961c80..04ee802 100644 --- a/crates/artist-dht/README.md +++ b/crates/artist-dht/README.md @@ -21,7 +21,10 @@ running peers itself. * Each peer has a stable 256-bit `NodeId` derived from its persistent `federation-net` endpoint id; records live on the `K = 8` XOR-closest nodes. -* Peers discover each other through automatic Hello/PeerExchange gossip; +* A peer joins the network knowing **only the network id**: on startup it + finds other peers through a rendezvous record in the public BitTorrent + Mainline DHT (see `federation-net`'s rendezvous). Within the network peers + then discover each other through automatic Hello/PeerExchange gossip; connections to newly learned peers are opened **on demand** from stored tickets. * Deletions propagate as **tombstones** (revision-based, they beat active @@ -33,16 +36,29 @@ running peers itself. Out of scope (by design): fuzzy search, content transfer, CRDTs, consensus, signatures on DHT records, Sybil protection, accounts, GUI. -## Bootstrapping limitation +## Bootstrapping -There is deliberately **no bootstrap server**, so a peer cannot join a -network it has no contact in: you must pass the ticket of *any* already -running peer via `--connect`. That first peer has no special role — once -contacts have spread through peer exchange, it can be switched off. +There is deliberately **no bootstrap server**. A peer joins a network knowing +only its `--network-id`: peers of the same network find each other through a +shared rendezvous record in the public BitTorrent Mainline DHT (the record's +signing key is derived from the network id). The first peer of a new network +simply publishes itself and waits; every later peer with the same id finds it +within a rendezvous round or two (typically well under a minute). No peer has +a special role. + +Because the rendezvous record is world-readable, the network id acts as a +public rendezvous token: anyone who knows it can discover and join the +network. Pick a unique, hard-to-guess name for a private network (e.g. +`myband-artists-prod-7f3a`). + +Discovery can be turned off with `--no-bootstrap`; then a peer can only join +through the ticket of *any* already running peer passed via `--connect` +(tickets also work in addition to discovery, e.g. on isolated networks +without internet access). ## Running the demo (three peers) -Start Peer A (alice): +Start the peers in any order — they only share the network id: ```bash cargo run -p artist-dht-cli -- \ @@ -51,30 +67,25 @@ cargo run -p artist-dht-cli -- \ --name alice ``` -Copy the printed `Ticket: fnet...`. Start Peer B (bob) with it: - ```bash cargo run -p artist-dht-cli -- \ --data-dir ./peer-b \ --network-id demo-artists \ - --name bob \ - --connect 'fnet...' + --name bob ``` -Start Peer C (charlie), connected **only to A** — it will learn about B via -peer exchange: - ```bash cargo run -p artist-dht-cli -- \ --data-dir ./peer-c \ --network-id demo-artists \ - --name charlie \ - --connect 'fnet...' + --name charlie ``` -`--connect` may be repeated to dial several peers. All three processes must -use the same `--network-id`; a peer from another network is rejected during -the transport handshake. +Within a minute `Peer connected: ...` lines appear on all three. All +processes must use the same `--network-id`; a peer from another network is +rejected during the transport handshake. (With `--no-bootstrap`, pass the +ticket printed by a running peer via `--connect 'fnet...'` instead; +`--connect` may be repeated.) ### Demo scenario @@ -130,6 +141,8 @@ use federation_net::NetworkId; let config = ArtistDhtConfig::builder() .data_dir("./peer-a") .network_id(NetworkId::from_name("demo-artists")) + // Optional: discover peers of this network via the mainline DHT. + .rendezvous(artist_dht::RendezvousConfig::default()) .build()?; let (service, mut events) = ArtistDhtService::start(config).await?; diff --git a/crates/artist-dht/src/config.rs b/crates/artist-dht/src/config.rs index 5050be6..b89341a 100644 --- a/crates/artist-dht/src/config.rs +++ b/crates/artist-dht/src/config.rs @@ -3,7 +3,7 @@ use std::path::PathBuf; use std::time::Duration; -use federation_net::NetworkId; +use federation_net::{NetworkId, RendezvousConfig}; use crate::error::{ArtistDhtError, Result}; @@ -40,6 +40,10 @@ pub struct ArtistDhtConfig { /// establishing a connection through relays can take much longer than a /// request over an existing one. pub transport_timeout: Duration, + /// Automatic peer discovery over the mainline DHT: peers of the same + /// network find each other knowing nothing but the network id. `None` + /// disables it; peers are then connected via tickets only. + pub rendezvous: Option, } impl ArtistDhtConfig { @@ -62,6 +66,7 @@ pub struct ArtistDhtConfigBuilder { request_timeout: Option, lookup_timeout: Option, transport_timeout: Option, + rendezvous: Option, } impl ArtistDhtConfigBuilder { @@ -107,6 +112,12 @@ impl ArtistDhtConfigBuilder { self } + /// Enables automatic peer discovery over the mainline DHT. + pub fn rendezvous(mut self, rendezvous: RendezvousConfig) -> Self { + self.rendezvous = Some(rendezvous); + self + } + /// Validates and builds the configuration. pub fn build(self) -> Result { let data_dir = self @@ -131,6 +142,7 @@ impl ArtistDhtConfigBuilder { request_timeout: self.request_timeout.unwrap_or(DEFAULT_REQUEST_TIMEOUT), lookup_timeout: self.lookup_timeout.unwrap_or(DEFAULT_LOOKUP_TIMEOUT), transport_timeout: self.transport_timeout.unwrap_or(DEFAULT_TRANSPORT_TIMEOUT), + rendezvous: self.rendezvous, }; for (name, value) in [ ("republish_interval", config.republish_interval), diff --git a/crates/artist-dht/src/lib.rs b/crates/artist-dht/src/lib.rs index 7d9e816..a129dc8 100644 --- a/crates/artist-dht/src/lib.rs +++ b/crates/artist-dht/src/lib.rs @@ -89,4 +89,4 @@ pub use service::{ }; // Re-exported types from the transport layer that appear in this API. -pub use federation_net::{EndpointId, NetworkId, PeerTicket}; +pub use federation_net::{EndpointId, NetworkId, PeerTicket, RendezvousConfig}; diff --git a/crates/artist-dht/src/service.rs b/crates/artist-dht/src/service.rs index 1cf02a1..7c091c5 100644 --- a/crates/artist-dht/src/service.rs +++ b/crates/artist-dht/src/service.rs @@ -106,11 +106,15 @@ impl ArtistDhtService { /// Starts the service: opens the database, starts the network engine, /// loads persisted contacts and spawns the maintenance tasks. pub async fn start(config: ArtistDhtConfig) -> Result<(Self, ArtistDhtEventReceiver)> { - let engine_config = NetworkConfig::builder() + let mut engine_builder = NetworkConfig::builder() .data_dir(&config.data_dir) .network_id(config.network_id) .schema_id(SchemaId::from_name(SCHEMA_NAME)) - .request_timeout(config.transport_timeout) + .request_timeout(config.transport_timeout); + if let Some(rendezvous) = config.rendezvous.clone() { + engine_builder = engine_builder.rendezvous(rendezvous); + } + let engine_config = engine_builder .build() .map_err(|err| ArtistDhtError::Network(err.to_string()))?; let (engine, net_events) = NetworkEngine::start(engine_config).await?; diff --git a/crates/federation-net/Cargo.toml b/crates/federation-net/Cargo.toml index 43fea27..b04f03c 100644 --- a/crates/federation-net/Cargo.toml +++ b/crates/federation-net/Cargo.toml @@ -10,6 +10,8 @@ rust-version.workspace = true iroh = { workspace = true } iroh-base = { workspace = true } iroh-tickets = { workspace = true } +mainline = { workspace = true } +futures = { workspace = true } tokio = { workspace = true } serde = { workspace = true } postcard = { workspace = true } @@ -21,5 +23,6 @@ rand = { workspace = true } [dev-dependencies] anyhow = { workspace = true } +mainline = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/federation-net/src/config.rs b/crates/federation-net/src/config.rs index 10cbf35..0b52c22 100644 --- a/crates/federation-net/src/config.rs +++ b/crates/federation-net/src/config.rs @@ -5,6 +5,7 @@ use std::time::Duration; use crate::error::{NetworkError, Result}; use crate::protocol::{NetworkId, SchemaId}; +use crate::rendezvous::RendezvousConfig; /// Default maximum size of an encoded domain message (256 KiB). pub const DEFAULT_MAX_MESSAGE_SIZE: usize = 256 * 1024; @@ -34,6 +35,9 @@ pub struct NetworkConfig { pub event_channel_capacity: usize, /// Maximum number of concurrently processed streams per peer. pub max_concurrent_streams_per_peer: usize, + /// Automatic peer discovery over the mainline DHT; `None` disables it and + /// peers are connected via tickets only. + pub rendezvous: Option, } impl NetworkConfig { @@ -56,6 +60,7 @@ pub struct NetworkConfigBuilder { request_timeout: Option, event_channel_capacity: Option, max_concurrent_streams_per_peer: Option, + rendezvous: Option, } impl NetworkConfigBuilder { @@ -101,6 +106,12 @@ impl NetworkConfigBuilder { self } + /// Enables automatic peer discovery over the mainline DHT. + pub fn rendezvous(mut self, rendezvous: RendezvousConfig) -> Self { + self.rendezvous = Some(rendezvous); + self + } + /// Validates the configuration and builds a [`NetworkConfig`]. pub fn build(self) -> Result { let data_dir = self @@ -156,6 +167,19 @@ impl NetworkConfigBuilder { )); } + if let Some(rendezvous) = &self.rendezvous { + if rendezvous.interval.is_zero() { + return Err(NetworkError::InvalidConfig( + "rendezvous interval must be greater than zero".into(), + )); + } + if rendezvous.entry_ttl.is_zero() { + return Err(NetworkError::InvalidConfig( + "rendezvous entry_ttl must be greater than zero".into(), + )); + } + } + Ok(NetworkConfig { data_dir, network_id, @@ -164,6 +188,7 @@ impl NetworkConfigBuilder { request_timeout, event_channel_capacity, max_concurrent_streams_per_peer, + rendezvous: self.rendezvous, }) } } @@ -223,4 +248,28 @@ mod tests { .is_err() ); } + + #[test] + fn rendezvous_is_disabled_by_default_and_validated_when_set() { + let config = base_builder().build().expect("valid config"); + assert!(config.rendezvous.is_none()); + + let config = base_builder() + .rendezvous(RendezvousConfig::default()) + .build() + .expect("valid config"); + assert!(config.rendezvous.is_some()); + + let zero_interval = RendezvousConfig { + interval: Duration::ZERO, + ..RendezvousConfig::default() + }; + assert!(base_builder().rendezvous(zero_interval).build().is_err()); + + let zero_ttl = RendezvousConfig { + entry_ttl: Duration::ZERO, + ..RendezvousConfig::default() + }; + assert!(base_builder().rendezvous(zero_ttl).build().is_err()); + } } diff --git a/crates/federation-net/src/engine.rs b/crates/federation-net/src/engine.rs index 1139912..329d43c 100644 --- a/crates/federation-net/src/engine.rs +++ b/crates/federation-net/src/engine.rs @@ -8,10 +8,10 @@ use std::time::Duration; use iroh::endpoint::{Connection, RecvStream, SendStream, VarInt, presets}; use iroh::protocol::{AcceptError, ProtocolHandler, Router}; -use iroh::{Endpoint, EndpointId}; +use iroh::{Endpoint, EndpointAddr, EndpointId}; use serde::Serialize; use serde::de::DeserializeOwned; -use tokio::sync::{Semaphore, mpsc}; +use tokio::sync::{Semaphore, mpsc, watch}; use tokio::task::JoinSet; use tokio::time::timeout; use tracing::{debug, info, warn}; @@ -24,6 +24,7 @@ use crate::protocol::{ ALPN, Handshake, HandshakeAck, HandshakeErrorCode, MAX_HANDSHAKE_FRAME_SIZE, MessageRejectReason, MessageRequest, MessageResponse, NetworkId, PROTOCOL_VERSION, SchemaId, }; +use crate::rendezvous::{RendezvousClient, RendezvousConfig, now_ms}; use crate::ticket::{PeerTicket, TICKET_VERSION}; use crate::wire; @@ -47,6 +48,8 @@ const REJECT_LINGER: Duration = Duration::from_secs(3); /// How long [`NetworkEngine::shutdown`] waits for background tasks before /// aborting them. const SHUTDOWN_TASK_GRACE: Duration = Duration::from_secs(5); +/// Upper bound of the rendezvous round delay while no peer is connected yet. +const RENDEZVOUS_LONELY_INTERVAL: Duration = Duration::from_secs(15); /// Bounds required of a domain message type. /// @@ -85,6 +88,9 @@ struct Shared { tasks: Mutex>, next_generation: AtomicU64, shutting_down: AtomicBool, + /// Broadcasts the start of the shutdown to long-running background + /// loops so they can stop promptly instead of being aborted. + shutdown_signal: watch::Sender, } impl Shared { @@ -396,6 +402,52 @@ impl Shared { Ok(()) } + /// Dials `addr`, runs the client handshake and registers the connection. + /// + /// An existing healthy connection to the same peer is reused. The remote + /// identity is taken from the authenticated Iroh connection, never from + /// `addr` itself. + async fn connect_to_addr(self: &Arc, addr: EndpointAddr) -> Result { + self.ensure_running()?; + let target = addr.id; + if lock(&self.peers).contains_key(&target) { + debug!(peer = %target, "reusing existing connection"); + return Ok(target); + } + + let request_timeout = self.config.request_timeout; + let connection = timeout(request_timeout, self.endpoint.connect(addr, ALPN)) + .await + .map_err(|_| NetworkError::Timeout)? + .map_err(|err| NetworkError::Transport(format!("failed to connect: {err}")))?; + let peer_id = connection.remote_id(); + + let handshake = timeout(request_timeout, self.run_client_handshake(&connection)) + .await + .map_err(|_| NetworkError::Timeout) + .and_then(|res| res); + if let Err(err) = handshake { + connection.close(CLOSE_CODE_HANDSHAKE_REJECTED, b"handshake failed"); + return Err(err); + } + + let generation = + self.register_connection(peer_id, connection.clone(), ConnectionDirection::Outgoing); + info!(peer = %peer_id, "peer connected (outgoing)"); + self.emit(NetworkEvent::PeerConnected { + peer_id, + direction: ConnectionDirection::Outgoing, + }) + .await; + let loop_shared = self.clone(); + self.spawn_task(async move { + loop_shared + .connection_loop(peer_id, connection, generation) + .await; + }); + Ok(peer_id) + } + /// Runs the client side of the handshake on a fresh outgoing connection. async fn run_client_handshake(&self, connection: &Connection) -> Result<()> { let (mut send, mut recv) = connection.open_bi().await.map_err(|err| { @@ -418,6 +470,65 @@ impl Shared { } } +/// Periodically publishes this peer to the network's rendezvous record and +/// dials every other peer found there. +/// +/// Failures of a single round or dial are logged and retried on the next +/// round; the loop only ends when the engine shuts down (the task is +/// aborted). +async fn rendezvous_loop( + shared: Arc>, + client: RendezvousClient, + config: RendezvousConfig, +) { + let mut shutdown = shared.shutdown_signal.subscribe(); + // Give the endpoint a moment to learn its relay and direct addresses so + // the very first published record is already dialable. + let _ = timeout(shared.config.request_timeout, shared.endpoint.online()).await; + let self_id = shared.endpoint.id(); + loop { + if shared.is_shutting_down() { + return; + } + let round = async { + let addr = shared.endpoint.addr(); + let self_addr = (!addr.is_empty()).then_some(addr); + match client.round(self_addr, now_ms()).await { + Ok(peers) => { + for peer_addr in peers { + let peer = peer_addr.id; + if peer == self_id || lock(&shared.peers).contains_key(&peer) { + continue; + } + debug!(peer = %peer, "rendezvous discovered a peer; connecting"); + if let Err(err) = shared.connect_to_addr(peer_addr).await { + // Stale entries (peers that left) fail here; they + // age out of the record by TTL. + debug!(peer = %peer, error = %err, "rendezvous connect attempt failed"); + } + } + } + Err(err) => debug!(error = %err, "rendezvous round failed"), + } + }; + tokio::select! { + _ = round => {} + _ = shutdown.changed() => return, + } + // Poll faster while the peer is still alone: joining a network + // should not have to wait a full interval for the next round. + let delay = if lock(&shared.peers).is_empty() { + config.interval.min(RENDEZVOUS_LONELY_INTERVAL) + } else { + config.interval + }; + tokio::select! { + _ = tokio::time::sleep(delay) => {} + _ = shutdown.changed() => return, + } + } +} + fn handshake_code_to_error(code: HandshakeErrorCode) -> NetworkError { match code { HandshakeErrorCode::UnsupportedProtocolVersion => NetworkError::UnsupportedProtocolVersion, @@ -506,12 +617,26 @@ impl NetworkEngine { tasks: Mutex::new(JoinSet::new()), next_generation: AtomicU64::new(0), shutting_down: AtomicBool::new(false), + shutdown_signal: watch::Sender::new(false), }); let handler = FederationProtocol { shared: Arc::downgrade(&shared), }; let router = Router::builder(endpoint).accept(ALPN, handler).spawn(); *lock(&shared.router) = Some(router); + if let Some(rendezvous) = shared.config.rendezvous.clone() { + match RendezvousClient::new(shared.config.network_id, &rendezvous) { + Ok(client) => { + let loop_shared = shared.clone(); + shared.spawn_task(async move { + rendezvous_loop(loop_shared, client, rendezvous).await; + }); + } + // Rendezvous is a convenience; the engine stays usable via + // tickets even when the DHT client cannot start. + Err(err) => warn!(error = %err, "peer rendezvous disabled"), + } + } info!( endpoint_id = %endpoint_id, network_id = %shared.config.network_id, @@ -587,54 +712,12 @@ impl NetworkEngine { if ticket.schema_id != shared.config.schema_id { return Err(NetworkError::SchemaMismatch); } - let target = ticket.endpoint_id(); - if target == self.endpoint_id() { + if ticket.endpoint_id() == self.endpoint_id() { return Err(NetworkError::InvalidTicket( "the ticket points to this peer itself".to_string(), )); } - if lock(&shared.peers).contains_key(&target) { - debug!(peer = %target, "reusing existing connection"); - return Ok(target); - } - - let request_timeout = shared.config.request_timeout; - let connection = timeout( - request_timeout, - shared.endpoint.connect(ticket.endpoint_addr, ALPN), - ) - .await - .map_err(|_| NetworkError::Timeout)? - .map_err(|err| NetworkError::Transport(format!("failed to connect: {err}")))?; - // The remote identity comes from the authenticated Iroh connection, - // never from the ticket payload. - let peer_id = connection.remote_id(); - - let handshake = timeout(request_timeout, shared.run_client_handshake(&connection)) - .await - .map_err(|_| NetworkError::Timeout) - .and_then(|res| res); - if let Err(err) = handshake { - connection.close(CLOSE_CODE_HANDSHAKE_REJECTED, b"handshake failed"); - return Err(err); - } - - let generation = - shared.register_connection(peer_id, connection.clone(), ConnectionDirection::Outgoing); - info!(peer = %peer_id, "peer connected (outgoing)"); - shared - .emit(NetworkEvent::PeerConnected { - peer_id, - direction: ConnectionDirection::Outgoing, - }) - .await; - let loop_shared = shared.clone(); - shared.spawn_task(async move { - loop_shared - .connection_loop(peer_id, connection, generation) - .await; - }); - Ok(peer_id) + shared.connect_to_addr(ticket.endpoint_addr).await } /// Sends a domain message to a connected peer. @@ -719,6 +802,7 @@ impl NetworkEngine { return Ok(()); } info!("network engine shutting down"); + let _ = shared.shutdown_signal.send(true); // Close all active connections. let states: Vec = lock(&shared.peers) diff --git a/crates/federation-net/src/lib.rs b/crates/federation-net/src/lib.rs index fee3e3f..bf1c8a4 100644 --- a/crates/federation-net/src/lib.rs +++ b/crates/federation-net/src/lib.rs @@ -13,6 +13,9 @@ //! * [`NetworkId`] — isolates independent P2P networks from each other. //! * [`SchemaId`] — isolates applications with incompatible message schemas. //! * [`PeerTicket`] — a shareable string invitation used to reach a peer. +//! * [`RendezvousConfig`] — optional automatic peer discovery: peers of a +//! network find each other through the BitTorrent Mainline DHT knowing +//! nothing but the network id. //! * [`NetworkEngine`] — the engine itself; [`NetworkEventReceiver`] delivers //! [`NetworkEvent`]s to the application. //! @@ -54,6 +57,7 @@ mod error; mod event; mod identity; mod protocol; +mod rendezvous; mod ticket; mod wire; @@ -65,6 +69,7 @@ pub use engine::{Message, NetworkEngine}; pub use error::{NetworkError, Result}; pub use event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver}; pub use protocol::{ALPN, NetworkId, PROTOCOL_VERSION, SchemaId}; +pub use rendezvous::{DEFAULT_RENDEZVOUS_ENTRY_TTL, DEFAULT_RENDEZVOUS_INTERVAL, RendezvousConfig}; pub use ticket::{PeerTicket, TICKET_VERSION}; // Re-exported Iroh types that appear in the public API. diff --git a/crates/federation-net/src/rendezvous.rs b/crates/federation-net/src/rendezvous.rs new file mode 100644 index 0000000..42446ec --- /dev/null +++ b/crates/federation-net/src/rendezvous.rs @@ -0,0 +1,344 @@ +//! Automatic peer discovery ("rendezvous") over the BitTorrent Mainline DHT. +//! +//! Every peer of a network periodically publishes its own [`EndpointAddr`] +//! into a shared BEP44 mutable record whose signing key is derived from the +//! [`NetworkId`], and reads the addresses other peers published there. +//! Knowing the network id is therefore enough to find and join the network — +//! no tickets and no dedicated bootstrap servers are required. +//! +//! The record lives in the public Mainline DHT (the same network BitTorrent +//! and pkarr use), so the mechanism needs no infrastructure of its own. The +//! flip side is that the network id acts as a public rendezvous token: anyone +//! who knows it can discover and join the network. Treat the id of a private +//! network like a shared secret; the application-level handshake still +//! rejects peers whose network id does not match exactly. +//! +//! Multiple peers write the same record concurrently. Writers merge every +//! record instance they can read before publishing, so a lost update leaves +//! at most a temporarily incomplete peer list — it converges as everyone +//! republishes — and one reachable entry is enough to join, because contact +//! exchange takes over after the first connection. + +use std::collections::HashMap; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use futures::StreamExt; +use iroh::{EndpointAddr, EndpointId}; +use mainline::async_dht::AsyncDht; +use mainline::{Dht, MutableItem, SigningKey}; +use serde::{Deserialize, Serialize}; +use tracing::{debug, warn}; + +use crate::error::{NetworkError, Result}; +use crate::protocol::NetworkId; + +/// Default interval between rendezvous rounds (read + publish). +pub const DEFAULT_RENDEZVOUS_INTERVAL: Duration = Duration::from_secs(60); +/// Default time after which a published peer entry is considered stale. +pub const DEFAULT_RENDEZVOUS_ENTRY_TTL: Duration = Duration::from_secs(30 * 60); + +/// Version of the rendezvous record wire format. +const RECORD_VERSION: u16 = 1; +/// Domain separation context for deriving the record signing key. +const KEY_DERIVATION_CONTEXT: &str = "federation-net:rendezvous:v1"; +/// BEP44 caps mutable values at 1000 bytes; stay safely below. +const MAX_RECORD_BYTES: usize = 900; + +/// Configuration of the mainline-DHT rendezvous. +/// +/// Passing a `RendezvousConfig` to +/// [`crate::NetworkConfigBuilder::rendezvous`] enables automatic peer +/// discovery; by default it is disabled and peers are connected via tickets. +#[derive(Debug, Clone)] +pub struct RendezvousConfig { + /// Interval between rendezvous rounds. Each round reads the record, + /// republishes it with this peer merged in and dials newly seen peers. + pub interval: Duration, + /// Entries older than this are dropped from the record. + pub entry_ttl: Duration, + /// Overrides the DHT bootstrap nodes (`host:port` strings). `None` uses + /// the public mainline defaults; tests point this at a local testnet. + pub dht_bootstrap: Option>, +} + +impl Default for RendezvousConfig { + fn default() -> Self { + Self { + interval: DEFAULT_RENDEZVOUS_INTERVAL, + entry_ttl: DEFAULT_RENDEZVOUS_ENTRY_TTL, + dht_bootstrap: None, + } + } +} + +/// One published peer: its dialable address and when it was last refreshed. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct RendezvousEntry { + addr: EndpointAddr, + last_seen_ms: u64, +} + +/// The record stored in the mainline DHT: a bounded list of recent peers. +#[derive(Debug, Serialize, Deserialize)] +struct RendezvousRecord { + version: u16, + entries: Vec, +} + +/// Derives the shared record signing key from the network id. +/// +/// The derivation is deterministic, so every member of the network can both +/// read and update the record. +fn signing_key(network_id: NetworkId) -> SigningKey { + let seed = blake3::derive_key(KEY_DERIVATION_CONTEXT, network_id.as_bytes()); + SigningKey::from_bytes(&seed) +} + +/// Current unix time in milliseconds. +pub(crate) fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or_default() +} + +fn decode_record(bytes: &[u8]) -> Option> { + let record: RendezvousRecord = postcard::from_bytes(bytes).ok()?; + (record.version == RECORD_VERSION).then_some(record.entries) +} + +/// Merges entries from every record instance seen this round with our own, +/// deduplicating by endpoint id (newest wins) and dropping stale entries. +fn merge_entries( + seen: Vec, + self_entry: Option, + now_ms: u64, + entry_ttl: Duration, +) -> Vec { + let ttl_ms = entry_ttl.as_millis() as u64; + let mut by_id: HashMap = HashMap::new(); + for mut entry in seen.into_iter().chain(self_entry) { + // A peer with a fast clock must not pin its entry into the future. + entry.last_seen_ms = entry.last_seen_ms.min(now_ms); + if entry.last_seen_ms + ttl_ms <= now_ms { + continue; + } + match by_id.get(&entry.addr.id) { + Some(existing) if existing.last_seen_ms >= entry.last_seen_ms => {} + _ => { + by_id.insert(entry.addr.id, entry); + } + } + } + let mut entries: Vec = by_id.into_values().collect(); + entries.sort_by(|a, b| { + b.last_seen_ms + .cmp(&a.last_seen_ms) + .then_with(|| a.addr.id.cmp(&b.addr.id)) + }); + entries +} + +/// Encodes the record, dropping the oldest entries until it fits the BEP44 +/// value size limit. +fn encode_record_capped(mut entries: Vec) -> Result> { + loop { + let record = RendezvousRecord { + version: RECORD_VERSION, + entries, + }; + let encoded = postcard::to_stdvec(&record).map_err(|err| { + NetworkError::Serialization(format!("failed to encode rendezvous record: {err}")) + })?; + if encoded.len() <= MAX_RECORD_BYTES { + return Ok(encoded); + } + entries = record.entries; + if entries.pop().is_none() { + return Err(NetworkError::Serialization( + "rendezvous record does not fit even when empty".to_string(), + )); + } + } +} + +/// A client of the shared rendezvous record of one network. +#[derive(Debug)] +pub(crate) struct RendezvousClient { + dht: AsyncDht, + key: SigningKey, + entry_ttl: Duration, +} + +impl RendezvousClient { + /// Binds a mainline DHT client for the rendezvous record of `network_id`. + pub(crate) fn new(network_id: NetworkId, config: &RendezvousConfig) -> Result { + let mut builder = Dht::builder(); + if let Some(nodes) = &config.dht_bootstrap { + builder.bootstrap(nodes); + } + let dht = builder + .build() + .map_err(|err| { + NetworkError::Transport(format!("failed to start mainline DHT client: {err}")) + })? + .as_async(); + Ok(Self { + dht, + key: signing_key(network_id), + entry_ttl: config.entry_ttl, + }) + } + + /// Runs one rendezvous round: reads every reachable instance of the + /// record, merges them with our own address, republishes the result and + /// returns the addresses of the other known peers. + /// + /// `self_addr` is `None` while the local endpoint does not know any + /// dialable address yet; the round then only reads. + pub(crate) async fn round( + &self, + self_addr: Option, + now_ms: u64, + ) -> Result> { + let public_key = self.key.verifying_key().to_bytes(); + let mut items = self.dht.get_mutable(&public_key, None, None); + let mut seen = Vec::new(); + let mut max_seq = 0i64; + while let Some(item) = items.next().await { + max_seq = max_seq.max(item.seq()); + match decode_record(item.value()) { + Some(entries) => seen.extend(entries), + None => debug!("ignoring malformed rendezvous record instance"), + } + } + + let self_id = self_addr.as_ref().map(|addr| addr.id); + let self_entry = self_addr.map(|addr| RendezvousEntry { + addr, + last_seen_ms: now_ms, + }); + let publish = self_entry.is_some(); + let merged = merge_entries(seen, self_entry, now_ms, self.entry_ttl); + + if publish { + let encoded = encode_record_capped(merged.clone())?; + // Strictly newer than every instance seen this round; concurrent + // writers race, but merging on read makes lost updates benign. + let item = MutableItem::new(self.key.clone(), &encoded, max_seq + 1, None); + if let Err(err) = self.dht.put_mutable(item, None).await { + warn!(error = %err, "failed to publish the rendezvous record"); + } + } + + Ok(merged + .into_iter() + .map(|entry| entry.addr) + .filter(|addr| Some(addr.id) != self_id) + .collect()) + } +} + +#[cfg(test)] +mod tests { + use std::net::{Ipv4Addr, SocketAddr}; + + use iroh::SecretKey; + + use super::*; + + fn addr(port: u16) -> EndpointAddr { + EndpointAddr::new(SecretKey::generate().public()) + .with_ip_addr(SocketAddr::from((Ipv4Addr::LOCALHOST, port))) + } + + fn entry(addr: EndpointAddr, last_seen_ms: u64) -> RendezvousEntry { + RendezvousEntry { addr, last_seen_ms } + } + + #[test] + fn signing_key_is_deterministic_and_network_specific() { + let a = signing_key(NetworkId::from_name("net-a")); + let b = signing_key(NetworkId::from_name("net-a")); + let c = signing_key(NetworkId::from_name("net-b")); + assert_eq!(a.to_bytes(), b.to_bytes()); + assert_ne!(a.to_bytes(), c.to_bytes()); + } + + #[test] + fn record_round_trips() { + let entries = vec![entry(addr(1000), 1), entry(addr(1001), 2)]; + let encoded = encode_record_capped(entries.clone()).expect("encode"); + let decoded = decode_record(&encoded).expect("decode"); + assert_eq!(decoded.len(), entries.len()); + assert_eq!(decoded[0].addr, entries[0].addr); + assert_eq!(decoded[1].last_seen_ms, entries[1].last_seen_ms); + } + + #[test] + fn malformed_and_wrong_version_records_are_ignored() { + assert!(decode_record(b"garbage").is_none()); + let record = RendezvousRecord { + version: RECORD_VERSION + 1, + entries: vec![], + }; + let encoded = postcard::to_stdvec(&record).expect("encode"); + assert!(decode_record(&encoded).is_none()); + } + + #[test] + fn merge_deduplicates_by_id_keeping_the_newest() { + let a = addr(1000); + let older = entry(a.clone(), 100); + let newer = entry(a.clone(), 200); + let merged = merge_entries( + vec![older, newer], + None, + 1000, + Duration::from_millis(10_000), + ); + assert_eq!(merged.len(), 1); + assert_eq!(merged[0].last_seen_ms, 200); + } + + #[test] + fn merge_drops_expired_and_clamps_future_entries() { + let now = 100_000; + let ttl = Duration::from_millis(1_000); + let expired = entry(addr(1000), now - 1_000); + let fresh = entry(addr(1001), now - 500); + let future = entry(addr(1002), now + 60_000); + let merged = merge_entries(vec![expired, fresh, future], None, now, ttl); + assert_eq!(merged.len(), 2); + assert!(merged.iter().all(|e| e.last_seen_ms <= now)); + } + + #[test] + fn merge_inserts_self_and_sorts_newest_first() { + let self_entry = entry(addr(1000), 300); + let other = entry(addr(1001), 200); + let merged = merge_entries( + vec![other], + Some(self_entry.clone()), + 300, + Duration::from_millis(10_000), + ); + assert_eq!(merged.len(), 2); + assert_eq!(merged[0].addr.id, self_entry.addr.id); + } + + #[test] + fn oversized_records_are_trimmed_to_the_size_limit() { + // Far more entries than can ever fit into one BEP44 value, sorted + // newest-first as `merge_entries` produces them. + let entries: Vec = (0..100) + .map(|i| entry(addr(1000 + i), 10_000 - i as u64)) + .collect(); + let encoded = encode_record_capped(entries).expect("encode"); + assert!(encoded.len() <= MAX_RECORD_BYTES); + let decoded = decode_record(&encoded).expect("decode"); + assert!(!decoded.is_empty()); + // Trimming drops from the tail, so the newest entry must survive. + assert_eq!(decoded[0].last_seen_ms, 10_000); + } +} diff --git a/crates/federation-net/tests/integration.rs b/crates/federation-net/tests/integration.rs index 9ed0151..159c893 100644 --- a/crates/federation-net/tests/integration.rs +++ b/crates/federation-net/tests/integration.rs @@ -5,7 +5,7 @@ use std::time::Duration; use federation_net::{ ConnectionDirection, EndpointId, NetworkConfig, NetworkEngine, NetworkError, NetworkEvent, - NetworkEventReceiver, NetworkId, PeerTicket, SchemaId, + NetworkEventReceiver, NetworkId, PeerTicket, RendezvousConfig, SchemaId, }; /// Hard cap on every test so a regression can never hang CI. @@ -297,3 +297,65 @@ async fn disconnect_removes_peer() { .await .expect("test timed out"); } + +#[tokio::test] +async fn rendezvous_discovers_peers_without_tickets() { + let _net = NET_LOCK.lock().await; + tokio::time::timeout(TEST_TIMEOUT, async { + // A local mainline DHT testnet replaces the public one, keeping the + // rendezvous traffic entirely on this machine. + let testnet = mainline::Testnet::builder(8) + .build() + .expect("mainline testnet"); + let config = |dir: &Path| { + NetworkConfig::builder() + .data_dir(dir) + .network_id(NetworkId::from_name("rendezvous-test-net")) + .schema_id(SchemaId::from_name("test-schema-v1")) + .request_timeout(Duration::from_secs(10)) + .rendezvous(RendezvousConfig { + interval: Duration::from_secs(2), + dht_bootstrap: Some(testnet.bootstrap.clone()), + ..RendezvousConfig::default() + }) + .build() + .expect("valid test config") + }; + + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + let (engine_a, mut events_a): (Engine, Events) = NetworkEngine::start(config(dir_a.path())) + .await + .expect("engine a starts"); + let (engine_b, mut events_b): (Engine, Events) = NetworkEngine::start(config(dir_b.path())) + .await + .expect("engine b starts"); + + // No tickets are exchanged: both peers only know the network id and + // must find each other through the rendezvous record. Discovery + // needs a few rendezvous rounds, so the wait is more generous than + // EVENT_TIMEOUT (which the usual helpers apply per event). + async fn wait_discovered(events: &mut Events, peer: EndpointId) { + loop { + match events.recv().await.expect("event channel open") { + NetworkEvent::PeerConnected { peer_id, .. } if peer_id == peer => return, + _ => {} + } + } + } + let discovery_timeout = Duration::from_secs(90); + tokio::time::timeout(discovery_timeout, async { + wait_discovered(&mut events_a, engine_b.endpoint_id()).await; + wait_discovered(&mut events_b, engine_a.endpoint_id()).await; + }) + .await + .expect("peers did not discover each other via rendezvous"); + assert_eq!(engine_a.connected_peers(), vec![engine_b.endpoint_id()]); + assert_eq!(engine_b.connected_peers(), vec![engine_a.endpoint_id()]); + + engine_a.shutdown().await.expect("shutdown a"); + engine_b.shutdown().await.expect("shutdown b"); + }) + .await + .expect("test timed out"); +}