fix: the KCP io throttle reset itself every cycle, so it never throttled

The send and recv arms shared one counter, and an ICMP error on a connected
socket is reported once and then cleared — so the steady state is an
alternation: the send succeeds and clears the counter, the next recv reports
the error and finds the counter at 1, and logs. Every error still wrote a
line, at the ~100/s the previous commit set out to stop, while the
persistent-failure and recovery branches were unreachable.

Use one LogThrottle per direction instead of a hand-rolled counter. That
removes the shared state the bug lived in, drops a third throttling mechanism
in favour of the one already added, and leaves the surrounding `if let Err`
untouched rather than reshaping it into a match.

Also fix test_udp_uat's socket-error arm, the untreated twin of the punch_udp
site: it had no backoff at all, so a persistent error re-armed recv
immediately and spun the loop at CPU speed, one warn line per iteration.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ

src/client.rs

@@ -332,6 +332,7 @@ async fn race_transports_prefer_webrtc<'a, T: 'a>(
 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 UDP_UAT_ERR_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);
 
@@ -5165,7 +5166,12 @@ async fn test_udp_uat(
                         }
                     }
                     Err(e) => {
-                        log::warn!("UDP NAT test socket error: {}", e);
+                        // Same ICMP-driven errors as punch_udp sees. Without a pause this arm
+                        // re-arms recv immediately and spins the loop at CPU speed.
+                        if let Some(n) = UDP_UAT_ERR_LOG.due() {
+                            log::warn!("UDP NAT test socket error x{n}, last: {e}");
+                        }
+                        hbb_common::sleep(0.01).await;
                     }
                 }
             }

src/common.rs

@@ -2653,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_sent:?} packets, {recv_errors} recv errors absorbed");
+                    bail!("UDP punch is timed out, stop sending packets after {:?} packets, {} recv errors absorbed", packets_sent, recv_errors);
                 }
                 let elapsed = last_send_time.elapsed();
 

src/kcp_stream.rs

@@ -19,6 +19,12 @@ pub struct KcpStream {
     stop_sender: Option<oneshot::Sender<()>>,
 }
 
+const KCP_IO_ERR_LOG_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5);
+static KCP_SEND_ERR_LOG: hbb_common::log_throttle::LogThrottle =
+    hbb_common::log_throttle::LogThrottle::new(KCP_IO_ERR_LOG_INTERVAL);
+static KCP_RECV_ERR_LOG: hbb_common::log_throttle::LogThrottle =
+    hbb_common::log_throttle::LogThrottle::new(KCP_IO_ERR_LOG_INTERVAL);
+
 impl KcpStream {
     // Engage KCP's built-in congestion control (nc=0) unless disabled by option: pure turbo
     // (nc=1) keeps blasting a full 1024-segment window through loss, which on constrained
@@ -36,33 +42,6 @@ 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()),
@@ -157,12 +136,11 @@ impl KcpStream {
             // 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
+            // These repeat every 10ms while the socket stays broken, so throttle the line —
-            // socket would otherwise write ~100 lines a second into the log file. One line
+            // debug output is written to the log file. One throttle per direction: an ICMP
-            // when a run starts, one per PERSIST run length so a stuck socket stays visible
+            // error on a connected socket is reported once and cleared, so the steady state is
-            // (the pong timeout takes 60s to reap it, and the session is frozen meanwhile),
+            // an alternation (send succeeds, the next recv reports the error), and a shared
-            // and one when it recovers carrying the total.
+            // counter would be reset by the succeeding direction on every cycle.
-            let mut fail_run = 0u64;
             loop {
                 tokio::select! {
                     _ = &mut stop_receiver => {
@@ -170,18 +148,16 @@ impl KcpStream {
                         break;
                     }
                     Some(data) = output.recv() => {
-                        match udp.send(&data.inner()).await {
+                        if let Err(e) = udp.send(&data.inner()).await {
-                            Ok(_) => Self::note_io_ok(&mut fail_run),
+                            if let Some(n) = KCP_SEND_ERR_LOG.due() {
-                            Err(e) => {
+                                log::debug!("KCP send error x{n} (treated as loss), last: {e}");
-                                Self::note_io_err(&mut fail_run, "send", &e);
-                                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                             }
+                            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;
                                 }
@@ -190,7 +166,9 @@ impl KcpStream {
                                     .await.ok();
                             }
                             Err(e) => {
-                                Self::note_io_err(&mut fail_run, "recv", &e);
+                                if let Some(n) = KCP_RECV_ERR_LOG.due() {
+                                    log::debug!("KCP recv error x{n} (treated as loss), last: {e}");
+                                }
                                 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                             }
                         }