ssh remoting: Fix heartbeat timer and exit conditions (#19557)

This contains a bunch of smallish but nasty fixes:

- Heartbeat timer was never reset after first heartbeat
- Use same return value when stderr is closed as when stdout is closed
- Always check proxy process status since it should also be done when we
get to this point (either it died and our task stopped, or our task
stopped and we dropped the process handle and it was killed on drop)
- make error messages less wrongly-specific


Release Notes:

- N/A

Co-authored-by: Bennet <bennet@zed.dev>

crates/remote/src/ssh_session.rs

@@ -741,8 +741,6 @@ impl SshRemoteClient {
                                 return Ok(());
                             }
 
-                            keepalive_timer.set(cx.background_executor().timer(HEARTBEAT_INTERVAL).fuse());
-
                             if missed_heartbeats != 0 {
                                 missed_heartbeats = 0;
                                 this.update(&mut cx, |this, mut cx| {
@@ -784,6 +782,8 @@ impl SshRemoteClient {
                             }
                         }
                     }
+
+                    keepalive_timer.set(cx.background_executor().timer(HEARTBEAT_INTERVAL).fuse());
                 }
             }
         })
@@ -874,7 +874,7 @@ impl SshRemoteClient {
                     .read(&mut stderr_buffer[stderr_offset..])
                     .await?;
                 if len == 0 {
-                    return Err(anyhow!("stderr is closed"));
+                    return anyhow::Ok(());
                 }
 
                 stderr_offset += len;
@@ -912,8 +912,9 @@ impl SshRemoteClient {
                 }
             };
 
+            let status = ssh_proxy_process.status().await?.code().unwrap_or(1);
             match result {
-                Ok(_) => Ok(ssh_proxy_process.status().await?.code().unwrap_or(1)),
+                Ok(_) => Ok(status),
                 Err(error) => Err(error),
             }
         })

crates/remote_server/src/unix.rs

@@ -475,19 +475,13 @@ pub fn execute_proxy(identifier: String, is_reconnecting: bool) -> Result<()> {
 
     if let Err(forwarding_result) = smol::block_on(async move {
         futures::select! {
-            result = stdin_task.fuse() => result,
-            result = stdout_task.fuse() => result,
-            result = stderr_task.fuse() => result,
+            result = stdin_task.fuse() => result.context("stdin_task failed"),
+            result = stdout_task.fuse() => result.context("stdout_task failed"),
+            result = stderr_task.fuse() => result.context("stderr_task failed"),
         }
     }) {
-        if let Some(error) = forwarding_result.downcast_ref::<std::io::Error>() {
-            if error.kind() == std::io::ErrorKind::UnexpectedEof {
-                log::error!("connection to server closed due to unexpected EOF");
-                return Err(anyhow!("connection to server closed"));
-            }
-        }
         log::error!(
-            "failed to forward messages: {:?}, terminating...",
+            "encountered error while forwarding messages: {:?}, terminating...",
             forwarding_result
         );
         return Err(forwarding_result);