DAP: Drain pending requests on recv failure

This matches <https://redirect.github.com/helix-editor/helix/pull/4852>
for the DAP transport: when there is a failure to receive a message from
the debugger we should drain all pending requests and respond to them
with the StreamClosed error.

This improves the behavior when a debugger fails to initialize, for
example starting debugpy without debugpy installed. Previously the UI
would freeze until the request timed out. Now it instantly prints a
statusline error saying that the debugger failed to start up.

helix-dap/src/transport.rs

@@ -236,25 +236,37 @@ impl Transport {
         }
     }
 
-    async fn recv_inner(
+    async fn recv(
         transport: Arc<Self>,
         mut server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
         client_tx: UnboundedSender<Payload>,
-    ) -> Result<()> {
+    ) {
         let mut recv_buffer = String::new();
         loop {
-            let msg = Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await?;
+            match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).await {
-            transport.process_server_message(&client_tx, msg).await?;
+                Ok(msg) => match transport.process_server_message(&client_tx, msg).await {
-        }
+                    Ok(_) => (),
-    }
+                    Err(err) => {
+                        error!("err: <- {err:?}");
+                        break;
+                    }
+                },
+                Err(err) => {
+                    if !matches!(err, Error::StreamClosed) {
+                        error!("Exiting after unexpected error: {err:?}");
+                    }
 
-    async fn recv(
+                    // Close any outstanding requests.
-        transport: Arc<Self>,
+                    for (id, tx) in transport.pending_requests.lock().await.drain() {
-        server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
+                        match tx.send(Err(Error::StreamClosed)).await {
-        client_tx: UnboundedSender<Payload>,
+                            Ok(_) => (),
-    ) {
+                            Err(_) => {
-        if let Err(err) = Self::recv_inner(transport, server_stdout, client_tx).await {
+                                error!("Could not close request on a closed channel (id={id})");
-            error!("err: <- {:?}", err);
+                            }
+                        }
+                    }
+                }
+            }
         }
     }