browser_control/sidecar/
mod.rs1use anyhow::{anyhow, Context, Result};
30use serde_json::{json, Value};
31use std::collections::HashMap;
32use std::path::PathBuf;
33use std::process::Stdio;
34use std::sync::atomic::{AtomicU64, Ordering};
35use std::sync::Arc;
36use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
37use tokio::process::{Child, Command};
38use tokio::sync::{oneshot, Mutex};
39
40pub mod assets;
41#[cfg(test)]
42mod tests;
43
44pub const DEFAULT_PLAYWRIGHT_VERSION: &str = "1.49.1";
47
48#[derive(Debug, Clone, Default)]
50pub struct SidecarConfig {
51 pub version: Option<String>,
53}
54
55impl SidecarConfig {
56 fn resolved_version(&self) -> &str {
57 self.version
58 .as_deref()
59 .unwrap_or(DEFAULT_PLAYWRIGHT_VERSION)
60 }
61}
62
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub enum Launcher {
66 Bun,
68 Node,
70}
71
72impl Launcher {
73 pub fn detect() -> Result<Launcher> {
75 if which::which("bun").is_ok() {
76 return Ok(Launcher::Bun);
77 }
78 if which::which("node").is_ok() && which::which("npm").is_ok() {
79 return Ok(Launcher::Node);
80 }
81 Err(anyhow!(
82 "neither `bun` nor `node`+`npm` is on PATH. \
83 Install Bun (https://bun.sh/) or Node.js (https://nodejs.org/)."
84 ))
85 }
86}
87
88type PendingMap = HashMap<u64, oneshot::Sender<Result<Value>>>;
91
92#[derive(Clone)]
95pub struct Sidecar {
96 next_id: Arc<AtomicU64>,
97 pending: Arc<Mutex<PendingMap>>,
98 write_tx: tokio::sync::mpsc::UnboundedSender<String>,
99 _inner: Arc<SidecarInner>,
101}
102
103struct SidecarInner {
106 #[allow(dead_code)] child: Mutex<Option<Child>>,
108 reader_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
109 writer_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
110}
111
112fn truncate_line(line: &str) -> std::borrow::Cow<'_, str> {
115 const MAX: usize = 200;
116 if line.len() <= MAX {
117 std::borrow::Cow::Borrowed(line)
118 } else {
119 let end = line
120 .char_indices()
121 .take_while(|(i, _)| *i < MAX)
122 .last()
123 .map(|(i, c)| i + c.len_utf8())
124 .unwrap_or(0);
125 std::borrow::Cow::Owned(format!("{}… ({} bytes total)", &line[..end], line.len()))
126 }
127}
128
129impl Drop for SidecarInner {
130 fn drop(&mut self) {
131 if let Ok(mut guard) = self.child.try_lock() {
134 if let Some(mut c) = guard.take() {
135 let _ = c.start_kill();
136 }
137 }
138 if let Ok(mut guard) = self.reader_handle.try_lock() {
139 if let Some(h) = guard.take() {
140 h.abort();
141 }
142 }
143 if let Ok(mut guard) = self.writer_handle.try_lock() {
144 if let Some(h) = guard.take() {
145 h.abort();
146 }
147 }
148 }
149}
150
151impl Sidecar {
152 pub async fn start(config: SidecarConfig) -> Result<Self> {
156 let launcher = Launcher::detect()?;
157 let cache_dir = assets::ensure_sidecar_dir(config.resolved_version())
158 .await
159 .context("preparing sidecar cache directory")?;
160 Self::install_deps(launcher, &cache_dir).await?;
161 Self::spawn(launcher, &cache_dir).await
162 }
163
164 async fn install_deps(launcher: Launcher, cache_dir: &PathBuf) -> Result<()> {
167 let marker = cache_dir.join("node_modules").join("playwright-core");
168 if tokio::fs::metadata(&marker).await.is_ok() {
169 return Ok(());
170 }
171 let (program, args) = match launcher {
172 Launcher::Bun => ("bun", vec!["install", "--silent"]),
173 Launcher::Node => ("npm", vec!["install", "--silent"]),
174 };
175 let status = Command::new(program)
176 .args(&args)
177 .current_dir(cache_dir)
178 .status()
179 .await
180 .with_context(|| format!("running `{program} {}` in {cache_dir:?}", args.join(" ")))?;
181 if !status.success() {
182 return Err(anyhow!(
183 "`{program} {}` in {cache_dir:?} exited with status {status}",
184 args.join(" ")
185 ));
186 }
187 Ok(())
188 }
189
190 async fn spawn(launcher: Launcher, cache_dir: &PathBuf) -> Result<Self> {
192 let (program, args) = match launcher {
193 Launcher::Bun => ("bun", vec!["run", "sidecar.mjs"]),
194 Launcher::Node => ("node", vec!["sidecar.mjs"]),
195 };
196 let mut child = Command::new(program)
197 .args(&args)
198 .current_dir(cache_dir)
199 .stdin(Stdio::piped())
200 .stdout(Stdio::piped())
201 .stderr(Stdio::inherit())
202 .kill_on_drop(true)
203 .spawn()
204 .with_context(|| format!("spawning `{program} {}`", args.join(" ")))?;
205
206 let mut child_stdin = child
207 .stdin
208 .take()
209 .ok_or_else(|| anyhow!("no child stdin"))?;
210 let child_stdout = child
211 .stdout
212 .take()
213 .ok_or_else(|| anyhow!("no child stdout"))?;
214
215 let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));
216 let (write_tx, mut write_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
217
218 let writer_handle = tokio::spawn(async move {
219 while let Some(line) = write_rx.recv().await {
220 if child_stdin.write_all(line.as_bytes()).await.is_err() {
221 break;
222 }
223 if child_stdin.write_all(b"\n").await.is_err() {
224 break;
225 }
226 let _ = child_stdin.flush().await;
227 }
228 });
229
230 let pending_r = pending.clone();
231 let reader_handle = tokio::spawn(async move {
232 let mut lines = BufReader::new(child_stdout).lines();
233 while let Ok(Some(line)) = lines.next_line().await {
234 if line.trim().is_empty() {
235 continue;
236 }
237 let v: Value = match serde_json::from_str(&line) {
238 Ok(v) => v,
239 Err(e) => {
243 tracing::warn!(
244 error = %e,
245 line = %truncate_line(&line),
246 "sidecar: dropping unparseable stdout line"
247 );
248 continue;
249 }
250 };
251 let id = match v.get("id").and_then(|x| x.as_u64()) {
252 Some(i) => i,
253 None => {
254 tracing::debug!(
255 line = %truncate_line(&line),
256 "sidecar: dropping idless stdout line"
257 );
258 continue;
259 }
260 };
261 let result = if let Some(err) = v.get("error") {
262 let msg = err
263 .get("message")
264 .and_then(|m| m.as_str())
265 .unwrap_or("(no message)");
266 Err(anyhow!("{msg}"))
267 } else {
268 Ok(v.get("result").cloned().unwrap_or(Value::Null))
269 };
270 let tx = {
271 let mut p = pending_r.lock().await;
272 p.remove(&id)
273 };
274 if let Some(tx) = tx {
275 let _ = tx.send(result);
276 }
277 }
278 let mut p = pending_r.lock().await;
280 for (_, tx) in p.drain() {
281 let _ = tx.send(Err(anyhow!("sidecar stdout closed")));
282 }
283 });
284
285 Ok(Self {
286 next_id: Arc::new(AtomicU64::new(1)),
287 pending,
288 write_tx,
289 _inner: Arc::new(SidecarInner {
290 child: Mutex::new(Some(child)),
291 reader_handle: Mutex::new(Some(reader_handle)),
292 writer_handle: Mutex::new(Some(writer_handle)),
293 }),
294 })
295 }
296
297 pub async fn connect(&self, endpoint: &str) -> Result<Value> {
301 self.call("connect", json!({ "endpoint": endpoint })).await
302 }
303
304 pub async fn call(&self, method: &str, params: Value) -> Result<Value> {
306 let id = self.next_id.fetch_add(1, Ordering::SeqCst);
307 let req = json!({ "id": id, "method": method, "params": params });
308 let line = serde_json::to_string(&req)?;
309 let (tx, rx) = oneshot::channel();
310 {
311 let mut p = self.pending.lock().await;
312 p.insert(id, tx);
313 }
314 if self.write_tx.send(line).is_err() {
315 let mut p = self.pending.lock().await;
316 p.remove(&id);
317 return Err(anyhow!("sidecar writer closed"));
318 }
319 match rx.await {
320 Ok(r) => r,
321 Err(_) => Err(anyhow!("sidecar response channel dropped")),
322 }
323 }
324}