port_forward_mux: a window violation drops the channel on the spot

Both demultiplexers only queued a `Violation` and left the entry until
the channel task woke and exited, so a peer that kept sending past the
window queued one more entry per frame in the meantime, bounded by
nothing. The entry now goes the moment `accept` fails; later frames for
that id are unknown-channel noise.

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

src/port_forward_mux.rs

@@ -551,7 +551,7 @@ mod tunnel {
             Ok(())
         }
 
-        fn on_frame(&self, ch: PortForwardChannel) -> Option<String> {
+        pub(super) fn on_frame(&self, ch: PortForwardChannel) -> Option<String> {
             match ch.union {
                 Some(port_forward_channel::Union::Opened(o)) => {
                     let refused = {
@@ -581,14 +581,18 @@ mod tunnel {
                         log::debug!("port forward data for unknown channel {}", d.channel_id);
                         return None;
                     };
-                    let msg = if e.window.lock().unwrap().accept(d.data.len()) {
-                        Inbound::Data(d.data)
-                    } else {
-                        log::warn!("port forward channel {} overran its window", d.channel_id);
-                        Inbound::Violation
-                    };
-                    if e.inbound.send(msg).is_err() {
-                        channels.remove(&d.channel_id);
+                    let accepted = e.window.lock().unwrap().accept(d.data.len());
+                    let delivered = accepted && e.inbound.send(Inbound::Data(d.data)).is_ok();
+                    if delivered {
+                        return None;
+                    }
+                    // Dropped here and now, so the peer cannot queue anything more
+                    // for this id while the task is still on its way out.
+                    if let Some(e) = channels.remove(&d.channel_id) {
+                        if !accepted {
+                            log::warn!("port forward channel {} overran its window", d.channel_id);
+                            e.inbound.send(Inbound::Violation).ok();
+                        }
                     }
                     None
                 }
@@ -1272,6 +1276,34 @@ mod tests {
             });
         }
 
+        #[test]
+        fn an_over_window_frame_drops_the_channel_at_once() {
+            rt().block_on(async {
+                let (ours, mut peer) = stream_pair().await;
+                let t = Tunnel::new();
+                t.claim();
+                let h = t.set_muxed(ours, NoUi::default());
+                let (_app, sock) = local_pair().await;
+                h.open("localhost", 1, sock, vec![]).unwrap();
+                let id = match recv_frame(&mut peer).await.union {
+                    Some(port_forward_channel::Union::Open(o)) => o.channel_id,
+                    other => panic!("expected open, got {:?}", other),
+                };
+                let mut ch = PortForwardChannel::new();
+                ch.set_data(PortForwardData {
+                    channel_id: id,
+                    data: Bytes::from(vec![0u8; CHANNEL_WINDOW as usize + 1]),
+                    ..Default::default()
+                });
+                h.on_frame(ch.clone());
+                // Gone on the spot: a peer that keeps sending past the window
+                // can no longer queue anything for this channel.
+                assert_eq!(h.live_channels(), 0);
+                h.on_frame(ch);
+                assert_eq!(h.live_channels(), 0);
+            });
+        }
+
         #[test]
         fn tunnel_death_closes_channels_and_resets_state() {
             rt().block_on(async {

src/server/port_forward_mux.rs

@@ -61,14 +61,18 @@ impl PortForwardMux {
                     return;
                 };
                 let accepted = entry.window.lock().unwrap().accept(len);
-                let msg = if accepted {
-                    Inbound::Data(d.data)
-                } else {
-                    log::warn!("port forward channel {} overran its window", d.channel_id);
-                    Inbound::Violation
+                let delivered = accepted && entry.inbound.send(Inbound::Data(d.data)).is_ok();
+                if delivered {
+                    return;
+                }
+                // Dropped here and now, so the peer cannot queue anything more
+                // for this id while the task is still on its way out.
+                let Some(entry) = self.channels.remove(&d.channel_id) else {
+                    return;
                 };
-                if entry.inbound.send(msg).is_err() {
-                    self.channels.remove(&d.channel_id);
+                if !accepted {
+                    log::warn!("port forward channel {} overran its window", d.channel_id);
+                    entry.inbound.send(Inbound::Violation).ok();
                 }
             }
             Some(port_forward_channel::Union::Close(c)) => {
@@ -418,6 +422,29 @@ mod tests {
         });
     }
 
+    #[test]
+    fn an_over_window_frame_drops_the_channel_at_once() {
+        rt().block_on(async {
+            let port = echo_target().await;
+            let (tx, mut rx) = mpsc::unbounded_channel();
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
+            mux.handle(open(1, port), || true);
+            opened(&next_frame(&mut rx).await);
+            let too_much = vec![0u8; CHANNEL_WINDOW as usize + 1];
+            mux.handle(data(1, &too_much), || true);
+            // Gone before the channel task has run: whatever the peer keeps
+            // sending for this id can no longer queue anything.
+            assert_eq!(mux.live_channels(), 0);
+            mux.handle(data(1, &too_much), || true);
+            assert_eq!(mux.live_channels(), 0);
+            let ch = next_frame(&mut rx).await;
+            match &ch.union {
+                Some(port_forward_channel::Union::Close(c)) => assert_eq!(c.channel_id, 1),
+                other => panic!("expected close, got {:?}", other),
+            }
+        });
+    }
+
     #[test]
     fn open_to_a_target_other_than_the_login_target_is_refused() {
         rt().block_on(async {