port_forward_mux: cap send credit and other final review fixes

Fix 1 (critical): clamp SendCredit to MAX_SEND_CREDIT (= CHANNEL_WINDOW)
in both new() and add(), so a peer with tunnel permission can no longer
advertise an unbounded window and force the controlled side's unbounded
FrameSink::Direct sink to buffer unlimited target data per channel.

Fix 2: rename the "starve" test to many_channels_echo_concurrently and
drop its (untrue) starvation claim, since it opens every channel before
the bulk transfer starts. Add a_channel_opened_during_a_bulk_transfer_
is_served_promptly, which opens the small channel while the bulk one is
demonstrably mid-flight.

Fix 3: only look up the tunnel permission for `open` frames in the
PortForwardChannel arm of on_message, instead of once per data frame.

Fix 4: two rustfmt deviations in connection.rs (matches! wrapping and a
tuple literal), fixed by hand without a blanket cargo fmt run.

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

src/port_forward_mux.rs

@@ -25,6 +25,10 @@ pub const MAX_FRAME: usize = 64 * 1024;
 pub const UPDATE_THRESHOLD: u32 = CHANNEL_WINDOW / 2;
 pub const MAX_CHANNELS: usize = 256;
 pub const DATA_QUEUE_FRAMES: usize = 128;
