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

Session 37: Transport-Node lifecycle integration

Browse Source 
- Added TransportHandle enum for polymorphic transport dispatch
- Node now owns transports via HashMap<TransportId, TransportHandle>
- 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<T> 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).
This commit is contained in:
Johnathan Corgan
2026-01-31 01:20:13 +00:00
parent d30865f60b
commit 4bbecccc11
5 changed files with 430 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config.rs
+202 -20
View File
@@ -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<String>,
/// 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<u16>,
}
@@ -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<T> {
/// Single unnamed instance (config fields directly under transport type).
Single(T),
/// Multiple named instances.
Named(HashMap<String, T>),
}
impl<T> TransportInstances<T> {
/// 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<Item = (Option<&str>, &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::<Vec<_>>()
.into_iter()
}
}
}
}
impl<T> Default for TransportInstances<T> {
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<UdpConfig>,
// Future transport types:
// #[serde(default, skip_serializing_if = "is_transport_empty")]
// pub tcp: TransportInstances<TcpConfig>,
//
// #[serde(default, skip_serializing_if = "is_transport_empty")]
// pub tor: TransportInstances<TorConfig>,
}
/// Helper for skip_serializing_if on TransportInstances.
fn is_transport_empty<T>(instances: &TransportInstances<T>) -> 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()));
}
}
src/lib.rs
+2 -2
View File
@@ -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;
src/node.rs
+155 -31
View File
@@ -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<TransportId>,
/// Active transports (owned by Node).
transports: HashMap<TransportId, TransportHandle>,
/// Active links.
links: HashMap<LinkId, Link>,
// === Packet Channel ===
/// Packet sender for transports.
packet_tx: Option<PacketTx>,
/// Packet receiver (for event loop).
packet_rx: Option<PacketRx>,
// === Peers ===
/// Authenticated peers.
peers: HashMap<NodeId, Peer>,
@@ -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<TransportHandle> {
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<Item = &TransportId> {
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]
src/transport/mod.rs
+71
View File
@@ -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<Vec<DiscoveredPeer>, 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<usize, TransportError> {
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
// ============================================================================
src/transport/udp.rs
-1
View File
@@ -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),
}