From 4bbecccc11305a8256f31f7cdebee0848ca89073 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 31 Jan 2026 01:20:13 +0000 Subject: [PATCH] Session 37: Transport-Node lifecycle integration - Added TransportHandle enum for polymorphic transport dispatch - Node now owns transports via HashMap - Added packet channel fields (packet_tx, packet_rx) to Node - Transport initialization in Node::start() with graceful degradation - Transport shutdown in Node::stop() before TUN cleanup - Factory method create_transports() instantiates from config Configuration redesign: - New transports section with TransportInstances enum - Single instance: config directly under transport type (no naming overhead) - Named instances: nested under instance names - #[serde(deny_unknown_fields)] ensures correct untagged enum matching - Instance names are Option<&str> - None for single, Some(name) for named Updated Node transport methods: - transport_count(), get_transport(), get_transport_mut() - transport_ids(), packet_rx() All 189 tests pass (4 new config parsing tests). --- src/config.rs | 222 +++++++++++++++++++++++++++++++++++++++---- src/lib.rs | 4 +- src/node.rs | 186 ++++++++++++++++++++++++++++++------ src/transport/mod.rs | 71 ++++++++++++++ src/transport/udp.rs | 1 - 5 files changed, 430 insertions(+), 54 deletions(-) diff --git a/src/config.rs b/src/config.rs index 03e77e4..12161cb 100644 --- a/src/config.rs +++ b/src/config.rs @@ -107,18 +107,15 @@ impl TunConfig { } } -/// UDP transport configuration (`udp.*`). +/// UDP transport instance configuration. #[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct UdpConfig { - /// Enable UDP transport (`udp.enabled`). - #[serde(default, skip_serializing_if = "std::ops::Not::not")] - pub enabled: bool, - - /// Bind address (`udp.bind_addr`). Defaults to "0.0.0.0:4000". + /// Bind address (`bind_addr`). Defaults to "0.0.0.0:4000". #[serde(default, skip_serializing_if = "Option::is_none")] pub bind_addr: Option, - /// UDP MTU (`udp.mtu`). Defaults to 1280 (IPv6 minimum). + /// UDP MTU (`mtu`). Defaults to 1280 (IPv6 minimum). #[serde(default, skip_serializing_if = "Option::is_none")] pub mtu: Option, } @@ -135,6 +132,123 @@ impl UdpConfig { } } +// ============================================================================ +// Transport Configuration +// ============================================================================ + +use std::collections::HashMap; + +/// Transport instances - either a single config or named instances. +/// +/// Allows both simple single-instance config: +/// ```yaml +/// transports: +/// udp: +/// bind_addr: "0.0.0.0:4000" +/// ``` +/// +/// And multiple named instances: +/// ```yaml +/// transports: +/// udp: +/// main: +/// bind_addr: "0.0.0.0:4000" +/// backup: +/// bind_addr: "192.168.1.100:4001" +/// ``` +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(untagged)] +pub enum TransportInstances { + /// Single unnamed instance (config fields directly under transport type). + Single(T), + /// Multiple named instances. + Named(HashMap), +} + +impl TransportInstances { + /// Get the number of instances. + pub fn len(&self) -> usize { + match self { + TransportInstances::Single(_) => 1, + TransportInstances::Named(map) => map.len(), + } + } + + /// Check if there are no instances. + pub fn is_empty(&self) -> bool { + match self { + TransportInstances::Single(_) => false, + TransportInstances::Named(map) => map.is_empty(), + } + } + + /// Iterate over all instances as (name, config) pairs. + /// + /// Single instances have `None` as the name. + /// Named instances have `Some(name)`. + pub fn iter(&self) -> impl Iterator, &T)> { + match self { + TransportInstances::Single(config) => { + vec![(None, config)].into_iter() + } + TransportInstances::Named(map) => { + map.iter() + .map(|(k, v)| (Some(k.as_str()), v)) + .collect::>() + .into_iter() + } + } + } +} + +impl Default for TransportInstances { + fn default() -> Self { + TransportInstances::Named(HashMap::new()) + } +} + +/// Transports configuration section. +/// +/// Each transport type can have either a single instance (config directly +/// under the type name) or multiple named instances. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct TransportsConfig { + /// UDP transport instances. + #[serde(default, skip_serializing_if = "is_transport_empty")] + pub udp: TransportInstances, + + // Future transport types: + // #[serde(default, skip_serializing_if = "is_transport_empty")] + // pub tcp: TransportInstances, + // + // #[serde(default, skip_serializing_if = "is_transport_empty")] + // pub tor: TransportInstances, +} + +/// Helper for skip_serializing_if on TransportInstances. +fn is_transport_empty(instances: &TransportInstances) -> bool { + instances.is_empty() +} + +impl TransportsConfig { + /// Check if any transports are configured. + pub fn is_empty(&self) -> bool { + self.udp.is_empty() + // && self.tcp.is_empty() + // && self.tor.is_empty() + } + + /// Merge another TransportsConfig into this one. + /// + /// Non-empty transport sections from `other` replace those in `self`. + pub fn merge(&mut self, other: TransportsConfig) { + if !other.udp.is_empty() { + self.udp = other.udp; + } + // Future: same for tcp, tor, etc. + } +} + /// Root configuration structure. #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct Config { @@ -146,9 +260,9 @@ pub struct Config { #[serde(default)] pub tun: TunConfig, - /// UDP transport configuration (`udp.*`). - #[serde(default)] - pub udp: UdpConfig, + /// Transport instances (`transports.*`). + #[serde(default, skip_serializing_if = "TransportsConfig::is_empty")] + pub transports: TransportsConfig, } impl Config { @@ -247,16 +361,8 @@ impl Config { if other.tun.mtu.is_some() { self.tun.mtu = other.tun.mtu; } - // Merge udp section - if other.udp.enabled { - self.udp.enabled = true; - } - if other.udp.bind_addr.is_some() { - self.udp.bind_addr = other.udp.bind_addr; - } - if other.udp.mtu.is_some() { - self.udp.mtu = other.udp.mtu; - } + // Merge transports section + self.transports.merge(other.transports); } /// Create an Identity from this configuration. @@ -497,4 +603,80 @@ node: // Empty nsec should not be serialized assert!(!yaml.contains("nsec:")); } + + #[test] + fn test_parse_transport_single_instance() { + let yaml = r#" +transports: + udp: + bind_addr: "0.0.0.0:4000" + mtu: 1400 +"#; + let config: Config = serde_yaml::from_str(yaml).unwrap(); + + assert_eq!(config.transports.udp.len(), 1); + let instances: Vec<_> = config.transports.udp.iter().collect(); + assert_eq!(instances.len(), 1); + assert_eq!(instances[0].0, None); // Single instance has no name + assert_eq!(instances[0].1.bind_addr(), "0.0.0.0:4000"); + assert_eq!(instances[0].1.mtu(), 1400); + } + + #[test] + fn test_parse_transport_named_instances() { + let yaml = r#" +transports: + udp: + main: + bind_addr: "0.0.0.0:4000" + backup: + bind_addr: "192.168.1.100:4001" + mtu: 1280 +"#; + let config: Config = serde_yaml::from_str(yaml).unwrap(); + + assert_eq!(config.transports.udp.len(), 2); + + let instances: std::collections::HashMap<_, _> = + config.transports.udp.iter().map(|(k, v)| (k, v)).collect(); + + // Named instances have Some(name) + assert!(instances.contains_key(&Some("main"))); + assert!(instances.contains_key(&Some("backup"))); + assert_eq!(instances[&Some("main")].bind_addr(), "0.0.0.0:4000"); + assert_eq!(instances[&Some("backup")].bind_addr(), "192.168.1.100:4001"); + assert_eq!(instances[&Some("backup")].mtu(), 1280); + } + + #[test] + fn test_parse_transport_empty() { + let yaml = r#" +transports: {} +"#; + let config: Config = serde_yaml::from_str(yaml).unwrap(); + assert!(config.transports.udp.is_empty()); + assert!(config.transports.is_empty()); + } + + #[test] + fn test_transport_instances_iter() { + // Single instance - no name + let single = TransportInstances::Single(UdpConfig { + bind_addr: Some("0.0.0.0:4000".to_string()), + mtu: None, + }); + let items: Vec<_> = single.iter().collect(); + assert_eq!(items.len(), 1); + assert_eq!(items[0].0, None); + + // Named instances - have names + let mut map = HashMap::new(); + map.insert("a".to_string(), UdpConfig::default()); + map.insert("b".to_string(), UdpConfig::default()); + let named = TransportInstances::Named(map); + let items: Vec<_> = named.iter().collect(); + assert_eq!(items.len(), 2); + // All named instances should have Some(name) + assert!(items.iter().all(|(name, _)| name.is_some())); + } } diff --git a/src/lib.rs b/src/lib.rs index 1cdcf0d..6b72d86 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,8 +33,8 @@ pub use bloom::{BloomError, BloomFilter, BloomState}; // Re-export transport types pub use transport::{ packet_channel, DiscoveredPeer, Link, LinkDirection, LinkId, LinkState, LinkStats, PacketRx, - PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, TransportId, - TransportState, TransportType, + PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, TransportHandle, + TransportId, TransportState, TransportType, }; pub use transport::udp::UdpTransport; diff --git a/src/node.rs b/src/node.rs index 6f58402..e6022ff 100644 --- a/src/node.rs +++ b/src/node.rs @@ -7,7 +7,10 @@ use crate::bloom::BloomState; use crate::cache::CoordCache; use crate::peer::Peer; -use crate::transport::{Link, LinkId, TransportId}; +use crate::transport::{ + packet_channel, Link, LinkId, PacketRx, PacketTx, TransportHandle, TransportId, +}; +use crate::transport::udp::UdpTransport; use crate::tree::TreeState; use crate::tun::{run_tun_reader, shutdown_tun_interface, TunDevice, TunError, TunState, TunTx}; use crate::{Config, ConfigError, Identity, IdentityError, NodeId}; @@ -15,7 +18,7 @@ use std::collections::HashMap; use std::fmt; use std::thread::{self, JoinHandle}; use thiserror::Error; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; /// Errors related to node operations. #[derive(Debug, Error)] @@ -134,11 +137,17 @@ pub struct Node { coord_cache: CoordCache, // === Transports & Links === - /// Active transport IDs. - transport_ids: Vec, + /// Active transports (owned by Node). + transports: HashMap, /// Active links. links: HashMap, + // === Packet Channel === + /// Packet sender for transports. + packet_tx: Option, + /// Packet receiver (for event loop). + packet_rx: Option, + // === Peers === /// Authenticated peers. peers: HashMap, @@ -207,8 +216,10 @@ impl Node { tree_state, bloom_state, coord_cache: CoordCache::with_defaults(), - transport_ids: Vec::new(), + transports: HashMap::new(), links: HashMap::new(), + packet_tx: None, + packet_rx: None, peers: HashMap::new(), max_peers: 128, max_links: 256, @@ -251,8 +262,10 @@ impl Node { tree_state, bloom_state: BloomState::new(node_id), coord_cache: CoordCache::with_defaults(), - transport_ids: Vec::new(), + transports: HashMap::new(), links: HashMap::new(), + packet_tx: None, + packet_rx: None, peers: HashMap::new(), max_peers: 128, max_links: 256, @@ -274,6 +287,55 @@ impl Node { Ok(node) } + /// Create transport instances from configuration. + /// + /// Returns a vector of TransportHandles for all configured transports. + fn create_transports(&mut self, packet_tx: &PacketTx) -> Vec { + let mut transports = Vec::new(); + + // Collect UDP configs with optional names to avoid borrow conflicts + let udp_instances: Vec<_> = self + .config + .transports + .udp + .iter() + .map(|(name, config)| (name.map(|s| s.to_string()), config.clone())) + .collect(); + + // Create UDP transport instances + for (name, udp_config) in udp_instances { + let transport_id = self.allocate_transport_id(); + let bind_addr = udp_config.bind_addr().to_string(); + let udp = UdpTransport::new( + transport_id, + udp_config, + packet_tx.clone(), + ); + transports.push(TransportHandle::Udp(udp)); + + // Log with name only if present (named instance) + if let Some(ref n) = name { + debug!( + transport_id = %transport_id, + name = %n, + bind_addr = %bind_addr, + "Created UDP transport" + ); + } else { + debug!( + transport_id = %transport_id, + bind_addr = %bind_addr, + "Created UDP transport" + ); + } + } + + // Future transports follow same pattern: + // for (name, tcp_config) in self.config.transports.tcp.iter() { ... } + + transports + } + // === Identity Accessors === /// Get this node's identity. @@ -383,9 +445,9 @@ impl Node { self.links.len() } - /// Number of transports. + /// Number of active transports. pub fn transport_count(&self) -> usize { - self.transport_ids.len() + self.transports.len() } // === Transport Management === @@ -397,21 +459,24 @@ impl Node { id } - /// Register a transport. - pub fn add_transport(&mut self, transport_id: TransportId) { - if !self.transport_ids.contains(&transport_id) { - self.transport_ids.push(transport_id); - } + /// Get a transport by ID. + pub fn get_transport(&self, id: &TransportId) -> Option<&TransportHandle> { + self.transports.get(id) } - /// Unregister a transport. - pub fn remove_transport(&mut self, transport_id: &TransportId) { - self.transport_ids.retain(|id| id != transport_id); + /// Get mutable transport by ID. + pub fn get_transport_mut(&mut self, id: &TransportId) -> Option<&mut TransportHandle> { + self.transports.get_mut(id) } - /// Get all transport IDs. - pub fn transport_ids(&self) -> &[TransportId] { - &self.transport_ids + /// Iterate over transport IDs. + pub fn transport_ids(&self) -> impl Iterator { + self.transports.keys() + } + + /// Get the packet receiver for the event loop. + pub fn packet_rx(&mut self) -> Option<&mut PacketRx> { + self.packet_rx.as_mut() } // === Link Management === @@ -579,7 +644,42 @@ impl Node { } } - // TODO: Initialize transports here + // Create packet channel for transport -> Node communication + const PACKET_BUFFER_SIZE: usize = 1024; + let (packet_tx, packet_rx) = packet_channel(PACKET_BUFFER_SIZE); + self.packet_tx = Some(packet_tx.clone()); + self.packet_rx = Some(packet_rx); + + // Initialize transports + let transport_handles = self.create_transports(&packet_tx); + + for mut handle in transport_handles { + let transport_id = handle.transport_id(); + let transport_type = handle.transport_type().name; + + match handle.start().await { + Ok(()) => { + info!( + transport_id = %transport_id, + transport_type, + "Transport started" + ); + self.transports.insert(transport_id, handle); + } + Err(e) => { + warn!( + transport_id = %transport_id, + transport_type, + error = %e, + "Transport failed to start, continuing without it" + ); + } + } + } + + if !self.transports.is_empty() { + info!(count = self.transports.len(), "Transports initialized"); + } self.state = NodeState::Running; info!(state = %self.state, "Node started"); @@ -597,6 +697,31 @@ impl Node { self.state = NodeState::Stopping; info!(state = %self.state, "Node stopping"); + // Shutdown transports first (they're packet producers) + let transport_ids: Vec<_> = self.transports.keys().cloned().collect(); + for transport_id in transport_ids { + if let Some(mut handle) = self.transports.remove(&transport_id) { + let transport_type = handle.transport_type().name; + match handle.stop().await { + Ok(()) => { + info!(transport_id = %transport_id, transport_type, "Transport stopped"); + } + Err(e) => { + warn!( + transport_id = %transport_id, + transport_type, + error = %e, + "Transport stop failed" + ); + } + } + } + } + + // Drop packet channels + self.packet_tx.take(); + self.packet_rx.take(); + // Shutdown TUN interface if let Some(name) = self.tun_name.take() { info!(name = %name, "Shutting down TUN interface"); @@ -620,8 +745,6 @@ impl Node { self.tun_state = TunState::Disabled; } - // TODO: Shutdown transports here - self.state = NodeState::Stopped; info!(state = %self.state, "Node stopped"); Ok(()) @@ -856,19 +979,20 @@ mod tests { fn test_node_transport_management() { let mut node = make_node(); + // Initially no transports (transports are created during start()) + assert_eq!(node.transport_count(), 0); + + // Allocating IDs still works let id1 = node.allocate_transport_id(); let id2 = node.allocate_transport_id(); + assert_ne!(id1, id2); - node.add_transport(id1); - node.add_transport(id2); - assert_eq!(node.transport_count(), 2); + // get_transport returns None when transport doesn't exist + assert!(node.get_transport(&id1).is_none()); + assert!(node.get_transport(&id2).is_none()); - // Adding same ID again doesn't duplicate - node.add_transport(id1); - assert_eq!(node.transport_count(), 2); - - node.remove_transport(&id1); - assert_eq!(node.transport_count(), 1); + // transport_ids() iterator is empty + assert_eq!(node.transport_ids().count(), 0); } #[test] diff --git a/src/transport/mod.rs b/src/transport/mod.rs index ee533c0..651fa95 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -7,6 +7,7 @@ pub mod udp; use secp256k1::XOnlyPublicKey; +use udp::UdpTransport; use std::fmt; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use thiserror::Error; @@ -752,6 +753,76 @@ pub trait Transport { fn discover(&self) -> Result, TransportError>; } +// ============================================================================ +// Transport Handle +// ============================================================================ + +/// Wrapper enum for concrete transport implementations. +/// +/// This enables polymorphic transport handling without trait objects, +/// supporting async methods that the sync Transport trait cannot express. +pub enum TransportHandle { + /// UDP/IP transport. + Udp(UdpTransport), + // Future: Tcp(TcpTransport), Tor(TorTransport), etc. +} + +impl TransportHandle { + /// Start the transport asynchronously. + pub async fn start(&mut self) -> Result<(), TransportError> { + match self { + TransportHandle::Udp(t) => t.start_async().await, + } + } + + /// Stop the transport asynchronously. + pub async fn stop(&mut self) -> Result<(), TransportError> { + match self { + TransportHandle::Udp(t) => t.stop_async().await, + } + } + + /// Send data to a remote address asynchronously. + pub async fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result { + match self { + TransportHandle::Udp(t) => t.send_async(addr, data).await, + } + } + + /// Get the transport ID. + pub fn transport_id(&self) -> TransportId { + match self { + TransportHandle::Udp(t) => t.transport_id(), + } + } + + /// Get the transport type metadata. + pub fn transport_type(&self) -> &TransportType { + match self { + TransportHandle::Udp(t) => t.transport_type(), + } + } + + /// Get current transport state. + pub fn state(&self) -> TransportState { + match self { + TransportHandle::Udp(t) => t.state(), + } + } + + /// Get the transport MTU. + pub fn mtu(&self) -> u16 { + match self { + TransportHandle::Udp(t) => t.mtu(), + } + } + + /// Check if transport is operational. + pub fn is_operational(&self) -> bool { + self.state().is_operational() + } +} + // ============================================================================ // Tests // ============================================================================ diff --git a/src/transport/udp.rs b/src/transport/udp.rs index 39483bf..d9851d9 100644 --- a/src/transport/udp.rs +++ b/src/transport/udp.rs @@ -288,7 +288,6 @@ mod tests { fn make_config(port: u16) -> UdpConfig { UdpConfig { - enabled: true, bind_addr: Some(format!("127.0.0.1:{}", port)), mtu: Some(1280), }