From 3d5406aa1c1733363bc57347e2565cc107edfa0e Mon Sep 17 00:00:00 2001 From: rustdesk Date: Thu, 6 Aug 2026 13:43:01 +0800 Subject: [PATCH] fix: bound log volume on sites whose rate a peer or retry loop controls Debug output goes to the log file, so a site that fires per received message or per retry lets someone else decide how much a machine writes to disk. The WebRTC work added the first such sites. - KCP io loop: absorbing ICMP errors as packet loss made a broken socket write ~100 lines a second for the 60s until the pong timeout reaps it. Log by run instead: one line when a run starts, one per ~5s while it persists so a stuck socket stays visible, and one on recovery with the total. - punch_udp: the recv error retries every 10ms for up to MAX_TIME, so one line per occurrence wrote thousands per punch. Log the first, report the count in the timeout message. - ICE candidate paths (client, mediator): the peer sets the candidate rate and the rendezvous route carrying them needs no prior punch, so throttle to one line a minute each with the suppressed count. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ --- libs/hbb_common | 2 +- src/client.rs | 33 +++++++++++++++++++++------ src/common.rs | 11 +++++++-- src/kcp_stream.rs | 46 ++++++++++++++++++++++++++++++++++---- src/rendezvous_mediator.rs | 22 +++++++++++++++--- 5 files changed, 97 insertions(+), 17 deletions(-) diff --git a/libs/hbb_common b/libs/hbb_common index 6aa8fbe46..d18dcee6a 160000 --- a/libs/hbb_common +++ b/libs/hbb_common @@ -1 +1 @@ -Subproject commit 6aa8fbe46b3773470ec7f15ead2702b445786814 +Subproject commit d18dcee6a1f6ee85f897747bb65967bec718a44c diff --git a/src/client.rs b/src/client.rs index 3d3a7feca..bb71de367 100644 --- a/src/client.rs +++ b/src/client.rs @@ -326,6 +326,15 @@ async fn race_transports_prefer_webrtc<'a, T: 'a>( } } +// A peer decides how many ICE candidates it sends, and the rendezvous route that carries them is +// reachable without a prior punch, so these sites would otherwise let someone else set how much +// this machine writes to its log file. One line a minute each, carrying the suppressed count. +use hbb_common::log_throttle::LogThrottle; +const ICE_LOG_INTERVAL: Duration = Duration::from_secs(60); +static REJECTED_ICE_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); +static UNEXPECTED_ICE_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); +static PENDING_ICE_FULL_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); + fn request_allows_tcp_punch(webrtc_sdp_offer: &str) -> bool { // WebRTC trickle ICE retains the rendezvous socket as its signaling bridge. Only a request // without an offer may close that socket and reuse its local address for TCP punching. @@ -728,7 +737,13 @@ impl Client { if let Err(err) = webrtc.add_remote_ice_candidate(&ice.candidate).await { - log::warn!("failed to add WebRTC ICE candidate: {}", err); + if let Some(n) = REJECTED_ICE_LOG.due() { + log::warn!( + "failed to add {} WebRTC ICE candidate(s), last: {}", + n, + err + ); + } } } } @@ -1157,16 +1172,20 @@ impl Client { // would discard exactly the ones that traverse NAT and keep the // host ones that only work on a shared LAN. if pending_webrtc_ice.len() >= Self::MAX_PENDING_WEBRTC_ICE { - log::warn!( - "WebRTC ICE pending buffer full ({}), dropping oldest candidate", - Self::MAX_PENDING_WEBRTC_ICE - ); + if let Some(n) = PENDING_ICE_FULL_LOG.due() { + log::warn!( + "WebRTC ICE pending buffer full ({}), evicted {} oldest", + Self::MAX_PENDING_WEBRTC_ICE, + n + ); + } pending_webrtc_ice.remove(0); } pending_webrtc_ice.push(ice.candidate); - } else { + } else if let Some(n) = UNEXPECTED_ICE_LOG.due() { log::debug!( - "dropping ICE candidate for unexpected WebRTC session key {}", + "dropped {} ICE candidate(s) for unexpected WebRTC session key, last: {}", + n, ice.session_key, ); } diff --git a/src/common.rs b/src/common.rs index 26f25791a..2bf64634a 100644 --- a/src/common.rs +++ b/src/common.rs @@ -2642,6 +2642,7 @@ pub async fn punch_udp( const MAX_INTERVAL: Duration = Duration::from_millis(200); const MAX_TIME: Duration = Duration::from_secs(20); let mut packets_sent = 0; + let mut recv_errors = 0u32; socket.send(&[]).await.ok(); packets_sent += 1; let mut last_send_time = Instant::now(); @@ -2652,7 +2653,7 @@ pub async fn punch_udp( tokio::select! { _ = hbb_common::sleep(retry_interval.as_secs_f32()) => { if tm.elapsed() > MAX_TIME { - bail!("UDP punch is timed out, stop sending packets after {:?} packets", packets_sent); + bail!("UDP punch is timed out, stop sending packets after {packets_sent:?} packets, {recv_errors} recv errors absorbed"); } let elapsed = last_send_time.elapsed(); @@ -2674,7 +2675,13 @@ pub async fn punch_udp( // is expected and surfaces as ConnectionReset/Refused on a connected // socket (notably 10054 on Windows). Treat it as loss and keep punching; // MAX_TIME above still bounds the whole attempt. - log::debug!("UDP punch recv error (treated as loss): {e}"); + // Log only the first: this retries every 10ms for up to MAX_TIME, so one + // line per occurrence would write thousands into the log file per punch. + // The count is reported once at the end. + recv_errors += 1; + if recv_errors == 1 { + log::debug!("UDP punch recv error (treated as loss): {e}"); + } hbb_common::sleep(0.01).await; } Ok(n) => { diff --git a/src/kcp_stream.rs b/src/kcp_stream.rs index fac98b89c..488552ac2 100644 --- a/src/kcp_stream.rs +++ b/src/kcp_stream.rs @@ -36,6 +36,33 @@ impl KcpStream { } } + // One line per run of socket failures rather than per failure: these are absorbed as packet + // loss and can repeat every 10ms, and debug output is written to the log file. + const KCP_IO_FAIL_PERSIST: u64 = 500; // ~5s at the 10ms retry interval + + fn note_io_err(fail_run: &mut u64, what: &str, e: &std::io::Error) { + *fail_run += 1; + if *fail_run == 1 { + log::debug!("KCP {} error (treated as loss): {:?}", what, e); + } else if *fail_run % Self::KCP_IO_FAIL_PERSIST == 0 { + // Still failing well past a transient ICMP: say so once per run length, else a + // socket that never recovers is silent until KCP reaps it 60s later. + log::warn!( + "KCP {} failing persistently: {} consecutive errors, last: {:?}", + what, + fail_run, + e + ); + } + } + + fn note_io_ok(fail_run: &mut u64) { + if *fail_run >= Self::KCP_IO_FAIL_PERSIST { + log::info!("KCP socket recovered after {} failed operations", fail_run); + } + *fail_run = 0; + } + fn create_framed(stream: stream::KcpStream, local_addr: Option) -> Stream { Stream::Tcp(FramedStream( tokio_util::codec::Framed::new(DynTcpStream(Box::new(stream)), BytesCodec::new()), @@ -129,6 +156,13 @@ impl KcpStream { // Treat socket errors as packet loss instead of tearing the session down; // a truly dead link is reaped by the KCP pong timeout / app-level timeouts. // The short sleep prevents a persistently failing socket from busy-spinning. + // + // Log by run, not per occurrence: the 10ms sleep means a persistently failing + // socket would otherwise write ~100 lines a second into the log file. One line + // when a run starts, one per PERSIST run length so a stuck socket stays visible + // (the pong timeout takes 60s to reap it, and the session is frozen meanwhile), + // and one when it recovers carrying the total. + let mut fail_run = 0u64; loop { tokio::select! { _ = &mut stop_receiver => { @@ -136,14 +170,18 @@ impl KcpStream { break; } Some(data) = output.recv() => { - if let Err(e) = udp.send(&data.inner()).await { - log::debug!("KCP send error (treated as loss): {:?}", e); - tokio::time::sleep(std::time::Duration::from_millis(10)).await; + match udp.send(&data.inner()).await { + Ok(_) => Self::note_io_ok(&mut fail_run), + Err(e) => { + Self::note_io_err(&mut fail_run, "send", &e); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } } } result = udp.recv_from(&mut buf) => { match result { Ok((size, _)) => { + Self::note_io_ok(&mut fail_run); if size < std::mem::size_of::() { continue; } @@ -152,7 +190,7 @@ impl KcpStream { .await.ok(); } Err(e) => { - log::debug!("KCP recv_from error (treated as loss): {:?}", e); + Self::note_io_err(&mut fail_run, "recv", &e); tokio::time::sleep(std::time::Duration::from_millis(10)).await; } } diff --git a/src/rendezvous_mediator.rs b/src/rendezvous_mediator.rs index 1351d75fc..f5d51525b 100644 --- a/src/rendezvous_mediator.rs +++ b/src/rendezvous_mediator.rs @@ -55,6 +55,15 @@ lazy_static::lazy_static! { static ref LAST_RELAY_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now())); static ref WEBRTC_ICE_TXS: Mutex>> = Default::default(); } +// The rendezvous ICE route is reachable without a prior punch and the peer decides how many +// candidates it sends, so these sites would let someone else set how much this machine writes to +// its log file. One line a minute each, carrying the suppressed count. +const ICE_LOG_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60); +static UNKNOWN_ICE_SESSION_LOG: hbb_common::log_throttle::LogThrottle = + hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL); +static REJECTED_REMOTE_ICE_LOG: hbb_common::log_throttle::LogThrottle = + hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL); + static SHOULD_EXIT: AtomicBool = AtomicBool::new(false); static MANUAL_RESTARTED: AtomicBool = AtomicBool::new(false); static SENT_REGISTER_PK: AtomicBool = AtomicBool::new(false); @@ -410,9 +419,10 @@ impl RendezvousMediator { let tx = WEBRTC_ICE_TXS.lock().await.get(&ice.session_key).cloned(); if let Some(tx) = tx { let _ = tx.send(ice.candidate); - } else { + } else if let Some(n) = UNKNOWN_ICE_SESSION_LOG.due() { log::debug!( - "dropping ICE candidate for unknown WebRTC session key {}", + "dropped {} ICE candidate(s) for unknown WebRTC session key, last: {}", + n, ice.session_key ); } @@ -718,7 +728,13 @@ impl RendezvousMediator { while let Some(candidate) = remote_ice_rx.recv().await { if let Err(err) = stream_for_remote_ice.add_remote_ice_candidate(&candidate).await { - log::warn!("failed to add remote WebRTC ICE candidate: {}", err); + if let Some(n) = REJECTED_REMOTE_ICE_LOG.due() { + log::warn!( + "failed to add {} remote WebRTC ICE candidate(s), last: {}", + n, + err + ); + } } } });