webrtc: end the ICE forwarder at gathering complete, keep the reassembly buffer

The local-candidate channel's only sender lived in the on_ice_candidate
handler until the pc reached a terminal state, so the forwarder loop a
caller writes on take_local_ice_rx sat parked on recv() for the whole
session. On the controlled side that loop holds its dedicated signaling
connection to hbbs, which hbbs closes after 90s idle, leaving a task and
a CLOSE_WAIT socket per session until the session ended. Gathering
complete is the agent's last word short of an ICE restart, which nothing
here does, so drop the sender there; the terminal-state handler's swap
stays for a pc that dies before it gets that far.

Reassembly handed the accumulated message over with mem::take, which
left the accumulator with no capacity and had every fragmented frame
regrow it from nothing, reallocating and recopying as it went. split()
hands over the same bytes and keeps the allocation, which reserve()
reclaims once the caller has dropped the message.

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

src/webrtc.rs

@@ -637,14 +637,20 @@ impl WebRTCStream {
         let pc = Arc::new(api.new_peer_connection(config).await?);
         // `on_*` handlers are never cleared — not by `close()`, not by `SESSIONS` eviction — so
         // this sender outlives the pc and would park a caller's ICE forwarder on `recv()` forever,
-        // holding the pc with it. The terminal-state handler below drops this closure to break it.
-        let local_ice_tx = ice_tx.clone();
+        // holding the pc with it. Gathering complete (`None`) is the agent's last word short of an
+        // ICE restart, which nothing here does, so the sender goes with it and the forwarder ends
+        // there rather than with the session; the terminal-state handler below drops this closure
+        // for a pc that dies before it gets that far.
+        let mut local_ice_tx = Some(ice_tx);
         pc.on_ice_candidate(Box::new(move |candidate| {
-            let local_ice_tx = local_ice_tx.clone();
+            let Some(candidate) = candidate else {
+                local_ice_tx = None;
+                return Box::pin(async {});
+            };
+            let Some(local_ice_tx) = local_ice_tx.clone() else {
+                return Box::pin(async {});
+            };
             Box::pin(async move {
-                let Some(candidate) = candidate else {
-                    return;
-                };
                 match candidate.to_json() {
                     Ok(candidate) => match serde_json::to_string(&candidate) {
                         Ok(candidate_json) => {
@@ -1426,7 +1432,10 @@ impl WebRTCStream {
             }
             acc.extend_from_slice(&scratch[1..n]);
             if header == FRAG_END {
-                let msg = std::mem::take(acc);
+                // Split, not take: the caller has dropped the message by the time the next
+                // fragment lands, so `reserve` reclaims this allocation instead of regrowing it
+                // from nothing for every fragmented frame.
+                let msg = acc.split();
                 return Some(Ok(msg));
             }
         }
@@ -2070,7 +2079,9 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc
     // The local-candidate channel must close when the pc does. Its only sender lives inside the
     // on_ice_candidate handler, which close() does not clear, so without the teardown the
     // receiver stays open forever — and the forwarder loop this API asks callers to write holds
-    // a stream clone while parked on recv(), keeping the pc alive past its own close.
+    // a stream clone while parked on recv(), keeping the pc alive past its own close. Gathering
+    // complete drops the sender as well, and may beat the close here; a pc closed before that
+    // has only the teardown, which is what this pins.
     #[tokio::test]
     async fn test_local_ice_channel_closes_with_the_peer_connection() {
         let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
@@ -2093,6 +2104,24 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc
             .unwrap();
     }
 
+    // The channel must close when gathering does, with the pc still open: nothing can follow the
+    // agent's last candidate, and a forwarder parked on recv() until the session ends holds its
+    // signaling connection for that whole time — long after the server's idle close, as a socket
+    // left in CLOSE_WAIT.
+    #[tokio::test]
+    async fn test_local_ice_channel_closes_when_gathering_completes() {
+        let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
+        let mut ice_rx = offerer.take_local_ice_rx().unwrap();
+        let drained = timeout(Duration::from_secs(20), async {
+            while ice_rx.recv().await.is_some() {}
+        })
+        .await;
+        // Close before asserting, or a failure here strands the pc and fails
+        // `test_cancelled_new_does_not_leak_the_pc` as well.
+        offerer.close().await;
+        drained.expect("local ICE channel outlived gathering");
+    }
+
     // Extra channels must not displace the bound one: the newcomer's on_open would push Open onto
     // the watch that gates send and recv, re-arming a session already latched Closed. Only the
     // bind-once guard is exercised; the ordered+reliable check cannot decide anything here.