From d17f6755f31519ace7f801fb13f2a67e86b85598 Mon Sep 17 00:00:00 2001 From: ppxb <317842449@qq.com> Date: Mon, 9 Nov 2026 12:10:31 +0800 Subject: [PATCH] fix(dap): avoid free-port race when starting debug adapters (#2262) --- crates/ide-dap/src/adapter.rs | 229 ++++++++++++++++++++++------------ crates/ide-dap/src/session.rs | 32 ++++- 2 files changed, 175 insertions(+), 86 deletions(-) diff --git a/crates/ide-dap/src/adapter.rs b/crates/ide-dap/src/adapter.rs index d517510e9..e74fa3380 100644 --- a/crates/ide-dap/src/adapter.rs +++ b/crates/ide-dap/src/adapter.rs @@ -166,6 +166,11 @@ pub fn parse_address(addr: &str) -> Option<(String, u16)> { } /// A port that was free a moment ago on the loopback interface. +/// +/// Inherently racy: another process can bind the port between the listener being +/// dropped here and the adapter binding it. Callers that hand the port to a child +/// retry on a bind conflict (see [`ProcessLauncher`]), or — better — let the child +/// pick `:0` and report the actual port back (as the Kotlin JDWP debuggee does). pub fn free_port() -> std::io::Result { let listener = std::net::TcpListener::bind(("127.0.0.1", 0))?; Ok(listener.local_addr()?.port()) @@ -394,6 +399,9 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); const EMBEDDED_START_TIMEOUT: Duration = Duration::from_secs(120); /// Probe of a `connect` address before starting the command. const PROBE_TIMEOUT: Duration = Duration::from_millis(500); +/// Attempts at a fresh port for a TCP adapter: the one `free_port` hands out can be +/// claimed by another process before the adapter binds it, so a new one usually wins. +const PORT_ATTEMPTS: u32 = 3; impl AdapterLauncher for ProcessLauncher { fn launch(&self, mut spec: AdapterSpec, reuse_port: Option) -> LaunchFuture { @@ -459,95 +467,140 @@ impl AdapterLauncher for ProcessLauncher { .ok_or_else(|| missing(&spec))? } }; - let port = match spec.transport { - // An embedded adapter started without `${port}` listens on its usual address. - Transport::Tcp => match &embedded { - Some((_, port)) if !spec.args.iter().any(|a| a.contains(PORT_PLACEHOLDER)) => Some(*port), - _ => Some(free_port()?), - }, - Transport::Stdio => None, - }; - let args = substitute_root(&spec.args, &spec.cwd); - let args = match port { - Some(port) => substitute_port(&args, port), - None => args, - }; - let mut cmd = Command::new(&binary); - if let Some(path) = adapter_path(&spec.env) { - cmd.env("PATH", path); - } - ide_core::process::prepare(cmd.as_std_mut()); - if spec.adapter == "rdbg" { - ide_core::process::own_group(cmd.as_std_mut()); - } - cmd.args(&args) - .envs(spec.env.iter().map(|(k, v)| (k, v))) - .current_dir(&spec.cwd) - .stdin(if port.is_some() { Stdio::null() } else { Stdio::piped() }) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true); - let mut child = - cmd.spawn().map_err(|e| IdeError::unavailable(format!("failed to start {}: {e}", binary.display())))?; - let (program_sink, program_output) = if SHARES_STDIO.contains(&spec.adapter.as_str()) { - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); - (Some(tx), Some(rx)) - } else { - (None, None) - }; - if let Some(err) = child.stderr.take() { - pump_lines(err, spec.adapter.clone(), stderr.clone(), program_sink.clone().map(|tx| (tx, "stderr"))); - } - match port { - None => { - let stdin = child.stdin.take().ok_or_else(|| IdeError::internal("adapter stdin unavailable"))?; - let stdout = child.stdout.take().ok_or_else(|| IdeError::internal("adapter stdout unavailable"))?; - Ok(AdapterIo { - reader: Box::new(stdout), - writer: Box::new(stdin), - child: Some(child), - port: None, - stderr, - program_output, - }) - } - Some(port) => { - if let Some(out) = child.stdout.take() { - pump_lines(out, spec.adapter.clone(), stderr.clone(), program_sink.map(|tx| (tx, "stdout"))); - } - let timeout = if embedded.is_some() { EMBEDDED_START_TIMEOUT } else { CONNECT_TIMEOUT }; - let stream = match connect_with_retries(LOOPBACK, port, Some(&mut child), timeout).await { - Ok(s) => s, - Err(e) => { - if spec.adapter == "rdbg" - && let Some(pid) = child.id() - { - ide_core::process::kill_tree_of(pid); - } - let _ = child.start_kill(); - let tail = stderr.lock().join("\n"); - return Err(if tail.is_empty() { - e - } else { - IdeError::unavailable(format!("{}\n{tail}", e.message)) - }); - } + let base_args = substitute_root(&spec.args, &spec.cwd); + match spec.transport { + Transport::Stdio => spawn_adapter(&spec, &binary, &base_args, None, CONNECT_TIMEOUT).await, + Transport::Tcp => { + // An embedded adapter started without `${port}` listens on its usual + // address; every other TCP adapter gets a port chosen here so it can be + // retried if another process takes it before the adapter binds. + let fixed = match &embedded { + Some((_, port)) if !spec.args.iter().any(|a| a.contains(PORT_PLACEHOLDER)) => Some(*port), + _ => None, }; - let (r, w) = stream.into_split(); - Ok(AdapterIo { - reader: Box::new(r), - writer: Box::new(w), - child: Some(child), - port: Some(port), - stderr, - program_output, - }) + let start_timeout = if embedded.is_some() { EMBEDDED_START_TIMEOUT } else { CONNECT_TIMEOUT }; + for attempt in 1..=PORT_ATTEMPTS { + let port = match fixed { + Some(port) => port, + None => free_port()?, + }; + let args = substitute_port(&base_args, port); + match spawn_adapter(&spec, &binary, &args, Some(port), start_timeout).await { + Ok(io) => return Ok(io), + Err(e) => { + if fixed.is_none() && attempt < PORT_ATTEMPTS && is_port_race(&e) { + tracing::warn!( + target: "ide_dap::adapter", + "debug adapter failed to start on port {port}: {}; retrying", + e.message + ); + continue; + } + return Err(e); + } + } + } + unreachable!("the retry loop returns on its last attempt") } } }) } } +/// Starts `binary` with `args`, then connects to it when `port` is set (a TCP adapter) +/// or hands back its stdio when it is not. The child is stopped if it never accepts. +async fn spawn_adapter( + spec: &AdapterSpec, + binary: &Path, + args: &[String], + port: Option, + start_timeout: Duration, +) -> IdeResult { + let stderr = Arc::new(parking_lot::Mutex::new(Vec::new())); + let mut cmd = Command::new(binary); + if let Some(path) = adapter_path(&spec.env) { + cmd.env("PATH", path); + } + ide_core::process::prepare(cmd.as_std_mut()); + if spec.adapter == "rdbg" { + ide_core::process::own_group(cmd.as_std_mut()); + } + cmd.args(args) + .envs(spec.env.iter().map(|(k, v)| (k, v))) + .current_dir(&spec.cwd) + .stdin(if port.is_some() { Stdio::null() } else { Stdio::piped() }) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let mut child = + cmd.spawn().map_err(|e| IdeError::unavailable(format!("failed to start {}: {e}", binary.display())))?; + let (program_sink, program_output) = if SHARES_STDIO.contains(&spec.adapter.as_str()) { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + if let Some(err) = child.stderr.take() { + pump_lines(err, spec.adapter.clone(), stderr.clone(), program_sink.clone().map(|tx| (tx, "stderr"))); + } + match port { + None => { + let stdin = child.stdin.take().ok_or_else(|| IdeError::internal("adapter stdin unavailable"))?; + let stdout = child.stdout.take().ok_or_else(|| IdeError::internal("adapter stdout unavailable"))?; + Ok(AdapterIo { + reader: Box::new(stdout), + writer: Box::new(stdin), + child: Some(child), + port: None, + stderr, + program_output, + }) + } + Some(port) => { + if let Some(out) = child.stdout.take() { + pump_lines(out, spec.adapter.clone(), stderr.clone(), program_sink.map(|tx| (tx, "stdout"))); + } + let stream = match connect_with_retries(LOOPBACK, port, Some(&mut child), start_timeout).await { + Ok(s) => s, + Err(e) => { + if spec.adapter == "rdbg" + && let Some(pid) = child.id() + { + ide_core::process::kill_tree_of(pid); + } + let _ = child.start_kill(); + let tail = stderr.lock().join("\n"); + return Err(if tail.is_empty() { + e + } else { + IdeError::unavailable(format!("{}\n{tail}", e.message)) + }); + } + }; + let (r, w) = stream.into_split(); + Ok(AdapterIo { + reader: Box::new(r), + writer: Box::new(w), + child: Some(child), + port: Some(port), + stderr, + program_output, + }) + } + } +} + +/// A failed adapter start that another process grabbing the free port can explain +/// (POSIX and Windows "address already in use" texts, or the adapter exiting because +/// it could not bind). Anything else is a real start failure and is not retried. +fn is_port_race(err: &IdeError) -> bool { + let message = err.message.to_ascii_lowercase(); + message.contains("address already in use") + || message.contains("addrinuse") + || message.contains("only one usage of each socket address") + || message.contains("exited before accepting connections") +} + fn pump_lines( stream: R, name: String, @@ -775,4 +828,16 @@ mod tests { assert_eq!(err.kind, ide_core::error::ErrorKind::Unavailable); assert!(err.message.contains("runs inside the Java language server"), "{}", err.message); } + + #[test] + fn port_conflicts_are_recognised_for_retry() { + let in_use = IdeError::unavailable( + "debug adapter exited before accepting connections (exit status: 1)\nbind: address already in use", + ); + assert!(is_port_race(&in_use)); + assert!(is_port_race(&IdeError::unavailable( + "Only one usage of each socket address (protocol/network address/port) is normally permitted." + ))); + assert!(!is_port_race(&IdeError::unavailable("failed to start codelldb: file not found"))); + } } diff --git a/crates/ide-dap/src/session.rs b/crates/ide-dap/src/session.rs index cfb4bfb1d..0bf73f0b0 100644 --- a/crates/ide-dap/src/session.rs +++ b/crates/ide-dap/src/session.rs @@ -791,7 +791,6 @@ fn forward_output(s: &Session, mut stream /// Launch with the IDE's resolved runtime and classpath; the adapter only attaches. pub(crate) async fn start_kotlin_debuggee(s: &Arc, args: &mut Value) -> ide_core::IdeResult<()> { use tokio::io::{AsyncBufReadExt, BufReader}; - let port = crate::adapter::free_port()?; let java = args.get("javaExec").and_then(Value::as_str).unwrap_or("java"); let strings = |key: &str| -> Vec { args.get(key) @@ -804,7 +803,9 @@ pub(crate) async fn start_kotlin_debuggee(s: &Arc, args: &mut Value) -> let mut cmd = tokio::process::Command::new(java); ide_core::process::prepare(cmd.as_std_mut()); cmd.args(strings("vmArguments")) - .arg(format!("-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:{port}")) + // Let the JVM pick its own port (`:0`) and report it below: pre-picking a free + // port here races with other processes that may take it before the JVM binds. + .arg("-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:0") .arg("-cp") .arg(classpath); let modules = strings("modulePaths"); @@ -837,14 +838,18 @@ pub(crate) async fn start_kotlin_debuggee(s: &Arc, args: &mut Value) -> } let mut stdout = BufReader::new(child.stdout.take().ok_or_else(|| ide_core::IdeError::internal("Missing JVM stdout"))?); - tokio::time::timeout(Duration::from_secs(15), async { + // The JVM chose the port (`:0` above) and reports it here; reading it back removes + // the free-port race window entirely. + let port = tokio::time::timeout(Duration::from_secs(15), async { loop { let mut line = String::new(); if stdout.read_line(&mut line).await? == 0 { return Err(ide_core::IdeError::unavailable("JVM exited before starting its debug listener")); } if line.contains("Listening for transport dt_socket") { - return Ok(()); + return parse_jdwp_port(&line).ok_or_else(|| { + ide_core::IdeError::unavailable(format!("JVM reported an unreadable debug port: {}", line.trim())) + }); } s.emit(DebugEvent::Output { category: "stdout".into(), output: line }); } @@ -858,3 +863,22 @@ pub(crate) async fn start_kotlin_debuggee(s: &Arc, args: &mut Value) -> args["timeout"] = json!(10000); Ok(()) } + +/// The port a JVM chose, from `Listening for transport dt_socket at address: ` +/// (some JVMs print `host:port` or `[::1]:port`, so take what follows the last colon). +fn parse_jdwp_port(line: &str) -> Option { + line.rsplit_once("address:")?.1.trim().rsplit(':').next()?.trim().parse().ok() +} + +#[cfg(test)] +mod tests { + use super::parse_jdwp_port; + + #[test] + fn reads_the_port_the_jvm_chose() { + assert_eq!(parse_jdwp_port("Listening for transport dt_socket at address: 51234"), Some(51234)); + assert_eq!(parse_jdwp_port("Listening for transport dt_socket at address: 127.0.0.1:51234"), Some(51234)); + assert_eq!(parse_jdwp_port("Listening for transport dt_socket at address: [::1]:51234"), Some(51234)); + assert_eq!(parse_jdwp_port("unrelated output"), None); + } +}