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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ

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!(
+                                if let Some(n) = PENDING_ICE_FULL_LOG.due() {
-                                    "WebRTC ICE pending buffer full ({}), dropping oldest candidate",
+                                    log::warn!(
-                                    Self::MAX_PENDING_WEBRTC_ICE
+                                        "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,
                             );
                         }

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) => {

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<SocketAddr>) -> 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 {
+                        match udp.send(&data.inner()).await {
-                            log::debug!("KCP send error (treated as loss): {:?}", e);
+                            Ok(_) => Self::note_io_ok(&mut fail_run),
-                            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+                            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::<KcpPacketHeader>() {
                                     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;
                             }
                         }

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<HashMap<String, mpsc::UnboundedSender<String>>> = 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
+                        );
+                    }
                 }
             }
         });