nineap/netrunner-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

diagnostics, recconections fix and bufferbloat fixes

Browse Source
This commit is contained in:
nineAp
2026-06-25 17:14:26 +07:00
parent b6056b2a66
commit 47208853c9
25 changed files with 1231 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/config.rs
+39 -13
View File
@@ -26,25 +26,51 @@ pub struct NetworkConfig {
pub udp_meta_light: usize,
}
/// MTU-aware defaults.
///
/// TCP socket buffers scale with the MTU so that the initial smoltcp window
/// always fits at least `WINDOW_SEGMENTS` back-to-back segments. Channel
/// capacity is sized to hold at most `CHANNEL_PACKETS` MTU-sized packets.
impl NetworkConfig {
pub fn new(system_mtu: usize) -> Self {
// Keep a minimum useful MTU even if the caller supplies something tiny.
let mtu = system_mtu.max(576);
// Minimum number of segments that must fit in a full TCP RX/TX buffer.
const BULK_WINDOW_SEGMENTS: usize = 128;
const LIGHT_WINDOW_SEGMENTS: usize = 32;
// How many MTU-sized packets the Tokio mpsc channel should hold.
// At MTU 1450 and 128 slots: ~185 KB per channel — enough to absorb
// ~15 ms of jitter at 100 Mbps without spawning backpressure tasks.
const CHANNEL_PACKETS: usize = 128;
// Payload bytes per segment (no IP/TCP headers in the smoltcp buffer).
let seg = mtu.saturating_sub(40).max(512); // subtract typical IP+TCP overhead
// Round up to the nearest 4 KB for alignment.
let round = |n: usize| ((n + 4095) / 4096) * 4096;
let tcp_heavy = round(seg * BULK_WINDOW_SEGMENTS);
let tcp_light = round(seg * LIGHT_WINDOW_SEGMENTS);
Self {
mtu: system_mtu,
connection_buf_size: 128 * 1024, // Уменьшили с 256KB
tcp_buffer_size: 128 * 1024, // Уменьшили с 256KB
udp_buffer_size: 64 * 1024, // Уменьшили со 128KB
tcp_chunk_size: 1024 * 64,
mtu,
connection_buf_size: tcp_heavy,
tcp_buffer_size: tcp_heavy,
udp_buffer_size: tcp_light,
// Read chunks up to one smoltcp frame payload; larger values just
// add latency without improving throughput.
tcp_chunk_size: 64 * 1024,
channel_capacity: 16,
channel_capacity: CHANNEL_PACKETS,
// Окна smoltcp (уменьшаем, чтобы не создавать огромные очереди)
tcp_rx_heavy: 128 * 1024,
tcp_tx_heavy: 128 * 1024,
tcp_rx_heavy: tcp_heavy,
tcp_tx_heavy: tcp_heavy,
tcp_rx_light: tcp_light,
tcp_tx_light: tcp_light,
tcp_rx_light: 32 * 1024, // Уменьшили с 64KB
tcp_tx_light: 32 * 1024,
udp_buf_heavy: 128 * 1024,
udp_buf_heavy: tcp_heavy,
udp_meta_heavy: 512,
udp_buf_light: 16 * 1024,
udp_meta_light: 32,
core/src/net/connection/connection.rs
+41 -22
View File
@@ -8,9 +8,10 @@ use crate::{
handler::{RemoteOpener, StreamHandler},
muxer::{MuxMessage, Muxer},
},
NetworkConfig, FALLBACK_CONNECT_TIMEOUT, LEG_RECONNECT_DELAY, LEG_STAGGER_DELAY,
MAX_TUNNEL_LEGS, SECURE_HANDSHAKE_TIMEOUT, STEALTH_FALLBACK_HOST, TLS_HELLO_TIMEOUT,
TOPOLOGY_PRINT_INTERVAL,
DNS_LOOKUP_TIMEOUT, FALLBACK_CONNECT_TIMEOUT, LEG_RECONNECT_DELAY, LEG_STAGGER_DELAY,
MAX_TUNNEL_LEGS, NETWORK_WATCHER_INTERVAL, SECURE_HANDSHAKE_TIMEOUT,
SESSION_CLEANUP_DELAY, STEALTH_FALLBACK_HOST, STEALTH_FALLBACK_SNI, TLS_HELLO_TIMEOUT,
TOPOLOGY_PRINT_INTERVAL, NetworkConfig,
},
nrxp::{Codec, Frame, FrameType, TlsBridge},
rawcast::{LocalProtocol, RawCastAdapter, RawCastFrame},
@@ -126,7 +127,7 @@ impl ClientHandler {
let mut conn = Connection::new(stream);
let mut session_keys = SessionKeys::new(true);
let ch =
TlsBridge::wrap_client_hello(&BrowserProfile::CHROME_131, "ubuntu.com", &session_keys);
TlsBridge::wrap_client_hello(&BrowserProfile::CHROME_131, STEALTH_FALLBACK_SNI, &session_keys);
conn.outbound
.write_all(&ch)
@@ -141,7 +142,7 @@ impl ClientHandler {
}
Ok(None) => {
let res = tokio::time::timeout(
std::time::Duration::from_secs(5),
TLS_HELLO_TIMEOUT,
conn.inbound.read_buf(&mut conn.read_buf),
)
.await;
@@ -214,7 +215,7 @@ impl ClientHandler {
let leg_name = format!("TCP-Leg-{}", leg_id);
let addrs_future = tokio::net::lookup_host(remote_proxy_addr);
let mut addrs = tokio::time::timeout(std::time::Duration::from_secs(3), addrs_future)
let mut addrs = tokio::time::timeout(DNS_LOOKUP_TIMEOUT, addrs_future)
.await
.map_err(|_| {
AppError::new(ERR_INFRA_TIMEOUT, "Сервер недоступен", "DNS Lookup Timeout")
@@ -230,7 +231,7 @@ impl ClientHandler {
})?;
let stream = tokio::time::timeout(
std::time::Duration::from_secs(5),
FALLBACK_CONNECT_TIMEOUT,
tokio::net::TcpStream::connect(addr),
)
.await
@@ -250,7 +251,6 @@ impl ClientHandler {
let (control_tx, control_rx) = mpsc::channel::<MuxMessage>(cap);
let (data_tx, data_rx) = mpsc::channel::<MuxMessage>(cap);
let control_tx_clone = control_tx.clone();
muxer.add_leg(leg_id, control_tx, data_tx);
let handler = Arc::new(StreamHandler::new(muxer.clone(), None));
@@ -272,7 +272,9 @@ impl ClientHandler {
};
let run_result = engine.run().await;
muxer.remove_leg(leg_id, &control_tx_clone);
// Use force_remove because the engine may have re-registered the leg internally
// (via reconnect + add_leg), making control_tx_clone stale for same_channel comparison.
muxer.force_remove_leg(leg_id);
run_result?;
Err(AppError::new(
@@ -286,7 +288,7 @@ impl ClientHandler {
remote_proxy_addr: &str,
mut rx_from_engine: mpsc::Receiver<RawCastFrame>,
tx_to_engine: mpsc::Sender<RawCastFrame>,
) -> Result<(), AppError> {
) -> Result<Arc<Muxer>, AppError> {
let session_id = SessionManager::generate_id();
let muxer = Arc::new(Muxer::new(true, session_id.clone()));
let registry: Arc<DashMap<u32, (u64, Ipv4Addr, u16, LocalProtocol)>> =
@@ -297,7 +299,7 @@ impl ClientHandler {
let watcher_muxer = muxer.clone();
tokio::spawn(async move {
let mut last_ip = Self::get_local_ip();
let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
let mut interval = tokio::time::interval(NETWORK_WATCHER_INTERVAL);
loop {
interval.tick().await;
let current_ip = Self::get_local_ip();
@@ -321,10 +323,31 @@ impl ClientHandler {
let sid = session_id.clone();
tokio::spawn(async move {
tokio::time::sleep(LEG_STAGGER_DELAY * id).await;
let mut attempt: u32 = 0;
loop {
if let Err(e) = Self::establish_leg(&addr, id, m.clone(), &sid).await {
attempt += 1;
error!("Leg {} disconnected: {}. Reconnecting in 2s...", id, e);
let rtt = crate::net::GLOBAL_MIN_RTT.load(std::sync::atomic::Ordering::Relaxed);
crate::net::diagnostics::DIAG_COUNTERS
.leg_disconnects
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
crate::net::diagnostics::send_diag_event(
crate::net::diagnostics::DiagnosticsEvent::LegDisconnected {
leg_id: id,
rtt_ms: rtt,
reason: e.to_string(),
},
);
tokio::time::sleep(LEG_RECONNECT_DELAY).await;
crate::net::diagnostics::send_diag_event(
crate::net::diagnostics::DiagnosticsEvent::LegReconnecting {
leg_id: id,
attempt,
},
);
} else {
attempt = 0;
}
}
});
@@ -431,15 +454,11 @@ impl ClientHandler {
});
}
FrameType::Data | FrameType::UdpData => {
if let Some(up_tx) = local_to_upload_tx.get(&local_socket_id) {
if let Err(mpsc::error::TrySendError::Full(_)) =
up_tx.try_send(payload)
{
netrunner_logger::trace!(
local_socket_id,
"⚠️ Upload stream buffer full, dropping frame (TCP will retransmit)"
);
}
if let Some(up_tx) = local_to_upload_tx.get(&local_socket_id).map(|r| r.value().clone()) {
// .send().await blocks the upload task when the muxer leg
// is saturated, creating back-pressure back to smoltcp
// (no drops → no unnecessary retransmits → lower jitter).
let _ = up_tx.send(payload).await;
}
}
FrameType::Close => {
@@ -468,7 +487,7 @@ impl ClientHandler {
}
});
Ok(())
Ok(muxer)
}
}
@@ -685,7 +704,7 @@ impl TunnelHandler for ServerHandler {
let sid = log_session_id;
let m = muxer.clone();
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(120)).await;
tokio::time::sleep(SESSION_CLEANUP_DELAY).await;
if m.active_legs_count() == 0 {
sm.remove(&sid);
}
core/src/net/connection/engine.rs
+40 -10
View File
@@ -15,7 +15,9 @@ use tracing::instrument;
use crate::{
net::{
connection::{handler::StreamHandler, muxer::MuxMessage},
NetworkConfig, HEALTH_CHECK_INTERVAL,
NetworkConfig, HEALTH_CHECK_INTERVAL, FALLBACK_CONNECT_TIMEOUT,
RECONNECT_BACKOFF_BASE, RECONNECT_BACKOFF_JITTER_MS,
TUNNEL_INTERLEAVE_CHUNK, TUNNEL_MAX_BUFFER_SIZE, TUNNEL_READ_RESERVE,
},
nrxp::{ErrorAction, FrameType, RxCodec, TxCodec, MAX_FRAME_PAYLOAD},
};
@@ -49,7 +51,7 @@ impl TunnelEngine {
) -> Result<(OwnedReadHalf, OwnedWriteHalf, RxCodec, TxCodec), AppError> {
info!("🔄 Attempting reconnect to {}", self.remote_addr);
let stream = tokio::time::timeout(
tokio::time::Duration::from_secs(5),
FALLBACK_CONNECT_TIMEOUT,
tokio::net::TcpStream::connect(&self.remote_addr),
)
.await
@@ -81,6 +83,14 @@ impl TunnelEngine {
self.leg_status = LegStatus::Reconnecting;
match self.attempt_reconnect().await {
Ok((new_in, new_out, new_rx, new_tx)) => {
let cap = crate::net::NetworkConfig::global().channel_capacity;
let (control_tx, control_rx) =
tokio::sync::mpsc::channel::<MuxMessage>(cap);
let (data_tx, data_rx) = tokio::sync::mpsc::channel::<MuxMessage>(cap);
self.muxer.add_leg(self.leg_id, control_tx, data_tx);
self.control_rx = Some(control_rx);
self.data_rx = Some(data_rx);
self.inbound = Some(new_in);
self.outbound = Some(new_out);
self.rx_codec = Some(new_rx);
@@ -90,8 +100,8 @@ impl TunnelEngine {
}
Err(e) => {
error!("Reconnect failed for leg {}: {}", self.leg_id, e);
let jitter = rand::random::<u64>() % 1000;
tokio::time::sleep(tokio::time::Duration::from_millis(2000 + jitter)).await;
let jitter = rand::random::<u64>() % RECONNECT_BACKOFF_JITTER_MS;
tokio::time::sleep(RECONNECT_BACKOFF_BASE + tokio::time::Duration::from_millis(jitter)).await;
continue;
}
}
@@ -119,10 +129,8 @@ impl TunnelEngine {
let mut reader_handle = tokio::spawn(async move {
let mut read_buf = read_buf;
let mut inbound = inbound;
const MAX_BUFFER_SIZE: usize = 1024 * 1024; // 1 MB
loop {
if read_buf.len() > MAX_BUFFER_SIZE {
if read_buf.len() > TUNNEL_MAX_BUFFER_SIZE {
error!("CRITICAL: Read buffer exceeded 1MB (OOM Protection). Dropping connection!");
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
@@ -134,7 +142,7 @@ impl TunnelEngine {
if read_buf.is_empty() {
read_buf.clear();
}
read_buf.reserve(16384);
read_buf.reserve(TUNNEL_READ_RESERVE);
tokio::select! {
_ = token_reader.cancelled() => {
@@ -184,7 +192,7 @@ impl TunnelEngine {
let mut heartbeat = tokio::time::interval(HEALTH_CHECK_INTERVAL);
let mut pending_data: Option<MuxMessage> = None;
const INTERLEAVE_CHUNK: usize = 16384;
let interleave_chunk = TUNNEL_INTERLEAVE_CHUNK;
loop {
tokio::select! {
@@ -196,13 +204,24 @@ impl TunnelEngine {
muxer_pong.record_ping_sent(leg_id);
let msg = MuxMessage { stream_id: 0, frame_type: FrameType::Heartbeat, data: Bytes::new() };
if let Err(e) = Self::handle_outbound(&mut outbound, &mut tx_codec, msg).await {
crate::net::diagnostics::send_diag_event(
crate::net::diagnostics::DiagnosticsEvent::TunnelWriteStuck {
leg_id, stream_id: 0,
},
);
return Err((e, control_rx, data_rx, tx_codec));
}
}
msg_opt = control_rx.recv() => {
if let Some(msg) = msg_opt {
let sid = msg.stream_id;
if let Err(e) = Self::handle_outbound(&mut outbound, &mut tx_codec, msg).await {
crate::net::diagnostics::send_diag_event(
crate::net::diagnostics::DiagnosticsEvent::TunnelWriteStuck {
leg_id, stream_id: sid,
},
);
return Err((e, control_rx, data_rx, tx_codec));
}
} else { break; }
@@ -212,7 +231,7 @@ impl TunnelEngine {
_ = std::future::ready(()), if pending_data.is_some() => {
let mut msg = pending_data.take().unwrap();
let chunk_size = std::cmp::min(msg.data.len(), INTERLEAVE_CHUNK);
let chunk_size = std::cmp::min(msg.data.len(), interleave_chunk);
let chunk_data = msg.data.split_to(chunk_size);
let chunk_msg = MuxMessage {
@@ -220,8 +239,14 @@ impl TunnelEngine {
frame_type: msg.frame_type.clone(),
data: chunk_data,
};
let chunk_sid = chunk_msg.stream_id;
if let Err(e) = Self::handle_outbound(&mut outbound, &mut tx_codec, chunk_msg).await {
crate::net::diagnostics::send_diag_event(
crate::net::diagnostics::DiagnosticsEvent::TunnelWriteStuck {
leg_id, stream_id: chunk_sid,
},
);
return Err((e, control_rx, data_rx, tx_codec));
}
@@ -366,6 +391,11 @@ impl TunnelEngine {
tokio::time::timeout(std::time::Duration::from_secs(20), write_future).await
{
error!(stream_id, "🔥 Physical leg STUCK on write. Killing leg.");
// Increment counter; the call site in run() emits the full event
// with the correct leg_id since handle_outbound is a static fn.
crate::net::diagnostics::DIAG_COUNTERS
.tunnel_write_stalls
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут отправки",
core/src/net/connection/handler.rs
+2 -2
View File
@@ -141,8 +141,8 @@ impl StreamHandler {
}
FrameType::Data | FrameType::UdpData => {
// Полностью синхронный вызов
self.muxer.dispatch_to_local(stream_id, frame.payload);
// MUST .await — maintains in-order delivery via back-pressure.
self.muxer.dispatch_to_local(stream_id, frame.payload).await;
}
FrameType::Close => {
core/src/net/connection/mod.rs
+1 -1
View File
@@ -5,4 +5,4 @@ mod handler;
mod muxer;
pub use connection::{ClientHandler, Connection, ServerHandler, SessionManager, TunnelHandler};
pub use muxer::GLOBAL_MIN_RTT;
pub use muxer::{Muxer, GLOBAL_MIN_RTT};
core/src/net/connection/muxer.rs
+141 -53
View File
@@ -7,7 +7,9 @@ use std::time::Instant;
use tokio::sync::mpsc::Sender;
use tokio_util::sync::CancellationToken;
use crate::net::{INITIAL_RTT_MS, MUXER_CONGESTION_WEIGHT};
use crate::net::{HEALTH_CHECK_TIMEOUT, MAX_TUNNEL_LEGS, MUXER_POOL_SIZE};
use crate::net::diagnostics::{self, DiagnosticsEvent, DIAG_COUNTERS, LegMetrics, TunnelMetrics};
use crate::nrxp::FrameType;
#[derive(Default, Debug)]
@@ -62,7 +64,7 @@ pub struct MuxMessage {
pub(crate) data: Bytes,
}
pub static GLOBAL_MIN_RTT: AtomicU32 = AtomicU32::new(250);
pub static GLOBAL_MIN_RTT: AtomicU32 = AtomicU32::new(INITIAL_RTT_MS);
#[derive(Clone)]
pub struct Muxer {
@@ -131,7 +133,7 @@ impl Muxer {
.map_or(false, |leg| leg.control_tx.same_channel(tx));
if should_remove {
self.legs.remove(&leg_id);
self.update_legs_cache(); // Обновляем Lock-Free кэш
self.update_legs_cache();
info!(
leg_id,
"MUXER: TCP leg removed safely, streams will re-balance"
@@ -139,6 +141,13 @@ impl Muxer {
}
}
pub fn force_remove_leg(&self, leg_id: u32) {
if self.legs.remove(&leg_id).is_some() {
self.update_legs_cache();
info!(leg_id, "MUXER: TCP leg force-removed on engine exit");
}
}
pub fn remove_all_legs(&self) {
self.legs.clear();
self.stream_bindings.clear();
@@ -174,9 +183,9 @@ impl Muxer {
.take(pool_size) // Берем только пул
.min_by(|a, b| {
let score_a = a.stats.rtt_ms.load(Ordering::Relaxed) as f64
+ (a.congestion_factor() * 2000.0);
+ (a.congestion_factor() * MUXER_CONGESTION_WEIGHT);
let score_b = b.stats.rtt_ms.load(Ordering::Relaxed) as f64
+ (b.congestion_factor() * 2000.0);
+ (b.congestion_factor() * MUXER_CONGESTION_WEIGHT);
score_a
.partial_cmp(&score_b)
.unwrap_or(std::cmp::Ordering::Equal)
@@ -244,36 +253,61 @@ impl Muxer {
Ok(())
}
Err(_) => {
DIAG_COUNTERS.upload_fails.fetch_add(1, Ordering::Relaxed);
diagnostics::send_diag_event(DiagnosticsEvent::UploadFailed {
stream_id,
reason: "data channel closed (leg dropped)".into(),
});
self.remove_leg(leg.id, &leg.control_tx);
Err(AppError::new(ERR_INFRA_TIMEOUT, "Обрыв", "Leg closed"))
}
}
} else {
// 💡 КОНТРОЛЬ: Используем .try_send() для мгновенной приоритетной отправки (Non-blocking)
match leg.control_tx.try_send(message) {
Ok(_) => {
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
if let Some(stream_ref) = self.streams.get(&stream_id) {
stream_ref
.value()
.1
.tx_bytes
.fetch_add(size, Ordering::Relaxed);
// Close and Heartbeat frames MUST be delivered reliably (.send().await).
// Close: dropping it leaks stream resources.
// Heartbeat (PONG): dropping it via try_send causes the health-check
// probe to time out after HEALTH_CHECK_TIMEOUT and evict a live leg.
let is_critical = matches!(message.frame_type, FrameType::Close | FrameType::Heartbeat);
if is_critical {
match leg.control_tx.send(message).await {
Ok(_) => {
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
if let Some(stream_ref) = self.streams.get(&stream_id) {
stream_ref.value().1.tx_bytes.fetch_add(size, Ordering::Relaxed);
}
Ok(())
}
Err(_) => {
self.remove_leg(leg.id, &leg.control_tx);
Err(AppError::new(ERR_INFRA_TIMEOUT, "Обрыв", "Leg closed"))
}
Ok(())
}
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
// Канал контроля не должен забиваться. Если это произошло, пакет сбрасывается,
// чтобы предотвратить зависание критических задач.
netrunner_logger::warn!(
stream_id,
"Control queue FULL! Dropping control frame to avoid deadlock."
);
Ok(())
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.remove_leg(leg.id, &leg.control_tx);
Err(AppError::new(ERR_INFRA_TIMEOUT, "Обрыв", "Leg closed"))
} else {
match leg.control_tx.try_send(message) {
Ok(_) => {
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
if let Some(stream_ref) = self.streams.get(&stream_id) {
stream_ref.value().1.tx_bytes.fetch_add(size, Ordering::Relaxed);
}
Ok(())
}
Err(tokio::sync::mpsc::error::TrySendError::Full(ref dropped)) => {
netrunner_logger::warn!(
stream_id,
"Control queue FULL! Dropping non-critical control frame."
);
DIAG_COUNTERS.control_full_drops.fetch_add(1, Ordering::Relaxed);
diagnostics::send_diag_event(DiagnosticsEvent::ControlChannelFull {
stream_id,
frame_type: format!("{:?}", dropped.frame_type),
});
Ok(())
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.remove_leg(leg.id, &leg.control_tx);
Err(AppError::new(ERR_INFRA_TIMEOUT, "Обрыв", "Leg closed"))
}
}
}
}
@@ -328,8 +362,11 @@ impl Muxer {
self.stream_bindings.remove(&stream_id);
}
#[inline]
pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
// ORDERING CONTRACT: callers MUST .await this; the caller (TunnelEngine reader)
// is a spawned task, so blocking here creates correct back-pressure all the way
// back to the kernel TCP socket buffer. Never spawn a task to deliver data
// from this function — that breaks in-order delivery guarantees.
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
let size = data.len() as u64;
let tx_and_stats = self.streams.get(&stream_id).map(|s| {
@@ -338,24 +375,10 @@ impl Muxer {
});
if let Some((tx, stats)) = tx_and_stats {
match tx.try_send(data) {
Ok(_) => {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
}
Err(tokio::sync::mpsc::error::TrySendError::Full(data)) => {
// Channel is full: spawn a task to wait for space.
// This keeps the reader loop unblocked while providing
// backpressure — the spawned future will be pending until
// TcpConnection drains the channel and makes room.
tokio::spawn(async move {
if tx.send(data).await.is_ok() {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
}
});
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
// Stream is gone; nothing to do.
}
// .send().await blocks until the receiver has space.
// If the receiver is closed the error is silently ignored.
if tx.send(data).await.is_ok() {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
}
}
}
@@ -391,17 +414,46 @@ impl Muxer {
frame_type: FrameType::Heartbeat,
data: Bytes::from("PING"),
};
if tx.try_send(msg).is_err() {
self.remove_leg(leg_id, &tx);
self.remove_stream(probe_stream_id);
continue;
match tx.try_send(msg) {
Ok(_) => {}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
// Writer is already dead — evict immediately without waiting 20s.
warn!(leg_id, "Health check: control channel closed, evicting dead leg");
self.remove_leg(leg_id, &tx);
self.remove_stream(probe_stream_id);
continue;
}
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
// Control channel is temporarily full: the writer is alive and
// busy processing other frames. Skip this probe cycle — evicting
// a healthy leg because its queue is momentarily saturated would
// cause a spurious reconnect.
self.remove_stream(probe_stream_id);
continue;
}
}
match tokio::time::timeout(crate::net::HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => trace!(leg_id, "✅ TCP Leg Health Check OK"),
_ => {
warn!(leg_id, "❌ TCP Leg Health Check FAIL/Timeout - Evicting");
self.remove_leg(leg_id, &tx);
// Before evicting, verify the muxer still holds the same control_tx
// we probed with. After an internal reconnect, add_leg replaces the
// entry with new channels, and the old probe belongs to a dead leg
// that the engine has already recycled — evicting the new leg here
// would be wrong.
let still_same = self
.legs
.get(&leg_id)
.map_or(false, |l| l.control_tx.same_channel(&tx));
if still_same {
warn!(leg_id, "❌ TCP Leg Health Check FAIL/Timeout - Evicting");
self.remove_leg(leg_id, &tx);
} else {
netrunner_logger::debug!(
leg_id,
"Health check probe timed out but leg already reconnected — skipping eviction"
);
}
}
}
self.remove_stream(probe_stream_id);
@@ -507,4 +559,40 @@ impl Muxer {
}
info!("\n{}", out);
}
/// Collect a point-in-time snapshot of tunnel metrics for diagnostics.
/// Lock-free: reads only atomics and the RwLock-protected legs cache.
pub fn snapshot_tunnel_metrics(&self) -> TunnelMetrics {
let global_min_rtt = crate::net::GLOBAL_MIN_RTT.load(Ordering::Relaxed);
let cached_legs = self.active_legs_cache.read().unwrap().clone();
let active_legs: Vec<LegMetrics> = cached_legs
.iter()
.map(|leg| {
let cap = leg.data_tx.max_capacity();
let free = leg.data_tx.capacity();
let filled = cap.saturating_sub(free);
let congestion_factor = if cap > 0 {
filled as f64 / cap as f64
} else {
0.0
};
LegMetrics {
leg_id: leg.id,
rtt_ms: leg.stats.rtt_ms.load(Ordering::Relaxed),
tx_mb: leg.stats.tx_bytes.load(Ordering::Relaxed) as f64 / 1_048_576.0,
rx_mb: leg.stats.rx_bytes.load(Ordering::Relaxed) as f64 / 1_048_576.0,
congestion_factor,
data_channel_free: free,
data_channel_capacity: cap,
}
})
.collect();
TunnelMetrics {
global_min_rtt_ms: global_min_rtt,
active_legs,
total_streams: self.streams.len(),
}
}
}
core/src/net/constants.rs
+57 -14
View File
@@ -1,32 +1,75 @@
use std::time::Duration;
// ── Connection pool ──────────────────────────────────────────────────────────
pub const MAX_SOCKETS: usize = 256;
pub const MAX_TUNNEL_LEGS: u32 = 4;
pub const MUXER_POOL_SIZE: usize = 3;
/// Weight applied to observed congestion when scoring tunnel legs.
pub const MUXER_CONGESTION_WEIGHT: f64 = 2000.0;
/// Initial RTT estimate used before any real measurement arrives.
pub const INITIAL_RTT_MS: u32 = 250;
// ── Timeouts ─────────────────────────────────────────────────────────────────
pub const TCP_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);
pub const UDP_IDLE_TIMEOUT: Duration = Duration::from_secs(15);
pub const GLOBAL_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
// 🔥 ФИКС: Ускоряем обнаружение мертвой сети при переключении Wi-Fi -> LTE
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(3); // Было 7
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(20); // Было 10
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(2); // Было 3
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(3);
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(20);
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(2);
pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
pub const TLS_HELLO_TIMEOUT: Duration = Duration::from_secs(10);
pub const SECURE_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);
pub const FALLBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
/// Timeout for resolving a proxy address via DNS.
pub const DNS_LOOKUP_TIMEOUT: Duration = Duration::from_secs(3);
/// Delay between leg reconnect attempts (base); actual delay adds random jitter.
pub const RECONNECT_BACKOFF_BASE: Duration = Duration::from_millis(2000);
/// Upper bound of the random jitter added to `RECONNECT_BACKOFF_BASE`.
pub const RECONNECT_BACKOFF_JITTER_MS: u64 = 1000;
/// How long to wait before removing an idle session after all legs drop.
pub const SESSION_CLEANUP_DELAY: Duration = Duration::from_secs(120);
/// How often the network-change watcher checks the local IP address.
pub const NETWORK_WATCHER_INTERVAL: Duration = Duration::from_secs(1);
pub const DNS_PORT: u16 = 53;
pub const HTTP_PORT: u16 = 80;
pub const HTTPS_PORT: u16 = 443;
pub const NETBIOS_PORTS: [u16; 2] = [137, 138];
// ── Logging ───────────────────────────────────────────────────────────────────
pub const LEG_STAGGER_DELAY: Duration = Duration::from_millis(1000);
pub const TOPOLOGY_PRINT_INTERVAL: Duration = Duration::from_secs(10);
/// How often the client engine logs traffic statistics.
pub const STATS_LOG_INTERVAL: Duration = Duration::from_secs(5);
// ── Authentication ───────────────────────────────────────────────────────────
pub const AUTH_TIME_STEP: u64 = 60;
pub const AUTH_WINDOW_SIZE: u64 = 2;
pub const LEG_STAGGER_DELAY: Duration = Duration::from_millis(1000); // Чуть ускорили старт
pub const TOPOLOGY_PRINT_INTERVAL: Duration = Duration::from_secs(10);
// ── Well-known ports ─────────────────────────────────────────────────────────
pub const DNS_PORT: u16 = 53;
pub const HTTP_PORT: u16 = 80;
pub const HTTPS_PORT: u16 = 443;
pub const HTTP_ALT_PORT: u16 = 8080;
pub const SSH_PORT: u16 = 22;
pub const RDP_PORT: u16 = 3389;
pub const VNC_PORT: u16 = 5900;
pub const RTMP_PORT: u16 = 1935;
pub const NTP_PORT: u16 = 123;
pub const NETBIOS_PORTS: [u16; 2] = [137, 138];
// ── TLS / stealth ────────────────────────────────────────────────────────────
/// Hostname used as the SNI in the stealth TLS ClientHello.
pub const STEALTH_FALLBACK_SNI: &str = "ubuntu.com";
pub const STEALTH_FALLBACK_HOST: &str = "ubuntu.com:443";
pub const FALLBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
pub const TLS_HELLO_TIMEOUT: Duration = Duration::from_secs(10);
pub const SECURE_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);
// ── Tunnel frame codec ───────────────────────────────────────────────────────
/// OOM guard: drop the leg if the read buffer grows past this.
pub const TUNNEL_MAX_BUFFER_SIZE: usize = 1024 * 1024;
/// Bytes reserved in the read buffer before each `read_buf` call.
pub const TUNNEL_READ_RESERVE: usize = 16 * 1024;
/// Maximum bytes written per stream in a single interleaved write pass.
pub const TUNNEL_INTERLEAVE_CHUNK: usize = 16 * 1024;
// ── Smoltcp socket defaults ──────────────────────────────────────────────────
/// Packet slots for the ICMP socket's RX and TX packet buffers.
pub const ICMP_META_SLOTS: usize = 4;
/// Byte capacity of the ICMP socket's RX and TX data buffers.
pub const ICMP_BUFFER_SIZE: usize = 512;
/// Log a bufferbloat warning when the application-layer queue exceeds this.
pub const BUFFERBLOAT_WARN_THRESHOLD: usize = 1024 * 1024;
core/src/net/diagnostics.rs
+247
View File
@@ -0,0 +1,247 @@
use std::{
collections::VecDeque,
sync::{
OnceLock,
atomic::{AtomicU64, Ordering},
Mutex,
},
time::{SystemTime, UNIX_EPOCH},
};
use serde::Serialize;
use tokio::sync::mpsc;
// ── Public event channel ──────────────────────────────────────────────────────
/// Global sender end of the diagnostics event channel.
/// Initialised once by `init_diagnostics()`; all producers call `send_diag_event()`.
static GLOBAL_DIAG_TX: OnceLock<DiagnosisTx> = OnceLock::new();
pub type DiagnosisTx = mpsc::UnboundedSender<DiagnosticsEvent>;
pub type DiagnosisRx = mpsc::UnboundedReceiver<DiagnosticsEvent>;
/// Call once at startup (from EngineBuilder on the client, from Network::run on
/// the server). Returns the receiver end that the consumer task/engine must hold.
pub fn init_diagnostics() -> DiagnosisRx {
let (tx, rx) = mpsc::unbounded_channel();
// Ignore the error if called twice (server may re-init across tests).
let _ = GLOBAL_DIAG_TX.set(tx);
rx
}
/// Fire-and-forget: enqueue a diagnostics event from anywhere in the codebase.
/// Does nothing if `init_diagnostics()` has not been called yet.
pub fn send_diag_event(event: DiagnosticsEvent) {
if let Some(tx) = GLOBAL_DIAG_TX.get() {
let _ = tx.send(event);
}
}
// ── Event types ───────────────────────────────────────────────────────────────
/// What triggered this diagnostics snapshot.
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum DiagnosticsEvent {
UploadFailed {
stream_id: u32,
reason: String,
},
DownloadBackpressure {
stream_id: u32,
/// How many background retry tasks are in flight for this stream.
queued_tasks: usize,
},
LegDisconnected {
leg_id: u32,
rtt_ms: u32,
reason: String,
},
LegReconnecting {
leg_id: u32,
attempt: u32,
},
StreamClosedWithError {
stream_id: u32,
up_bytes: u64,
down_bytes: u64,
error: String,
},
ControlChannelFull {
stream_id: u32,
frame_type: String,
},
TunnelWriteStuck {
leg_id: u32,
stream_id: u32,
},
}
// ── Per-snapshot sub-structs (all Serialize) ──────────────────────────────────
#[derive(Debug, Clone, Serialize)]
pub struct EngineMetrics {
pub rx_total_mb: f64,
pub tx_total_mb: f64,
pub rx_speed_mb_s: f64,
pub tx_speed_mb_s: f64,
pub rx_packets: u64,
pub tx_packets: u64,
/// Current depth of the smoltcp ChannelDevice RX queue (packets).
pub device_rx_queue_depth: usize,
/// Current depth of the smoltcp ChannelDevice TX queue (packets).
pub device_tx_queue_depth: usize,
/// Free slots in the engine→TUN writer channel.
pub tun_tx_channel_free: usize,
/// Free slots in the TUN reader→engine channel.
pub tun_rx_channel_free: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct LegMetrics {
pub leg_id: u32,
pub rtt_ms: u32,
pub tx_mb: f64,
pub rx_mb: f64,
/// 0.0 = idle, 1.0 = data channel completely full.
pub congestion_factor: f64,
pub data_channel_free: usize,
pub data_channel_capacity: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct TunnelMetrics {
pub global_min_rtt_ms: u32,
pub active_legs: Vec<LegMetrics>,
pub total_streams: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct SocketMetrics {
pub stream_id: u32,
pub state: String,
pub send_queue_bytes: usize,
pub send_capacity_bytes: usize,
pub recv_queue_bytes: usize,
pub recv_capacity_bytes: usize,
/// Bytes held in the single pending_chunk (partial write to smoltcp TX buf).
pub pending_chunk_bytes: usize,
/// Whether the upload path is currently blocked by smoltcp backpressure.
pub tx_congested: bool,
pub total_up_bytes: u64,
pub total_down_bytes: u64,
}
/// Cumulative error/event counters at the moment of the snapshot.
#[derive(Debug, Clone, Serialize, Default)]
pub struct ErrorCounters {
pub upload_fails: u64,
pub download_backpressure_events: u64,
pub leg_disconnects: u64,
pub control_channel_full_drops: u64,
pub tunnel_write_stalls: u64,
pub stream_errors: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct DiagnosticsSnapshot {
pub timestamp_ms: u64,
pub trigger: DiagnosticsEvent,
/// Engine-side metrics (traffic, device queue depths, channel free space).
/// Present only on the client; `None` on server snapshots.
pub engine: Option<EngineMetrics>,
pub tunnel: TunnelMetrics,
/// Per-socket smoltcp state. Present only on the client engine.
pub sockets: Vec<SocketMetrics>,
/// Running totals of all error/event counters up to this snapshot.
pub error_totals: ErrorCounters,
}
// ── Atomic error counters (global, updated at event sites) ───────────────────
pub struct DiagnosticsCounters {
pub upload_fails: AtomicU64,
pub download_backpressure: AtomicU64,
pub leg_disconnects: AtomicU64,
pub control_full_drops: AtomicU64,
pub tunnel_write_stalls: AtomicU64,
pub stream_errors: AtomicU64,
}
impl DiagnosticsCounters {
const fn new() -> Self {
Self {
upload_fails: AtomicU64::new(0),
download_backpressure: AtomicU64::new(0),
leg_disconnects: AtomicU64::new(0),
control_full_drops: AtomicU64::new(0),
tunnel_write_stalls: AtomicU64::new(0),
stream_errors: AtomicU64::new(0),
}
}
pub fn snapshot(&self) -> ErrorCounters {
ErrorCounters {
upload_fails: self.upload_fails.load(Ordering::Relaxed),
download_backpressure_events: self.download_backpressure.load(Ordering::Relaxed),
leg_disconnects: self.leg_disconnects.load(Ordering::Relaxed),
control_channel_full_drops: self.control_full_drops.load(Ordering::Relaxed),
tunnel_write_stalls: self.tunnel_write_stalls.load(Ordering::Relaxed),
stream_errors: self.stream_errors.load(Ordering::Relaxed),
}
}
}
/// Process-global counters incremented at every event site.
pub static DIAG_COUNTERS: DiagnosticsCounters = DiagnosticsCounters::new();
// ── DiagnosticsStore — holds the last N snapshots ─────────────────────────────
pub struct DiagnosticsStore {
snapshots: Mutex<VecDeque<DiagnosticsSnapshot>>,
max_snapshots: usize,
}
impl DiagnosticsStore {
pub fn new(max_snapshots: usize) -> Self {
Self {
snapshots: Mutex::new(VecDeque::new()),
max_snapshots,
}
}
pub fn push(&self, snap: DiagnosticsSnapshot) {
let mut q = self.snapshots.lock().unwrap();
if q.len() >= self.max_snapshots {
q.pop_front();
}
q.push_back(snap);
}
/// Returns all stored snapshots as a pretty-printed JSON array.
pub fn get_all_json(&self) -> String {
let q = self.snapshots.lock().unwrap();
let items: Vec<&DiagnosticsSnapshot> = q.iter().collect();
serde_json::to_string_pretty(&items)
.unwrap_or_else(|e| format!("{{\"error\":\"{e}\"}}"))
}
/// Returns only the most recent snapshot as pretty-printed JSON, or `"null"`.
pub fn get_latest_json(&self) -> String {
let q = self.snapshots.lock().unwrap();
match q.back() {
Some(s) => serde_json::to_string_pretty(s)
.unwrap_or_else(|e| format!("{{\"error\":\"{e}\"}}")) ,
None => "null".to_string(),
}
}
}
// ── Helpers ───────────────────────────────────────────────────────────────────
pub fn current_timestamp_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
core/src/net/mod.rs
+2 -1
View File
@@ -1,9 +1,10 @@
mod config;
mod connection;
mod constants;
pub mod diagnostics;
pub use config::NetworkConfig;
pub use connection::{
ClientHandler, Connection, ServerHandler, SessionManager, TunnelHandler, GLOBAL_MIN_RTT,
ClientHandler, Connection, Muxer, ServerHandler, SessionManager, TunnelHandler, GLOBAL_MIN_RTT,
};
pub use constants::*;