review: time the inactivity window off receive progress, bound the parting send

Two things the review found, both on the controlling side.

The 30s inactivity window still ran off completed messages alone, so the probe
added for the fast path did not fix what it was added for: a message larger than
the transport's fragment size yields nothing until its last fragment, and a peer
sending one steadily was still timed out mid-transfer. It is now timed off
whichever is later, a completed message or receive progress. Transports that
report no progress leave that at its starting value, so nothing else moves.

The parting close-reason send for KCP waited on send capacity with no deadline
of its own, and a queue a dead peer will never drain held the finished session's
thread until the endpoint reaped the connection a minute later. Bounded once the
peer has been declared gone. Still attempted rather than skipped: if the loss was
one-way the peer does receive it, and drops its side immediately instead of
waiting out its own timeout - which is also the one case where the note below
resolves itself.

Recorded from the same review, for the case none of this targets - a peer that
is alive behind a path that broke for five to ten seconds and then healed.
Giving up cannot deliver a close there, because the path is still down at that
moment, so the controlled side keeps the old connection until its own 30s
expires. For up to twenty of those seconds it holds two authorised connections:
its connection manager lists both, and the stale one reports a growing delay
that pins the shared frame rate low for the new one. Input is unaffected
throughout and both recover once the stale connection goes, so this trades
twenty-two seconds of a frozen, uncontrollable session for a controllable one
that looks wrong for a while. Closing the displaced connection is controlled-side
work and belongs with the rest of it, not here.

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

src/client/io_loop.rs

@@ -15,6 +15,7 @@ use crate::{
 // Restart msgbox text is kept as a legacy UI fallback; Flutter handles the type as a control event.
 const RESTART_REMOTE_DEVICE_NO_DATA_TIMEOUT: Duration = Duration::from_secs(5);
 const KCP_CLOSE_REASON_FLUSH_DELAY: Duration = Duration::from_millis(30);
+const KCP_CLOSE_REASON_GONE_TIMEOUT: u64 = 500;
 // A peer that is killed, logged out or rebooted sends nothing at all over UDP, so the session
 // sees silence and only a timeout ends it. These bound that wait for the two transports that
 // have a liveness signal of their own; TCP and WebSocket are left exactly as they were, and the
@@ -266,6 +267,8 @@ impl<T: InvokeUiSession> Remote<T> {
                 let mut last_recv_time = Instant::now();
                 let mut webrtc_suspect_since: Option<Instant> = None;
                 let mut last_rx_progress = peer.rx_progress();
+                let mut last_rx_progress_at = last_recv_time;
+                let mut peer_gone = false;
 
                 loop {
                     tokio::select! {
@@ -311,7 +314,12 @@ impl<T: InvokeUiSession> Remote<T> {
                             self.handle_local_clipboard_msg(&mut peer, _msg).await;
                         }
                         _ = self.timer.tick() => {
-                            if last_recv_time.elapsed() >= SEC30 {
+                            // Not `last_recv_time` alone: a message larger than the transport's
+                            // fragment size yields nothing until its last fragment, so a peer
+                            // sending one steadily - a clipboard image is the case that occurs -
+                            // would otherwise be timed out mid-transfer. Transports that report no
+                            // progress leave this at its starting value and are unaffected.
+                            if last_recv_time.max(last_rx_progress_at).elapsed() >= SEC30 {
                                 self.handler.msgbox("error", "Connection Error", "Timeout", "");
                                 break;
                             }
@@ -337,12 +345,15 @@ impl<T: InvokeUiSession> Remote<T> {
                             let rx_progress = peer.rx_progress();
                             let progressed = rx_progress.is_some() && rx_progress != last_rx_progress;
                             last_rx_progress = rx_progress;
+                            if progressed {
+                                last_rx_progress_at = Instant::now();
+                            }
                             if peer.webrtc_disconnected() && !progressed {
                                 webrtc_suspect_since.get_or_insert_with(Instant::now);
                             } else {
                                 webrtc_suspect_since = None;
                             }
-                            let peer_gone = webrtc_suspect_since
+                            peer_gone = webrtc_suspect_since
                                 .map_or(false, |since| since.elapsed() >= WEBRTC_SUSPECT_GRACE)
                                 || kcp
                                     .as_ref()
@@ -398,6 +409,15 @@ impl<T: InvokeUiSession> Remote<T> {
                     s.send(()).ok();
                 }
                 if kcp.is_some() {
+                    // Bounded once the peer has been declared gone: KCP waits for send capacity
+                    // with no deadline of its own, and a queue that a dead peer will never drain
+                    // would hold this thread until the endpoint reaps the connection. Still
+                    // attempted rather than skipped - if the loss was one-way the peer does get
+                    // it, and drops its side of the session instead of waiting out its own
+                    // timeout.
+                    if peer_gone {
+                        peer.set_send_timeout(KCP_CLOSE_REASON_GONE_TIMEOUT);
+                    }
                     // Send the close reason if it hasn't been sent yet, as KCP cannot detect the socket close event.
                     self.send_close_reason(&mut peer, "kcp").await;
                     // KCP does not send messages immediately, so wait to ensure the last message is sent.