debugger: Kill debug sessions on app quit (#33273)

Before this PR force quitting Zed would leave hanging debug adapter
processes and not allow debug adapters to clean up their sessions
properly.

This PR fixes this problem by sending a disconnect/terminate to all
debug adapters and force shutting down their processes after they
respond.

Co-authored-by: Cole Miller \<cole@zed.dev\>

Release Notes:

- debugger: Shutdown and clean up debug processes when force quitting
Zed

---------

Co-authored-by: Conrad Irwin <conrad.irwin@gmail.com>
Co-authored-by: Remco Smits <djsmits12@gmail.com>

crates/dap/src/client.rs

@@ -163,8 +163,9 @@ impl DebugAdapterClient {
         self.sequence_count.fetch_add(1, Ordering::Relaxed)
     }
 
-    pub async fn shutdown(&self) -> Result<()> {
-        self.transport_delegate.shutdown().await
+    pub fn kill(&self) {
+        log::debug!("Killing DAP process");
+        self.transport_delegate.transport.lock().kill();
     }
 
     pub fn has_adapter_logs(&self) -> bool {
@@ -315,8 +316,6 @@ mod tests {
             },
             response
         );
-
-        client.shutdown().await.unwrap();
     }
 
     #[gpui::test]
@@ -368,8 +367,6 @@ mod tests {
             called_event_handler.load(std::sync::atomic::Ordering::SeqCst),
             "Event handler was not called"
         );
-
-        client.shutdown().await.unwrap();
     }
 
     #[gpui::test]
@@ -433,7 +430,5 @@ mod tests {
             called_event_handler.load(std::sync::atomic::Ordering::SeqCst),
             "Event handler was not called"
         );
-
-        client.shutdown().await.unwrap();
     }
 }

crates/dap/src/transport.rs

@@ -63,7 +63,7 @@ pub trait Transport: Send + Sync {
             Box<dyn AsyncRead + Unpin + Send + 'static>,
         )>,
     >;
