Implement fast userspace multihop mesh routing
Replace the one-intermediate-peer relay with protocol-scoped connectivity graphs and precomputed shortest-path/ECMP forwarding snapshots. Independent transport readers forward opaque transit frames without a plugin or TUN round trip. Carry source, destination, a bounded hop limit and stable flow tags; preserve end-to-end WireGuard links across topology changes. Classify IP flows before encryption and preserve their tags through the WireGuard pending queue. Add offline four-agent path-change coverage, loop and isolation tests, and an opt-in release forwarding microbenchmark. Bump control/data ALPNs while preserving persistent network identities and state. Also include the pending Windows Mainline idle-timeout fix and its regression test, using a reproducible vendored dependency patch. Validation: fmt, workspace Clippy with warnings denied, and 307 release tests passed. Two pre-existing Windows SQLite wipe failures were excluded; public DHT and the manual benchmark remain ignored by default. The forwarding microbenchmark measured 103 ns (64 B) and 202 ns (1280 B) per transit packet, excluding encryption and socket I/O.
This commit is contained in:
@@ -28,7 +28,7 @@
|
||||
//! system level and signed by its holder, and that is what is compared
|
||||
//! against.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::net::Ipv4Addr;
|
||||
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex, RwLock};
|
||||
@@ -39,6 +39,7 @@ use bytes::Bytes;
|
||||
use iroh::EndpointId;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use tsunagi::dataplane::routing::{FlowId, flow::ip_flow};
|
||||
use tsunagi::dataplane::transport::{SharedLink, TransportError};
|
||||
use tsunagi::dataplane::{PacketSink, PluginError};
|
||||
use tsunagi::identity::NetworkId;
|
||||
@@ -54,6 +55,66 @@ const TIMER_INTERVAL: Duration = Duration::from_millis(250);
|
||||
/// Scratch space for one encapsulate or decapsulate call.
|
||||
const SCRATCH: usize = 4096;
|
||||
|
||||
/// Mirrors boringtun 0.7's bounded pending-packet FIFO with opaque flow tags.
|
||||
/// All three operations share the same mutex as the cryptographic state.
|
||||
struct FlowTunnel {
|
||||
tunn: Tunn,
|
||||
queued: VecDeque<FlowId>,
|
||||
}
|
||||
|
||||
fn is_data(result: &TunnResult<'_>) -> bool {
|
||||
matches!(result, TunnResult::WriteToNetwork(bytes) if bytes.starts_with(&[4, 0, 0, 0]))
|
||||
}
|
||||
|
||||
impl FlowTunnel {
|
||||
fn encapsulate<'a>(
|
||||
&mut self,
|
||||
packet: &[u8],
|
||||
scratch: &'a mut [u8],
|
||||
flow: FlowId,
|
||||
) -> (TunnResult<'a>, FlowId) {
|
||||
let result = self.tunn.encapsulate(packet, scratch);
|
||||
if is_data(&result) {
|
||||
return (result, flow);
|
||||
}
|
||||
// MAX_QUEUE_DEPTH in boringtun 0.7: the newest packet is dropped at 256.
|
||||
if self.queued.len() < 256 {
|
||||
self.queued.push_back(flow);
|
||||
}
|
||||
(result, 0)
|
||||
}
|
||||
|
||||
fn decapsulate<'a>(
|
||||
&mut self,
|
||||
packet: &[u8],
|
||||
scratch: &'a mut [u8],
|
||||
) -> (TunnResult<'a>, FlowId) {
|
||||
let result = self.tunn.decapsulate(None, packet, scratch);
|
||||
let mut flow = 0;
|
||||
if packet.is_empty() {
|
||||
if is_data(&result) {
|
||||
flow = self.queued.pop_front().unwrap_or(0);
|
||||
} else if let Some(pending) = self.queued.pop_front() {
|
||||
// Without a session, draining pops the oldest IP packet and
|
||||
// encapsulate queues it again at the back.
|
||||
self.queued.push_back(pending);
|
||||
}
|
||||
}
|
||||
(result, flow)
|
||||
}
|
||||
|
||||
fn update_timers<'a>(&mut self, scratch: &'a mut [u8]) -> TunnResult<'a> {
|
||||
let result = self.tunn.update_timers(scratch);
|
||||
if matches!(
|
||||
result,
|
||||
TunnResult::Err(boringtun::noise::errors::WireGuardError::ConnectionExpired)
|
||||
) {
|
||||
self.queued.clear();
|
||||
}
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
/// Counters for one peer's tunnel.
|
||||
#[derive(Debug, Default)]
|
||||
struct PeerCounters {
|
||||
@@ -120,7 +181,7 @@ struct Peer {
|
||||
/// allocates it, the peer signs the claim, and every protocol carries
|
||||
/// traffic for the same address.
|
||||
overlay_v4: Mutex<Option<Ipv4Addr>>,
|
||||
tunn: Mutex<Tunn>,
|
||||
tunn: Mutex<FlowTunnel>,
|
||||
link: SharedLink,
|
||||
counters: Arc<PeerCounters>,
|
||||
task: Mutex<Option<JoinHandle<()>>>,
|
||||
@@ -165,7 +226,7 @@ impl Peer {
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
};
|
||||
PeerHealth {
|
||||
since_handshake: guard.time_since_last_handshake(),
|
||||
since_handshake: guard.tunn.time_since_last_handshake(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -254,32 +315,34 @@ impl WireguardDevice {
|
||||
return false;
|
||||
};
|
||||
|
||||
let mut scratch = vec![0u8; SCRATCH];
|
||||
let mut scratch = [0u8; SCRATCH];
|
||||
let flow = ip_flow(packet);
|
||||
// The encryption is the whole of what this protocol contributes, so
|
||||
// it happens here rather than anywhere the packet passes through.
|
||||
// The lock is released before the send: a slow link must not hold up
|
||||
// the tunnel's timers.
|
||||
let len = {
|
||||
let result = {
|
||||
let mut tunn = match peer.tunn.lock() {
|
||||
Ok(guard) => guard,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
};
|
||||
match tunn.encapsulate(packet, &mut scratch) {
|
||||
TunnResult::WriteToNetwork(out) => Some(out.len()),
|
||||
let (result, selected_flow) = tunn.encapsulate(packet, &mut scratch, flow);
|
||||
match result {
|
||||
TunnResult::WriteToNetwork(out) => Some((out.len(), selected_flow)),
|
||||
// No session yet, so nothing to send. Counted as dropped
|
||||
// rather than reported: the handshake is in flight and the
|
||||
// next packet will go.
|
||||
_ => None,
|
||||
}
|
||||
};
|
||||
let Some(len) = len else {
|
||||
let Some((len, selected_flow)) = result else {
|
||||
peer.counters
|
||||
.dropped_no_session
|
||||
.fetch_add(1, Ordering::Relaxed);
|
||||
return false;
|
||||
};
|
||||
|
||||
send_to_peer(&peer, &scratch[..len]);
|
||||
send_to_peer_flow(&peer, &scratch[..len], selected_flow);
|
||||
peer.counters.tx_packets.fetch_add(1, Ordering::Relaxed);
|
||||
peer.counters
|
||||
.tx_bytes
|
||||
@@ -320,7 +383,10 @@ impl WireguardDevice {
|
||||
endpoint_id,
|
||||
public_key,
|
||||
overlay_v4: Mutex::new(overlay_v4),
|
||||
tunn: Mutex::new(tunn),
|
||||
tunn: Mutex::new(FlowTunnel {
|
||||
tunn,
|
||||
queued: VecDeque::new(),
|
||||
}),
|
||||
link,
|
||||
counters: Arc::new(PeerCounters::default()),
|
||||
task: Mutex::new(None),
|
||||
@@ -412,9 +478,7 @@ fn write_lock<T>(lock: &RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
|
||||
|
||||
/// Asks boringtun for a handshake initiation and sends it.
|
||||
///
|
||||
/// Encapsulating an empty packet is how the protocol state machine is told
|
||||
/// "there is something to say"; with no session yet it answers with the
|
||||
/// handshake initiation.
|
||||
/// This does not enqueue an empty IP packet in boringtun's pending queue.
|
||||
fn kick_handshake(peer: &Peer) {
|
||||
let mut scratch = vec![0u8; SCRATCH];
|
||||
let len = {
|
||||
@@ -422,7 +486,7 @@ fn kick_handshake(peer: &Peer) {
|
||||
Ok(guard) => guard,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
};
|
||||
match tunn.encapsulate(&[], &mut scratch) {
|
||||
match tunn.tunn.format_handshake_initiation(&mut scratch, false) {
|
||||
TunnResult::WriteToNetwork(out) => Some(out.len()),
|
||||
_ => None,
|
||||
}
|
||||
@@ -434,7 +498,11 @@ fn kick_handshake(peer: &Peer) {
|
||||
|
||||
/// Sends whatever boringtun produced, without holding the tunnel lock.
|
||||
fn send_to_peer(peer: &Peer, payload: &[u8]) {
|
||||
match peer.link.send(Bytes::copy_from_slice(payload)) {
|
||||
send_to_peer_flow(peer, payload, 0);
|
||||
}
|
||||
|
||||
fn send_to_peer_flow(peer: &Peer, payload: &[u8], flow: FlowId) {
|
||||
match peer.link.send_flow(Bytes::copy_from_slice(payload), flow) {
|
||||
Ok(()) => {}
|
||||
Err(TransportError::TooLarge { .. }) => {
|
||||
peer.counters
|
||||
@@ -449,12 +517,12 @@ fn send_to_peer(peer: &Peer, payload: &[u8]) {
|
||||
}
|
||||
|
||||
async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
|
||||
let mut scratch = vec![0u8; SCRATCH];
|
||||
loop {
|
||||
let Some(datagram) = peer.link.recv().await else {
|
||||
return;
|
||||
};
|
||||
|
||||
let mut scratch = vec![0u8; SCRATCH];
|
||||
// boringtun may need several passes: a handshake reply first, then
|
||||
// any packets that were queued while the session was coming up.
|
||||
let mut input: Option<&[u8]> = Some(&datagram);
|
||||
@@ -464,8 +532,9 @@ async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
|
||||
Ok(guard) => guard,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
};
|
||||
match tunn.decapsulate(None, input.unwrap_or(&[]), &mut scratch) {
|
||||
TunnResult::WriteToNetwork(out) => Outcome::ToNetwork(out.len()),
|
||||
let (result, flow) = tunn.decapsulate(input.unwrap_or(&[]), &mut scratch);
|
||||
match result {
|
||||
TunnResult::WriteToNetwork(out) => Outcome::ToNetwork(out.len(), flow),
|
||||
// The source boringtun reports is not consulted here:
|
||||
// whether the peer may use it is checked where the
|
||||
// claims live.
|
||||
@@ -480,8 +549,8 @@ async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
|
||||
};
|
||||
|
||||
match outcome {
|
||||
Outcome::ToNetwork(len) => {
|
||||
send_to_peer(&peer, &scratch[..len]);
|
||||
Outcome::ToNetwork(len, flow) => {
|
||||
send_to_peer_flow(&peer, &scratch[..len], flow);
|
||||
// Keep draining with an empty datagram, as boringtun asks.
|
||||
input = None;
|
||||
continue;
|
||||
@@ -515,7 +584,7 @@ async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
|
||||
}
|
||||
|
||||
enum Outcome {
|
||||
ToNetwork(usize),
|
||||
ToNetwork(usize, FlowId),
|
||||
ToTunnel(usize),
|
||||
Done,
|
||||
Failed,
|
||||
@@ -550,3 +619,73 @@ async fn drive_timers(inner: Arc<Inner>) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod flow_tests {
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
use super::*;
|
||||
|
||||
fn tunnel(ours: &WgSecretKey, theirs: &WgSecretKey) -> FlowTunnel {
|
||||
FlowTunnel {
|
||||
tunn: Tunn::new(
|
||||
ours.to_static_secret(),
|
||||
theirs.public().into_x25519(),
|
||||
None,
|
||||
None,
|
||||
1,
|
||||
None,
|
||||
),
|
||||
queued: VecDeque::new(),
|
||||
}
|
||||
}
|
||||
fn wire(result: TunnResult<'_>) -> Vec<u8> {
|
||||
match result {
|
||||
TunnResult::WriteToNetwork(bytes) => bytes.to_vec(),
|
||||
_ => panic!("expected WireGuard frame"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flow_tags_survive_pending_queue_overflow_handshake_and_encryption() {
|
||||
let ak = WgSecretKey::from_bytes(&[1; 32]);
|
||||
let bk = WgSecretKey::from_bytes(&[2; 32]);
|
||||
let mut a = tunnel(&ak, &bk);
|
||||
let mut b = tunnel(&bk, &ak);
|
||||
let mut scratch = [0u8; SCRATCH];
|
||||
let mut packet = [0u8; 40];
|
||||
packet[0] = 0x45;
|
||||
packet[2..4].copy_from_slice(&40u16.to_be_bytes());
|
||||
packet[9] = 6;
|
||||
packet[12..20].copy_from_slice(&[10, 0, 0, 1, 10, 0, 0, 2]);
|
||||
let (result, flow) = a.encapsulate(&packet, &mut scratch, 100);
|
||||
assert_eq!(
|
||||
flow, 0,
|
||||
"handshake does not masquerade as application traffic"
|
||||
);
|
||||
let hello = wire(result);
|
||||
for flow in 101..400 {
|
||||
let _ = a.encapsulate(&packet, &mut scratch, flow);
|
||||
}
|
||||
assert_eq!(a.queued.len(), 256);
|
||||
let response = wire(b.decapsulate(&hello, &mut scratch).0);
|
||||
let keepalive = wire(a.decapsulate(&response, &mut scratch).0);
|
||||
let _ = b.decapsulate(&keepalive, &mut scratch);
|
||||
for expected in 100..356 {
|
||||
let (result, flow) = a.decapsulate(&[], &mut scratch);
|
||||
assert_eq!(flow, expected);
|
||||
let ciphertext = wire(result);
|
||||
match b.decapsulate(&ciphertext, &mut scratch).0 {
|
||||
TunnResult::WriteToTunnelV4(bytes, _) => assert_eq!(bytes, packet),
|
||||
_ => panic!("queued IP packet must decrypt"),
|
||||
}
|
||||
}
|
||||
assert!(a.queued.is_empty());
|
||||
assert!(matches!(
|
||||
a.decapsulate(&[], &mut scratch).0,
|
||||
TunnResult::Done
|
||||
));
|
||||
let (result, flow) = a.encapsulate(&packet, &mut scratch, 12345);
|
||||
assert!(is_data(&result));
|
||||
assert_eq!(flow, 12345);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1510,3 +1510,120 @@ async fn a_peer_with_no_direct_path_is_reached_through_one_that_has_both() {
|
||||
b.shutdown().await;
|
||||
middle.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_chain_routes_through_two_transit_peers_without_touching_their_tuns() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("wg-multihop");
|
||||
let mut agents = Vec::new();
|
||||
let mut blocked = Vec::new();
|
||||
for index in 0..4 {
|
||||
let cuts = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new()));
|
||||
agents.push(WgAgent::spawn_cut_off(&discovery, &format!("mh{index}"), cuts.clone()).await);
|
||||
blocked.push(cuts);
|
||||
}
|
||||
// The authenticated control plane stays connected. The only available
|
||||
// data links are A--B--C--D; A and D need two transit routers.
|
||||
for (i, cuts) in blocked.iter().enumerate() {
|
||||
for (j, agent) in agents.iter().enumerate() {
|
||||
if i.abs_diff(j) > 1 {
|
||||
cuts.lock().unwrap().insert(agent.endpoint_id());
|
||||
}
|
||||
}
|
||||
}
|
||||
let mut id = None;
|
||||
for agent in &agents {
|
||||
id = Some(agent.agent.join_network(&name, &secret).await.unwrap());
|
||||
}
|
||||
let id = id.unwrap();
|
||||
for agent in &agents {
|
||||
wait_for_peers(&agent.agent, id, 3).await;
|
||||
agent.wait_for_tunnels(id, 3).await;
|
||||
}
|
||||
let a = &agents[0];
|
||||
let d = &agents[3];
|
||||
let a_addr = a.overlay(id).await;
|
||||
let d_addr = d.overlay(id).await;
|
||||
let a_tun = a.tun(id).await;
|
||||
let d_tun = d.tun(id).await;
|
||||
for (from, to, source, destination) in [
|
||||
(&a_tun, &d_tun, a_addr, d_addr),
|
||||
(&d_tun, &a_tun, d_addr, a_addr),
|
||||
] {
|
||||
let packet = tcp_packet(source, destination, 1280);
|
||||
from.push_from_os(packet.clone());
|
||||
let received = tokio::time::timeout(tsunagi::testing::DEADLINE, to.pop_to_os())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(received, packet);
|
||||
}
|
||||
for middle in &agents[1..3] {
|
||||
assert!(
|
||||
middle
|
||||
.agent
|
||||
.network_status(id)
|
||||
.await
|
||||
.unwrap()
|
||||
.relay
|
||||
.forwarded
|
||||
> 0
|
||||
);
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(100), middle.tun(id).await.pop_to_os())
|
||||
.await
|
||||
.is_err(),
|
||||
"transit ciphertext must never enter the intermediate host's TUN"
|
||||
);
|
||||
}
|
||||
// A direct shortcut wins immediately; losing it restores the chain using
|
||||
// the same end-to-end tunnels and overlay addresses.
|
||||
for allow_direct in [true, false] {
|
||||
for (from, to) in [(0, 3), (3, 0)] {
|
||||
if allow_direct {
|
||||
blocked[from]
|
||||
.lock()
|
||||
.unwrap()
|
||||
.remove(&agents[to].endpoint_id());
|
||||
} else {
|
||||
blocked[from]
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(agents[to].endpoint_id());
|
||||
}
|
||||
agents[from].agent.recheck_network(id).await.unwrap();
|
||||
}
|
||||
for (from, to) in [(a, d), (d, a)] {
|
||||
wait_until("route follows the changed topology", || async {
|
||||
let view = from.plugin.overview(id)?;
|
||||
let tunnel = view
|
||||
.peers
|
||||
.iter()
|
||||
.find(|peer| peer.endpoint_id == to.endpoint_id())?
|
||||
.tunnel
|
||||
.as_ref()?;
|
||||
let relayed = tunnel.path.contains("relay 3 hops");
|
||||
(tunnel.health.is_up()
|
||||
&& if allow_direct {
|
||||
!tunnel.path.contains("relay") && !tunnel.path.contains("unreachable")
|
||||
} else {
|
||||
relayed
|
||||
})
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
}
|
||||
let packet = tcp_packet(a_addr, d_addr, 1280);
|
||||
a_tun.push_from_os(packet.clone());
|
||||
assert_eq!(
|
||||
tokio::time::timeout(tsunagi::testing::DEADLINE, d_tun.pop_to_os())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap(),
|
||||
packet
|
||||
);
|
||||
}
|
||||
for agent in agents {
|
||||
agent.shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,6 +33,7 @@ dns-publish = ["dep:zbus"]
|
||||
testing = ["dep:tempfile", "dep:tracing-subscriber"]
|
||||
|
||||
[dependencies]
|
||||
arc-swap = "1.9"
|
||||
iroh.workspace = true
|
||||
# `net` for the local control interface: a Unix socket on Unix, a named pipe
|
||||
# on Windows. Both live behind the same `net` feature.
|
||||
|
||||
@@ -264,10 +264,16 @@ struct Runtime {
|
||||
/// else, so there is no hearsay to weigh and nothing to poison. Soft
|
||||
/// state — an entry that stops being refreshed expires, which is how a
|
||||
/// relay that has gone stops being chosen without anybody revoking it.
|
||||
reachable: HashMap<EndpointId, (HashSet<EndpointId>, std::time::Instant)>,
|
||||
reachable: HashMap<
|
||||
EndpointId,
|
||||
(
|
||||
HashSet<crate::proto::message::ReachableLink>,
|
||||
std::time::Instant,
|
||||
),
|
||||
>,
|
||||
/// What this agent last told its peers it could reach, so it is only
|
||||
/// said again when it changed.
|
||||
announced_reach: HashSet<EndpointId>,
|
||||
announced_reach: HashSet<crate::proto::message::ReachableLink>,
|
||||
/// Links currently being opened, so we do not start two.
|
||||
opening: HashSet<(EndpointId, String)>,
|
||||
link_results_tx: mpsc::Sender<LinkOutcome>,
|
||||
@@ -328,7 +334,7 @@ impl Runtime {
|
||||
dial_results_tx,
|
||||
dial_results_rx,
|
||||
links: HashMap::new(),
|
||||
hub: crate::dataplane::relay::RelayHub::new(network_id),
|
||||
hub: crate::dataplane::relay::RelayHub::new(network_id, local_id),
|
||||
reachable: HashMap::new(),
|
||||
announced_reach: HashSet::new(),
|
||||
opening: HashSet::new(),
|
||||
@@ -360,6 +366,11 @@ impl Runtime {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = self.shutdown.wait() => break,
|
||||
_ = self.hub.changed() => {
|
||||
self.ensure_links();
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
}
|
||||
command = commands.recv() => match command {
|
||||
Some(command) => self.handle_command(command).await,
|
||||
None => break,
|
||||
@@ -387,6 +398,8 @@ impl Runtime {
|
||||
_ = ticker.tick() => {
|
||||
self.discovery_round().await;
|
||||
self.ensure_links();
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -401,6 +414,7 @@ impl Runtime {
|
||||
// Stop accepting session events first: nothing is going to act on them
|
||||
// any more, and a sender blocked on a full queue would stall shutdown.
|
||||
self.session_events_rx.close();
|
||||
self.hub.close();
|
||||
self.links.clear();
|
||||
let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
|
||||
for peer in peers {
|
||||
@@ -1122,7 +1136,7 @@ impl Runtime {
|
||||
|
||||
let mut dead: Vec<(EndpointId, String)> = Vec::new();
|
||||
for (key, link) in &self.links {
|
||||
if link.is_closed() {
|
||||
if link.is_closed() || self.no_direct_data(key.0) {
|
||||
dead.push(key.clone());
|
||||
}
|
||||
}
|
||||
@@ -1307,28 +1321,28 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// Tells peers which peers this agent has a live link with, when that
|
||||
/// has changed.
|
||||
///
|
||||
/// Only what it can see for itself, and only to the peers it is
|
||||
/// talking to: one hop, no flooding, nothing second-hand. A peer that
|
||||
/// hears this is the only one that can use it, because it is also the
|
||||
/// only one that could route through this agent.
|
||||
/// Publishes only locally observed, protocol-specific transport edges.
|
||||
fn announce_reach(&mut self, force: bool) {
|
||||
let live: HashSet<EndpointId> = self
|
||||
let live: HashSet<_> = self
|
||||
.links
|
||||
.iter()
|
||||
.filter(|(_, link)| !link.is_closed())
|
||||
.map(|((peer, _), _)| *peer)
|
||||
.map(
|
||||
|((peer, protocol), _)| crate::proto::message::ReachableLink {
|
||||
peer: *peer.as_bytes(),
|
||||
protocol: protocol.clone(),
|
||||
},
|
||||
)
|
||||
.collect();
|
||||
if !force && live == self.announced_reach {
|
||||
return;
|
||||
}
|
||||
self.announced_reach = live.clone();
|
||||
let message = ControlMessage::Reachable {
|
||||
peers: live.iter().map(|peer| *peer.as_bytes()).collect(),
|
||||
};
|
||||
let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
|
||||
let mut links: Vec<_> = live.into_iter().collect();
|
||||
links.sort_by(|a, b| (&a.protocol, a.peer).cmp(&(&b.protocol, b.peer)));
|
||||
links.truncate(self.params.limits.max_state_records);
|
||||
let message = ControlMessage::Reachable { links };
|
||||
let peers: Vec<_> = self.sessions.keys().copied().collect();
|
||||
for peer in peers {
|
||||
if let Err(err) = self.send_to(peer, message.clone()) {
|
||||
tracing::debug!(%err, "could not queue a reachability announcement");
|
||||
@@ -1336,99 +1350,72 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// Chooses, for every peer with no direct link, somebody to go through.
|
||||
///
|
||||
/// Local and deterministic: the lowest endpoint id among the peers
|
||||
/// that both this agent has a link with and that say they have a link
|
||||
/// with the destination. Nothing is agreed with anybody — two agents
|
||||
/// may well pick differently, and traffic each way finds its own path.
|
||||
/// Builds the graph off the packet path. Every remote row comes from that
|
||||
/// member's authenticated control session; stale or incompatible rows never
|
||||
/// enter the graph. The hub replaces an immutable table only on changes.
|
||||
fn update_paths(&mut self) {
|
||||
self.reachable
|
||||
.retain(|_, (_, heard)| heard.elapsed() < REACH_EXPIRY);
|
||||
|
||||
let served = self.served_protocols();
|
||||
let live: HashSet<EndpointId> = self
|
||||
.links
|
||||
.iter()
|
||||
.filter(|(_, link)| !link.is_closed())
|
||||
.map(|((peer, _), _)| *peer)
|
||||
.collect();
|
||||
|
||||
// Every peer this agent would carry traffic with: a session says
|
||||
// who it is and what it speaks, which is also what a protocol
|
||||
// needs before it can have a link at all.
|
||||
let wanted: Vec<(EndpointId, String)> = self
|
||||
.sessions
|
||||
.values()
|
||||
.flat_map(|session| {
|
||||
let peer = session.peer;
|
||||
session
|
||||
.capabilities
|
||||
.iter()
|
||||
.filter(|capability| capability.enabled)
|
||||
.map(move |capability| (peer, capability.protocol.clone(), capability.version))
|
||||
})
|
||||
.filter(|(_, protocol, version)| {
|
||||
served
|
||||
.iter()
|
||||
.any(|(name, ours)| name == protocol && ours == version)
|
||||
})
|
||||
.map(|(peer, protocol, _)| (peer, protocol))
|
||||
.collect();
|
||||
|
||||
for (peer, protocol) in wanted {
|
||||
if live.contains(&peer) {
|
||||
// There is a direct link; a hop is only for when there is
|
||||
// not, and holding a stale one would keep a peer that has
|
||||
// gone looking reachable.
|
||||
if self.hub.has_link(peer, &protocol) {
|
||||
self.hub.set_hop(peer, &protocol, None);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
let hop = self
|
||||
self.reachable.retain(|peer, (_, heard)| {
|
||||
heard.elapsed() < REACH_EXPIRY && self.sessions.contains_key(peer)
|
||||
});
|
||||
for (protocol, version) in self.served_protocols() {
|
||||
let peers: Vec<_> = self
|
||||
.sessions
|
||||
.values()
|
||||
.filter(|session| {
|
||||
session.capabilities.iter().any(|capability| {
|
||||
capability.enabled
|
||||
&& capability.protocol == protocol
|
||||
&& capability.version == version
|
||||
})
|
||||
})
|
||||
.map(|session| session.peer)
|
||||
.collect();
|
||||
let mut members: HashSet<_> = peers.iter().map(|peer| *peer.as_bytes()).collect();
|
||||
members.insert(*self.local_id.as_bytes());
|
||||
let graph = self
|
||||
.reachable
|
||||
.iter()
|
||||
.filter(|(middle, (reaches, _))| {
|
||||
live.contains(*middle) && reaches.contains(&peer) && **middle != peer
|
||||
.filter(|(peer, _)| members.contains(peer.as_bytes()))
|
||||
.map(|(peer, (links, _))| {
|
||||
let mut neighbors: Vec<_> = links
|
||||
.iter()
|
||||
.filter(|link| {
|
||||
link.protocol == protocol
|
||||
&& members.contains(&link.peer)
|
||||
&& link.peer != *peer.as_bytes()
|
||||
})
|
||||
.map(|link| link.peer)
|
||||
.collect();
|
||||
neighbors.sort_unstable();
|
||||
(*peer.as_bytes(), neighbors)
|
||||
})
|
||||
.map(|(middle, _)| *middle)
|
||||
.min_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
|
||||
|
||||
let known = self.hub.has_link(peer, &protocol);
|
||||
if !known && hop.is_none() {
|
||||
// Nothing to reach it with and nothing to offer a
|
||||
// protocol; inventing a link object here would only look
|
||||
// like connectivity.
|
||||
continue;
|
||||
.collect();
|
||||
self.hub.set_topology(&protocol, graph, members);
|
||||
for peer in peers {
|
||||
if self.hub.has_link(peer, &protocol) || !self.hub.reachable(peer, &protocol) {
|
||||
continue;
|
||||
}
|
||||
let Some(plugin) = self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.find(|plugin| plugin.protocol_id() == protocol)
|
||||
.cloned()
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let link = self.hub.link(peer, &protocol);
|
||||
let path = link.path_description();
|
||||
let max_datagram = link.max_datagram_size();
|
||||
plugin.on_peer_link(self.network_id, peer, link);
|
||||
self.emit(Event::DataLinkUp {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
protocol: protocol.clone(),
|
||||
path,
|
||||
max_datagram,
|
||||
});
|
||||
}
|
||||
self.hub.set_hop(peer, &protocol, hop);
|
||||
if known {
|
||||
continue;
|
||||
}
|
||||
// A peer reachable only through somebody still gets a link, so
|
||||
// its tunnel can be built: the protocol is not told how its
|
||||
// datagrams travel, only that this is the way to that peer.
|
||||
let Some(plugin) = self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.find(|plugin| plugin.protocol_id() == protocol)
|
||||
.cloned()
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let link = self.hub.link(peer, &protocol);
|
||||
let path = link.path_description();
|
||||
let max_datagram = link.max_datagram_size();
|
||||
plugin.on_peer_link(self.network_id, peer, link);
|
||||
self.emit(Event::DataLinkUp {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
protocol,
|
||||
path,
|
||||
max_datagram,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1718,6 +1705,8 @@ impl Runtime {
|
||||
}
|
||||
self.metrics.disconnects += 1;
|
||||
self.drop_links_for(peer);
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
for plugin in &self.params.plugins {
|
||||
plugin.on_peer_gone(self.network_id, peer);
|
||||
}
|
||||
@@ -1748,6 +1737,7 @@ impl Runtime {
|
||||
}
|
||||
self.dispatch_capabilities(peer, &capabilities);
|
||||
self.ensure_links();
|
||||
self.update_paths();
|
||||
}
|
||||
ControlMessage::Ping { seq, payload } => {
|
||||
let pong = ControlMessage::Pong {
|
||||
@@ -1772,11 +1762,8 @@ impl Runtime {
|
||||
}
|
||||
// What that peer can reach, from that peer. Kept with the time
|
||||
// it was heard, so it can go stale on its own.
|
||||
ControlMessage::Reachable { peers } => {
|
||||
let heard: HashSet<EndpointId> = peers
|
||||
.iter()
|
||||
.filter_map(|raw| EndpointId::from_bytes(raw).ok())
|
||||
.collect();
|
||||
ControlMessage::Reachable { links } => {
|
||||
let heard = links.iter().cloned().collect();
|
||||
self.reachable
|
||||
.insert(peer, (heard, std::time::Instant::now()));
|
||||
self.update_paths();
|
||||
|
||||
@@ -480,3 +480,5 @@ impl std::fmt::Debug for AgentConfig {
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
/// Maximum number of transport links a routed datagram may traverse.
|
||||
pub const ROUTING_HOP_LIMIT: u8 = 16;
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
//! recorded and surfaced, the control plane keeps running.
|
||||
|
||||
pub mod relay;
|
||||
pub mod routing;
|
||||
pub mod transport;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,287 @@
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
use super::*;
|
||||
use crate::identity::{NetworkKeys, NetworkName, NetworkSecret};
|
||||
use std::sync::atomic::AtomicBool;
|
||||
|
||||
fn network() -> NetworkId {
|
||||
NetworkKeys::derive(
|
||||
&NetworkName::new("routing-tests").unwrap(),
|
||||
&NetworkSecret::from_bytes([3; 32]).unwrap(),
|
||||
)
|
||||
.network_id()
|
||||
}
|
||||
fn peer(seed: u8) -> EndpointId {
|
||||
iroh::SecretKey::from_bytes(&[seed; 32]).public()
|
||||
}
|
||||
fn id(seed: u8) -> PeerId {
|
||||
*peer(seed).as_bytes()
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Wire {
|
||||
peer: EndpointId,
|
||||
sent: Mutex<Vec<Bytes>>,
|
||||
feed: mpsc::Sender<Bytes>,
|
||||
inbound: tokio::sync::Mutex<mpsc::Receiver<Bytes>>,
|
||||
dead: AtomicBool,
|
||||
count: AtomicU64,
|
||||
record: bool,
|
||||
}
|
||||
impl Wire {
|
||||
fn new(seed: u8, record: bool) -> Arc<Self> {
|
||||
let (feed, inbound) = mpsc::channel(32);
|
||||
Arc::new(Self {
|
||||
peer: peer(seed),
|
||||
sent: Mutex::default(),
|
||||
feed,
|
||||
inbound: tokio::sync::Mutex::new(inbound),
|
||||
dead: AtomicBool::new(false),
|
||||
count: AtomicU64::new(0),
|
||||
record,
|
||||
})
|
||||
}
|
||||
fn take(&self) -> Vec<Bytes> {
|
||||
std::mem::take(&mut self.sent.lock().unwrap())
|
||||
}
|
||||
}
|
||||
impl PacketLink for Wire {
|
||||
fn network(&self) -> NetworkId {
|
||||
network()
|
||||
}
|
||||
fn peer(&self) -> EndpointId {
|
||||
self.peer
|
||||
}
|
||||
fn max_datagram_size(&self) -> usize {
|
||||
MAX_DATA_DATAGRAM
|
||||
}
|
||||
fn send(&self, frame: Bytes) -> Result<(), TransportError> {
|
||||
if self.is_closed() {
|
||||
return Err(TransportError::Closed);
|
||||
}
|
||||
self.count.fetch_add(1, Ordering::Relaxed);
|
||||
if self.record {
|
||||
self.sent.lock().unwrap().push(frame);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn recv(&self) -> BoxFuture<'_, Option<Bytes>> {
|
||||
Box::pin(async move { self.inbound.lock().await.recv().await })
|
||||
}
|
||||
fn closed(&self) -> BoxFuture<'_, ()> {
|
||||
Box::pin(std::future::pending())
|
||||
}
|
||||
fn is_closed(&self) -> bool {
|
||||
self.dead.load(Ordering::Relaxed)
|
||||
}
|
||||
fn path_description(&self) -> String {
|
||||
"test-direct".into()
|
||||
}
|
||||
}
|
||||
|
||||
fn topology(hub: &RelayHub, protocol: &str, edges: &[(u8, &[u8])], members: &[u8]) {
|
||||
hub.set_topology(
|
||||
protocol,
|
||||
edges
|
||||
.iter()
|
||||
.map(|(a, bs)| (id(*a), bs.iter().map(|b| id(*b)).collect()))
|
||||
.collect(),
|
||||
members.iter().map(|p| id(*p)).collect(),
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ecmp_is_per_flow_direct_wins_and_existing_tunnel_survives_changes() {
|
||||
let hub = RelayHub::new(network(), peer(1));
|
||||
let b = Wire::new(2, true);
|
||||
let c = Wire::new(3, true);
|
||||
hub.set_direct(peer(2), "ip", b.clone());
|
||||
hub.set_direct(peer(3), "ip", c.clone());
|
||||
topology(&hub, "ip", &[(2, &[4]), (3, &[4])], &[1, 2, 3, 4]);
|
||||
let link = hub.link(peer(4), "ip");
|
||||
let mut chosen = HashMap::new();
|
||||
for _ in 0..8 {
|
||||
for flow in 0..100 {
|
||||
link.send_flow(Bytes::from_static(b"ciphertext"), flow)
|
||||
.unwrap();
|
||||
}
|
||||
for (hop, wire) in [(2, &b), (3, &c)] {
|
||||
for frame in wire.take() {
|
||||
let header = envelope::decode(&frame).unwrap();
|
||||
assert_eq!(header.source, id(1));
|
||||
assert_eq!(header.destination, id(4));
|
||||
if let Some(old) = chosen.insert(header.flow, hop) {
|
||||
assert_eq!(old, hop);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
chosen.values().copied().collect::<HashSet<_>>(),
|
||||
HashSet::from([2, 3])
|
||||
);
|
||||
let direct = Wire::new(4, true);
|
||||
hub.set_direct(peer(4), "ip", direct.clone());
|
||||
link.send_flow(Bytes::from_static(b"direct"), 8).unwrap();
|
||||
assert_eq!(direct.take().len(), 1);
|
||||
assert!(b.take().is_empty() && c.take().is_empty());
|
||||
hub.clear_direct(peer(4), "ip");
|
||||
b.dead.store(true, Ordering::Relaxed);
|
||||
for flow in 0..100 {
|
||||
link.send_flow(Bytes::from_static(b"fallback"), flow)
|
||||
.unwrap();
|
||||
}
|
||||
assert_eq!(c.take().len(), 100);
|
||||
hub.clear_direct(peer(2), "ip");
|
||||
topology(&hub, "ip", &[(3, &[])], &[1, 2, 3, 4]);
|
||||
assert!(link.send(Bytes::new()).is_err());
|
||||
assert!(!link.is_closed());
|
||||
assert!(Arc::ptr_eq(&link, &hub.link(peer(4), "ip")));
|
||||
hub.remove_peer(peer(4));
|
||||
assert!(link.is_closed());
|
||||
tokio::time::timeout(std::time::Duration::from_secs(1), link.closed())
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(link.recv().await.is_none());
|
||||
hub.close();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn transport_reader_forwards_without_any_plugin_reader_and_isolates_protocols() {
|
||||
let hub = RelayHub::new(network(), peer(2));
|
||||
let a = Wire::new(1, true);
|
||||
let c = Wire::new(3, true);
|
||||
hub.set_direct(peer(1), "ip", a.clone());
|
||||
hub.set_direct(peer(3), "ip", c.clone());
|
||||
topology(&hub, "ip", &[(3, &[4])], &[1, 2, 3, 4]);
|
||||
let frame = envelope::encode(id(1), id(4), 91, b"end-to-end encrypted");
|
||||
let pointer = frame.as_ptr();
|
||||
a.feed.send(frame).await.unwrap();
|
||||
tokio::time::timeout(std::time::Duration::from_secs(1), async {
|
||||
while hub.counters().forwarded == 0 {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let sent = c.take();
|
||||
assert_eq!(sent[0].as_ptr(), pointer);
|
||||
assert_eq!(
|
||||
envelope::decode(&sent[0]).unwrap().remaining,
|
||||
ROUTING_HOP_LIMIT - 1
|
||||
);
|
||||
assert_eq!(&sent[0][RELAY_OVERHEAD..], b"end-to-end encrypted");
|
||||
let isolated = hub.link(peer(4), "other-ip");
|
||||
assert!(isolated.send(Bytes::new()).is_err());
|
||||
assert_eq!(hub.counters().forwarded, 1);
|
||||
hub.close();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn local_delivery_is_bounded_and_malformed_or_unknown_sources_are_dropped() {
|
||||
let hub = RelayHub::new(network(), peer(2));
|
||||
let a = Wire::new(1, true);
|
||||
hub.set_direct(peer(1), "ip", a);
|
||||
let link = hub.link(peer(1), "ip");
|
||||
let plane = link.plane.clone();
|
||||
for _ in 0..INBOX + 1 {
|
||||
plane.receive(id(1), envelope::encode(id(1), id(2), 0, b"ip"));
|
||||
}
|
||||
assert_eq!(hub.counters().dropped_congested, 1);
|
||||
for _ in 0..INBOX {
|
||||
assert_eq!(link.recv().await.unwrap(), Bytes::from_static(b"ip"));
|
||||
}
|
||||
plane.receive(id(1), Bytes::from_static(b"broken"));
|
||||
plane.receive(id(1), envelope::encode(id(9), id(2), 0, b"unknown member"));
|
||||
assert_eq!(hub.counters().dropped_unknown, 2);
|
||||
plane.receive(
|
||||
id(1),
|
||||
envelope::encode(id(1), id(9), 0, b"unknown destination"),
|
||||
);
|
||||
assert_eq!(hub.counters().dropped_no_link, 1);
|
||||
hub.close();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inconsistent_tables_cannot_loop_beyond_the_hop_limit() {
|
||||
let a = RelayHub::new(network(), peer(1));
|
||||
let b = RelayHub::new(network(), peer(2));
|
||||
let ab = Wire::new(2, true);
|
||||
let ba = Wire::new(1, true);
|
||||
a.set_direct(peer(2), "ip", ab.clone());
|
||||
b.set_direct(peer(1), "ip", ba.clone());
|
||||
// Deliberately inconsistent snapshots while updates are in flight.
|
||||
topology(&a, "ip", &[(2, &[4])], &[1, 2, 3, 4]);
|
||||
topology(&b, "ip", &[(1, &[4])], &[1, 2, 3, 4]);
|
||||
let ap = a.link(peer(4), "ip").plane.clone();
|
||||
let bp = b.link(peer(4), "ip").plane.clone();
|
||||
let mut frame = envelope::encode(id(3), id(4), 7, b"transit");
|
||||
for turn in 0..ROUTING_HOP_LIMIT {
|
||||
let (plane, incoming, wire) = if turn % 2 == 0 {
|
||||
(&ap, id(2), &ab)
|
||||
} else {
|
||||
(&bp, id(1), &ba)
|
||||
};
|
||||
plane.receive(incoming, frame);
|
||||
let sent = wire.take();
|
||||
if turn == ROUTING_HOP_LIMIT - 1 {
|
||||
assert!(sent.is_empty());
|
||||
break;
|
||||
}
|
||||
assert_eq!(sent.len(), 1);
|
||||
frame = sent.into_iter().next().unwrap();
|
||||
}
|
||||
assert_eq!(
|
||||
a.counters().forwarded + b.counters().forwarded,
|
||||
u64::from(ROUTING_HOP_LIMIT - 1)
|
||||
);
|
||||
assert_eq!(
|
||||
a.counters().dropped_hop_limit + b.counters().dropped_hop_limit,
|
||||
1
|
||||
);
|
||||
a.close();
|
||||
b.close();
|
||||
}
|
||||
|
||||
/// Measures the actual synchronous transit routine, including header checks,
|
||||
/// snapshot lookup, ECMP, in-place TTL update and mock transport submission.
|
||||
/// Packet construction, encryption and socket I/O are outside this measurement.
|
||||
#[tokio::test]
|
||||
#[ignore = "manual release-mode forwarding microbenchmark"]
|
||||
async fn forwarding_benchmark() {
|
||||
let hub = RelayHub::new(network(), peer(2));
|
||||
let wire = Wire::new(3, false);
|
||||
hub.set_direct(peer(3), "ip", wire.clone());
|
||||
let alternate = Wire::new(5, false);
|
||||
hub.set_direct(peer(5), "ip", alternate.clone());
|
||||
topology(&hub, "ip", &[(3, &[4]), (5, &[4])], &[1, 2, 3, 4, 5]);
|
||||
let plane = hub.link(peer(4), "ip").plane.clone();
|
||||
let source = id(1);
|
||||
let destination = id(4);
|
||||
let incoming = id(1);
|
||||
for size in [64, 1280] {
|
||||
let mut elapsed = std::time::Duration::ZERO;
|
||||
const COUNT: usize = 10000;
|
||||
const BATCHES: usize = 100;
|
||||
for _ in 0..BATCHES {
|
||||
let frames: Vec<_> = (0..COUNT)
|
||||
.map(|flow| envelope::encode(source, destination, flow as u64, &vec![42; size]))
|
||||
.collect();
|
||||
let start = std::time::Instant::now();
|
||||
for frame in frames {
|
||||
plane.receive(incoming, std::hint::black_box(frame));
|
||||
}
|
||||
elapsed += start.elapsed();
|
||||
}
|
||||
let count = (COUNT * BATCHES) as f64;
|
||||
println!(
|
||||
"transit {size} B: {:.1} ns/packet, {:.2} Mpps (1M packets, two equal paths)",
|
||||
elapsed.as_nanos() as f64 / count,
|
||||
count / elapsed.as_secs_f64() / 1e6
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
wire.count.load(Ordering::Relaxed) + alternate.count.load(Ordering::Relaxed),
|
||||
2000000
|
||||
);
|
||||
hub.close();
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
//! Fixed header. Transit changes one byte in an owned receive buffer.
|
||||
use super::{FlowId, PeerId};
|
||||
use crate::config::{MAX_DATA_DATAGRAM, ROUTING_HOP_LIMIT};
|
||||
use bytes::{BufMut, Bytes, BytesMut};
|
||||
|
||||
pub(crate) const HEADER: usize = 2 + 32 + 32 + 8;
|
||||
const VERSION: u8 = 1;
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub(crate) struct Header {
|
||||
pub source: PeerId,
|
||||
pub destination: PeerId,
|
||||
pub remaining: u8,
|
||||
pub flow: FlowId,
|
||||
}
|
||||
|
||||
pub(crate) fn decode(bytes: &[u8]) -> Option<Header> {
|
||||
if bytes.len() < HEADER
|
||||
|| bytes.len() > MAX_DATA_DATAGRAM
|
||||
|| bytes[0] != VERSION
|
||||
|| bytes[1] == 0
|
||||
|| bytes[1] > ROUTING_HOP_LIMIT
|
||||
{
|
||||
return None;
|
||||
}
|
||||
Some(Header {
|
||||
source: bytes[2..34].try_into().ok()?,
|
||||
destination: bytes[34..66].try_into().ok()?,
|
||||
remaining: bytes[1],
|
||||
flow: u64::from_be_bytes(bytes[66..74].try_into().ok()?),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn encode(source: PeerId, destination: PeerId, flow: FlowId, payload: &[u8]) -> Bytes {
|
||||
let mut bytes = BytesMut::with_capacity(HEADER + payload.len());
|
||||
bytes.put_u8(VERSION);
|
||||
bytes.put_u8(ROUTING_HOP_LIMIT);
|
||||
bytes.extend_from_slice(&source);
|
||||
bytes.extend_from_slice(&destination);
|
||||
bytes.put_u64(flow);
|
||||
bytes.extend_from_slice(payload);
|
||||
bytes.freeze()
|
||||
}
|
||||
|
||||
pub(crate) fn decrement(bytes: Bytes) -> Bytes {
|
||||
let mut bytes = match bytes.try_into_mut() {
|
||||
Ok(bytes) => bytes,
|
||||
Err(bytes) => BytesMut::from(bytes.as_ref()),
|
||||
};
|
||||
// Called only after decode and after excluding remaining <= 1.
|
||||
bytes[1] -= 1;
|
||||
bytes.freeze()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
use super::*;
|
||||
#[test]
|
||||
fn transit_reuses_the_buffer_and_rejects_invalid_headers() {
|
||||
let frame = encode([1; 32], [2; 32], 17, &[42; 1280]);
|
||||
let address = frame.as_ptr();
|
||||
let frame = decrement(frame);
|
||||
assert_eq!(
|
||||
frame.as_ptr(),
|
||||
address,
|
||||
"no payload copy for an owned buffer"
|
||||
);
|
||||
assert_eq!(decode(&frame).unwrap().remaining, ROUTING_HOP_LIMIT - 1);
|
||||
assert_eq!(&frame[HEADER..], &[42; 1280]);
|
||||
for length in 0..HEADER {
|
||||
assert!(decode(&frame[..length]).is_none());
|
||||
}
|
||||
for ttl in [0, ROUTING_HOP_LIMIT + 1] {
|
||||
let mut bad = frame.to_vec();
|
||||
bad[1] = ttl;
|
||||
assert!(decode(&bad).is_none());
|
||||
}
|
||||
let mut bad = frame.to_vec();
|
||||
bad[0] = 99;
|
||||
assert!(decode(&bad).is_none());
|
||||
assert!(decode(&vec![1; MAX_DATA_DATAGRAM + 1]).is_none());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
//! Optional IP flow classifier for IP plugins, called before encryption.
|
||||
//! The router itself treats this identifier and the payload as opaque.
|
||||
|
||||
use super::FlowId;
|
||||
|
||||
fn hash(parts: &[&[u8]]) -> FlowId {
|
||||
let mut value = 0xcbf29ce484222325u64;
|
||||
for part in parts {
|
||||
for byte in *part {
|
||||
value = (value ^ u64::from(*byte)).wrapping_mul(0x100000001b3);
|
||||
}
|
||||
}
|
||||
value
|
||||
}
|
||||
|
||||
/// Stable directional IP flow hash, independent of TCP sequence numbers,
|
||||
/// checksums and packet contents. Fragmented traffic uses addresses/protocol
|
||||
/// instead of ports: even non-initial fragments and subsequent fragmented
|
||||
/// datagrams stay together. This deliberately coalesces fragmented flows.
|
||||
/// Malformed packets get a coarse hash; parsing is bounded and never allocates.
|
||||
pub fn ip_flow(packet: &[u8]) -> FlowId {
|
||||
match packet.first().map(|b| b >> 4) {
|
||||
Some(4) if packet.len() >= 20 => {
|
||||
let addresses = &packet[12..20];
|
||||
let protocol = &packet[9..10];
|
||||
let offset = usize::from(packet[0] & 15) * 4;
|
||||
if u16::from_be_bytes([packet[6], packet[7]]) & 0x3fff != 0 {
|
||||
return hash(&[addresses, protocol]);
|
||||
}
|
||||
if offset >= 20
|
||||
&& matches!(packet[9], 6 | 17)
|
||||
&& let Some(ports) = packet.get(offset..offset + 4)
|
||||
{
|
||||
return hash(&[addresses, protocol, ports]);
|
||||
}
|
||||
hash(&[addresses, protocol])
|
||||
}
|
||||
Some(6) if packet.len() >= 40 => {
|
||||
let addresses = &packet[8..40];
|
||||
let mut protocol = packet[6];
|
||||
let mut offset = 40;
|
||||
for _ in 0..8 {
|
||||
match protocol {
|
||||
44 => {
|
||||
if let Some(fragment) = packet.get(offset..offset + 8) {
|
||||
return hash(&[addresses, &fragment[..1]]);
|
||||
}
|
||||
break;
|
||||
}
|
||||
0 | 43 | 60 | 51 => {
|
||||
let Some(header) = packet.get(offset..offset + 2) else {
|
||||
break;
|
||||
};
|
||||
let size = if protocol == 51 {
|
||||
(usize::from(header[1]) + 2) * 4
|
||||
} else {
|
||||
(usize::from(header[1]) + 1) * 8
|
||||
};
|
||||
protocol = header[0];
|
||||
offset += size;
|
||||
}
|
||||
6 | 17 => {
|
||||
if let Some(ports) = packet.get(offset..offset + 4) {
|
||||
return hash(&[addresses, &[protocol], ports]);
|
||||
}
|
||||
break;
|
||||
}
|
||||
_ => break,
|
||||
}
|
||||
}
|
||||
hash(&[addresses, &[protocol]])
|
||||
}
|
||||
_ => 0,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn tcp_flow_ignores_packet_contents_but_distinguishes_ports() {
|
||||
let mut packet = [0u8; 80];
|
||||
packet[0] = 0x45;
|
||||
packet[9] = 6;
|
||||
packet[12..24].copy_from_slice(&[10, 0, 0, 1, 10, 0, 0, 2, 0, 22, 1, 2]);
|
||||
let flow = ip_flow(&packet);
|
||||
packet[4] = 123; // IP id changes for each unfragmented packet.
|
||||
packet[24..].fill(47);
|
||||
assert_eq!(flow, ip_flow(&packet));
|
||||
packet[23] = 3;
|
||||
assert_ne!(flow, ip_flow(&packet));
|
||||
packet[6] = 0x20;
|
||||
let fragment_flow = ip_flow(&packet);
|
||||
packet[6] = 0;
|
||||
packet[7] = 1;
|
||||
packet[20..].fill(99);
|
||||
assert_eq!(fragment_flow, ip_flow(&packet));
|
||||
packet[4] = 97;
|
||||
assert_eq!(fragment_flow, ip_flow(&packet));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ipv6_extensions_and_fragments_are_bounded_and_stable() {
|
||||
let mut packet = [0u8; 80];
|
||||
packet[0] = 0x60;
|
||||
packet[6] = 0;
|
||||
packet[40] = 17;
|
||||
packet[48..52].copy_from_slice(&[1, 2, 3, 4]);
|
||||
let flow = ip_flow(&packet);
|
||||
packet[52..].fill(5);
|
||||
assert_eq!(flow, ip_flow(&packet));
|
||||
packet[50] = 9;
|
||||
assert_ne!(flow, ip_flow(&packet));
|
||||
packet[6] = 44;
|
||||
packet[40] = 6;
|
||||
let flow = ip_flow(&packet);
|
||||
packet[42] = 32;
|
||||
packet[48..].fill(3);
|
||||
assert_eq!(flow, ip_flow(&packet));
|
||||
for len in 0..packet.len() {
|
||||
let _ = ip_flow(&packet[..len]);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
//! Transport-independent routing over opaque identities. Tables are built on
|
||||
//! topology changes, never on the packet path. No WireGuard or iroh APIs here.
|
||||
pub(crate) mod envelope;
|
||||
pub mod flow;
|
||||
|
||||
use std::collections::{BTreeSet, HashMap, VecDeque};
|
||||
|
||||
/// Opaque authenticated identity; the transport adapter validates its keys.
|
||||
pub type PeerId = [u8; 32];
|
||||
/// Stable flow identifier, carried unchanged between transit routers.
|
||||
pub type FlowId = u64;
|
||||
|
||||
/// A shortest route with all equal-cost first hops in stable order.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Route {
|
||||
/// Number of transport links to the destination.
|
||||
pub hops: u8,
|
||||
/// Equal-cost first hops. Direct destinations have one entry.
|
||||
pub next_hops: Box<[PeerId]>,
|
||||
}
|
||||
|
||||
/// Immutable shortest-path table, replaced when connectivity changes.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct RoutingTable {
|
||||
routes: HashMap<PeerId, Route>,
|
||||
}
|
||||
|
||||
impl RoutingTable {
|
||||
/// Computes shortest directed paths bounded by `hop_limit`.
|
||||
/// The caller supplies only authenticated, fresh, compatible edges.
|
||||
pub fn build(local: PeerId, graph: &HashMap<PeerId, Vec<PeerId>>, hop_limit: u8) -> Self {
|
||||
let mut distances = HashMap::from([(local, 0u8)]);
|
||||
let mut first_hops: HashMap<PeerId, BTreeSet<PeerId>> = HashMap::new();
|
||||
let mut pending = VecDeque::from([local]);
|
||||
while let Some(node) = pending.pop_front() {
|
||||
let distance = distances[&node];
|
||||
if distance >= hop_limit {
|
||||
continue;
|
||||
}
|
||||
let inherited = first_hops.get(&node).cloned().unwrap_or_default();
|
||||
for &next in graph.get(&node).into_iter().flatten() {
|
||||
let next_distance = distance + 1;
|
||||
match distances.get(&next) {
|
||||
Some(&known) if known < next_distance => continue,
|
||||
None => {
|
||||
distances.insert(next, next_distance);
|
||||
pending.push_back(next);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
let hops = first_hops.entry(next).or_default();
|
||||
if node == local {
|
||||
hops.insert(next);
|
||||
} else {
|
||||
hops.extend(inherited.iter().copied());
|
||||
}
|
||||
}
|
||||
}
|
||||
Self {
|
||||
routes: first_hops
|
||||
.into_iter()
|
||||
.map(|(peer, hops)| {
|
||||
(
|
||||
peer,
|
||||
Route {
|
||||
hops: distances[&peer],
|
||||
next_hops: hops.into_iter().collect(),
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
/// Precomputed route to a destination.
|
||||
pub fn get(&self, destination: &PeerId) -> Option<&Route> {
|
||||
self.routes.get(destination)
|
||||
}
|
||||
/// All reachable destinations, excluding ourselves.
|
||||
pub fn iter(&self) -> impl Iterator<Item = (&PeerId, &Route)> {
|
||||
self.routes.iter()
|
||||
}
|
||||
}
|
||||
|
||||
/// Stable ECMP mixing, without random state or hashing the payload per hop.
|
||||
pub(crate) fn mix64(mut value: u64) -> u64 {
|
||||
value ^= value >> 30;
|
||||
value = value.wrapping_mul(0xbf58476d1ce4e5b9);
|
||||
value ^= value >> 27;
|
||||
value = value.wrapping_mul(0x94d049bb133111eb);
|
||||
value ^ (value >> 31)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
use super::*;
|
||||
fn p(n: u8) -> PeerId {
|
||||
[n; 32]
|
||||
}
|
||||
#[test]
|
||||
fn shortest_paths_keep_equal_cost_hops_and_prefer_direct_links() {
|
||||
let mut graph = HashMap::from([
|
||||
(p(1), vec![p(3), p(2)]),
|
||||
(p(2), vec![p(4)]),
|
||||
(p(3), vec![p(4)]),
|
||||
(p(4), vec![p(5)]),
|
||||
(p(5), vec![p(1)]),
|
||||
]);
|
||||
let routes = RoutingTable::build(p(1), &graph, 16);
|
||||
assert_eq!(routes.get(&p(5)).unwrap().hops, 3);
|
||||
assert_eq!(&*routes.get(&p(5)).unwrap().next_hops, &[p(2), p(3)]);
|
||||
assert!(routes.get(&p(1)).is_none());
|
||||
assert!(routes.get(&p(9)).is_none());
|
||||
graph.get_mut(&p(1)).unwrap().push(p(5));
|
||||
assert_eq!(
|
||||
RoutingTable::build(p(1), &graph, 16).get(&p(5)).unwrap(),
|
||||
&Route {
|
||||
hops: 1,
|
||||
next_hops: Box::new([p(5)])
|
||||
}
|
||||
);
|
||||
}
|
||||
#[test]
|
||||
fn topology_removal_and_hop_budget_remove_impossible_routes() {
|
||||
let mut graph = HashMap::from([(p(1), vec![p(2)]), (p(2), vec![p(3)]), (p(3), vec![p(4)])]);
|
||||
assert!(RoutingTable::build(p(1), &graph, 2).get(&p(4)).is_none());
|
||||
assert!(RoutingTable::build(p(1), &graph, 3).get(&p(4)).is_some());
|
||||
graph.remove(&p(2));
|
||||
assert!(RoutingTable::build(p(1), &graph, 16).get(&p(4)).is_none());
|
||||
}
|
||||
}
|
||||
@@ -88,6 +88,16 @@ pub trait PacketLink: Send + Sync + std::fmt::Debug + 'static {
|
||||
/// handed to the transport, nothing more.
|
||||
fn send(&self, payload: Bytes) -> Result<(), TransportError>;
|
||||
|
||||
/// Sends a datagram with an opaque stable flow identifier. Plugins derive
|
||||
/// this before encryption; transports that do not route can ignore it.
|
||||
fn send_flow(
|
||||
&self,
|
||||
payload: Bytes,
|
||||
_flow: super::routing::FlowId,
|
||||
) -> Result<(), TransportError> {
|
||||
self.send(payload)
|
||||
}
|
||||
|
||||
/// Receives the next datagram, or `None` once the link is finished.
|
||||
fn recv(&self) -> BoxFuture<'_, Option<Bytes>>;
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ use crate::error::ProtocolError;
|
||||
/// The version in the ALPN is the wire-compatibility version of the control
|
||||
/// protocol. It is independent of the network identity scheme version, so
|
||||
/// bumping it must not change any existing [`crate::NetworkId`].
|
||||
pub const ALPN: &[u8] = b"tsunagi/ctrl/1";
|
||||
pub const ALPN: &[u8] = b"tsunagi/ctrl/2";
|
||||
|
||||
/// ALPN of the tsunagi data plane.
|
||||
///
|
||||
@@ -29,11 +29,10 @@ pub const ALPN: &[u8] = b"tsunagi/ctrl/1";
|
||||
/// saturated or broken data plane cannot disturb control traffic, and the
|
||||
/// transport underneath can be replaced without touching the control protocol.
|
||||
///
|
||||
/// Version 3 fragments logical datagrams below the peer-relay envelope, so
|
||||
/// the overlay's MTU is independent of the current QUIC path MTU. Older data
|
||||
/// versions cannot form a data link; control and persistent identities remain
|
||||
/// compatible. Both ends, including any intermediate peer, must be upgraded.
|
||||
pub const DATA_ALPN: &[u8] = b"tsunagi/data/3";
|
||||
/// Version 4 adds source, destination, hop limit and stable flow id above
|
||||
/// transport fragmentation. All members must upgrade together. Persistent
|
||||
/// identities and network configuration do not change.
|
||||
pub const DATA_ALPN: &[u8] = b"tsunagi/data/4";
|
||||
|
||||
/// Largest plugin protocol identifier accepted when opening a data channel.
|
||||
pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
|
||||
@@ -42,7 +41,7 @@ pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
|
||||
pub const MAX_SIGNATURE_LEN: usize = 64;
|
||||
|
||||
/// Control protocol version carried inside the handshake.
|
||||
pub const PROTOCOL_VERSION: u16 = 1;
|
||||
pub const PROTOCOL_VERSION: u16 = 2;
|
||||
|
||||
/// First message of the handshake, sent by the initiator.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
@@ -154,8 +153,8 @@ pub enum ControlMessage {
|
||||
/// and repairs itself; and volatile, so it lives here and not in
|
||||
/// signed state, which is for what has to survive a member being away.
|
||||
Reachable {
|
||||
/// The peers, as raw endpoint keys.
|
||||
peers: Vec<[u8; 32]>,
|
||||
/// Direct links, scoped to the plugin protocol they actually carry.
|
||||
links: Vec<ReachableLink>,
|
||||
},
|
||||
/// Graceful goodbye.
|
||||
///
|
||||
@@ -183,6 +182,15 @@ pub struct PeerHint {
|
||||
pub addrs: Vec<String>,
|
||||
}
|
||||
|
||||
/// One directly observed edge, advertised only by its authenticated source.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
pub struct ReachableLink {
|
||||
/// The other end of this transport link.
|
||||
pub peer: [u8; 32],
|
||||
/// Protocol carried by the link.
|
||||
pub protocol: String,
|
||||
}
|
||||
|
||||
/// Longest address text accepted in a hint.
|
||||
pub const MAX_HINT_ADDR_LEN: usize = 128;
|
||||
|
||||
@@ -256,10 +264,18 @@ pub fn validate(message: &ControlMessage, limits: &Limits) -> Result<(), Protoco
|
||||
}
|
||||
}
|
||||
}
|
||||
ControlMessage::Reachable { peers } => {
|
||||
// One entry per member at most; a sender claiming more than a
|
||||
// network can hold is not describing this network.
|
||||
check_len("reachable.peers", peers.len(), limits.max_state_records)?;
|
||||
ControlMessage::Reachable { links } => {
|
||||
check_len("reachable.links", links.len(), limits.max_state_records)?;
|
||||
for link in links {
|
||||
if link.protocol.is_empty() {
|
||||
return Err(ProtocolError::Malformed("empty routing protocol"));
|
||||
}
|
||||
check_len(
|
||||
"reachable.protocol",
|
||||
link.protocol.len(),
|
||||
MAX_PROTOCOL_ID_LEN,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
ControlMessage::Announce(announcement) => {
|
||||
check_len(
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
//! The real Mainline UDP receive loop must be quiet while idle on Windows too.
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
|
||||
use std::net::Ipv4Addr;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
|
||||
struct SocketWarnings(Arc<AtomicUsize>);
|
||||
|
||||
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for SocketWarnings {
|
||||
fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) {
|
||||
if event.metadata().target() == "mainline::rpc::socket"
|
||||
&& *event.metadata().level() <= tracing::Level::WARN
|
||||
{
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_idle_dht_socket_stays_quiet_and_still_answers_packets() {
|
||||
// Mainline owns an OS thread, so a thread-local subscriber cannot see its
|
||||
// events. This integration-test binary has just this test and installs
|
||||
// its own subscriber; it cannot affect the application or other binaries.
|
||||
let warnings = Arc::new(AtomicUsize::new(0));
|
||||
tracing::subscriber::set_global_default(
|
||||
tracing_subscriber::registry().with(SocketWarnings(Arc::clone(&warnings))),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
let dht = mainline::Dht::builder()
|
||||
.no_bootstrap()
|
||||
.server_mode()
|
||||
.bind_address(Ipv4Addr::LOCALHOST)
|
||||
.port(0)
|
||||
.build()
|
||||
.unwrap()
|
||||
.as_async();
|
||||
|
||||
let address = dht.info().await.local_addr();
|
||||
// Each info response comes from the actor after another receive-loop
|
||||
// iteration. With no peers or packets, these exercise real socket
|
||||
// read timeouts, without an arbitrary sleep or public bootstrap.
|
||||
for _ in 0..8 {
|
||||
assert_eq!(dht.info().await.local_addr(), address);
|
||||
}
|
||||
assert_eq!(warnings.load(Ordering::Relaxed), 0, "idle receive warnings");
|
||||
|
||||
let client = tokio::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
let ping = b"d1:ad2:id20:abcdefghijklmnopqrste1:q4:ping1:t4:ping1:y1:qe";
|
||||
client.send_to(ping, address).await.unwrap();
|
||||
let mut response = [0u8; 2048];
|
||||
let (len, from) = client.recv_from(&mut response).await.unwrap();
|
||||
assert_eq!(from, std::net::SocketAddr::V4(address));
|
||||
let response = &response[..len];
|
||||
assert!(response.windows(9).any(|field| field == b"1:t4:ping"));
|
||||
assert!(response.windows(6).any(|field| field == b"1:y1:r"));
|
||||
assert_eq!(warnings.load(Ordering::Relaxed), 0);
|
||||
})
|
||||
.await
|
||||
.expect("the DHT receive loop keeps making progress");
|
||||
}
|
||||
Reference in New Issue
Block a user