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

select of download/upload seperated, write timeout, muxer leg kills timeout, pong await fix

Browse Source
This commit is contained in:
nineAp
2026-06-26 05:48:12 +07:00
parent 47208853c9
commit 6735c481dc
7 changed files with 224 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/bridge.rs
+66 -41
View File
@@ -1,14 +1,13 @@
use std::sync::Arc;
use std::time::Duration;
use crate::net::connection::muxer::{MuxMessage, Muxer};
use crate::net::connection::muxer::Muxer;
use crate::net::{NetworkConfig, BRIDGE_IDLE_TIMEOUT};
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::time::timeout;
use tokio_util::sync::CancellationToken;
struct StreamGuard {
stream_id: u32,
@@ -23,10 +22,10 @@ impl Drop for StreamGuard {
}
pub(crate) async fn run_tcp_bridge<R, W>(
stream_id: u32,
mut reader: R,
mut writer: W,
reader: R,
writer: W,
muxer: Arc<Muxer>,
mut v_rx: mpsc::Receiver<Bytes>,
v_rx: mpsc::Receiver<Bytes>,
) where
R: tokio::io::AsyncReadExt + Unpin,
W: tokio::io::AsyncWriteExt + Unpin,
@@ -36,50 +35,76 @@ pub(crate) async fn run_tcp_bridge<R, W>(
muxer: muxer.clone(),
};
let buf_size = NetworkConfig::global().tcp_buffer_size;
let token = CancellationToken::new();
// Создаем отдельный канал для упорядоченной отправки в туннель
let (tx_to_mux, mut rx_from_bridge) = mpsc::channel::<Bytes>(16);
// Задача-отправщик: гарантирует порядок и не блокирует основной цикл моста
let m_clone = muxer.clone();
tokio::spawn(async move {
while let Some(data) = rx_from_bridge.recv().await {
if let Err(_) = m_clone.send_data_safe(stream_id, data, false).await {
break;
}
}
});
let mut buf = BytesMut::with_capacity(buf_size);
loop {
if buf.capacity() < 16384 {
buf.reserve(buf_size);
}
tokio::select! {
// Читаем из Интернета -> В очередь отправки (Upload)
res = reader.read_buf(&mut buf) => {
match res {
Ok(0) => break,
Ok(_) => {
let data = buf.split().freeze();
if tx_to_mux.send(data).await.is_err() { break; }
// Upload: internet → tunnel.
// Runs concurrently with download so a congested muxer path does not
// prevent downstream data from being delivered.
let upload = {
let muxer = muxer.clone();
let token = token.clone();
async move {
let mut reader = reader;
let mut buf = BytesMut::with_capacity(buf_size);
loop {
if buf.capacity() < 16384 {
buf.reserve(buf_size);
}
tokio::select! {
biased;
_ = token.cancelled() => break,
res = reader.read_buf(&mut buf) => match res {
Ok(0) | Err(_) => break,
Ok(_) => {
let data = buf.split().freeze();
if muxer.send_data_safe(stream_id, data, false).await.is_err() {
break;
}
}
}
Err(_) => break,
}
}
// Читаем из Туннеля -> В Интернет (Download)
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
if data.is_empty() { continue; }
if writer.write_all(&data).await.is_err() { break; }
token.cancel();
}
};
// Download: tunnel → internet.
// write_all has a hard timeout so a slow local app (full socket buffer)
// does not block the pipeline indefinitely and starve other streams on
// the same tunnel leg.
let download = {
let token = token.clone();
async move {
let mut writer = writer;
let mut v_rx = v_rx;
loop {
tokio::select! {
biased;
_ = token.cancelled() => break,
maybe_data = v_rx.recv() => match maybe_data {
None => break,
Some(data) => {
if data.is_empty() { continue; }
match timeout(crate::net::BRIDGE_STREAM_WRITE_TIMEOUT, writer.write_all(&data)).await {
Ok(Ok(_)) => {}
_ => break,
}
}
}
None => break,
}
}
token.cancel();
}
};
// Both halves run concurrently via the outer select. When either half
// finishes (connection closed, error, or write timeout), the token
// cancels the other half so cleanup is prompt.
tokio::select! {
_ = upload => {}
_ = download => {}
}
token.cancel();
}
pub(crate) async fn run_udp_bridge(
stream_id: u32,
core/src/net/connection/connection.rs
+16 -12
View File
@@ -230,19 +230,23 @@ impl ClientHandler {
)
})?;
let stream = tokio::time::timeout(
FALLBACK_CONNECT_TIMEOUT,
tokio::net::TcpStream::connect(addr),
)
.await
.map_err(|_| {
AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут подключения",
"Connection timeout",
)
})?
let socket = (if addr.is_ipv4() {
tokio::net::TcpSocket::new_v4()
} else {
tokio::net::TcpSocket::new_v6()
})
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сокета", e.to_string()))?;
// Limit OS TCP send buffer to reduce bufferbloat on the tunnel leg.
// Default buffers (4–8 MB) can hold seconds of data at mobile speeds.
let _ = socket.set_send_buffer_size(crate::net::TUNNEL_SOCKET_SNDBUF);
let _ = socket.set_recv_buffer_size(crate::net::TUNNEL_SOCKET_RCVBUF);
let stream = tokio::time::timeout(FALLBACK_CONNECT_TIMEOUT, socket.connect(addr))
.await
.map_err(|_| {
AppError::new(ERR_INFRA_TIMEOUT, "Таймаут подключения", "Connection timeout")
})?
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сокета", e.to_string()))?;
let (inbound, outbound, rx_codec, tx_codec) =
Self::perform_handshake(stream, session_id, leg_id).await?;
core/src/net/connection/engine.rs
+69 -13
View File
@@ -15,9 +15,10 @@ use tracing::instrument;
use crate::{
net::{
connection::{handler::StreamHandler, muxer::MuxMessage},
NetworkConfig, HEALTH_CHECK_INTERVAL, FALLBACK_CONNECT_TIMEOUT,
RECONNECT_BACKOFF_BASE, RECONNECT_BACKOFF_JITTER_MS,
TUNNEL_INTERLEAVE_CHUNK, TUNNEL_MAX_BUFFER_SIZE, TUNNEL_READ_RESERVE,
NetworkConfig, FALLBACK_CONNECT_TIMEOUT, HEALTH_CHECK_INTERVAL,
MAX_INTERNAL_RECONNECT_ATTEMPTS, MAX_RECONNECT_BACKOFF_MS, 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},
};
@@ -50,19 +51,41 @@ impl TunnelEngine {
&mut self,
) -> Result<(OwnedReadHalf, OwnedWriteHalf, RxCodec, TxCodec), AppError> {
info!("🔄 Attempting reconnect to {}", self.remote_addr);
let stream = tokio::time::timeout(
FALLBACK_CONNECT_TIMEOUT,
tokio::net::TcpStream::connect(&self.remote_addr),
)
.await
.map_err(|_| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сети", "Reconnect timeout"))?
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сети", e.to_string()))?;
// Re-resolve the hostname each time so a server IP change or DNS
// failover is picked up automatically.
let mut addrs = tokio::net::lookup_host(&self.remote_addr)
.await
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "DNS при реконнекте", e.to_string()))?;
let addr = addrs.next().ok_or_else(|| {
AppError::new(ERR_INFRA_TIMEOUT, "Нет IP", "No IPs for reconnect addr")
})?;
let socket = (if addr.is_ipv4() {
tokio::net::TcpSocket::new_v4()
} else {
tokio::net::TcpSocket::new_v6()
})
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сокет", e.to_string()))?;
let _ = socket.set_send_buffer_size(crate::net::TUNNEL_SOCKET_SNDBUF);
let _ = socket.set_recv_buffer_size(crate::net::TUNNEL_SOCKET_RCVBUF);
let stream = tokio::time::timeout(FALLBACK_CONNECT_TIMEOUT, socket.connect(addr))
.await
.map_err(|_| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сети", "Reconnect timeout"))?
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сети", e.to_string()))?;
crate::net::ClientHandler::perform_handshake(stream, &self.session_id, self.leg_id).await
}
#[instrument(skip_all, fields(leg_id = self.leg_id))]
pub async fn run(mut self) -> Result<(), AppError> {
// Tracks consecutive internal reconnect failures. Resets to 0 on
// success. When it reaches MAX_INTERNAL_RECONNECT_ATTEMPTS the engine
// returns Err so the outer establish_leg loop gets control: it re-runs
// DNS, resets its own counters, and emits proper diagnostic events.
let mut internal_attempt: u32 = 0;
loop {
// Проверяем наличие всех необходимых ресурсов
if self.inbound.is_none()
@@ -83,6 +106,8 @@ impl TunnelEngine {
self.leg_status = LegStatus::Reconnecting;
match self.attempt_reconnect().await {
Ok((new_in, new_out, new_rx, new_tx)) => {
internal_attempt = 0; // successful reconnect — reset counter
let cap = crate::net::NetworkConfig::global().channel_capacity;
let (control_tx, control_rx) =
tokio::sync::mpsc::channel::<MuxMessage>(cap);
@@ -99,9 +124,40 @@ impl TunnelEngine {
info!("✅ Leg {} reconnected successfully", self.leg_id);
}
Err(e) => {
error!("Reconnect failed for leg {}: {}", self.leg_id, e);
internal_attempt += 1;
// Emit a diagnostic event so the snapshot system (and
// operator dashboards) can see we're stuck, even though
// the outer establish_leg loop hasn't returned yet.
crate::net::diagnostics::send_diag_event(
crate::net::diagnostics::DiagnosticsEvent::LegReconnecting {
leg_id: self.leg_id,
attempt: internal_attempt,
},
);
if internal_attempt >= MAX_INTERNAL_RECONNECT_ATTEMPTS {
// Give up so the outer loop re-runs DNS, resets
// its state, and records the failure in counters.
error!(
"Leg {} giving up after {} consecutive reconnect failures — handing off to outer loop",
self.leg_id, internal_attempt
);
return Err(e);
}
error!(
"Reconnect failed for leg {} (attempt {}/{}): {}",
self.leg_id, internal_attempt, MAX_INTERNAL_RECONNECT_ATTEMPTS, e
);
// Exponential back-off: 2 s, 4 s, 8 s, 16 s, 30 s (cap).
// The shift is capped at 4 to avoid overflow (2^4 = 16).
let exp_ms = RECONNECT_BACKOFF_BASE.as_millis() as u64
* (1u64 << internal_attempt.saturating_sub(1).min(4));
let jitter = rand::random::<u64>() % RECONNECT_BACKOFF_JITTER_MS;
tokio::time::sleep(RECONNECT_BACKOFF_BASE + tokio::time::Duration::from_millis(jitter)).await;
let backoff_ms = (exp_ms + jitter).min(MAX_RECONNECT_BACKOFF_MS);
tokio::time::sleep(tokio::time::Duration::from_millis(backoff_ms)).await;
continue;
}
}
@@ -178,7 +234,7 @@ impl TunnelEngine {
let m = muxer.clone();
tokio::spawn(async move { m.record_pong(leg_id).await; });
}
handler.handle(frame).await;
let _ = handler.handle(frame).await;
}
}
}
core/src/net/connection/handler.rs
+1 -1
View File
@@ -112,7 +112,7 @@ impl StreamHandler {
});
} else if payload == b"PONG" {
trace!(stream_id, "🤝 [Tunnel] PONG received");
self.muxer.dispatch_to_local(stream_id, frame.payload);
self.muxer.dispatch_to_local(stream_id, frame.payload).await;
} else {
if self.opener.is_some() {
trace!(
core/src/net/connection/muxer.rs
+30 -12
View File
@@ -8,7 +8,7 @@ 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::{DISPATCH_TO_LOCAL_TIMEOUT, HEALTH_CHECK_TIMEOUT, MAX_TUNNEL_LEGS};
use crate::net::diagnostics::{self, DiagnosticsEvent, DIAG_COUNTERS, LegMetrics, TunnelMetrics};
use crate::nrxp::FrameType;
@@ -173,14 +173,12 @@ impl Muxer {
}
}
// 3. O(N) поиск лучшей леги без сортировки всего вектора
// Мы берем подмножество (pool) и сразу ищем в нем минимум
let pool_size = std::cmp::min(legs.len(), MUXER_POOL_SIZE);
// Используем min_by, чтобы найти лучший вариант за один проход
// 3. O(N) поиск лучшей леги без сортировки всего вектора.
// Consider all available legs so the 4th leg is not permanently starved.
// MUXER_POOL_SIZE is kept for topology printing but no longer limits
// leg selection: sticky bindings already prevent hot-leg thrashing.
let selected_leg = legs
.iter()
.take(pool_size) // Берем только пул
.min_by(|a, b| {
let score_a = a.stats.rtt_ms.load(Ordering::Relaxed) as f64
+ (a.congestion_factor() * MUXER_CONGESTION_WEIGHT);
@@ -206,8 +204,18 @@ 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;
let measured = start_time.elapsed().as_millis() as u32;
if let Some(leg) = self.legs.get(&leg_id) {
let current = leg.stats.rtt_ms.load(Ordering::Relaxed);
// EWMA with α=0.25: new = (3·old + measured) / 4.
// A single noisy heartbeat (e.g. 300 ms on a 50 ms baseline)
// only moves the stored RTT to ~112 ms instead of jumping
// straight to 300 ms, preventing unnecessary leg re-selection.
let rtt = if current == crate::net::INITIAL_RTT_MS {
measured // first real measurement: accept immediately
} else {
(current.saturating_mul(3).saturating_add(measured)) / 4
};
leg.stats.rtt_ms.store(rtt, Ordering::Relaxed);
let min_rtt = self
.legs
@@ -366,6 +374,11 @@ impl Muxer {
// 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.
//
// TIMEOUT GUARD: if the stream's receive channel stays full for longer than
// DISPATCH_TO_LOCAL_TIMEOUT the stream is forcibly closed. Without this a
// single slow consumer (app socket buffer full, background app paused, etc.)
// would block the engine reader and starve every other stream on the same leg.
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
let size = data.len() as u64;
@@ -375,10 +388,15 @@ impl Muxer {
});
if let Some((tx, stats)) = tx_and_stats {
// .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);
match tokio::time::timeout(DISPATCH_TO_LOCAL_TIMEOUT, tx.send(data)).await {
Ok(Ok(_)) => { stats.rx_bytes.fetch_add(size, Ordering::Relaxed); }
Ok(Err(_)) => { /* receiver already closed — stream gone */ }
Err(_) => {
// Bridge isn't consuming: app socket full or app paused too long.
// Close the stream to free the leg for all other streams.
warn!(stream_id, "dispatch_to_local: stream stalled for {:?}, closing", DISPATCH_TO_LOCAL_TIMEOUT);
self.remove_stream(stream_id);
}
}
}
}
core/src/net/constants.rs
+24
View File
@@ -17,6 +17,14 @@ 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);
/// Max time to wait for a local app socket to accept downloaded data.
/// If the app's receive buffer stays full longer than this, the connection
/// is closed to unblock the tunnel leg for other streams.
pub const BRIDGE_STREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(30);
/// Max time dispatch_to_local will block waiting for a stream's receive channel.
/// Protects the engine reader (and thus the entire tunnel leg) from being stuck
/// behind one slow stream's backlog. On timeout the stream is forcibly closed.
pub const DISPATCH_TO_LOCAL_TIMEOUT: Duration = Duration::from_secs(10);
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);
@@ -26,6 +34,12 @@ pub const DNS_LOOKUP_TIMEOUT: Duration = Duration::from_secs(3);
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;
/// After this many consecutive internal reconnect failures the engine gives up
/// and returns Err to the outer establish_leg loop, which re-runs DNS resolution
/// and resets all counters. 10 × ~18 s ≈ 3 minutes max stuck-silent time.
pub const MAX_INTERNAL_RECONNECT_ATTEMPTS: u32 = 10;
/// Cap for exponential reconnect backoff inside the engine (milliseconds).
pub const MAX_RECONNECT_BACKOFF_MS: u64 = 30_000;
/// 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.
@@ -66,6 +80,16 @@ 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;
// ── Tunnel leg TCP socket tuning ─────────────────────────────────────────────
/// OS-level TCP send buffer for each tunnel leg. The default (4–8 MB on
/// Linux/Android) can hold seconds of data at typical mobile speeds, causing
/// severe jitter. 256 KB limits extra queuing to ~80 ms at 25 Mbit/s per leg
/// while still providing enough headroom for TCP slow-start.
pub const TUNNEL_SOCKET_SNDBUF: u32 = 256 * 1024;
/// OS-level TCP receive buffer for each tunnel leg. Larger than the send
/// buffer so the receiver can absorb bursts without dropping packets.
pub const TUNNEL_SOCKET_RCVBUF: u32 = 512 * 1024;
// ── Smoltcp socket defaults ──────────────────────────────────────────────────
/// Packet slots for the ICMP socket's RX and TX packet buffers.
pub const ICMP_META_SLOTS: usize = 4;