Forward ChannelMsg::Close to channel before dropping sender (#674)

## Problem

When the remote side sends `CHANNEL_CLOSE`, the client and server
handlers call `self.channels.remove()` which drops the mpsc `Sender`,
causing `Channel::wait()` to return `None`. No `ChannelMsg::Close` is
ever delivered through the channel.

In contrast, the `CHANNEL_EOF` handler correctly sends `ChannelMsg::Eof`
through the channel's mpsc before returning. This inconsistency means
consumers have no way to distinguish a clean remote close from an
unexpected channel teardown.

## RFC context

[RFC 4254 Section
5.3](https://www.rfc-editor.org/rfc/rfc4254#section-5.3) specifies:

> Upon receiving this message, a party MUST send back an
SSH_MSG_CHANNEL_CLOSE unless it has already sent this message for the
channel. The channel is considered closed for a party when it has both
sent and received SSH_MSG_CHANNEL_CLOSE, and the party may then reuse
the channel number.

Importantly, the RFC also states:

> It is not mandatory to send SSH_MSG_CHANNEL_EOF before
SSH_MSG_CHANNEL_CLOSE.

This means `CHANNEL_CLOSE` can arrive as the first and only termination
signal on a channel -- without a preceding `CHANNEL_EOF`. Without this
fix, that close is invisible to consumers: `Channel::wait()` returns
`None` with no way to distinguish "remote cleanly closed" from
"something dropped the channel handle." The `ChannelMsg::Close` variant
exists in the enum but was never delivered through this path.

## Behavioral change

This changes the observable behavior of `Channel::wait()`: consumers now
receive `Some(ChannelMsg::Close)` before `None` when the remote end
sends `CHANNEL_CLOSE`.

Existing `while let Some(msg) = ch.wait().await` loops with `_ => {}`
catch-alls are **unaffected** -- `Close` is silently ignored, the loop
continues, and the next `wait()` returns `None`. However, loops with
catch-all panics that don't handle `Close` will now panic where they
previously exited cleanly via `None`. Two such patterns in the test
suite are hardened in this patch.

## Changes

All changes are in the `russh` crate. No public API signatures are
changed.

### Core fix (client and server)

In both `client/encrypted.rs` and `server/encrypted.rs`, the
`CHANNEL_CLOSE` handler now sends `ChannelMsg::Close` through the
channel's mpsc sender before removing the channel entry from the map.
This matches the existing `CHANNEL_EOF` pattern in both files.

### Documentation

Updated doc comments on `Channel::wait()` and `ChannelReadHalf::wait()`
to document that `ChannelMsg::Close` is delivered before `None` when the
remote end sends `CHANNEL_CLOSE`.

### Tests

- **`test_channel_objects`**: Updated to assert `ChannelMsg::Close` is
received before `None` on the client side.
- **`test_server_receives_close_on_client_close`**: New test validating
the server-side fix -- client opens a channel and closes it, server
observes `ChannelMsg::Close` before `None`.
- **`test_backpressure`**: Added `Some(ChannelMsg::Close) | None =>
break` to the server handler's match arm.
- **`test_data_stream`**: Added `ChannelMsg::Close` alongside
`ChannelMsg::Eof` in the break arm.

---------

Co-authored-by: Eugene <inbox@null.page>

Author: coreyleavitt

russh/src/client/encrypted.rs

@@ -394,6 +394,12 @@ impl Session {
                     // will not be released.
                     enc.close(channel_num)?;
                 }
+                // Forward the close to the channel before removing it, so that
+                // consumers waiting on `Channel::wait()` receive an explicit
+                // `ChannelMsg::Close` instead of just seeing `None`.
+                if let Some(chan) = self.channels.get(&channel_num) {
+                    let _ = chan.send(ChannelMsg::Close).await;
+                }
                 self.channels.remove(&channel_num);
                 client.channel_close(channel_num, self).await
             }

russh/src/server/encrypted.rs

@@ -602,6 +602,12 @@ impl Session {
                 if let Some(ref mut enc) = self.common.encrypted {
                     enc.channels.remove(&channel_num);
                 }
+                // Forward the close to the channel before removing it, so that
+                // consumers waiting on `Channel::wait()` receive an explicit
+                // `ChannelMsg::Close` instead of just seeing `None`.
+                if let Some(chan) = self.channels.get(&channel_num) {
+                    chan.send(ChannelMsg::Close).await.unwrap_or(())
+                }
                 self.channels.remove(&channel_num);
                 debug!("handler.channel_close {channel_num:?}");
                 handler.channel_close(channel_num, self).await

russh/src/tests.rs

@@ -455,6 +455,13 @@ mod channels {
                     panic!("Unexpected message {msg:?}");
                 }
 
+                // After the server closes the channel, we should receive an
+                // explicit Close message before the channel stream ends.
+                let msg = ch.wait().await.unwrap();
+                assert!(
+                    matches!(msg, ChannelMsg::Close),
+                    "expected Close, got {msg:?}"
+                );
                 assert!(ch.wait().await.is_none());
                 c
             },
@@ -463,6 +470,78 @@ mod channels {
         .await;
     }
 
+    /// Verify that the server-side CHANNEL_CLOSE handler delivers
+    /// `ChannelMsg::Close` before the channel stream ends.
+    #[tokio::test]
+    async fn test_server_receives_close_on_client_close() {
+        #[derive(Debug)]
+        struct Client {}
+
+        impl client::Handler for Client {
+            type Error = crate::Error;
+
+            async fn check_server_key(
+                &mut self,
+                _server_public_key: &crate::keys::ssh_key::PublicKey,
+            ) -> Result<bool, Self::Error> {
+                Ok(true)
+            }
+        }
+
+        struct ServerHandle {
+            channel: Option<tokio::sync::oneshot::Sender<Channel<server::Msg>>>,
+        }
+
+        impl server::Handler for ServerHandle {
+            type Error = crate::Error;
+
+            async fn auth_publickey(
+                &mut self,
+                _: &str,
+                _: &crate::keys::ssh_key::PublicKey,
+            ) -> Result<server::Auth, Self::Error> {
+                Ok(server::Auth::Accept)
+            }
+
+            async fn channel_open_session(
+                &mut self,
+                channel: Channel<server::Msg>,
+                _session: &mut server::Session,
+            ) -> Result<bool, Self::Error> {
+                if let Some(tx) = self.channel.take() {
+                    tx.send(channel).unwrap();
+                }
+                Ok(true)
+            }
+        }
+
+        let (tx, rx) = tokio::sync::oneshot::channel::<Channel<server::Msg>>();
+        let sh = ServerHandle { channel: Some(tx) };
+
+        test_session(
+            Client {},
+            sh,
+            |c| async move {
+                let ch = c.channel_open_session().await.unwrap();
+                ch.close().await.unwrap();
+                c
+            },
+            |s| async move {
+                let mut ch = rx.await.unwrap();
+                // The server should receive an explicit Close message
+                // when the client closes the channel.
+                let msg = ch.wait().await.unwrap();
+                assert!(
+                    matches!(msg, ChannelMsg::Close),
+                    "expected Close, got {msg:?}"
+                );
+                assert!(ch.wait().await.is_none());
+                s
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     async fn test_channel_window_size() {
         #[derive(Debug)]

russh/tests/test_backpressure.rs

@@ -135,6 +135,7 @@ impl russh::server::Handler for Server {
             while let Ok(_) = rx.changed().await {
                 match channel.wait().await {
                     Some(ChannelMsg::Data { .. }) => (),
+                    Some(ChannelMsg::Close) | None => break,
                     other => panic!("unexpected message {other:?}"),
                 }
             }

russh/tests/test_data_stream.rs

@@ -72,7 +72,7 @@ impl ChannelDataCopy for ChannelHalves {
                     match msg {
                         ChannelMsg::WindowAdjusted { .. } => {}
                         ChannelMsg::Data { data } => buf.extend(&*data),
-                        ChannelMsg::Eof => break,
+                        ChannelMsg::Eof | ChannelMsg::Close => break,
                         msg => panic!("Got unexpected message: {msg:?}"),
                     }
                 }