mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
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
This commit is contained in:
@@ -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() {
|
||||
|
||||
+41
-1
@@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
+37
-2
@@ -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() {
|
||||
|
||||
Reference in New Issue
Block a user