281 lines
10 KiB
Rust
281 lines
10 KiB
Rust
//! Session-owned asynchronous helper. Taking the process out of the slot makes cancellation safe:
|
|
//! dropping a request destroys that process group, and the next request starts fresh.
|
|
use anyhow::{anyhow, Context, Result};
|
|
use serde_json::{json, Value};
|
|
use std::{
|
|
path::{Path, PathBuf},
|
|
process::Stdio,
|
|
time::Duration,
|
|
};
|
|
use tokio::{
|
|
io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
|
|
process::{Child, ChildStdin, ChildStdout},
|
|
};
|
|
#[derive(Debug, Default)]
|
|
pub struct BrowserClient {
|
|
state: tokio::sync::Mutex<Option<Helper>>,
|
|
timeout: Option<Duration>,
|
|
}
|
|
#[derive(Debug)]
|
|
struct Helper {
|
|
child: Child,
|
|
input: ChildStdin,
|
|
output: BufReader<ChildStdout>,
|
|
next: u64,
|
|
}
|
|
impl Drop for Helper {
|
|
fn drop(&mut self) {
|
|
#[cfg(unix)]
|
|
if let Some(pid) = self.child.id() {
|
|
// Playwright launches Chromium in a detached process group. Killing only
|
|
// the Node helper group leaves those browser processes behind.
|
|
let descendants = browser_descendants(pid);
|
|
unsafe {
|
|
libc::kill(-(pid as i32), libc::SIGKILL);
|
|
for child in descendants.into_iter().rev() {
|
|
libc::kill(child as i32, libc::SIGKILL);
|
|
}
|
|
}
|
|
}
|
|
let _ = self.child.start_kill();
|
|
}
|
|
}
|
|
#[cfg(unix)]
|
|
fn browser_descendants(root: u32) -> Vec<u32> {
|
|
let Ok(output) = std::process::Command::new("ps")
|
|
.args(["-axo", "pid=,ppid="])
|
|
.output()
|
|
else {
|
|
return vec![];
|
|
};
|
|
let pairs = String::from_utf8_lossy(&output.stdout)
|
|
.lines()
|
|
.filter_map(|line| {
|
|
let mut fields = line.split_whitespace();
|
|
Some((
|
|
fields.next()?.parse::<u32>().ok()?,
|
|
fields.next()?.parse::<u32>().ok()?,
|
|
))
|
|
})
|
|
.collect::<Vec<_>>();
|
|
let mut found = vec![root];
|
|
let mut index = 0;
|
|
while index < found.len() {
|
|
let parent = found[index];
|
|
for &(pid, ppid) in &pairs {
|
|
if ppid == parent && !found.contains(&pid) {
|
|
found.push(pid);
|
|
}
|
|
}
|
|
index += 1;
|
|
}
|
|
found.remove(0);
|
|
found
|
|
}
|
|
impl BrowserClient {
|
|
pub async fn is_started(&self) -> bool {
|
|
self.state.lock().await.is_some()
|
|
}
|
|
pub async fn close(&self) {
|
|
let mut state = self.state.lock().await;
|
|
if let Some(mut helper) = state.take() {
|
|
// Graceful Chromium shutdown flushes the persistent profile.
|
|
let _ = helper
|
|
.input
|
|
.write_all(b"{\"id\":\"close\",\"op\":\"close\"}\n")
|
|
.await;
|
|
let mut line = String::new();
|
|
let _ =
|
|
tokio::time::timeout(Duration::from_secs(3), helper.output.read_line(&mut line))
|
|
.await;
|
|
}
|
|
}
|
|
pub async fn request(
|
|
&self,
|
|
cwd: &Path,
|
|
profile: Option<PathBuf>,
|
|
mut req: Value,
|
|
) -> Result<Value> {
|
|
let mut slot = self.state.lock().await;
|
|
let mut helper = match slot.take() {
|
|
Some(h) => h,
|
|
None => spawn(cwd, profile).await?,
|
|
};
|
|
helper.next += 1;
|
|
let id = helper.next.to_string();
|
|
req["id"] = json!(id);
|
|
let reply = tokio::time::timeout(self.timeout.unwrap_or(Duration::from_secs(60)), async {
|
|
helper
|
|
.input
|
|
.write_all(format!("{}\n", req).as_bytes())
|
|
.await?;
|
|
helper.input.flush().await?;
|
|
let mut line = Vec::new();
|
|
if (&mut helper.output)
|
|
.take(4 * 1024 * 1024 + 1)
|
|
.read_until(b'\n', &mut line)
|
|
.await?
|
|
== 0
|
|
{
|
|
return Err(anyhow!(
|
|
"browser helper exited; next request will restart it"
|
|
));
|
|
}
|
|
if line.len() > 4 * 1024 * 1024 {
|
|
return Err(anyhow!("browser response too large"));
|
|
}
|
|
let value: Value = serde_json::from_slice(&line).context("browser response JSON")?;
|
|
if value["id"].as_str() != Some(&id) {
|
|
return Err(anyhow!("browser response id mismatch"));
|
|
}
|
|
Ok(value)
|
|
})
|
|
.await
|
|
.map_err(|_| {
|
|
anyhow!("browser helper timed out; its process was stopped; result may be unknown")
|
|
})??;
|
|
*slot = Some(helper);
|
|
Ok(reply)
|
|
}
|
|
}
|
|
async fn spawn(cwd: &Path, profile: Option<PathBuf>) -> Result<Helper> {
|
|
if local_browser_enabled() {
|
|
let script = crate::browser::find_helper_script(cwd)
|
|
.ok_or_else(|| anyhow!("{}", crate::browser::INSTALL_HINT))?;
|
|
return spawn_script(cwd, &script, profile).await;
|
|
}
|
|
crate::box_runtime::BoxHub::new().ensure_browser_ready().await?;
|
|
let mut command = tokio::process::Command::new("docker");
|
|
command.args(["exec", "-i", "-w", "/workspace", "-e", "GROKBOY_BROWSER_CDP=http://127.0.0.1:9222",
|
|
"grokboy-box", "flock", "-n", "/home/box/.grokboy-browser.lock", "node", "/opt/grokboy/playwright/browser_helper.mjs"]);
|
|
start_helper(command).await
|
|
}
|
|
|
|
/// Local browser is an explicit compatibility mode, never a Docker fallback.
|
|
pub(crate) fn local_browser_enabled() -> bool {
|
|
std::env::var("GROKBOY_BROWSER_SURFACE").as_deref() == Ok("local")
|
|
}
|
|
|
|
async fn spawn_script(cwd: &Path, script: &Path, profile: Option<PathBuf>) -> Result<Helper> {
|
|
let mut command = tokio::process::Command::new("node");
|
|
command.arg(script).current_dir(cwd);
|
|
if let Some(profile) = profile {
|
|
command.env("GROKBOY_BROWSER_PROFILE", profile);
|
|
}
|
|
start_helper(command).await
|
|
}
|
|
async fn start_helper(mut command: tokio::process::Command) -> Result<Helper> {
|
|
command.stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true);
|
|
#[cfg(unix)]
|
|
command.process_group(0);
|
|
let mut child = command.spawn().context("start Playwright helper")?;
|
|
let input = child.stdin.take().unwrap();
|
|
let output = BufReader::new(child.stdout.take().unwrap());
|
|
if let Some(mut err) = child.stderr.take() {
|
|
tokio::spawn(async move {
|
|
let _ = tokio::io::copy(&mut err, &mut tokio::io::sink()).await;
|
|
});
|
|
}
|
|
Ok(Helper {
|
|
child,
|
|
input,
|
|
output,
|
|
next: 0,
|
|
})
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
#[tokio::test]
|
|
async fn dropping_helper_kills_detached_browser_descendants() {
|
|
if std::process::Command::new("node").arg("--version").output().is_err() {
|
|
eprintln!("SKIPPED: detached browser cleanup test requires Node.js on PATH");
|
|
return;
|
|
}
|
|
let dir = std::env::temp_dir().join(format!("gb-detached-{}", uuid::Uuid::new_v4()));
|
|
std::fs::create_dir_all(&dir).unwrap();
|
|
let script = dir.join("fake.cjs");
|
|
std::fs::write(&script, "const c=require('child_process').spawn(process.execPath,['-e','setInterval(()=>{},1000)'],{detached:true,stdio:'ignore'});require('fs').writeFileSync('child.pid',String(c.pid));setInterval(()=>{},1000)").unwrap();
|
|
let helper = spawn_script(&dir, &script, None).await.unwrap();
|
|
tokio::time::timeout(Duration::from_secs(5), async {
|
|
while !dir.join("child.pid").exists() {
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
let pid = std::fs::read_to_string(dir.join("child.pid"))
|
|
.unwrap()
|
|
.parse::<i32>()
|
|
.unwrap();
|
|
drop(helper);
|
|
tokio::time::timeout(Duration::from_secs(5), async {
|
|
while unsafe { libc::kill(pid, 0) } == 0 {
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
}
|
|
})
|
|
.await
|
|
.expect("detached browser child must exit when helper is dropped");
|
|
std::fs::remove_dir_all(dir).unwrap();
|
|
}
|
|
#[tokio::test]
|
|
async fn helper_timeout_mismatched_id_and_cancellation_allow_restart() {
|
|
if std::process::Command::new("node")
|
|
.arg("--version")
|
|
.output()
|
|
.is_err()
|
|
{
|
|
return;
|
|
}
|
|
let dir = std::env::temp_dir().join(format!("grokboy-helper-{}", uuid::Uuid::new_v4()));
|
|
std::fs::create_dir_all(&dir).unwrap();
|
|
let script = dir.join("fake.cjs");
|
|
std::fs::write(&script, "setInterval(()=>{},1000)").unwrap();
|
|
let client = BrowserClient {
|
|
state: tokio::sync::Mutex::new(Some(spawn_script(&dir, &script, None).await.unwrap())),
|
|
timeout: Some(Duration::from_millis(250)),
|
|
};
|
|
let error = client
|
|
.request(&dir, None, json!({"op":"ping"}))
|
|
.await
|
|
.unwrap_err();
|
|
assert!(error.to_string().contains("timed out"));
|
|
assert!(!client.is_started().await);
|
|
std::fs::write(
|
|
&script,
|
|
"process.stdin.on('data',()=>console.log(JSON.stringify({id:'wrong',ok:true})))",
|
|
)
|
|
.unwrap();
|
|
*client.state.lock().await = Some(spawn_script(&dir, &script, None).await.unwrap());
|
|
assert!(client
|
|
.request(&dir, None, json!({"op":"ping"}))
|
|
.await
|
|
.unwrap_err()
|
|
.to_string()
|
|
.contains("id mismatch"));
|
|
assert!(!client.is_started().await);
|
|
std::fs::write(&script, "setInterval(()=>{},1000)").unwrap();
|
|
*client.state.lock().await = Some(spawn_script(&dir, &script, None).await.unwrap());
|
|
assert!(tokio::time::timeout(
|
|
Duration::from_millis(20),
|
|
client.request(&dir, None, json!({"op":"ping"}))
|
|
)
|
|
.await
|
|
.is_err());
|
|
assert!(!client.is_started().await);
|
|
// Restart the real local helper after failures; ping requires no Chromium.
|
|
let real_script = crate::browser::find_helper_script(&dir).unwrap();
|
|
*client.state.lock().await = Some(spawn_script(&dir, &real_script, None).await.unwrap());
|
|
assert_eq!(
|
|
client
|
|
.request(&dir, None, json!({"op":"ping"}))
|
|
.await
|
|
.unwrap()["pong"],
|
|
true
|
|
);
|
|
client.close().await;
|
|
std::fs::remove_dir_all(dir).unwrap();
|
|
}
|
|
}
|