oimwiodev/chrome-use
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: harden Lightpanda startup timeouts (#762)

Browse Source
This commit is contained in:
mikewong23571
2026-03-13 11:36:51 -05:00
committed by GitHub
parent 056024b447
commit d40fd4009d
6 changed files with 689 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/native/cdp/lightpanda.rs
+335 -154
View File
@@ -1,13 +1,23 @@
use std::collections::VecDeque;
use std::io::{BufRead, BufReader};
use std::net::TcpListener;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use super::discovery::discover_cdp_url_with_request_timeout;
const LIGHTPANDA_STARTUP_TIMEOUT: Duration = Duration::from_secs(10);
const LIGHTPANDA_POLL_INTERVAL: Duration = Duration::from_millis(100);
const LIGHTPANDA_DISCOVERY_TIMEOUT: Duration = Duration::from_millis(500);
const LIGHTPANDA_SESSION_TIMEOUT_SECS: u64 = 604800; // 1 week, the documented maximum
const MAX_LOG_LINES: usize = 40;
pub struct LightpandaProcess {
child: Child,
pub ws_url: String,
_stderr_drain: Option<std::thread::JoinHandle<()>>,
_log_drainers: Vec<std::thread::JoinHandle<()>>,
}
impl LightpandaProcess {
@@ -30,8 +40,68 @@ pub struct LightpandaLaunchOptions {
pub port: Option<u16>,
}
fn build_lightpanda_serve_args(port: u16, proxy: Option<&str>) -> Vec<String> {
let mut args = vec![
"serve".to_string(),
"--host".to_string(),
"127.0.0.1".to_string(),
"--port".to_string(),
port.to_string(),
"--timeout".to_string(),
LIGHTPANDA_SESSION_TIMEOUT_SECS.to_string(),
];
if let Some(proxy) = proxy {
args.push("--http_proxy".to_string());
args.push(proxy.to_string());
}
args
}
#[derive(Clone, Default)]
struct LaunchLogBuffer {
stdout: Arc<Mutex<VecDeque<String>>>,
stderr: Arc<Mutex<VecDeque<String>>>,
}
impl LaunchLogBuffer {
fn push_stdout(&self, line: String) {
push_bounded(&self.stdout, line);
}
fn push_stderr(&self, line: String) {
push_bounded(&self.stderr, line);
}
fn snapshot_stdout(&self) -> Vec<String> {
self.stdout
.lock()
.expect("stdout log buffer poisoned")
.iter()
.cloned()
.collect()
}
fn snapshot_stderr(&self) -> Vec<String> {
self.stderr
.lock()
.expect("stderr log buffer poisoned")
.iter()
.cloned()
.collect()
}
}
fn push_bounded(buffer: &Mutex<VecDeque<String>>, line: String) {
let mut guard = buffer.lock().expect("log buffer poisoned");
if guard.len() >= MAX_LOG_LINES {
guard.pop_front();
}
guard.push_back(line);
}
pub fn find_lightpanda() -> Option<PathBuf> {
// Check PATH via `which`
#[cfg(unix)]
{
if let Ok(output) = Command::new("which").arg("lightpanda").output() {
@@ -61,7 +131,6 @@ pub fn find_lightpanda() -> Option<PathBuf> {
}
}
// Common install locations
if let Some(home) = dirs::home_dir() {
let candidates = [
home.join(".lightpanda/lightpanda"),
@@ -74,13 +143,12 @@ pub fn find_lightpanda() -> Option<PathBuf> {
}
}
// npm package binary: @lightpanda/browser installs to node_modules/.bin
// Not checked here since the user would typically have it in PATH.
None
}
pub fn launch_lightpanda(options: &LightpandaLaunchOptions) -> Result<LightpandaProcess, String> {
pub async fn launch_lightpanda(
options: &LightpandaLaunchOptions,
) -> Result<LightpandaProcess, String> {
let binary_path = match &options.executable_path {
Some(p) => PathBuf::from(p),
None => find_lightpanda().ok_or(
@@ -95,31 +163,7 @@ pub fn launch_lightpanda(options: &LightpandaLaunchOptions) -> Result<Lightpanda
.map(|a| a.port())
.map_err(|e| format!("Failed to find an available port for Lightpanda: {}", e))?,
};
let port_str = port.to_string();
let mut args = vec![
"serve".to_string(),
"--host".to_string(),
"127.0.0.1".to_string(),
"--port".to_string(),
port_str,
];
if let Some(ref proxy) = options.proxy {
args.push("--http_proxy".to_string());
args.push(proxy.clone());
}
// Use the maximum inactivity timeout (1 week) so the connection stays alive
// during long sessions. Lightpanda treats 0 as "timeout immediately" rather
// than "no timeout".
args.push("--timeout".to_string());
args.push("604800".to_string());
// Lightpanda defaults to --log_level warn, which suppresses the "server
// running" info message we need to detect the listen address.
args.push("--log_level".to_string());
args.push("info".to_string());
let args = build_lightpanda_serve_args(port, options.proxy.as_deref());
let mut child = Command::new(&binary_path)
.args(&args)
@@ -129,174 +173,273 @@ pub fn launch_lightpanda(options: &LightpandaLaunchOptions) -> Result<Lightpanda
.spawn()
.map_err(|e| format!("Failed to launch Lightpanda at {:?}: {}", binary_path, e))?;
// Lightpanda logs to stderr
let stderr = child.stderr.take().ok_or_else(|| {
let _ = child.kill();
"Failed to capture Lightpanda stderr".to_string()
})?;
let reader = BufReader::new(stderr);
let (log_buffer, log_drainers) = start_log_drainers(&mut child)?;
let (address, reader) = match wait_for_address(reader) {
Ok(result) => result,
Err(e) => {
let _ = child.kill();
return Err(e);
}
};
let ws_url = format!("ws://{}", address);
let drain = std::thread::spawn(move || {
let mut reader = reader;
let mut buf = String::new();
loop {
buf.clear();
match reader.read_line(&mut buf) {
Ok(0) | Err(_) => break,
Ok(_) => {}
let ws_url =
match wait_for_lightpanda_ready(&mut child, port, &log_buffer, LIGHTPANDA_STARTUP_TIMEOUT)
.await
{
Ok(url) => url,
Err(e) => {
let _ = child.kill();
let _ = child.wait();
return Err(e);
}
}
});
};
Ok(LightpandaProcess {
child,
ws_url,
_stderr_drain: Some(drain),
_log_drainers: log_drainers,
})
}
/// Parse Lightpanda's stderr for the server address.
/// Lightpanda outputs lines like:
/// INFO app : server running . . . address = 127.0.0.1:9222
///
/// Returns the address and the reader so the caller can keep the pipe alive.
fn wait_for_address(
mut reader: BufReader<std::process::ChildStderr>,
) -> Result<(String, BufReader<std::process::ChildStderr>), String> {
let deadline = std::time::Instant::now() + Duration::from_secs(30);
let mut stderr_lines: Vec<String> = Vec::new();
let mut buf = String::new();
fn start_log_drainers(
child: &mut Child,
) -> Result<(LaunchLogBuffer, Vec<std::thread::JoinHandle<()>>), String> {
let stdout = child.stdout.take().ok_or_else(|| {
let _ = child.kill();
"Failed to capture Lightpanda stdout".to_string()
})?;
let stderr = child.stderr.take().ok_or_else(|| {
let _ = child.kill();
"Failed to capture Lightpanda stderr".to_string()
})?;
let logs = LaunchLogBuffer::default();
let stdout_logs = logs.clone();
let stderr_logs = logs.clone();
let stdout_handle =
std::thread::spawn(move || drain_reader(stdout, move |line| stdout_logs.push_stdout(line)));
let stderr_handle =
std::thread::spawn(move || drain_reader(stderr, move |line| stderr_logs.push_stderr(line)));
Ok((logs, vec![stdout_handle, stderr_handle]))
}
fn drain_reader<R, F>(reader: R, mut push: F)
where
R: std::io::Read,
F: FnMut(String),
{
for line in BufReader::new(reader).lines() {
match line {
Ok(line) => push(line),
Err(_) => break,
}
}
}
async fn wait_for_lightpanda_ready(
child: &mut Child,
port: u16,
logs: &LaunchLogBuffer,
startup_timeout: Duration,
) -> Result<String, String> {
let deadline = std::time::Instant::now() + startup_timeout;
let mut last_probe_error = None;
loop {
if std::time::Instant::now() > deadline {
if let Ok(Some(status)) = child.try_wait() {
// Give the drainer threads a brief window to flush the last log lines
// before we snapshot them. This is best-effort: lines written just
// before exit may still be missing, but the most useful output (early
// startup errors) will already be in the buffer.
tokio::time::sleep(Duration::from_millis(25)).await;
return Err(lightpanda_launch_error(
"Timeout waiting for Lightpanda server address",
&stderr_lines,
&format!(
"Lightpanda exited before CDP became ready (status: {})",
status
),
logs,
last_probe_error.as_deref(),
));
}
buf.clear();
match reader.read_line(&mut buf) {
Ok(0) => {
return Err(lightpanda_launch_error(
"Lightpanda exited before providing server address",
&stderr_lines,
));
}
Ok(_) => {
let line = buf.trim_end().to_string();
if let Some(address) = extract_address(&line) {
return Ok((address, reader));
}
stderr_lines.push(line);
}
Err(e) => {
return Err(format!("Failed to read Lightpanda stderr: {}", e));
}
match discover_cdp_url_with_request_timeout(port, LIGHTPANDA_DISCOVERY_TIMEOUT).await {
Ok(ws_url) => return Ok(ws_url),
Err(err) => last_probe_error = Some(err),
}
if std::time::Instant::now() >= deadline {
return Err(lightpanda_launch_error(
&format!(
"Timed out after {}ms waiting for Lightpanda CDP endpoint on port {}",
startup_timeout.as_millis(),
port
),
logs,
last_probe_error.as_deref(),
));
}
tokio::time::sleep(LIGHTPANDA_POLL_INTERVAL).await;
}
}
fn extract_address(line: &str) -> Option<String> {
// Lightpanda uses logfmt (`address=...`) in release, pretty (`address = ...`) in debug.
for pattern in &["address=", "address = "] {
if let Some(idx) = line.find(pattern) {
let addr = line[idx + pattern.len()..].trim().to_string();
// logfmt lines may have subsequent key=value pairs
let addr = addr.split_whitespace().next().unwrap_or("").to_string();
if !addr.is_empty() {
return Some(addr);
}
}
}
None
}
fn lightpanda_launch_error(
message: &str,
logs: &LaunchLogBuffer,
last_probe_error: Option<&str>,
) -> String {
let stdout_lines = logs.snapshot_stdout();
let stderr_lines = logs.snapshot_stderr();
let mut details = Vec::new();
fn lightpanda_launch_error(message: &str, stderr_lines: &[String]) -> String {
if stderr_lines.is_empty() {
return format!("{} (no stderr output from Lightpanda)", message);
if let Some(err) = last_probe_error {
details.push(format!("Last probe error: {}", err));
}
let last_lines: Vec<&String> = stderr_lines.iter().rev().take(5).collect();
format!(
"{}\nLightpanda stderr (last {} lines):\n {}",
message,
last_lines.len(),
last_lines
.into_iter()
.rev()
.map(|s| s.as_str())
.collect::<Vec<_>>()
.join("\n ")
)
if !stderr_lines.is_empty() {
details.push(format!(
"Lightpanda stderr (last {} lines):\n {}",
stderr_lines.len(),
stderr_lines.join("\n ")
));
}
if !stdout_lines.is_empty() {
details.push(format!(
"Lightpanda stdout (last {} lines):\n {}",
stdout_lines.len(),
stdout_lines.join("\n ")
));
}
if details.is_empty() {
format!("{} (no stdout/stderr output from Lightpanda)", message)
} else {
format!("{}\n{}", message, details.join("\n"))
}
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener as TokioTcpListener;
#[test]
fn test_extract_address_pretty_debug_build() {
assert_eq!(
extract_address(" address = 127.0.0.1:9222"),
Some("127.0.0.1:9222".to_string())
);
fn unused_port() -> u16 {
std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
#[test]
fn test_extract_address_logfmt_release_build() {
assert_eq!(
extract_address(
"$time=1234 $scope=app $level=info $msg=\"server running\" address=127.0.0.1:9222"
),
Some("127.0.0.1:9222".to_string())
async fn serve_json_version_once_after_delay(port: u16, delay_ms: u64, body: &'static str) {
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
let listener = TokioTcpListener::bind(("127.0.0.1", port)).await.unwrap();
let (mut socket, _) = listener.accept().await.unwrap();
let mut buf = [0u8; 1024];
let _ = socket.read(&mut buf).await;
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\nContent-Type: application/json\r\n\r\n{}",
body.len(),
body
);
socket.write_all(response.as_bytes()).await.unwrap();
}
#[test]
fn test_extract_address_pretty_inline() {
assert_eq!(
extract_address("INFO app : server running address = 127.0.0.1:4567"),
Some("127.0.0.1:4567".to_string())
);
#[tokio::test]
async fn waits_for_ready_without_logs() {
let port = unused_port();
tokio::spawn(serve_json_version_once_after_delay(
port,
150,
r#"{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/"}"#,
));
let mut child = Command::new("/bin/sh")
.args(["-c", "sleep 5"])
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let (logs, _drainers) = start_log_drainers(&mut child).unwrap();
let ws_url = wait_for_lightpanda_ready(&mut child, port, &logs, LIGHTPANDA_STARTUP_TIMEOUT)
.await
.unwrap();
assert_eq!(ws_url, "ws://127.0.0.1:9222/");
let _ = child.kill();
let _ = child.wait();
}
#[test]
fn test_extract_address_no_match() {
assert_eq!(extract_address("INFO app : starting up..."), None);
#[tokio::test]
async fn child_exit_surfaces_logs() {
let port = unused_port();
let mut child = Command::new("/bin/sh")
.args(["-c", "echo boom >&2; sleep 0.1; exit 23"])
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let (logs, _drainers) = start_log_drainers(&mut child).unwrap();
let err = wait_for_lightpanda_ready(&mut child, port, &logs, LIGHTPANDA_STARTUP_TIMEOUT)
.await
.unwrap_err();
assert!(err.contains("Lightpanda exited before CDP became ready"));
assert!(err.contains("boom"));
}
#[tokio::test]
async fn timeout_reports_last_probe_error() {
let port = unused_port();
let mut child = Command::new("/bin/sh")
.args(["-c", "sleep 30"])
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let timeout = Duration::from_millis(300);
let (logs, _drainers) = start_log_drainers(&mut child).unwrap();
let err = tokio::time::timeout(
Duration::from_secs(2),
wait_for_lightpanda_ready(&mut child, port, &logs, timeout),
)
.await
.expect("ready wait should return before outer timeout")
.unwrap_err();
assert!(err.contains("Timed out after 300ms waiting for Lightpanda CDP endpoint"));
assert!(
err.contains("Failed to connect to CDP") || err.contains("Timeout connecting to CDP")
);
let _ = child.kill();
let _ = child.wait();
}
#[test]
fn test_find_lightpanda_returns_none_when_missing() {
// On most CI/dev machines Lightpanda won't be installed
// Just verify the function doesn't panic
let _ = find_lightpanda();
}
#[test]
fn test_lightpanda_launch_error_no_stderr() {
let msg = lightpanda_launch_error("Lightpanda exited", &[]);
assert!(msg.contains("no stderr output"));
fn test_lightpanda_launch_error_no_logs() {
let logs = LaunchLogBuffer::default();
let msg = lightpanda_launch_error("Lightpanda exited", &logs, None);
assert!(msg.contains("no stdout/stderr output"));
}
#[test]
fn test_lightpanda_launch_error_with_lines() {
let lines = vec![
"INFO starting up".to_string(),
"ERROR bind failed: address in use".to_string(),
];
let msg = lightpanda_launch_error("Lightpanda exited", &lines);
assert!(msg.contains("bind failed"));
assert!(msg.contains("last 2 lines"));
let logs = LaunchLogBuffer::default();
logs.push_stdout("stdout line".to_string());
logs.push_stderr("stderr line".to_string());
let msg = lightpanda_launch_error("Lightpanda exited", &logs, Some("connect failed"));
assert!(msg.contains("stdout line"));
assert!(msg.contains("stderr line"));
assert!(msg.contains("Last probe error: connect failed"));
}
#[test]
@@ -306,4 +449,42 @@ mod tests {
assert!(opts.proxy.is_none());
assert!(opts.port.is_none());
}
#[test]
fn test_build_lightpanda_serve_args_sets_explicit_session_timeout() {
let args = build_lightpanda_serve_args(9222, None);
assert_eq!(
args,
vec![
"serve".to_string(),
"--host".to_string(),
"127.0.0.1".to_string(),
"--port".to_string(),
"9222".to_string(),
"--timeout".to_string(),
"604800".to_string(),
]
);
}
#[test]
fn test_build_lightpanda_serve_args_with_proxy() {
let args = build_lightpanda_serve_args(9333, Some("http://127.0.0.1:8080"));
assert_eq!(
args,
vec![
"serve".to_string(),
"--host".to_string(),
"127.0.0.1".to_string(),
"--port".to_string(),
"9333".to_string(),
"--timeout".to_string(),
"604800".to_string(),
"--http_proxy".to_string(),
"http://127.0.0.1:8080".to_string(),
]
);
}
}