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

all unbounded and remove AI hallucinations

Browse Source
This commit is contained in:
Kirill
2026-04-12 15:14:42 +07:00
parent 9233641e67
commit 11b810ad65
16 changed files with 315 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/config.rs
-21
View File
@@ -10,17 +10,6 @@ pub struct NetworkConfig {
pub tcp_buffer_size: usize,
pub udp_buffer_size: usize,
// --- Очереди MPSC (Разделенные) ---
// Для Клиента (мобильный интернет)
pub client_muxer_capacity: usize, // Подушка для 10 ног
pub client_tun_capacity: usize, // Стык TUN <-> Engine
pub client_stream_capacity: usize, // Быстрый Backpressure для сокета
pub client_virtual_stream_capacity: usize,
// Для Сервера (Дата-центр)
pub server_muxer_capacity: usize, // Огромная очередь для входящего трафика
pub server_stream_capacity: usize, // Чтобы сервер не тормозил на отдачу
pub tcp_chunk_size: usize,
// Буферы сокетов smoltcp
@@ -44,16 +33,6 @@ impl NetworkConfig {
udp_buffer_size: 64 * 1024,
tcp_chunk_size: system_mtu - 100,
// 🔥 Зажимаем программные очереди (Убиваем Hidden Bloat)
client_muxer_capacity: 8,
client_tun_capacity: 16,
client_stream_capacity: 16,
client_virtual_stream_capacity: 32,
server_muxer_capacity: 64, // Кардинально режем
server_stream_capacity: 32,
// 🔥 Расширяем TCP окна под BBR (Разблокируем Gigabit на дальние дистанции)
tcp_rx_heavy: 256 * 1024, //512
tcp_tx_heavy: 256 * 1024, // 1 MB
core/src/net/connection/bridge.rs
+18 -12
View File
@@ -7,7 +7,7 @@ use crate::nrxp::FrameType;
use bytes::{Bytes, BytesMut};
use netrunner_logger::{debug, error, info, warn};
use tokio::net::UdpSocket;
use tokio::sync::mpsc;
use tokio::sync::mpsc; // Подтянет UnboundedReceiver
use tokio::time::timeout;
struct StreamGuard {
@@ -27,7 +27,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
mut reader: R,
mut writer: W,
muxer: Arc<Muxer>,
mut v_rx: mpsc::Receiver<Bytes>,
mut v_rx: mpsc::UnboundedReceiver<Bytes>, // 🔥 Перешли на Unbounded
) where
R: tokio::io::AsyncReadExt + Unpin,
W: tokio::io::AsyncWriteExt + Unpin,
@@ -43,7 +43,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
buf.reserve(NetworkConfig::global().tcp_buffer_size);
let select_res = timeout(BRIDGE_IDLE_TIMEOUT, async {
tokio::select! {
// 1. Читаем из реального сокета интернета -> шлем в туннель
res = reader.read_buf(&mut buf) => {
match res {
Ok(0) => {
@@ -57,6 +57,8 @@ pub(crate) async fn run_tcp_bridge<R, W>(
data: buf.split().freeze(),
};
// send_to_network теперь внутри использует UnboundedSender,
// поэтому этот вызов больше никогда не заблокирует цикл из-за Full
if muxer.send_to_network(msg).await.is_err() {
return Ok(false);
}
@@ -69,7 +71,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
}
}
// 2. Получаем из туннеля -> пишем в реальный сокет
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
@@ -104,11 +106,14 @@ pub(crate) async fn run_tcp_bridge<R, W>(
}
}
let _ = muxer.send_to_network(MuxMessage {
stream_id,
frame_type: FrameType::Close,
data: Bytes::new(),
});
// Сообщаем другой стороне, что стрим закрыт
let _ = muxer
.send_to_network(MuxMessage {
stream_id,
frame_type: FrameType::Close,
data: Bytes::new(),
})
.await;
tokio::time::sleep(Duration::from_millis(50)).await;
}
@@ -117,7 +122,7 @@ pub(crate) async fn run_udp_bridge(
stream_id: u32,
socket: UdpSocket,
muxer: Arc<Muxer>,
mut v_rx: mpsc::Receiver<Bytes>,
mut v_rx: mpsc::UnboundedReceiver<Bytes>, // 🔥 Перешли на Unbounded
) {
let _guard = StreamGuard {
stream_id,
@@ -132,11 +137,12 @@ pub(crate) async fn run_udp_bridge(
loop {
let select_res = timeout(BRIDGE_IDLE_TIMEOUT, async {
tokio::select! {
// Из интернета в туннель
res = socket.recv(&mut buf) => {
match res {
Ok(n) if n > 0 => {
let data = Bytes::copy_from_slice(&buf[..n]);
// Внутри muxer теперь тоже Unbounded
if let Err(e) = muxer.send_data_safe(stream_id, data, true).await {
error!(stream_id, "UDP Failed to send to tunnel: {}", e);
return Err(e);
@@ -151,7 +157,7 @@ pub(crate) async fn run_udp_bridge(
}
}
// Из туннеля в интернет
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
core/src/net/connection/connection.rs
+32 -54
View File
@@ -6,7 +6,7 @@ use crate::{
connection::{
engine::TunnelEngine,
handler::{RemoteOpener, StreamHandler},
muxer::Muxer,
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,
@@ -65,7 +65,6 @@ pub trait TunnelHandler {
async fn run(self) -> Result<(), String>;
}
// ⚠️ Connection теперь содержит только транспорт, без Codec
pub struct Connection {
pub(crate) inbound: OwnedReadHalf,
pub(crate) outbound: OwnedWriteHalf,
@@ -104,7 +103,6 @@ impl ClientHandler {
.next()
.ok_or_else(|| format!("No IPs found for {}", remote_proxy_addr))?;
// --- Шаг 1: Создание сырого сокета через socket2 ---
let domain = if addr.is_ipv4() {
Domain::IPV4
} else {
@@ -113,11 +111,9 @@ impl ClientHandler {
let socket =
Socket::new(domain, Type::STREAM, Some(Protocol::TCP)).map_err(|e| e.to_string())?;
// --- Шаг 2: Настройка кроссплатформенных опций ---
socket.set_nonblocking(true).map_err(|e| e.to_string())?;
socket.set_nodelay(true).map_err(|e| e.to_string())?;
// 🔥 Шаг 3: Установка TCP_NOTSENT_LOWAT через libc (Linux, Android, iOS, macOS)
#[cfg(any(
target_os = "linux",
target_os = "android",
@@ -126,24 +122,16 @@ impl ClientHandler {
))]
unsafe {
use std::os::fd::AsRawFd;
let lowat: libc::c_int = 16384; // 16 КБ
let ret = libc::setsockopt(
let lowat: libc::c_int = 16384;
let _ = libc::setsockopt(
socket.as_raw_fd(),
libc::IPPROTO_TCP,
libc::TCP_NOTSENT_LOWAT,
&lowat as *const _ as *const libc::c_void,
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
);
if ret != 0 {
warn!(
"Could not set TCP_NOTSENT_LOWAT: {}",
std::io::Error::last_os_error()
);
}
}
// --- Шаг 4: Подключение и конвертация в TcpStream для Tokio ---
let _ = socket.connect(&addr.into());
let std_stream: std::net::TcpStream = socket.into();
let stream = TcpStream::from_std(std_stream).map_err(|e| e.to_string())?;
@@ -186,17 +174,19 @@ impl ClientHandler {
let codec = Codec::new(cipher, session_keys.get_auth_key());
let (rx_codec, mut tx_codec) = codec.split();
let handshake_payload = Bytes::from(format!("{}:{}", session_id, leg_id));
let encrypted_handshake = tx_codec
.encode_frame(0, FrameType::Handshake, handshake_payload)
.map_err(|e| format!("Failed to encrypt Handshake: {:?}", e))?;
// 🔥 ФИКС: Используем Heartbeat для первой авторизации соединения
let auth_payload = Bytes::from(format!("{}:{}", session_id, leg_id));
let encrypted_auth = tx_codec
.encode_frame(0, FrameType::Heartbeat, auth_payload)
.map_err(|e| format!("Failed to encrypt Auth: {:?}", e))?;
conn.outbound
.write_all(&encrypted_handshake)
.write_all(&encrypted_auth)
.await
.map_err(|e| e.to_string())?;
let (control_tx, control_rx) = mpsc::channel(NetworkConfig::global().client_muxer_capacity);
let (data_tx, data_rx) = mpsc::channel(NetworkConfig::global().client_muxer_capacity);
let (control_tx, control_rx) = mpsc::unbounded_channel::<MuxMessage>();
let (data_tx, data_rx) = mpsc::unbounded_channel::<MuxMessage>();
muxer.add_leg(leg_id, control_tx, data_tx);
let handler = Arc::new(StreamHandler::new(muxer.clone(), None));
@@ -218,8 +208,8 @@ impl ClientHandler {
pub async fn connect(
remote_proxy_addr: &str,
mut rx_from_engine: mpsc::Receiver<RawCastFrame>,
tx_to_engine: mpsc::Sender<RawCastFrame>,
mut rx_from_engine: mpsc::UnboundedReceiver<RawCastFrame>,
tx_to_engine: mpsc::UnboundedSender<RawCastFrame>,
) -> Result<(), String> {
let session_id = SessionManager::generate_id();
let muxer = Arc::new(Muxer::new(true, session_id.clone()));
@@ -276,8 +266,7 @@ impl ClientHandler {
),
);
let (v_tx, mut v_rx) =
mpsc::channel(NetworkConfig::global().client_stream_capacity);
let (v_tx, mut v_rx) = mpsc::unbounded_channel::<Bytes>();
muxer_inner.register_stream(global_stream_id, v_tx);
let tx_to_tun = tx_to_engine.clone();
@@ -302,7 +291,7 @@ impl ClientHandler {
port,
proto == LocalProtocol::Udp,
) {
let _ = tx_to_tun.send(raw).await;
let _ = tx_to_tun.send(raw);
}
}
}
@@ -341,6 +330,7 @@ impl ClientHandler {
Ok(())
}
}
pub struct ServerHandler {
pub(crate) conn: Connection,
pub(crate) session_manager: Arc<SessionManager>,
@@ -392,7 +382,6 @@ impl TunnelHandler for ServerHandler {
} = self.conn;
let mut session_keys = SessionKeys::new(false);
// --- PHASE 1: TLS Hello & Protocol Identification ---
let hello = loop {
let buf_snapshot = read_buf.clone().freeze();
@@ -430,8 +419,7 @@ impl TunnelHandler for ServerHandler {
}
}
Err(e) => {
let strategy = e.execute_strategy();
if strategy == ErrorAction::Redirect {
if e.execute_strategy() == ErrorAction::Redirect {
Self::handle_stealth_fallback(inbound, outbound, buf_snapshot).await;
}
return Ok(());
@@ -439,13 +427,11 @@ impl TunnelHandler for ServerHandler {
}
};
// --- PHASE 2: Send Server Hello ---
outbound
.write_all(&hello)
.await
.map_err(|e| e.to_string())?;
// --- PHASE 3: Secure Handshake & Engine Startup ---
let (tx_key, tx_iv, rx_key, rx_iv) = session_keys.get_aead_parameters();
let mut cipher = ChaChaCipher::new();
cipher.set_keys(tx_key, tx_iv, rx_key, rx_iv);
@@ -453,53 +439,45 @@ impl TunnelHandler for ServerHandler {
let codec = Codec::new(cipher, session_keys.get_auth_key());
let (mut rx_codec, tx_codec) = codec.split();
// 🔥 ФИКС: Ожидаем Heartbeat в качестве авторизационного фрейма!
let (session_id, leg_id) = loop {
// ✅ ИСПРАВЛЕНИЕ: Теперь мы не игнорируем Err!
match rx_codec.decode_inbound(&mut read_buf) {
Ok(Some(frame)) => {
if frame.header.frame_type == FrameType::Handshake {
let parts: Vec<&str> = std::str::from_utf8(&frame.payload)
.unwrap_or("")
.split(':')
.collect();
if parts.len() == 2 {
if frame.header.frame_type == FrameType::Heartbeat {
let payload_str = std::str::from_utf8(&frame.payload).unwrap_or("");
let parts: Vec<&str> = payload_str.split(':').collect();
if parts.len() == 2 && parts[1].parse::<u32>().is_ok() {
let sid = parts[0].to_string();
let lid: u32 = parts[1].parse().unwrap_or(0);
info!(
"🤝 Secure Handshake verified! Session: {}, Leg: {}",
sid, lid
);
let lid: u32 = parts[1].parse().unwrap();
info!("🤝 Secure Auth verified! Session: {}, Leg: {}", sid, lid);
break (sid, lid);
}
}
return Err("Expected Handshake frame".into());
return Err("Expected Auth payload in first Heartbeat frame".into());
}
Ok(None) => {
// Ждем новых данных из сети
let n = tokio::time::timeout(
SECURE_HANDSHAKE_TIMEOUT,
inbound.read_buf(&mut read_buf),
)
.await
.map_err(|_| "Timeout waiting for Handshake")?
.map_err(|_| "Timeout waiting for Auth")?
.map_err(|e| e.to_string())?;
if n == 0 {
return Err("Client closed connection before Handshake".into());
return Err("Client closed connection before Auth".into());
}
}
Err(e) => {
// Если криптография сломалась, сразу рвем соединение и логируем
error!("❌ Secure Handshake Failed: {:?}", e);
error!("❌ Secure Auth Failed: {:?}", e);
return Err("Dropped by security strategy (Auth Phase)".into());
}
}
};
// --- PHASE 4: Engine Startup ---
let muxer = self.session_manager.get_or_create(&session_id);
let (control_tx, control_rx) = mpsc::channel(NetworkConfig::global().server_muxer_capacity);
let (data_tx, data_rx) = mpsc::channel(NetworkConfig::global().server_muxer_capacity);
let (control_tx, control_rx) = mpsc::unbounded_channel::<MuxMessage>();
let (data_tx, data_rx) = mpsc::unbounded_channel::<MuxMessage>();
muxer.add_leg(leg_id, control_tx, data_tx);
let opener = Arc::new(RemoteOpener {
core/src/net/connection/engine.rs
+13 -33
View File
@@ -5,7 +5,7 @@ use netrunner_logger::{debug, error, info};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::tcp::{OwnedReadHalf, OwnedWriteHalf},
sync::mpsc::Receiver,
sync::mpsc::UnboundedReceiver,
};
use tokio_util::sync::CancellationToken;
@@ -23,8 +23,8 @@ pub(crate) struct TunnelEngine {
pub rx_codec: RxCodec,
pub tx_codec: TxCodec,
pub read_buf: BytesMut,
pub control_rx: Receiver<MuxMessage>,
pub data_rx: Receiver<MuxMessage>,
pub control_rx: UnboundedReceiver<MuxMessage>,
pub data_rx: UnboundedReceiver<MuxMessage>,
pub handler: Arc<StreamHandler>,
pub leg_id: u32,
pub muxer: Arc<crate::net::connection::muxer::Muxer>,
@@ -82,9 +82,7 @@ impl TunnelEngine {
Ok(Some(frame)) => frames.push(frame),
Ok(None) => break,
Err(e) => {
if e.action == ErrorAction::Wait {
break;
}
if e.action == ErrorAction::Wait { break; }
if e.action == ErrorAction::Drop {
error!("CRITICAL: Crypto tampering or sync lost. Hard dropping tunnel!");
return Err("Crypto drop".into());
@@ -96,15 +94,13 @@ impl TunnelEngine {
}
for frame in frames {
// ИСПРАВЛЕНИЕ: Перехватываем PONG для замера RTT, но НЕ ДЕЛАЕМ continue,
// чтобы фрейм дошел до хендлера (если нужно)
if frame.header.frame_type == FrameType::Handshake && frame.payload.as_ref() == b"PONG" {
muxer.record_pong(leg_id).await;
} else if frame.header.frame_type == FrameType::Heartbeat {
// Хартбиты тоже можем использовать для RTT, если они двусторонние
muxer.record_pong(leg_id).await;
// Любой Heartbeat – это подтверждение жизни туннеля, обновляем пинг
if frame.header.frame_type == FrameType::Heartbeat {
let m = muxer.clone();
tokio::spawn(async move {
m.record_pong(leg_id).await;
});
}
handler.handle(frame).await;
}
}
@@ -120,32 +116,21 @@ impl TunnelEngine {
loop {
tokio::select! {
biased;
_ = token_writer.cancelled() => {
info!("Writer Task: Shutdown signal received.");
break;
}
_ = token_writer.cancelled() => break,
msg_opt = control_rx.recv() => {
if let Some(msg) = msg_opt {
Self::handle_outbound(&mut outbound, &mut tx_codec, msg).await?;
} else {
break;
}
} else { break; }
}
_ = heartbeat.tick() => {
muxer_pong.record_ping_sent(leg_id);
let msg = MuxMessage { stream_id: 0, frame_type: FrameType::Heartbeat, data: Bytes::new() };
Self::handle_outbound(&mut outbound, &mut tx_codec, msg).await?;
}
msg_opt = data_rx.recv() => {
if let Some(msg) = msg_opt {
Self::handle_outbound(&mut outbound, &mut tx_codec, msg).await?;
} else {
break;
}
} else { break; }
}
}
}
@@ -160,7 +145,6 @@ impl TunnelEngine {
if let Err(e) = &res {
error!("TunnelEngine critical failure: {}", e);
}
res
}
@@ -172,10 +156,8 @@ impl TunnelEngine {
let mut data = msg.data;
let stream_id = msg.stream_id;
let frame_type = msg.frame_type;
let mut packets = Vec::new();
// ИСПРАВЛЕНИЕ: Чанкуем ТОЛЬКО Data. Контрольные фреймы отправляем целиком.
if frame_type == FrameType::Data {
while !data.is_empty() {
let chunk_size = std::cmp::min(data.len(), NetworkConfig::global().tcp_chunk_size);
@@ -190,7 +172,6 @@ impl TunnelEngine {
}
}
} else {
// Для UdpData, Connect, Handshake, Heartbeat, Close чанкование запрещено
match tx_codec.encode_frame(stream_id, frame_type.clone(), data) {
Ok(pkt) => packets.push(pkt),
Err(e) => {
@@ -206,7 +187,6 @@ impl TunnelEngine {
e.to_string()
})?;
}
Ok(())
}
}
core/src/net/connection/handler.rs
+32 -32
View File
@@ -19,7 +19,12 @@ pub struct RemoteOpener {
}
impl RemoteOpener {
pub async fn open_tcp(&self, stream_id: u32, target: String, v_rx: mpsc::Receiver<Bytes>) {
pub async fn open_tcp(
&self,
stream_id: u32,
target: String,
v_rx: mpsc::UnboundedReceiver<Bytes>,
) {
let muxer = self.muxer.clone();
tokio::spawn(async move {
info!(stream_id, "🌐 [Remote] Connecting to {}", target);
@@ -48,7 +53,12 @@ impl RemoteOpener {
});
}
pub async fn open_udp(&self, stream_id: u32, target: String, v_rx: mpsc::Receiver<Bytes>) {
pub async fn open_udp(
&self,
stream_id: u32,
target: String,
v_rx: mpsc::UnboundedReceiver<Bytes>,
) {
let muxer = self.muxer.clone();
tokio::spawn(async move {
info!(stream_id, "🚀 [Remote] Binding UDP for {}", target);
@@ -78,30 +88,29 @@ impl StreamHandler {
match frame.header.frame_type {
FrameType::Heartbeat => {
if self.opener.is_some() {
trace!(stream_id, "💓 [Server] Ping received, sending Pong");
let payload = frame.payload.as_ref();
if payload == b"PING" {
trace!(stream_id, "🤝 [Tunnel] PING received, replying PONG");
let _ = self
.muxer
.send_control(stream_id, FrameType::Heartbeat, Bytes::new())
.send_control(stream_id, FrameType::Heartbeat, Bytes::from("PONG"))
.await;
} else if payload == b"PONG" {
trace!(stream_id, "🤝 [Tunnel] PONG received");
self.muxer.dispatch_to_local(stream_id, frame.payload).await;
} else {
trace!(stream_id, "💓 [Client] Pong received");
}
}
// ИСПРАВЛЕНИЕ: Добавлена обработка PING/PONG для Health Check
FrameType::Handshake => {
if frame.payload.as_ref() == b"PING" {
trace!(
stream_id,
"🤝 [Tunnel] Health Check PING received, replying PONG"
);
let _ = self
.muxer
.send_control(stream_id, FrameType::Handshake, Bytes::from("PONG"))
.await;
} else if frame.payload.as_ref() == b"PONG" {
trace!(stream_id, "🤝 [Tunnel] Health Check PONG received");
if self.opener.is_some() {
trace!(
stream_id,
"💓 [Server] Standard Heartbeat received, sending reply"
);
let _ = self
.muxer
.send_control(stream_id, FrameType::Heartbeat, Bytes::new())
.await;
} else {
trace!(stream_id, "💓 [Client] Standard Heartbeat reply received");
}
}
}
@@ -122,8 +131,6 @@ impl StreamHandler {
debug!(stream_id, "🏁 [Tunnel] Peer closed stream");
self.muxer.remove_stream(stream_id);
}
_ => debug!(stream_id, "Unhandled frame: {:?}", frame.header.frame_type),
}
}
@@ -131,13 +138,7 @@ impl StreamHandler {
let target = String::from_utf8_lossy(&payload).to_string();
if let Some(opener) = &self.opener {
let capacity = if is_udp {
NetworkConfig::global().server_stream_capacity
} else {
NetworkConfig::global().server_stream_capacity
};
let (v_tx, v_rx) = mpsc::channel(capacity);
let (v_tx, v_rx) = mpsc::unbounded_channel::<Bytes>();
self.muxer.register_stream(stream_id, v_tx);
if is_udp {
@@ -150,7 +151,6 @@ impl StreamHandler {
stream_id,
"⚠️ [Tunnel] Rejected incoming connection to {} (Client mode)", target
);
let _ = self
.muxer
.send_control(stream_id, FrameType::Close, Bytes::new())
core/src/net/connection/muxer.rs
+44 -62
View File
@@ -4,7 +4,7 @@ use netrunner_logger::{debug, info, trace, warn};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::mpsc::Sender;
use tokio::sync::mpsc::UnboundedSender;
use crate::net::{HEALTH_CHECK_TIMEOUT, MAX_TUNNEL_LEGS, MUXER_POOL_SIZE};
use crate::nrxp::FrameType;
@@ -24,23 +24,14 @@ pub struct StreamStats {
#[derive(Clone)]
struct MuxLeg {
control_tx: Sender<MuxMessage>,
data_tx: Sender<MuxMessage>,
control_tx: UnboundedSender<MuxMessage>,
data_tx: UnboundedSender<MuxMessage>,
stats: Arc<LegStats>,
}
impl MuxLeg {
/// Возвращает коэффициент загруженности очереди (0.0 - пуста, 1.0 - полна)
fn congestion_factor(&self) -> f64 {
let cap = self.data_tx.capacity() as f64;
let max = crate::net::NetworkConfig::global().client_muxer_capacity as f64;
// Защита от деления на ноль, если конфигурация задана криво
if max <= 0.0 {
return 0.0;
}
1.0 - (cap / max)
0.0
}
}
@@ -70,7 +61,7 @@ pub struct MuxMessage {
#[derive(Clone)]
pub struct Muxer {
legs: Arc<DashMap<u32, MuxLeg>>,
streams: Arc<DashMap<u32, (Sender<Bytes>, Arc<StreamStats>)>>,
streams: Arc<DashMap<u32, (UnboundedSender<Bytes>, Arc<StreamStats>)>>,
stream_bindings: Arc<DashMap<u32, u32>>,
pending_pings: Arc<DashMap<u32, Instant>>,
id_gen: Arc<IdGenerator>,
@@ -92,8 +83,8 @@ impl Muxer {
pub fn add_leg(
&self,
leg_id: u32,
control_tx: Sender<MuxMessage>,
data_tx: Sender<MuxMessage>,
control_tx: UnboundedSender<MuxMessage>,
data_tx: UnboundedSender<MuxMessage>,
) {
if self.legs.len() >= MAX_TUNNEL_LEGS as usize {
warn!(leg_id, "MUXER: Max legs reached: {}", MAX_TUNNEL_LEGS);
@@ -113,10 +104,8 @@ impl Muxer {
pub fn remove_leg(&self, leg_id: u32) {
self.legs.remove(&leg_id);
// Удаляем все привязки стримов к этой ноге, чтобы они перебалансировались
self.stream_bindings
.retain(|_, target_leg| *target_leg != leg_id);
info!(leg_id, "MUXER: Leg removed and bindings cleared");
self.stream_bindings.clear();
info!(leg_id, "MUXER: Leg removed, all streams re-balanced");
}
pub fn active_legs_count(&self) -> usize {
@@ -155,7 +144,6 @@ impl Muxer {
candidates.sort_by(|(_, leg_a), (_, leg_b)| {
let rtt_a = leg_a.stats.rtt_ms.load(Ordering::Relaxed) as f64;
let rtt_b = leg_b.stats.rtt_ms.load(Ordering::Relaxed) as f64;
let score_a = rtt_a + (leg_a.congestion_factor() * 2000.0);
let score_b = rtt_b + (leg_b.congestion_factor() * 2000.0);
@@ -164,10 +152,6 @@ impl Muxer {
.unwrap_or(std::cmp::Ordering::Equal)
});
// ИСПРАВЛЕНИЕ: Убрал агрессивную блокировку "в 2 раза быстрее",
// из-за которой вторая нога голодала и отваливалась по таймаутам.
// Теперь балансировка полагается на congestion_factor и скоринг плавно.
let pool_size = std::cmp::min(candidates.len(), MUXER_POOL_SIZE);
let (selected_id, selected_leg) = candidates[stream_id as usize % pool_size].clone();
self.stream_bindings.insert(stream_id, selected_id);
@@ -182,7 +166,6 @@ impl Muxer {
pub async fn record_pong(&self, leg_id: u32) {
if let Some((_, start_time)) = self.pending_pings.remove(&leg_id) {
let rtt = start_time.elapsed().as_millis() as u32;
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(rtt, Ordering::Relaxed);
trace!(leg_id, rtt, "💓 [Muxer] RTT updated for leg");
@@ -202,11 +185,11 @@ impl Muxer {
FrameType::Connect
| FrameType::Close
| FrameType::UdpConnect
| FrameType::Handshake => leg.control_tx.clone(),
| FrameType::Heartbeat => leg.control_tx.clone(),
_ => leg.data_tx.clone(),
};
if target_tx.send(message).await.is_err() {
if target_tx.send(message).is_err() {
self.remove_leg(leg_id);
return Err(format!("MUXER: Leg {} died during send", leg_id));
}
@@ -234,7 +217,6 @@ impl Muxer {
} else {
FrameType::Data
};
self.send_to_network(MuxMessage {
stream_id,
frame_type,
@@ -257,7 +239,7 @@ impl Muxer {
.await
}
pub fn register_stream(&self, stream_id: u32, tx: Sender<Bytes>) {
pub fn register_stream(&self, stream_id: u32, tx: UnboundedSender<Bytes>) {
self.streams
.insert(stream_id, (tx, Arc::new(StreamStats::default())));
}
@@ -273,14 +255,12 @@ impl Muxer {
.streams
.get(&stream_id)
.map(|s| (s.value().0.clone(), s.value().1.clone()));
if let Some((tx, stats)) = stream_opt {
let size = data.len() as u64;
if tx.send(data).await.is_ok() {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
} else {
if tx.send(data).is_err() {
self.remove_stream(stream_id);
} else {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
}
}
}
@@ -296,7 +276,6 @@ impl Muxer {
}
pub async fn perform_health_check(&self) {
// Берем список ID заранее, чтобы не держать lock DashMap
let leg_ids: Vec<u32> = self.legs.iter().map(|kv| *kv.key()).collect();
for leg_id in leg_ids {
@@ -306,20 +285,21 @@ impl Muxer {
let tx = leg.control_tx.clone();
let probe_stream_id = self.id_gen.next();
let (probe_tx, mut probe_rx) = tokio::sync::mpsc::channel(2);
let (probe_tx, mut probe_rx) = tokio::sync::mpsc::unbounded_channel();
self.register_stream(probe_stream_id, probe_tx);
self.record_ping_sent(leg_id);
let msg = MuxMessage {
stream_id: probe_stream_id,
frame_type: FrameType::Handshake,
frame_type: FrameType::Heartbeat, // 🔥 Теперь Heartbeat
data: Bytes::from("PING"),
};
let start = std::time::Instant::now();
// 🔥 ФИКС: Если мы даже PING не можем отправить в очередь - нога уже труп
if tx.try_send(msg).is_err() {
warn!(leg_id, "❌ MUXER: Leg queue overflow, killing leg");
if tx.send(msg).is_err() {
warn!(leg_id, "❌ MUXER: Leg channel dropped, killing leg");
self.remove_leg(leg_id);
self.remove_stream(probe_stream_id);
continue;
@@ -327,14 +307,9 @@ impl Muxer {
match tokio::time::timeout(HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => {
let rtt = start.elapsed().as_millis() as u32;
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(rtt, Ordering::Relaxed);
debug!(leg_id, rtt, "✅ Leg Health Check OK");
}
debug!(leg_id, "✅ Leg Health Check OK");
}
_ => {
// 🔥 ФИКС: Если таймаут - УДАЛЯЕМ НОГУ. Хватит быть зомби.
warn!(leg_id, "❌ Leg Health Check Timeout - Evicting leg");
self.remove_leg(leg_id);
}
@@ -359,15 +334,16 @@ impl Muxer {
}
pub fn print_topology_tree(&self) {
println!(
"\n🌐 Netrunner Tunnel Topology [Session: {}]",
let mut out = String::new();
out.push_str(&format!(
"🌐 Netrunner Tunnel Topology [Session: {}]\n",
self.session_id
);
));
let mut total_tx = 0;
let mut total_rx = 0;
let mut legs_info = Vec::new();
for kv in self.legs.iter() {
let id = kv.key();
let stats = &kv.value().stats;
@@ -395,23 +371,29 @@ impl Muxer {
));
}
println!(
"├─ 📊 Global Traffic: ⇡ {} | ⇣ {}",
out.push_str(&format!(
"├─ 📊 Global Traffic: ⇡ {} | ⇣ {}\n",
Self::format_size(total_tx),
Self::format_size(total_rx)
);
));
out.push_str(&format!(
"├─ 🦵 Physical Legs (Active: {})\n",
legs_info.len()
));
println!("├─ 🦵 Physical Legs (Active: {})", legs_info.len());
for (i, info) in legs_info.iter().enumerate() {
if i == legs_info.len() - 1 {
println!("{}", info.replace("├─", "└─"));
out.push_str(&format!("{}\n", info.replace("├─", "└─")));
} else {
println!("{}", info);
out.push_str(&format!("{}\n", info));
}
}
let streams_count = self.streams.len();
println!("└─ 🔀 Virtual Streams (Active: {})", streams_count);
out.push_str(&format!(
"└─ 🔀 Virtual Streams (Active: {})\n",
streams_count
));
let mut count = 0;
for kv in self.streams.iter() {
@@ -428,14 +410,14 @@ impl Muxer {
" ├─"
};
println!(
"{} Stream {:<4} ─ ⇡ {:<9} | ⇣ {}",
out.push_str(&format!(
"{} Stream {:<4} ─ ⇡ {:<9} | ⇣ {}\n",
prefix,
id,
Self::format_size(tx),
Self::format_size(rx)
);
));
}
println!();
info!("\n{}", out);
}
}
core/src/net/constants.rs
+2 -2
View File
@@ -9,8 +9,8 @@ pub const MUXER_POOL_SIZE: usize = 3; // Из скольких лучших Leg-
pub const TCP_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20); // Время на установку SYN/ACK
pub const UDP_IDLE_TIMEOUT: Duration = Duration::from_secs(60); // Смерть UDP сессии без данных
pub const GLOBAL_IDLE_TIMEOUT: Duration = Duration::from_secs(120); // Очистка Tracker-ом
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(5); // Частота пинга Leg-ов
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(3);
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(7); // Частота пинга Leg-ов
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(10);
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(3); // Пауза перед реконнектом Leg
pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(30); // Таймаут задач-бриджей