fix: address PR review feedback

- Make REQUEST_TIMEOUT configurable via open_with_timeout() instead of
  #[cfg(test)] hack
- Check that `leaks` tool is installed before running probes
- Replace custom json_escape with serde_json::json! macro
- Use 2ms timeout in leak-check xtask (no point waiting on a dropped stream)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: psteinroe

crates/pgls_cli/src/service/mod.rs

@@ -99,12 +99,10 @@ pub struct SocketTransport {
     runtime: Runtime,
     write_send: Sender<(Vec<u8>, bool)>,
     pending_requests: PendingRequests,
+    request_timeout: Duration,
 }
 
-#[cfg(not(test))]
-const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
-#[cfg(test)]
-const REQUEST_TIMEOUT: Duration = Duration::from_millis(50);
+const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
 
 /// Stores a handle to the map of pending requests, and clears the map
 /// automatically when the handle is dropped
@@ -137,6 +135,19 @@ impl Drop for PendingRequests {
 
 impl SocketTransport {
     pub fn open<R, W>(runtime: Runtime, socket_read: R, socket_write: W) -> Self
+    where
+        R: AsyncRead + Unpin + Send + 'static,
+        W: AsyncWrite + Unpin + Send + 'static,
+    {
+        Self::open_with_timeout(runtime, socket_read, socket_write, DEFAULT_REQUEST_TIMEOUT)
+    }
+
+    pub fn open_with_timeout<R, W>(
+        runtime: Runtime,
+        socket_read: R,
+        socket_write: W,
+        request_timeout: Duration,
+    ) -> Self
     where
         R: AsyncRead + Unpin + Send + 'static,
         W: AsyncWrite + Unpin + Send + 'static,
@@ -177,6 +188,7 @@ impl SocketTransport {
             runtime,
             write_send,
             pending_requests: pending_requests_2,
+            request_timeout,
         }
     }
 }
@@ -227,7 +239,7 @@ impl WorkspaceTransport for SocketTransport {
                         Err(_) => Err(TransportError::ChannelClosed),
                     }
                 }
-                _ = sleep(REQUEST_TIMEOUT) => {
+                _ = sleep(self.request_timeout) => {
                     Err(TransportError::Timeout)
                 }
             }
