From a317896367828ca5838f3c4725784d1082c480ae Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 31 Jan 2026 04:16:11 +0000 Subject: [PATCH] Add transport instance naming and reorder node startup - Add name field to UdpTransport for named config instances - Add name() and local_addr() accessors to TransportHandle - Update UdpTransport logging to include instance name - Remove redundant Node-level transport startup logging - Reorder Node::start() to initialize transports before TUN --- src/node.rs | 94 ++++++++++++++++---------------------------- src/transport/mod.rs | 14 +++++++ src/transport/udp.rs | 56 +++++++++++++++++--------- 3 files changed, 86 insertions(+), 78 deletions(-) diff --git a/src/node.rs b/src/node.rs index e6022ff..693027f 100644 --- a/src/node.rs +++ b/src/node.rs @@ -18,7 +18,7 @@ use std::collections::HashMap; use std::fmt; use std::thread::{self, JoinHandle}; use thiserror::Error; -use tracing::{debug, info, warn}; +use tracing::{info, warn}; /// Errors related to node operations. #[derive(Debug, Error)] @@ -305,29 +305,8 @@ impl Node { // 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(), - ); + let udp = UdpTransport::new(transport_id, name, 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: @@ -597,6 +576,38 @@ impl Node { } self.state = NodeState::Starting; + // 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 first (before TUN) + 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; + let name = handle.name().map(|s| s.to_string()); + + match handle.start().await { + Ok(()) => { + self.transports.insert(transport_id, handle); + } + Err(e) => { + if let Some(ref n) = name { + warn!(transport_type, name = %n, error = %e, "Transport failed to start"); + } else { + warn!(transport_type, error = %e, "Transport failed to start"); + } + } + } + } + + if !self.transports.is_empty() { + info!(count = self.transports.len(), "Transports initialized"); + } + // Initialize TUN interface if configured if self.config.tun.enabled { let address = *self.identity.address(); @@ -644,43 +655,6 @@ impl Node { } } - // 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"); Ok(()) diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 651fa95..7c7d5b0 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -796,6 +796,13 @@ impl TransportHandle { } } + /// Get the instance name (if configured as a named instance). + pub fn name(&self) -> Option<&str> { + match self { + TransportHandle::Udp(t) => t.name(), + } + } + /// Get the transport type metadata. pub fn transport_type(&self) -> &TransportType { match self { @@ -817,6 +824,13 @@ impl TransportHandle { } } + /// Get the local bound address (only valid after start). + pub fn local_addr(&self) -> Option { + match self { + TransportHandle::Udp(t) => t.local_addr(), + } + } + /// Check if transport is operational. pub fn is_operational(&self) -> bool { self.state().is_operational() diff --git a/src/transport/udp.rs b/src/transport/udp.rs index d9851d9..fc03a84 100644 --- a/src/transport/udp.rs +++ b/src/transport/udp.rs @@ -21,6 +21,8 @@ use tracing::{debug, info, warn}; pub struct UdpTransport { /// Unique transport identifier. transport_id: TransportId, + /// Optional instance name (for named instances in config). + name: Option, /// Configuration. config: UdpConfig, /// Current state. @@ -37,9 +39,15 @@ pub struct UdpTransport { impl UdpTransport { /// Create a new UDP transport. - pub fn new(transport_id: TransportId, config: UdpConfig, packet_tx: PacketTx) -> Self { + pub fn new( + transport_id: TransportId, + name: Option, + config: UdpConfig, + packet_tx: PacketTx, + ) -> Self { Self { transport_id, + name, config, state: TransportState::Configured, socket: None, @@ -49,6 +57,11 @@ impl UdpTransport { } } + /// Get the instance name (if configured as a named instance). + pub fn name(&self) -> Option<&str> { + self.name.as_deref() + } + /// Get the local bound address (only valid after start). pub fn local_addr(&self) -> Option { self.local_addr @@ -102,12 +115,18 @@ impl UdpTransport { self.recv_task = Some(recv_task); self.state = TransportState::Up; - info!( - transport_id = %self.transport_id, - local_addr = %self.local_addr.unwrap(), - mtu = self.config.mtu(), - "UDP transport started" - ); + if let Some(ref name) = self.name { + info!( + name = %name, + local_addr = %self.local_addr.unwrap(), + "UDP transport started" + ); + } else { + info!( + local_addr = %self.local_addr.unwrap(), + "UDP transport started" + ); + } Ok(()) } @@ -296,7 +315,7 @@ mod tests { #[tokio::test] async fn test_start_stop() { let (tx, _rx) = packet_channel(100); - let mut transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let mut transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); assert_eq!(transport.state(), TransportState::Configured); @@ -311,7 +330,7 @@ mod tests { #[tokio::test] async fn test_double_start_fails() { let (tx, _rx) = packet_channel(100); - let mut transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let mut transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); transport.start_async().await.unwrap(); @@ -324,7 +343,7 @@ mod tests { #[tokio::test] async fn test_stop_not_started_fails() { let (tx, _rx) = packet_channel(100); - let mut transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let mut transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); let result = transport.stop_async().await; assert!(matches!(result, Err(TransportError::NotStarted))); @@ -335,8 +354,8 @@ mod tests { let (tx1, _rx1) = packet_channel(100); let (tx2, mut rx2) = packet_channel(100); - let mut t1 = UdpTransport::new(TransportId::new(1), make_config(0), tx1); - let mut t2 = UdpTransport::new(TransportId::new(2), make_config(0), tx2); + let mut t1 = UdpTransport::new(TransportId::new(1), None, make_config(0), tx1); + let mut t2 = UdpTransport::new(TransportId::new(2), None, make_config(0), tx2); t1.start_async().await.unwrap(); t2.start_async().await.unwrap(); @@ -370,8 +389,8 @@ mod tests { let (tx1, mut rx1) = packet_channel(100); let (tx2, mut rx2) = packet_channel(100); - let mut t1 = UdpTransport::new(TransportId::new(1), make_config(0), tx1); - let mut t2 = UdpTransport::new(TransportId::new(2), make_config(0), tx2); + let mut t1 = UdpTransport::new(TransportId::new(1), None, make_config(0), tx1); + let mut t2 = UdpTransport::new(TransportId::new(2), None, make_config(0), tx2); t1.start_async().await.unwrap(); t2.start_async().await.unwrap(); @@ -408,6 +427,7 @@ mod tests { let (tx, _rx) = packet_channel(100); let mut transport = UdpTransport::new( TransportId::new(1), + None, UdpConfig { mtu: Some(100), ..make_config(0) @@ -430,7 +450,7 @@ mod tests { #[tokio::test] async fn test_send_not_started() { let (tx, _rx) = packet_channel(100); - let transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); let result = transport .send_async(&TransportAddr::from_string("127.0.0.1:9999"), b"test") @@ -442,7 +462,7 @@ mod tests { #[tokio::test] async fn test_discover_returns_empty() { let (tx, _rx) = packet_channel(100); - let transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); // Discovery returns empty until multicast/DNS-SD is implemented let peers = transport.discover().unwrap(); @@ -452,7 +472,7 @@ mod tests { #[test] fn test_transport_type() { let (tx, _rx) = packet_channel(100); - let transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); assert_eq!(transport.transport_type().name, "udp"); assert!(!transport.transport_type().connection_oriented); @@ -462,7 +482,7 @@ mod tests { #[test] fn test_sync_methods_return_not_supported() { let (tx, _rx) = packet_channel(100); - let mut transport = UdpTransport::new(TransportId::new(1), make_config(0), tx); + let mut transport = UdpTransport::new(TransportId::new(1), None, make_config(0), tx); assert!(matches!( transport.start(),