fix(ssh): report a failed backend upload instead of waiting forever

The IDE backend is uploaded by streaming it into `gzip -dc > file` on the
host. When that command exits early (no gzip, a folder it cannot write, a
full disk), it stops reading: once the channel window (about 2 MB) is
used up, the next write waited for window space forever, so Remote
Development stayed on "Installing the IDE backend on the host…" at a few
percent and never showed the host's error. Each write, and closing the
input, now races the command's exit; an early exit ends the upload with
the command's status and stderr.

Refs #1330

crates/ide-ssh/src/session.rs

@@ -267,19 +267,29 @@ impl Session {
                 }),
             )
             .await?;
+        let failed = |code: Option<u32>| {
+            let err = String::from_utf8_lossy(&stderr.lock()).trim().to_owned();
+            io(format!("cannot write {remote} on {} ({code:?}): {err}", self.label))
+        };
+        // A command that exits early (no gzip, a folder it cannot write) stops reading: a
+        // write would wait for channel window forever, so each one races the exit.
+        let mut exit = std::pin::pin!(process.wait());
         let mut sent = 0;
         for chunk in packed.chunks(256 * 1024) {
-            process.write(chunk).await?;
+            tokio::select! {
+                written = process.write(chunk) => written?,
+                code = &mut exit => return Err(failed(code)),
+            }
             sent += chunk.len();
             progress(sent as f64 / packed.len().max(1) as f64);
         }
-        process.eof().await?;
-        match process.wait().await {
+        tokio::select! {
+            closed = process.eof() => closed?,
+            code = &mut exit => return Err(failed(code)),
+        }
+        match exit.await {
             Some(0) => Ok(()),
-            code => {
-                let err = String::from_utf8_lossy(&stderr.lock()).trim().to_owned();
-                Err(io(format!("cannot write {remote} on {} ({code:?}): {err}", self.label)))
-            }
+            code => Err(failed(code)),
         }
     }
 

crates/ide-ssh/tests/session.rs

@@ -269,3 +269,26 @@ async fn runs_commands_and_streams_processes() {
     assert_eq!(process.wait().await, Some(3));
     assert_eq!(*seen.lock(), ("ran: python3 main.py".to_owned(), "oops".to_owned()));
 }
+
+#[tokio::test]
+async fn upload_reports_a_command_that_stops_reading() {
+    let env = env().await;
+    let session = env.connect_trusted(&env.target(SshAuth::Password), Some("secret")).await.ok().unwrap();
+    // Incompressible and larger than the channel window: the command exits (status 3, `oops`)
+    // without reading it, as when the host has no gzip or the folder is not writable.
+    let mut state = 0x9e37_79b9_7f4a_7c15_u64;
+    let data: Vec<u8> = (0..8 << 20)
+        .map(|_| {
+            state ^= state << 13;
+            state ^= state >> 7;
+            state ^= state << 17;
+            state as u8
+        })
+        .collect();
+    let put = session.put("/home/dev/.openidea/server/openidea-server", &data);
+    let err = tokio::time::timeout(std::time::Duration::from_secs(20), put)
+        .await
+        .expect("the upload must end when the command exits")
+        .unwrap_err();
+    assert!(err.message.contains("oops"), "{}", err.message);
+}