@@ -490,6 +502,7 @@ impl FromStr for TransportHeader {
 #[cfg(test)]
 mod tests {
     use std::fmt;
+    use std::time::Duration;
 
     use pgls_workspace::TransportError;
     use pgls_workspace::workspace::{TransportRequest, WorkspaceTransport};
@@ -523,7 +536,7 @@ mod tests {
         let (stream, peer) = duplex(1024);
         drop(peer);
         let (read, write) = split(stream);
-        SocketTransport::open(runtime, read, write)
+        SocketTransport::open_with_timeout(runtime, read, write, Duration::from_millis(50))
     }
 
     #[test]

xtask/src/leak_check.rs

@@ -8,7 +8,6 @@ use anyhow::{bail, Context};
 use pgls_cli::SocketTransport;
 #[cfg(unix)]
 use pgls_workspace::workspace::{TransportRequest, WorkspaceTransport};
-#[cfg(unix)]
 use serde_json::Value;
 #[cfg(unix)]
 use tokio::net::UnixStream;
@@ -25,6 +24,10 @@ impl flags::LeakCheck {
             bail!("`xtask leak-check` is currently implemented only for macOS (`leaks` tool).");
         }
 
+        if Command::new("leaks").arg("--help").output().is_err() {
+            bail!("`leaks` not found — install Xcode Command Line Tools (`xcode-select --install`)");
+        }
+
         let iterations = self.iterations.unwrap_or(DEFAULT_ITERATIONS);
         let pause = Duration::from_millis(self.pause_ms.unwrap_or(DEFAULT_PAUSE_MS));
         let probe = self.probe.unwrap_or_else(|| "lsp".to_string());
@@ -126,7 +129,8 @@ fn run_cli_timeout_probe(iterations: usize) -> anyhow::Result<()> {
     drop(stream_b);
 
     let (read, write) = stream_a.into_split();
-    let transport = SocketTransport::open(runtime, read, write);
+    let transport =
+        SocketTransport::open_with_timeout(runtime, read, write, Duration::from_millis(2));
 
     let mut channel_closed = 0usize;
     let mut timed_out = 0usize;
@@ -256,38 +260,32 @@ fn wait_for_socket(pid: u32) -> anyhow::Result<()> {
 fn run_lsp_churn(stdin: &mut ChildStdin, iterations: usize, pause: Duration) -> anyhow::Result<()> {
     let uri = "file:///tmp/pgls-leak-check.sql";
 
-    send_lsp_message(
+    send_lsp_json(
         stdin,
-        r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"capabilities":{},"rootUri":null}}"#,
+        serde_json::json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"capabilities":{},"rootUri":null}}),
     )?;
-    send_lsp_message(
+    send_lsp_json(
         stdin,
-        r#"{"jsonrpc":"2.0","method":"initialized","params":{}}"#,
+        serde_json::json!({"jsonrpc":"2.0","method":"initialized","params":{}}),
     )?;
 
     for i in 0..iterations {
-        let open_text = json_escape(&format!("select {i} as value;"));
-        let changed_text = json_escape(&format!("select {i} as value, {} as extra;", i + 1));
+        let open_text = format!("select {i} as value;");
+        let changed_text = format!("select {i} as value, {} as extra;", i + 1);
 
-        send_lsp_message(
+        send_lsp_json(
             stdin,
-            &format!(
-                r#"{{"jsonrpc":"2.0","method":"textDocument/didOpen","params":{{"textDocument":{{"uri":"{uri}","languageId":"sql","version":1,"text":"{open_text}"}}}}}}"#
-            ),
+            serde_json::json!({"jsonrpc":"2.0","method":"textDocument/didOpen","params":{"textDocument":{"uri":uri,"languageId":"sql","version":1,"text":open_text}}}),
         )?;
 
-        send_lsp_message(
+        send_lsp_json(
             stdin,
-            &format!(
-                r#"{{"jsonrpc":"2.0","method":"textDocument/didChange","params":{{"textDocument":{{"uri":"{uri}","version":2}},"contentChanges":[{{"text":"{changed_text}"}}]}}}}"#
-            ),
+            serde_json::json!({"jsonrpc":"2.0","method":"textDocument/didChange","params":{"textDocument":{"uri":uri,"version":2},"contentChanges":[{"text":changed_text}]}}),
         )?;
 
-        send_lsp_message(
+        send_lsp_json(
             stdin,
-            &format!(
-                r#"{{"jsonrpc":"2.0","method":"textDocument/didClose","params":{{"textDocument":{{"uri":"{uri}"}}}}}}"#
-            ),
+            serde_json::json!({"jsonrpc":"2.0","method":"textDocument/didClose","params":{"textDocument":{"uri":uri}}}),
         )?;
 
         thread::sleep(pause);
@@ -297,15 +295,19 @@ fn run_lsp_churn(stdin: &mut ChildStdin, iterations: usize, pause: Duration) ->
 }
 
 fn send_shutdown_and_exit(stdin: &mut ChildStdin) -> anyhow::Result<()> {
-    send_lsp_message(
+    send_lsp_json(
+        stdin,
+        serde_json::json!({"jsonrpc":"2.0","id":2,"method":"shutdown","params":null}),
+    )?;
+    send_lsp_json(
         stdin,
-        r#"{"jsonrpc":"2.0","id":2,"method":"shutdown","params":null}"#,
+        serde_json::json!({"jsonrpc":"2.0","method":"exit","params":null}),
     )?;
-    send_lsp_message(stdin, r#"{"jsonrpc":"2.0","method":"exit","params":null}"#)?;
     Ok(())
 }
 
-fn send_lsp_message(stdin: &mut ChildStdin, payload: &str) -> anyhow::Result<()> {
+fn send_lsp_json(stdin: &mut ChildStdin, value: Value) -> anyhow::Result<()> {
+    let payload = serde_json::to_string(&value).context("failed to serialize LSP message")?;
     let header = format!("Content-Length: {}\r\n\r\n", payload.len());
     stdin
         .write_all(header.as_bytes())
@@ -316,21 +318,6 @@ fn send_lsp_message(stdin: &mut ChildStdin, payload: &str) -> anyhow::Result<()>
     stdin.flush().context("failed to flush LSP payload")
 }
 
-fn json_escape(s: &str) -> String {
-    let mut out = String::with_capacity(s.len());
-    for c in s.chars() {
-        match c {
-            '"' => out.push_str("\\\""),
-            '\\' => out.push_str("\\\\"),
-            '\n' => out.push_str("\\n"),
-            '\r' => out.push_str("\\r"),
-            '\t' => out.push_str("\\t"),
-            _ => out.push(c),
-        }
-    }
-    out
-}
-
 fn run_leaks(pid: u32) -> anyhow::Result<String> {
     let output = Command::new("leaks")
         .arg(pid.to_string())