fix: heartbeat race condition, lock contention, add resilience layer, bump to v0.1.5

Author: Seggys116

CHANGELOG.md

@@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.1.5] - 2026-02-14
+
+### Added
+- Resilience module with circuit breaker to prevent infinite demux restart loops
+- Demux task watchdog for automatic recovery from task failures
+- Tunnel metrics tracking for observability (demux restarts, channel full events, frames dropped, lock wait times, heartbeat timeouts)
+- Configurable channel sizes for connection management (`connection_channel_size` in source and dest configs)
+- Circuit breaker configuration options (`circuit_breaker_window_secs`, `circuit_breaker_max_restarts`)
+- Dedicated writer task for lock-free frame sending via unbounded write queue
+
+### Fixed
+- Heartbeat race condition where pong could arrive after timeout check began, now uses atomic flag checked atomically before timeout
+- Potential deadlock in destination response channel by switching to unbounded channel (monitored via metrics for backpressure)
+- Lock contention during frame writes by implementing dedicated writer task that minimizes critical section to just write/flush operations
+- Clippy dead_code warnings by adding appropriate allow attributes to config fields and public API methods not yet used
+
+### Changed
+- Source demux task now takes ownership of tunnel read half for lock-free operation (no mutex on read path)
+- Connection channels now use configurable size (default 1024) instead of hardcoded 100
+- Heartbeat pong detection now atomic (flag cleared before ping sent, checked atomically after timeout)
+- Frame sending now uses unbounded write queue instead of direct writes (eliminates per-frame lock acquisition)
+
 ## [0.1.4] - 2026-02-13
 
 ### Added

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "utun"
-version = "0.1.4"
+version = "0.1.5"
 edition = "2021"
 authors = ["UTun Contributors"]
 description = "Quantum-safe tunnel system"

Cargo.toml

