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

warnings fix

Browse Source
This commit is contained in:
nineAp
2026-06-29 18:07:53 +07:00
parent 6735c481dc
commit 38dfd588c1
6 changed files with 58 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/connection.rs
+20 -10
View File
@@ -8,10 +8,10 @@ use crate::{
handler::{RemoteOpener, StreamHandler},
muxer::{MuxMessage, Muxer},
},
DNS_LOOKUP_TIMEOUT, FALLBACK_CONNECT_TIMEOUT, LEG_RECONNECT_DELAY, LEG_STAGGER_DELAY,
MAX_TUNNEL_LEGS, NETWORK_WATCHER_INTERVAL, SECURE_HANDSHAKE_TIMEOUT,
NetworkConfig, 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,
TOPOLOGY_PRINT_INTERVAL,
},
nrxp::{Codec, Frame, FrameType, TlsBridge},
rawcast::{LocalProtocol, RawCastAdapter, RawCastFrame},
@@ -72,7 +72,6 @@ impl SessionManager {
info!("📊 --- SERVER GLOBAL SESSIONS REPORT ---");
for entry in self.sessions.iter() {
let session_id = entry.key();
let muxer = entry.value();
muxer.print_topology_tree();
}
@@ -111,7 +110,7 @@ impl ClientHandler {
}
pub async fn perform_handshake(
mut stream: tokio::net::TcpStream,
stream: tokio::net::TcpStream,
session_id: &str,
leg_id: u32,
) -> Result<
@@ -126,8 +125,11 @@ impl ClientHandler {
stream.set_nodelay(true).unwrap_or_default();
let mut conn = Connection::new(stream);
let mut session_keys = SessionKeys::new(true);
let ch =
TlsBridge::wrap_client_hello(&BrowserProfile::CHROME_131, STEALTH_FALLBACK_SNI, &session_keys);
let ch = TlsBridge::wrap_client_hello(
&BrowserProfile::CHROME_131,
STEALTH_FALLBACK_SNI,
&session_keys,
);
conn.outbound
.write_all(&ch)
@@ -244,7 +246,11 @@ impl ClientHandler {
let stream = tokio::time::timeout(FALLBACK_CONNECT_TIMEOUT, socket.connect(addr))
.await
.map_err(|_| {
AppError::new(ERR_INFRA_TIMEOUT, "Таймаут подключения", "Connection timeout")
AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут подключения",
"Connection timeout",
)
})?
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сокета", e.to_string()))?;
@@ -332,7 +338,8 @@ impl ClientHandler {
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);
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);
@@ -458,7 +465,10 @@ impl ClientHandler {
});
}
FrameType::Data | FrameType::UdpData => {
if let Some(up_tx) = local_to_upload_tx.get(&local_socket_id).map(|r| r.value().clone()) {
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).
core/src/net/connection/engine.rs
+3 -4
View File
@@ -15,10 +15,9 @@ use tracing::instrument;
use crate::{
net::{
connection::{handler::StreamHandler, muxer::MuxMessage},
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,
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},
};
core/src/net/connection/muxer.rs
+28 -9
View File
@@ -7,9 +7,9 @@ use std::time::Instant;
use tokio::sync::mpsc::Sender;
use tokio_util::sync::CancellationToken;
use crate::net::diagnostics::{self, DiagnosticsEvent, LegMetrics, TunnelMetrics, DIAG_COUNTERS};
use crate::net::{DISPATCH_TO_LOCAL_TIMEOUT, MAX_TUNNEL_LEGS};
use crate::net::{INITIAL_RTT_MS, MUXER_CONGESTION_WEIGHT};
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;
#[derive(Default, Debug)]
@@ -212,7 +212,7 @@ impl Muxer {
// 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
measured // first real measurement: accept immediately
} else {
(current.saturating_mul(3).saturating_add(measured)) / 4
};
@@ -282,7 +282,11 @@ impl Muxer {
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);
stream_ref
.value()
.1
.tx_bytes
.fetch_add(size, Ordering::Relaxed);
}
Ok(())
}
@@ -296,7 +300,11 @@ impl Muxer {
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);
stream_ref
.value()
.1
.tx_bytes
.fetch_add(size, Ordering::Relaxed);
}
Ok(())
}
@@ -305,7 +313,9 @@ impl Muxer {
stream_id,
"Control queue FULL! Dropping non-critical control frame."
);
DIAG_COUNTERS.control_full_drops.fetch_add(1, Ordering::Relaxed);
DIAG_COUNTERS
.control_full_drops
.fetch_add(1, Ordering::Relaxed);
diagnostics::send_diag_event(DiagnosticsEvent::ControlChannelFull {
stream_id,
frame_type: format!("{:?}", dropped.frame_type),
@@ -389,12 +399,18 @@ impl Muxer {
if let Some((tx, stats)) = tx_and_stats {
match tokio::time::timeout(DISPATCH_TO_LOCAL_TIMEOUT, tx.send(data)).await {
Ok(Ok(_)) => { stats.rx_bytes.fetch_add(size, Ordering::Relaxed); }
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);
warn!(
stream_id,
"dispatch_to_local: stream stalled for {:?}, closing",
DISPATCH_TO_LOCAL_TIMEOUT
);
self.remove_stream(stream_id);
}
}
@@ -436,7 +452,10 @@ impl Muxer {
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");
warn!(
leg_id,
"Health check: control channel closed, evicting dead leg"
);
self.remove_leg(leg_id, &tx);
self.remove_stream(probe_stream_id);
continue;