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

bufferbloat fixes and reconnect

Browse Source
This commit is contained in:
Kirill
2026-04-15 13:33:38 +07:00
parent fac16c5296
commit f5f1d20f54
8 changed files with 164 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/connection.rs
+54 -44
View File
@@ -83,6 +83,12 @@ impl Connection {
pub struct ClientHandler;
impl ClientHandler {
fn get_local_ip() -> Option<std::net::IpAddr> {
let socket = std::net::UdpSocket::bind("0.0.0.0:0").ok()?;
socket.connect("8.8.8.8:80").ok()?;
socket.local_addr().ok().map(|a| a.ip())
}
async fn establish_leg(
remote_proxy_addr: &str,
leg_id: u32,
@@ -91,49 +97,25 @@ impl ClientHandler {
) -> Result<(), String> {
let leg_name = format!("TCP-Leg-{}", leg_id);
let mut addrs = tokio::net::lookup_host(remote_proxy_addr)
let addrs_future = tokio::net::lookup_host(remote_proxy_addr);
let mut addrs = tokio::time::timeout(std::time::Duration::from_secs(3), addrs_future)
.await
.map_err(|_| "DNS Lookup Timeout".to_string())?
.map_err(|e| format!("DNS resolution failed: {}", e))?;
let addr = addrs
.next()
.ok_or_else(|| format!("No IPs found for {}", remote_proxy_addr))?;
// 🔥 ФИКС: Используем Tokio TcpSocket вместо socket2.
// Это позволяет корректно отменять подключение при зависании ОС.
let socket = if addr.is_ipv4() {
tokio::net::TcpSocket::new_v4()
} else {
tokio::net::TcpSocket::new_v6()
}
let stream = tokio::time::timeout(
std::time::Duration::from_secs(5),
tokio::net::TcpStream::connect(addr),
)
.await
.map_err(|_| "Connection timeout".to_string())?
.map_err(|e| e.to_string())?;
socket.set_nodelay(true).unwrap_or_default();
#[cfg(any(
target_os = "linux",
target_os = "android",
target_os = "ios",
target_os = "macos"
))]
unsafe {
use std::os::unix::io::AsRawFd;
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,
);
}
// 🔥 ФИКС: Добавляем Timeout на подключение (5 секунд)!
// Если при смене сети пакеты идут в никуда, мы быстро прерываем попытку и пробуем снова.
let stream = tokio::time::timeout(std::time::Duration::from_secs(5), socket.connect(addr))
.await
.map_err(|_| "Connection timed out (Network changed?)".to_string())?
.map_err(|e| e.to_string())?;
stream.set_nodelay(true).unwrap_or_default();
let mut conn = Connection::new(stream);
let mut session_keys = SessionKeys::new(true);
@@ -154,7 +136,6 @@ impl ClientHandler {
break;
}
Ok(None) => {
// Также добавляем Timeout на чтение ответа от сервера
let res = tokio::time::timeout(
std::time::Duration::from_secs(5),
conn.inbound.read_buf(&mut conn.read_buf),
@@ -191,6 +172,9 @@ impl ClientHandler {
let cap = NetworkConfig::global().channel_capacity;
let (control_tx, control_rx) = mpsc::channel::<MuxMessage>(cap);
let (data_tx, data_rx) = mpsc::channel::<MuxMessage>(cap);
// Клонируем Sender до передачи в add_leg, чтобы использовать для remove_leg
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));
@@ -206,7 +190,14 @@ impl ClientHandler {
handler,
muxer: muxer.clone(),
};
engine.run().await.map_err(|e| e.to_string())?;
let run_result = engine.run().await;
// 🔥 ФИКС: Гарантированно вычищаем ногу по завершению работы,
// используя безопасный same_channel чек
muxer.remove_leg(leg_id, &control_tx_clone);
run_result.map_err(|e| e.to_string())?;
Err(format!("{} Engine stopped", leg_name))
}
@@ -221,6 +212,26 @@ impl ClientHandler {
Arc::new(DashMap::new());
let local_to_global: Arc<DashMap<u64, u32>> = Arc::new(DashMap::new());
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));
loop {
interval.tick().await;
let current_ip = Self::get_local_ip();
// 🔥 ФИКС: Trigger also if we completely lost network (current_ip == None)
if current_ip != last_ip {
netrunner_logger::warn!(
"🌐 Network Change Detected: {:?} -> {:?}",
last_ip,
current_ip
);
watcher_muxer.remove_all_legs();
last_ip = current_ip;
}
}
});
for id in 0..MAX_TUNNEL_LEGS {
let addr = remote_proxy_addr.to_string();
let m = muxer.clone();
@@ -241,7 +252,7 @@ impl ClientHandler {
while let Some(m_stats) = m_weak.upgrade() {
tokio::time::sleep(TOPOLOGY_PRINT_INTERVAL).await;
if m_stats.active_legs_count() == 0 {
break;
continue;
}
m_stats.perform_health_check().await;
m_stats.print_topology_tree();
@@ -335,6 +346,7 @@ impl ClientHandler {
Ok(())
}
}
pub struct ServerHandler {
pub(crate) conn: Connection,
pub(crate) session_manager: Arc<SessionManager>,
@@ -363,18 +375,15 @@ impl ServerHandler {
if let Ok(Ok(target_server)) = target_stream {
let (mut server_read, mut server_write) = target_server.into_split();
// Отправляем всё, что уже успели прочитать от клиента
if !initial_data.is_empty() {
if server_write.write_all(&initial_data).await.is_err() {
return;
}
}
// 🔥 ПРАВИЛЬНЫЙ способ связать 4 половинки сокетов в Tokio
let client_to_server = tokio::io::copy(&mut client_inbound, &mut server_write);
let server_to_client = tokio::io::copy(&mut server_read, &mut client_outbound);
// Ждем, пока любое из направлений не закроется
let _ = tokio::join!(client_to_server, server_to_client);
debug!("Stealth fallback connection closed.");
} else {
@@ -395,7 +404,6 @@ impl TunnelHandler for ServerHandler {
} = self.conn;
let mut session_keys = SessionKeys::new(false);
// --- ФАЗА 1: ПЕРВЫЙ ПАКЕТ И АВТОРИЗАЦИЯ ---
let hello = loop {
let buf_snapshot = read_buf.clone().freeze();
@@ -411,7 +419,6 @@ impl TunnelHandler for ServerHandler {
break sh;
}
Err(e) => {
// 🔥 ФИКС: ЛЮБАЯ ошибка (включая неверный Auth Tag) ведет на донора!
warn!("❌ Unauthorized/Invalid ClientHello. Triggering Stealth Fallback. Reason: {:?}", e.stage);
Self::handle_stealth_fallback(inbound, outbound, buf_snapshot).await;
return Ok(());
@@ -433,7 +440,6 @@ impl TunnelHandler for ServerHandler {
}
}
Err(_) => {
// 🔥 ФИКС: Мусор или сканер протоколов (не TLS) -> ведет на донора
warn!("❌ Handshake parse failed (Not a valid TLS probe). Triggering Stealth Fallback.");
Self::handle_stealth_fallback(inbound, outbound, buf_snapshot).await;
return Ok(());
@@ -492,6 +498,7 @@ impl TunnelHandler for ServerHandler {
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 opener = Arc::new(RemoteOpener {
@@ -513,7 +520,10 @@ impl TunnelHandler for ServerHandler {
};
let res = engine.run().await;
muxer.remove_leg(leg_id);
// 🔥 ФИКС: Безопасное удаление ноги и на сервере
muxer.remove_leg(leg_id, &control_tx_clone);
if muxer.active_legs_count() == 0 {
self.session_manager.remove(&session_id);
}
core/src/net/connection/engine.rs
+8 -4
View File
@@ -181,10 +181,14 @@ impl TunnelEngine {
}
for pkt in packets {
outbound.write_all(&pkt).await.map_err(|e| {
error!(stream_id, error = %e, "Failed to write encrypted data to network");
e.to_string()
})?;
// 🔥 ФИКС: Увеличен таймаут до 10 секунд (Mobile RRC Transitions)
let write_future = outbound.write_all(&pkt);
if let Err(_) =
tokio::time::timeout(std::time::Duration::from_secs(10), write_future).await
{
error!(stream_id, "🔥 Physical leg STUCK on write. Killing leg.");
return Err("Leg write timeout".into());
}
}
Ok(())
}
core/src/net/connection/muxer.rs
+51 -42
View File
@@ -89,7 +89,7 @@ impl Muxer {
control_tx: Sender<MuxMessage>,
data_tx: Sender<MuxMessage>,
) {
if self.legs.len() >= MAX_TUNNEL_LEGS as usize {
if self.legs.len() >= MAX_TUNNEL_LEGS as usize && !self.legs.contains_key(&leg_id) {
warn!(leg_id, "MUXER: Max legs reached: {}", MAX_TUNNEL_LEGS);
return;
}
@@ -105,12 +105,30 @@ impl Muxer {
info!(leg_id, "MUXER: Leg registered (Total: {})", self.legs.len());
}
pub fn remove_leg(&self, leg_id: u32) {
self.legs.remove(&leg_id);
info!(
leg_id,
"MUXER: Leg removed, streams will re-balance dynamically"
);
// 🔥 ФИКС: Умное удаление ноги. Мы проверяем, не была ли эта нога уже перезаписана
// более новым подключением (чтобы не удалить живую ногу по ошибке)
pub fn remove_leg(&self, leg_id: u32, tx: &Sender<MuxMessage>) {
let should_remove = if let Some(leg) = self.legs.get(&leg_id) {
leg.control_tx.same_channel(tx)
} else {
false
};
if should_remove {
self.legs.remove(&leg_id);
info!(leg_id, "MUXER: Leg removed safely, streams will re-balance");
} else {
trace!(
leg_id,
"MUXER: Leg removal skipped (already removed or replaced by a new connection)"
);
}
}
pub fn remove_all_legs(&self) {
warn!("🚨 MUXER: Emergency reset! Removing all legs due to network change.");
self.legs.clear();
self.stream_bindings.clear();
}
pub fn active_legs_count(&self) -> usize {
@@ -201,9 +219,13 @@ impl Muxer {
_ => &leg.data_tx,
};
// 🔥 ФИКС: Если канал данных забит, дропаем пакет, чтобы не ломать туннель
match target_tx.try_send(message.clone()) {
Ok(_) => {
match tokio::time::timeout(
std::time::Duration::from_secs(2),
target_tx.send(message.clone()),
)
.await
{
Ok(Ok(_)) => {
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
if let Some(stream_ref) = self.streams.get(&message.stream_id) {
stream_ref
@@ -214,26 +236,16 @@ impl Muxer {
}
return Ok(());
}
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
if matches!(message.frame_type, FrameType::Data | FrameType::UdpData) {
// Канал данных физически забит (слабый интернет). СБРАСЫВАЕМ пакет.
// TCP smoltcp на клиенте не получит ACK и сделает ретрансмиссию.
trace!(
message.stream_id,
"Physical TX full! Dropped packet to prevent deadlock."
);
return Ok(());
} else {
// Контрольные фреймы (Connect, Close, Ping) обязательны к доставке.
if target_tx.send(message.clone()).await.is_err() {
self.remove_leg(leg_id);
continue;
}
return Ok(());
}
Ok(Err(_)) => {
self.remove_leg(leg_id, &leg.control_tx);
continue;
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.remove_leg(leg_id);
Err(_) => {
warn!(
message.stream_id,
"Physical TX full for 2s! Leg {} is dead. Evicting.", leg_id
);
self.remove_leg(leg_id, &leg.control_tx);
continue;
}
}
@@ -291,10 +303,6 @@ impl Muxer {
if let Some((tx, stats)) = stream_opt {
let size = data.len() as u64;
// 🔥 ФИКС: Используем try_send вместо await!
// Если локальный сокет тупит, а очередь полна (2MB), мы сбрасываем пакет.
// Внешний сервер увидит потерю пакета и замедлит передачу.
match tx.try_send(data) {
Ok(_) => {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
@@ -332,8 +340,7 @@ impl Muxer {
let tx = leg.control_tx.clone();
let probe_stream_id = self.id_gen.next();
let cap = crate::net::NetworkConfig::global().channel_capacity;
let (probe_tx, mut probe_rx) = tokio::sync::mpsc::channel(cap);
let (probe_tx, mut probe_rx) = tokio::sync::mpsc::channel(10);
self.register_stream(probe_stream_id, probe_tx);
self.record_ping_sent(leg_id);
@@ -344,18 +351,20 @@ impl Muxer {
data: Bytes::from("PING"),
};
if tx.send(msg).await.is_err() {
warn!(leg_id, "❌ MUXER: Leg channel dropped, killing leg");
self.remove_leg(leg_id);
if tx.try_send(msg).is_err() {
warn!(leg_id, "❌ MUXER: Leg channel overflow/blocked, evicting");
self.remove_leg(leg_id, &tx);
self.remove_stream(probe_stream_id);
continue;
}
match tokio::time::timeout(HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => debug!(leg_id, "✅ Leg Health Check OK"),
match tokio::time::timeout(crate::net::HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => {
trace!(leg_id, "✅ Leg Health Check OK");
}
_ => {
warn!(leg_id, "❌ Leg Health Check Timeout - Evicting leg");
self.remove_leg(leg_id);
warn!(leg_id, "❌ Leg Health Check FAIL/Timeout - Evicting");
self.remove_leg(leg_id, &tx);
}
}
self.remove_stream(probe_stream_id);