From 5879d06447726a5bfe61566d81d347e2ca3d3be3 Mon Sep 17 00:00:00 2001 From: adios2d6 <31654864+lighterEB@users.noreply.github.com> Date: Thu, 12 Mar 2026 07:53:38 +0800 Subject: [PATCH] fix: drain tunnel pipes to prevent zombie process (#735) * fix(tunnel): drain ngrok stdout/stderr to prevent zombie process * fix: limit stderr lines read on startup failure to prevent OOM * fix: drain pipes in cloudflare and custom tunnel to prevent zombie process * style: fix formatting in custom tunnel * test: add regression test for stdout drain preventing zombie process * style: apply rustfmt --- src/tunnel/cloudflare.rs | 38 +++++++++++++++++++++++++++++++++++- src/tunnel/custom.rs | 42 +++++++++++++++++++++++++++++++++++++++- src/tunnel/ngrok.rs | 39 +++++++++++++++++++++++++++++++++++-- 3 files changed, 115 insertions(+), 4 deletions(-) diff --git a/src/tunnel/cloudflare.rs b/src/tunnel/cloudflare.rs index 38f0cd97..2c0ceb2a 100644 --- a/src/tunnel/cloudflare.rs +++ b/src/tunnel/cloudflare.rs @@ -49,6 +49,8 @@ impl Tunnel for CloudflareTunnel { .kill_on_drop(true) .spawn()?; + let stdout = child.stdout.take(); + // cloudflared prints the public URL on stderr let stderr = child .stderr @@ -82,8 +84,42 @@ impl Tunnel for CloudflareTunnel { } if public_url.is_empty() { + let error_detail = if let Some(stdout) = stdout { + let mut out_reader = tokio::io::BufReader::new(stdout).lines(); + let mut lines = Vec::new(); + while lines.len() < 10 { + match tokio::time::timeout( + tokio::time::Duration::from_secs(1), + out_reader.next_line(), + ) + .await + { + Ok(Ok(Some(line))) => lines.push(line), + _ => break, + } + } + lines.join("\n") + } else { + String::new() + }; + child.kill().await.ok(); - bail!("cloudflared did not produce a public URL within 30s. Is the token valid?"); + if error_detail.is_empty() { + bail!("cloudflared did not produce a public URL within 30s"); + } else { + bail!("cloudflared failed to start: {error_detail}"); + } + } + + // Drain stderr in the background to prevent SIGPIPE/buffer stalls. + tokio::spawn(async move { while let Ok(Some(_)) = reader.next_line().await {} }); + + // Drain stdout silently. + if let Some(stdout) = stdout { + tokio::spawn(async move { + let mut out_reader = tokio::io::BufReader::new(stdout).lines(); + while let Ok(Some(_)) = out_reader.next_line().await {} + }); } if let Ok(mut guard) = self.url.write() { diff --git a/src/tunnel/custom.rs b/src/tunnel/custom.rs index 1cb71b0f..9a2be403 100644 --- a/src/tunnel/custom.rs +++ b/src/tunnel/custom.rs @@ -69,10 +69,13 @@ impl Tunnel for CustomTunnel { .kill_on_drop(true) .spawn()?; + let stdout = child.stdout.take(); + let stderr = child.stderr.take(); + let mut public_url = format!("http://{local_host}:{local_port}"); if self.url_pattern.is_some() - && let Some(stdout) = child.stdout.take() + && let Some(stdout) = stdout { let mut reader = tokio::io::BufReader::new(stdout).lines(); let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(15); @@ -100,6 +103,22 @@ impl Tunnel for CustomTunnel { Err(_) => {} } } + // Drain remaining stdout to prevent SIGPIPE/buffer stalls. + tokio::spawn(async move { while let Ok(Some(_)) = reader.next_line().await {} }); + } else if let Some(stdout) = stdout { + // No url_pattern: still drain stdout to prevent pipe stalls. + tokio::spawn(async move { + let mut reader = tokio::io::BufReader::new(stdout).lines(); + while let Ok(Some(_)) = reader.next_line().await {} + }); + } + + // Drain stderr silently. + if let Some(stderr) = stderr { + tokio::spawn(async move { + let mut reader = tokio::io::BufReader::new(stderr).lines(); + while let Ok(Some(_)) = reader.next_line().await {} + }); } if let Ok(mut guard) = self.url.write() { @@ -246,4 +265,25 @@ mod tests { fn extract_url_none_when_absent() { assert_eq!(extract_url("no url here"), None); } + + #[tokio::test] + async fn stdout_drain_prevents_zombie() { + // `yes` floods stdout indefinitely; without the drain task the pipe + // buffer fills (64 KB) and the child blocks on write(), becoming a + // zombie. With draining the child stays alive and stop() can kill it. + let tunnel = CustomTunnel::new("yes".into(), None, None); + let url = tunnel.start("127.0.0.1", 19999).await.unwrap(); + assert_eq!(url, "http://127.0.0.1:19999"); + + // Give the drain task time to consume some output. + tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; + + // Child should still be alive (not blocked/zombie). + assert!( + tunnel.health_check().await, + "yes process should still be alive" + ); + + tunnel.stop().await.unwrap(); + } } diff --git a/src/tunnel/ngrok.rs b/src/tunnel/ngrok.rs index 2b0e0df9..80a5cc46 100644 --- a/src/tunnel/ngrok.rs +++ b/src/tunnel/ngrok.rs @@ -54,7 +54,7 @@ impl Tunnel for NgrokTunnel { .stdout .take() .ok_or_else(|| anyhow::anyhow!("Failed to capture ngrok stdout"))?; - + let stderr = child.stderr.take(); let mut reader = tokio::io::BufReader::new(stdout).lines(); let mut public_url = String::new(); @@ -84,8 +84,43 @@ impl Tunnel for NgrokTunnel { } if public_url.is_empty() { + let error_detail = if let Some(stderr) = stderr { + let mut err_reader = tokio::io::BufReader::new(stderr).lines(); + let mut lines = Vec::new(); + while lines.len() < 10 { + match tokio::time::timeout( + tokio::time::Duration::from_secs(1), + err_reader.next_line(), + ) + .await + { + Ok(Ok(Some(line))) => lines.push(line), + _ => break, + } + } + lines.join("\n") + } else { + String::new() + }; child.kill().await.ok(); - bail!("ngrok did not produce a public URL within 15s. Is the auth token valid?"); + if error_detail.is_empty() { + bail!("ngrok did not produce a public URL within 15s"); + } else { + bail!("ngrok failed to start: {error_detail}"); + } + } + + // Drain stdout silently — ngrok only emits low-level connection events + // to stdout; the pipe must be consumed to prevent SIGPIPE/buffer stalls. + tokio::spawn(async move { while let Ok(Some(_)) = reader.next_line().await {} }); + + // Drain stderr silently — with --log stdout all meaningful output goes + // to stdout; stderr only needs to be consumed to prevent pipe stalls. + if let Some(stderr) = stderr { + tokio::spawn(async move { + let mut err_reader = tokio::io::BufReader::new(stderr).lines(); + while let Ok(Some(_)) = err_reader.next_line().await {} + }); } if let Ok(mut guard) = self.url.write() {