fix(dap): avoid free-port race when starting debug adapters (#2262)

Цей коміт міститься в:
ppxb authored and GitHub committed 2026-11-09 12:10:31 +08:00
1 джерело f1c10f4d84
коміт d17f6755f3
2 files changed
+175 -86

No files matched your search

+147 -82
Переглянути файл
@@ -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<u16> {
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<u16>) -> 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<u16>,
start_timeout: Duration,
) -> IdeResult<AdapterIo> {
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<R: AsyncRead + Unpin + Send + 'static>(
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")));
}
}
+28 -4
Переглянути файл
@@ -791,7 +791,6 @@ fn forward_output<R: AsyncRead + Unpin + Send + 'static>(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<Session>, 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<String> {
args.get(key)
@@ -804,7 +803,9 @@ pub(crate) async fn start_kotlin_debuggee(s: &Arc<Session>, 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<Session>, 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<Session>, args: &mut Value) ->
args["timeout"] = json!(10000);
Ok(())
}
/// The port a JVM chose, from `Listening for transport dt_socket at address: <port>`
/// (some JVMs print `host:port` or `[::1]:port`, so take what follows the last colon).
fn parse_jdwp_port(line: &str) -> Option<u16> {
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);
}
}