-    fn kill(&self);
+    fn kill(&mut self);
     #[cfg(any(test, feature = "test-support"))]
     fn as_fake(&self) -> &FakeTransport {
         unreachable!()
@@ -93,12 +93,18 @@ async fn start(
 
 pub(crate) struct TransportDelegate {
     log_handlers: LogHandlers,
-    pending_requests: Requests,
+    pub(crate) pending_requests: Requests,
     pub(crate) transport: Mutex<Box<dyn Transport>>,
-    server_tx: smol::lock::Mutex<Option<Sender<Message>>>,
+    pub(crate) server_tx: smol::lock::Mutex<Option<Sender<Message>>>,
     tasks: Mutex<Vec<Task<()>>>,
 }
 
+impl Drop for TransportDelegate {
+    fn drop(&mut self) {
+        self.transport.lock().kill()
+    }
+}
+
 impl TransportDelegate {
     pub(crate) async fn start(binary: &DebugAdapterBinary, cx: &mut AsyncApp) -> Result<Self> {
         let log_handlers: LogHandlers = Default::default();
@@ -354,7 +360,6 @@ impl TransportDelegate {
         let mut content_length = None;
         loop {
             buffer.truncate(0);
-
             match reader.read_line(buffer).await {
                 Ok(0) => return ConnectionResult::ConnectionReset,
                 Ok(_) => {}
@@ -412,21 +417,6 @@ impl TransportDelegate {
         ConnectionResult::Result(message)
     }
 
-    pub async fn shutdown(&self) -> Result<()> {
-        log::debug!("Start shutdown client");
-
-        if let Some(server_tx) = self.server_tx.lock().await.take().as_ref() {
-            server_tx.close();
-        }
-
-        self.pending_requests.lock().clear();
-        self.transport.lock().kill();
-
-        log::debug!("Shutdown client completed");
-
-        anyhow::Ok(())
-    }
-
     pub fn has_adapter_logs(&self) -> bool {
         self.transport.lock().has_adapter_logs()
     }
@@ -546,7 +536,7 @@ impl Transport for TcpTransport {
         true
     }
 
-    fn kill(&self) {
+    fn kill(&mut self) {
         if let Some(process) = &mut *self.process.lock() {
             process.kill();
         }
@@ -613,13 +603,13 @@ impl Transport for TcpTransport {
 impl Drop for TcpTransport {
     fn drop(&mut self) {
         if let Some(mut p) = self.process.lock().take() {
-            p.kill();
+            p.kill()
         }
     }
 }
 
 pub struct StdioTransport {
-    process: Mutex<Child>,
+    process: Mutex<Option<Child>>,
     _stderr_task: Option<Task<()>>,
 }
 
@@ -660,7 +650,7 @@ impl StdioTransport {
             ))
         });
 
-        let process = Mutex::new(process);
+        let process = Mutex::new(Some(process));
 
         Ok(Self {
             process,
@@ -674,8 +664,10 @@ impl Transport for StdioTransport {
         false
     }
 
-    fn kill(&self) {
-        self.process.lock().kill()
+    fn kill(&mut self) {
+        if let Some(process) = &mut *self.process.lock() {
+            process.kill();
+        }
     }
 
     fn connect(
@@ -686,8 +678,9 @@ impl Transport for StdioTransport {
             Box<dyn AsyncRead + Unpin + Send + 'static>,
         )>,
     > {
-        let mut process = self.process.lock();
         let result = util::maybe!({
+            let mut guard = self.process.lock();
+            let process = guard.as_mut().context("oops")?;
             Ok((
                 Box::new(process.stdin.take().context("Cannot reconnect")?) as _,
                 Box::new(process.stdout.take().context("Cannot reconnect")?) as _,
@@ -703,7 +696,9 @@ impl Transport for StdioTransport {
 
 impl Drop for StdioTransport {
     fn drop(&mut self) {
-        self.process.get_mut().kill();
+        if let Some(process) = &mut *self.process.lock() {
+            process.kill();
+        }
     }
 }
 
@@ -723,6 +718,7 @@ pub struct FakeTransport {
 
     stdin_writer: Option<PipeWriter>,
     stdout_reader: Option<PipeReader>,
+    message_handler: Option<Task<Result<()>>>,
 }
 
 #[cfg(any(test, feature = "test-support"))]
@@ -774,18 +770,19 @@ impl FakeTransport {
         let (stdin_writer, stdin_reader) = async_pipe::pipe();
         let (stdout_writer, stdout_reader) = async_pipe::pipe();
 
-        let this = Self {
+        let mut this = Self {
             request_handlers: Arc::new(Mutex::new(HashMap::default())),
             response_handlers: Arc::new(Mutex::new(HashMap::default())),
             stdin_writer: Some(stdin_writer),
             stdout_reader: Some(stdout_reader),
+            message_handler: None,
         };
 
         let request_handlers = this.request_handlers.clone();
         let response_handlers = this.response_handlers.clone();
         let stdout_writer = Arc::new(smol::lock::Mutex::new(stdout_writer));
 
-        cx.background_spawn(async move {
+        this.message_handler = Some(cx.background_spawn(async move {
             let mut reader = BufReader::new(stdin_reader);
             let mut buffer = String::new();
 
@@ -833,7 +830,6 @@ impl FakeTransport {
                                             .unwrap();
 
                                     let mut writer = stdout_writer.lock().await;
-
                                     writer
                                         .write_all(
                                             TransportDelegate::build_rpc_message(message)
@@ -870,8 +866,7 @@ impl FakeTransport {
                     }
                 }
             }
-        })
-        .detach();
+        }));
 
         Ok(this)
     }
@@ -904,7 +899,9 @@ impl Transport for FakeTransport {
         false
     }
 
-    fn kill(&self) {}
+    fn kill(&mut self) {
+        self.message_handler.take();
+    }
 
     #[cfg(any(test, feature = "test-support"))]
     fn as_fake(&self) -> &FakeTransport {