+/// Never keep more than our own advertised window in flight, whatever the peer
+/// offers. The controlled side's sink is unbounded, so credit is the only bound
+/// on how much target data it buffers, and the peer chooses that number.
+pub const MAX_SEND_CREDIT: u32 = CHANNEL_WINDOW;
 
 pub fn effective_window(advertised: u32) -> u32 {
     advertised.max(INITIAL_WINDOW)
@@ -86,7 +90,7 @@ pub struct SendCredit {
 impl SendCredit {
     pub fn new(initial: u32) -> Self {
         Self {
-            credit: Mutex::new(initial),
+            credit: Mutex::new(initial.min(MAX_SEND_CREDIT)),
             notify: Notify::new(),
         }
     }
@@ -112,7 +116,7 @@ impl SendCredit {
     pub fn add(&self, n: u32) {
         {
             let mut credit = self.credit.lock().unwrap();
-            *credit = credit.saturating_add(n);
+            *credit = credit.saturating_add(n).min(MAX_SEND_CREDIT);
         }
         self.notify.notify_waiters();
         self.notify.notify_one();
@@ -757,6 +761,19 @@ mod tests {
         });
     }
 
+    #[test]
+    fn send_credit_is_capped_whatever_the_peer_advertises() {
+        rt().block_on(async {
+            let credit = SendCredit::new(u32::MAX);
+            assert_eq!(credit.take(usize::MAX).await, MAX_SEND_CREDIT as usize);
+            // A flood of window updates cannot lift it past the cap either.
+            for _ in 0..10 {
+                credit.add(u32::MAX);
+            }
+            assert_eq!(credit.take(usize::MAX).await, MAX_SEND_CREDIT as usize);
+        });
+    }
+
     #[test]
     fn raise_initial_rebases_credit_from_initial_window() {
         rt().block_on(async {
@@ -1246,7 +1263,7 @@ mod tests {
         }
 
         #[test]
-        fn many_channels_echo_concurrently_and_a_bulk_one_does_not_starve_them() {
+        fn many_channels_echo_concurrently() {
             rt().block_on(async {
                 let (h, port) = muxed_tunnel().await;
                 let mut apps = Vec::new();
@@ -1255,7 +1272,8 @@ mod tests {
                     h.open("127.0.0.1", port as i32, sock, vec![i]).unwrap();
                     apps.push(app);
                 }
-                // Channel 0 streams 4 MiB; the others each expect their one byte back promptly.
+                // Twenty channels round-trip concurrently: channel 0 streams 4 MiB
+                // while the other nineteen each exchange one byte.
                 // Read and write the bulk socket from separate tasks: the echo can only
                 // drain if this side keeps reading while it writes.
                 let bulk = vec![0xAB; 4 << 20];
@@ -1291,6 +1309,41 @@ mod tests {
             });
         }
 
+        #[test]
+        fn a_channel_opened_during_a_bulk_transfer_is_served_promptly() {
+            rt().block_on(async {
+                let (h, port) = muxed_tunnel().await;
+                let (bulk_app, bulk_sock) = local_pair().await;
+                h.open("127.0.0.1", port as i32, bulk_sock, vec![]).unwrap();
+                let bulk = vec![0xAB; 4 << 20];
+                let (mut bulk_rd, mut bulk_wr) = bulk_app.into_split();
+                let bulk_writer = {
+                    let bulk = bulk.clone();
+                    tokio::spawn(async move {
+                        bulk_wr.write_all(&bulk).await.unwrap();
+                        // Holding the write half open: dropping it half-closes
+                        // the socket, which ends the channel by design.
+                        bulk_wr
+                    })
+                };
+                // Wait until a mebibyte is back, so the bulk channel is
+                // demonstrably mid-flight before anything else is opened.
+                let mut back = vec![0u8; 1 << 20];
+                bulk_rd.read_exact(&mut back).await.unwrap();
+                let (mut app, sock) = local_pair().await;
+                h.open("127.0.0.1", port as i32, sock, vec![42]).unwrap();
+                let mut b = [0u8; 1];
+                tokio::time::timeout(std::time::Duration::from_secs(2), app.read_exact(&mut b))
+                    .await
+                    .expect("a channel opened during a bulk transfer starved")
+                    .unwrap();
+                assert_eq!(b[0], 42);
+                let mut rest = vec![0u8; bulk.len() - (1 << 20)];
+                bulk_rd.read_exact(&mut rest).await.unwrap();
+                let _bulk_wr = bulk_writer.await.unwrap();
+            });
+        }
+
         #[test]
         fn a_local_half_close_ends_the_whole_channel() {
             rt().block_on(async {

src/server/connection.rs

@@ -3902,8 +3902,6 @@ impl Connection {
                     }
                 }
                 Some(message::Union::PortForwardChannel(ch)) => {
-                    let permitted =
-                        Self::permission(keys::OPTION_ENABLE_TUNNEL, &self.control_permissions);
                     // Only open/close can change the target set; sweeping after every
                     // data frame would walk the whole table per 64 KiB.
                     let may_change_targets = matches!(
@@ -3911,6 +3909,10 @@ impl Connection {
                         Some(port_forward_channel::Union::Open(_))
                             | Some(port_forward_channel::Union::Close(_))
                     );
+                    // Only `open` consults permissions; short-circuit so the lookup
+                    // isn't made per 64 KiB of data.
+                    let permitted = matches!(ch.union, Some(port_forward_channel::Union::Open(_)))
+                        && Self::permission(keys::OPTION_ENABLE_TUNNEL, &self.control_permissions);
                     if let Some(mux) = self.port_forward_mux.as_mut() {
                         mux.handle(ch, permitted);
                         if may_change_targets {
@@ -5808,7 +5810,10 @@ impl Connection {
     }
 
     fn is_port_forward_scoped_message(msg: &Message) -> bool {
-        matches!(msg.union.as_ref(), Some(message::Union::PortForwardChannel(_)))
+        matches!(
+            msg.union.as_ref(),
+            Some(message::Union::PortForwardChannel(_))
+        )
     }
 
     fn is_terminal_scoped_message(msg: &Message) -> bool {
@@ -7460,7 +7465,10 @@ mod test {
                         }),
                         None,
                     ),
-                    (msg(|m| m.set_port_forward_channel(PortForwardChannel::new())), None),
+                    (
+                        msg(|m| m.set_port_forward_channel(PortForwardChannel::new())),
+                        None,
+                    ),
                 ],
             ),
         ];