@@ -18,6 +18,7 @@ pub enum ConfigError {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct Config {
     pub source: Option<SourceConfig>,
     pub dest: Option<DestConfig>,
@@ -28,6 +29,7 @@ pub struct Config {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct SourceConfig {
     #[serde(default = "default_listen_ip")]
     pub listen_ip: String,
@@ -76,6 +78,17 @@ pub struct SourceConfig {
     #[serde(default = "default_frame_buffer_size")]
     pub frame_buffer_size: usize,
 
+    // Channel configuration
+    #[serde(default = "default_connection_channel_size")]
+    pub connection_channel_size: usize,
+
+    // Circuit breaker configuration
+    #[serde(default = "default_circuit_breaker_window_secs")]
+    pub circuit_breaker_window_secs: u64,
+
+    #[serde(default = "default_circuit_breaker_max_restarts")]
+    pub circuit_breaker_max_restarts: usize,
+
     #[serde(default)]
     pub allowed_outbound: AllowedOutboundConfig,
 
@@ -92,6 +105,7 @@ pub struct SourceConfig {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct DestConfig {
     #[serde(default = "default_listen_ip")]
     pub listen_ip: String,
@@ -109,6 +123,10 @@ pub struct DestConfig {
     #[serde(default = "default_target_timeout")]
     pub target_connect_timeout_ms: u64,
 
+    // Channel configuration
+    #[serde(default = "default_connection_channel_size")]
+    pub connection_channel_size: usize,
+
     #[serde(default)]
     pub connection_filter: ConnectionFilterConfig,
 
@@ -122,6 +140,7 @@ pub struct DestConfig {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct AuthConfig {
     #[serde(default = "default_mtls")]
     pub use_mtls: bool,
@@ -139,6 +158,7 @@ pub struct AuthConfig {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct CryptoConfig {
     #[serde(default = "default_kem_mode")]
     pub kem_mode: KemMode,
@@ -220,6 +240,7 @@ pub struct ExposedPortConfig {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct ServiceConfig {
     pub name: String,
     pub port: u16,
@@ -231,6 +252,7 @@ pub struct ServiceConfig {
 }
 
 impl ServiceConfig {
+    #[allow(dead_code)] // Public API method
     pub fn get_protocol(&self) -> crate::tunnel::Protocol {
         match self.protocol.to_lowercase().as_str() {
             "udp" => crate::tunnel::Protocol::Udp,
@@ -246,6 +268,7 @@ pub struct ConnectionFilterConfig {
 }
 
 #[derive(Debug, Deserialize)]
+#[allow(dead_code)] // Config fields loaded from TOML
 pub struct LoggingConfig {
     #[serde(default = "default_log_level")]
     pub level: String,
@@ -361,6 +384,15 @@ fn default_max_reconnect_delay() -> u64 {
 fn default_frame_buffer_size() -> usize {
     1000
 }
+fn default_connection_channel_size() -> usize {
+    1024
+}
+fn default_circuit_breaker_window_secs() -> u64 {
+    60
+}
+fn default_circuit_breaker_max_restarts() -> usize {
+    5
+}
 
 /// Load configuration from file
 pub fn load_config(path: &Path) -> Result<Config, ConfigError> {
@@ -455,6 +487,7 @@ fn validate_config(config: &Config) -> Result<(), ConfigError> {
 
 impl SourceConfig {
     /// Check if IP is allowed for outbound connections
+    #[allow(dead_code)] // Public API method
     pub fn is_ip_allowed(&self, ip: std::net::IpAddr) -> bool {
         for cidr in &self.allowed_outbound.allowed_ips {
             if let Ok(network) = cidr.parse::<IpNetwork>() {
@@ -469,6 +502,7 @@ impl SourceConfig {
 
 impl DestConfig {
     /// Check if source IP is allowed
+    #[allow(dead_code)] // Public API method
     pub fn is_source_allowed(&self, ip: std::net::IpAddr) -> bool {
         for cidr in &self.connection_filter.allowed_source_ips {
             if let Ok(network) = cidr.parse::<IpNetwork>() {

src/config.rs

@@ -4,6 +4,9 @@
 //! post-quantum cryptography (ML-KEM-768 + Classic McEliece 460896) for
 //! key exchange and AES-256-GCM for symmetric encryption.
 
+// Allow dead code for library API methods not yet used
+#![allow(dead_code)]
+
 pub mod cert;
 pub mod config;
 pub mod crypto;

src/lib.rs

@@ -1,3 +1,6 @@
+// Allow dead code for library API methods not yet used
+#![allow(dead_code)]
+
 use clap::Parser;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;

src/main.rs

@@ -16,6 +16,7 @@ pub enum ConnectionState {
     Closed,
 }
 
+#[allow(dead_code)] // Fields used by public API methods
 pub struct Connection {
     id: u32,
     remote_addr: SocketAddr,
@@ -36,8 +37,18 @@ impl Connection {
         _protocol: Protocol,
         _service_port: u16,
     ) -> (Self, mpsc::Sender<Frame>, mpsc::Receiver<Frame>) {
-        let (tx_to_tunnel, rx_to_tunnel) = mpsc::channel(100);
-        let (tx_from_tunnel, rx_from_tunnel) = mpsc::channel(100);
+        Self::new_with_channel_size(id, remote_addr, _protocol, _service_port, 1024)
+    }
+
+    pub fn new_with_channel_size(
+        id: u32,
+        remote_addr: SocketAddr,
+        _protocol: Protocol,
+        _service_port: u16,
+        channel_size: usize,
+    ) -> (Self, mpsc::Sender<Frame>, mpsc::Receiver<Frame>) {
+        let (tx_to_tunnel, rx_to_tunnel) = mpsc::channel(channel_size);
+        let (tx_from_tunnel, rx_from_tunnel) = mpsc::channel(channel_size);
 
         let conn = Self {
             id,
@@ -85,32 +96,38 @@ impl Connection {
         self.id
     }
 
+    #[allow(dead_code)] // Public API method
     pub fn remote_addr(&self) -> SocketAddr {
         self.remote_addr
     }
 
+    #[allow(dead_code)] // Public API method
     pub async fn set_service(&self, name: String, target: SocketAddr) {
         let mut sn = self.service_name.write().await;
         *sn = Some(name);
         let mut ta = self.target_addr.write().await;
         *ta = Some(target);
     }
 
+    #[allow(dead_code)] // Public API method
     pub async fn service_name(&self) -> Option<String> {
         self.service_name.read().await.clone()
     }
 
+    #[allow(dead_code)] // Public API method
     pub async fn target_addr(&self) -> Option<SocketAddr> {
         *self.target_addr.read().await
     }
 
+    #[allow(dead_code)] // Public API method
     pub async fn send_to_tunnel(&self, frame: Frame) -> Result<(), ConnectionError> {
         self.tx_to_tunnel
             .send(frame)
             .await
             .map_err(|_| ConnectionError::ChannelClosed)
     }
 
+    #[allow(dead_code)] // Public API method
     pub async fn recv_from_tunnel(&self) -> Result<Frame, ConnectionError> {
         let mut rx_guard = self.rx_from_tunnel.write().await;
         if let Some(ref mut rx) = *rx_guard {
@@ -166,15 +183,25 @@ pub struct ConnectionManager {
     connections: RwLock<HashMap<u32, Arc<Connection>>>,
     max_connections: usize,
     connection_timeout: Duration,
+    channel_size: usize,
 }
 
 impl ConnectionManager {
     pub fn new(max_connections: usize, connection_timeout_ms: u64) -> Self {
+        Self::new_with_channel_size(max_connections, connection_timeout_ms, 1024)
+    }
+
+    pub fn new_with_channel_size(
+        max_connections: usize,
+        connection_timeout_ms: u64,
+        channel_size: usize,
+    ) -> Self {
         Self {
             next_id: AtomicU32::new(1),
             connections: RwLock::new(HashMap::new()),
             max_connections,
             connection_timeout: Duration::from_millis(connection_timeout_ms),
+            channel_size,
         }
     }
 
@@ -194,7 +221,13 @@ impl ConnectionManager {
         }
 
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
-        let (conn, tx, rx) = Connection::new(id, remote_addr, protocol, service_port);
+        let (conn, tx, rx) = Connection::new_with_channel_size(
+            id,
+            remote_addr,
+            protocol,
+            service_port,
+            self.channel_size,
+        );
         let conn = Arc::new(conn);
 
         connections.insert(id, conn.clone());

src/tunnel/connection.rs

@@ -115,17 +115,22 @@ pub struct DestContainer {
     /// Maps connection_id -> target connection write half for data forwarding
     target_connections: Arc<RwLock<HashMap<u32, TargetConnection>>>,
     /// Channel to send response frames back to the tunnel
-    response_tx: Arc<RwLock<Option<mpsc::Sender<Frame>>>>,
+    response_tx: Arc<RwLock<Option<mpsc::UnboundedSender<Frame>>>>,
     /// Maximum handshake message size (depends on KEM mode)
     max_handshake_size: u32,
+    /// Channel size configuration
+    channel_size: usize,
 }
 
 impl DestContainer {
     pub async fn new(config: DestConfig, crypto_config: CryptoConfig) -> Result<Self, DestError> {
+        let channel_size = config.connection_channel_size;
+
         let key_manager = Arc::new(KeyManager::new(3600, 300)); // 1 hour rotation, 5 min window
-        let connection_manager = Arc::new(ConnectionManager::new(
+        let connection_manager = Arc::new(ConnectionManager::new_with_channel_size(
             config.max_connections_per_service * config.exposed_services.len(),
             config.connection_timeout_ms,
+            channel_size,
         ));
         let service_registry = Arc::new(ServiceRegistry::new(config.exposed_services.clone()));
 
@@ -149,6 +154,7 @@ impl DestContainer {
             target_connections: Arc::new(RwLock::new(HashMap::new())),
             response_tx: Arc::new(RwLock::new(None)),
             max_handshake_size,
+            channel_size,
         })
     }
 
@@ -211,14 +217,21 @@ impl DestContainer {
         drop(codec);
 
         // Create channel for response frames from target connections
-        let (response_tx, mut response_rx) = mpsc::channel::<Frame>(256);
+        // Use unbounded to prevent backpressure deadlock (monitored via metrics)
+        let (response_tx, mut response_rx) = mpsc::unbounded_channel::<Frame>();
 
         // Store the response sender
         {
             let mut tx = self.response_tx.write().await;
             *tx = Some(response_tx.clone());
         }
 
+        // Log channel configuration
+        tracing::info!(
+            "Response channel configured as unbounded for connection from {}",
+            addr
+        );
+
         // Handle incoming frames
         let connection_manager = self.connection_manager.clone();
         let service_registry = self.service_registry.clone();
@@ -326,7 +339,7 @@ impl DestContainer {
 
             // Send response if any (through the channel)
             if let Some(resp_frame) = response {
-                if response_tx.send(resp_frame).await.is_err() {
+                if response_tx.send(resp_frame).is_err() {
                     tracing::error!("Failed to send response - channel closed");
                     break;
                 }
@@ -460,7 +473,7 @@ impl DestContainer {
         connection_manager: &Arc<ConnectionManager>,
         service_registry: &Arc<ServiceRegistry>,
         metrics: &Arc<DestMetrics>,
-        response_tx: &mpsc::Sender<Frame>,
+        response_tx: &mpsc::UnboundedSender<Frame>,
     ) -> Option<Frame> {
         match frame.frame_type() {
             FrameType::Connect => {
@@ -544,7 +557,7 @@ impl DestContainer {
                                 if let Ok(mut close_frame) = Frame::new_data(connection_id, 0, &[])
                                 {
                                     close_frame.set_fin();
-                                    let _ = response_tx_clone.send(close_frame).await;
+                                    let _ = response_tx_clone.send(close_frame);
                                 }
                                 break;
                             }
@@ -553,7 +566,7 @@ impl DestContainer {
                                 conn_clone.record_receive(n);
                                 if let Ok(data_frame) = Frame::new_data(connection_id, 0, &buf[..n])
                                 {
-                                    if response_tx_clone.send(data_frame).await.is_err() {
+                                    if response_tx_clone.send(data_frame).is_err() {
                                         tracing::error!(
                                             "Failed to send data frame - channel closed"
                                         );
@@ -715,6 +728,7 @@ impl Clone for DestContainer {
             target_connections: self.target_connections.clone(),
             response_tx: self.response_tx.clone(),
             max_handshake_size: self.max_handshake_size,
+            channel_size: self.channel_size,
         }
     }
 }