laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add transport instance naming and reorder node startup

Browse Source 
- 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
This commit is contained in:
Johnathan Corgan
2026-01-31 04:16:11 +00:00
parent 4bbecccc11
commit a317896367
3 changed files with 86 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node.rs
+34 -60
View File
@@ -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(())
src/transport/mod.rs
+14
View File
@@ -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<std::net::SocketAddr> {
match self {
TransportHandle::Udp(t) => t.local_addr(),
}
}
/// Check if transport is operational.
pub fn is_operational(&self) -> bool {
self.state().is_operational()
src/transport/udp.rs
+38 -18
View File
@@ -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<String>,
/// 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<String>,
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<SocketAddr> {
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(),