From 26db54155ef303b9ba585f8560840189ad0b1681 Mon Sep 17 00:00:00 2001
From: Michael Davis <mcarsondavis@gmail.com>
Date: Tue, 4 Feb 2025 09:09:54 -0500
Subject: [PATCH] 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 | 38 +++++++++++++++++++++++++-------------
 1 file changed, 25 insertions(+), 13 deletions(-)

diff --git a/helix-dap/src/transport.rs b/helix-dap/src/transport.rs
index 1b1fca5f..c1e15b3f 100644
--- a/helix-dap/src/transport.rs
+++ b/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?;
-            transport.process_server_message(&client_tx, msg).await?;
-        }
-    }
+            match Self::recv_server_message(&mut server_stdout, &mut recv_buffer).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(
-        transport: Arc<Self>,
-        server_stdout: Box<dyn AsyncBufRead + Unpin + Send>,
-        client_tx: UnboundedSender<Payload>,
-    ) {
-        if let Err(err) = Self::recv_inner(transport, server_stdout, client_tx).await {
-            error!("err: <- {:?}", err);
+                    // Close any outstanding requests.
+                    for (id, tx) in transport.pending_requests.lock().await.drain() {
+                        match tx.send(Err(Error::StreamClosed)).await {
+                            Ok(_) => (),
+                            Err(_) => {
+                                error!("Could not close request on a closed channel (id={id})");
+                            }
+                        }
+                    }
+                }
+            }
         }
     }