use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{anyhow, Result};
use regex::Regex;
use serde_json::{json, Value};
use crate::bidi::BidiClient;
use crate::cdp::CdpClient;
use crate::detect::Engine;
use crate::errors::SessionError;
use crate::session::freshness;
use crate::session::targets::{open_bidi, open_cdp, BidiContext, CdpTarget};
pub enum PageSession {
Cdp(CdpPage),
Bidi(BidiPage),
}
pub struct CdpPage {
pub client: CdpClient,
pub session_id: String,
pub target_id: String,
}
pub struct BidiPage {
pub client: Arc<BidiClient>,
pub context: String,
owns_session: bool,
}
impl PageSession {
pub async fn attach(endpoint: &str, engine: Engine, url_regex: Option<&str>) -> Result<Self> {
let pattern = url_regex.map(Regex::new).transpose()?;
match engine {
Engine::Cdp => {
let client = open_cdp(endpoint).await?;
let target_id = pick_cdp_page(&client, pattern.as_ref()).await?;
let session_id = client.attach_to_target(&target_id).await?;
let _ = client
.send_with_session("Inspector.enable", json!({}), Some(&session_id))
.await;
Ok(PageSession::Cdp(CdpPage {
client,
session_id,
target_id,
}))
}
Engine::Bidi => {
let client = Arc::new(open_bidi(endpoint).await?);
client.session_new().await?;
let context = pick_bidi_context(&client, pattern.as_ref()).await?;
Ok(PageSession::Bidi(BidiPage {
client,
context,
owns_session: true,
}))
}
}
}
pub async fn from_bidi_cache(client: Arc<BidiClient>, url_regex: Option<&str>) -> Result<Self> {
let pattern = url_regex.map(Regex::new).transpose()?;
let context = pick_bidi_context(&client, pattern.as_ref()).await?;
Ok(PageSession::Bidi(BidiPage {
client,
context,
owns_session: false,
}))
}
pub async fn attach_for_origin(endpoint: &str, engine: Engine, origin: &str) -> Result<Self> {
let want =
url::Url::parse(origin).map_err(|e| anyhow!("invalid origin URL `{origin}`: {e}"))?;
let origin_root = origin_root_url(&want);
match engine {
Engine::Cdp => {
let client = open_cdp(endpoint).await?;
let target_id = match find_cdp_target_for_origin(&client, &want).await? {
Some(id) => id,
None => create_cdp_tab(&client, &origin_root).await?,
};
let session_id = client.attach_to_target(&target_id).await?;
let _ = client
.send_with_session("Inspector.enable", json!({}), Some(&session_id))
.await;
Ok(PageSession::Cdp(CdpPage {
client,
session_id,
target_id,
}))
}
Engine::Bidi => {
let client = Arc::new(open_bidi(endpoint).await?);
client.session_new().await?;
let context = match find_bidi_context_for_origin(&client, &want).await? {
Some(c) => c,
None => create_bidi_tab(&client, &origin_root).await?,
};
Ok(PageSession::Bidi(BidiPage {
client,
context,
owns_session: true,
}))
}
}
}
pub async fn evaluate(&self, expression: &str, await_promise: bool) -> Result<Value> {
self.evaluate_with_timeout(expression, await_promise, None)
.await
}
pub async fn evaluate_with_timeout(
&self,
expression: &str,
await_promise: bool,
timeout: Option<Duration>,
) -> Result<Value> {
let target_id = self.target_id();
let url = None;
match self {
PageSession::Cdp(p) => {
let inner = async {
let v = p
.client
.send_with_session(
"Runtime.evaluate",
json!({
"expression": expression,
"returnByValue": true,
"awaitPromise": await_promise,
}),
Some(&p.session_id),
)
.await?;
Ok::<Value, anyhow::Error>(v["result"]["value"].clone())
};
crate::session::crash::evaluate_with_crash_detection(
&p.client,
&p.target_id,
Some(&p.session_id),
inner,
timeout,
)
.await
}
PageSession::Bidi(p) => {
let inner = async {
let _ = await_promise; let v = p.client.script_evaluate(&p.context, expression).await?;
Ok::<Value, anyhow::Error>(v["result"]["value"].clone())
};
match timeout {
None => inner.await,
Some(d) => match tokio::time::timeout(d, inner).await {
Ok(r) => r,
Err(_) => Err(SessionError::TabHung {
target_id,
url,
timeout_ms: d.as_millis() as u64,
hint: "op-timeout",
}
.into()),
},
}
}
}
}
pub fn target_id(&self) -> Option<String> {
match self {
PageSession::Cdp(p) => Some(p.target_id.clone()),
PageSession::Bidi(p) => Some(p.context.clone()),
}
}
pub async fn navigate(&self, url: &str) -> Result<()> {
match self {
PageSession::Cdp(p) => {
p.client
.send_with_session("Page.navigate", json!({"url": url}), Some(&p.session_id))
.await?;
Ok(())
}
PageSession::Bidi(p) => {
p.client.browsing_context_navigate(&p.context, url).await?;
Ok(())
}
}
}
pub async fn ensure_fresh(&self, max_age: Duration) -> Result<()> {
let info_value = self
.evaluate_with_timeout(
freshness::PAGE_FRESHNESS_EXPR,
false,
Some(freshness::CHECK_TIMEOUT),
)
.await?;
let info = freshness::parse_page_freshness(info_value)?;
if !info.should_reload(max_age) {
return Ok(());
}
tracing::info!(
target = "session",
url = %info.href,
age_ms = info.age_ms,
max_age_ms = max_age.as_millis(),
"reloading stale page before reading page context"
);
tokio::time::timeout(freshness::RELOAD_READY_TIMEOUT, self.navigate(&info.href)).await??;
self.wait_until_ready().await
}
async fn wait_until_ready(&self) -> Result<()> {
let deadline = Instant::now() + freshness::RELOAD_READY_TIMEOUT;
loop {
let value = self
.evaluate_with_timeout(
freshness::READY_STATE_EXPR,
false,
Some(freshness::CHECK_TIMEOUT),
)
.await?;
if freshness::is_ready(&value) {
return Ok(());
}
if Instant::now() >= deadline {
tracing::warn!(
target = "session",
"page reload did not reach document.readyState=complete before continuing"
);
return Ok(());
}
tokio::time::sleep(freshness::READY_POLL_INTERVAL).await;
}
}
pub async fn screenshot(&self, full_page: bool) -> Result<String> {
match self {
PageSession::Cdp(p) => {
let v = p
.client
.send_with_session(
"Page.captureScreenshot",
json!({
"format": "png",
"captureBeyondViewport": full_page,
}),
Some(&p.session_id),
)
.await?;
v["data"]
.as_str()
.map(|s| s.to_string())
.ok_or_else(|| anyhow!("no screenshot data"))
}
PageSession::Bidi(p) => {
let _ = full_page; p.client
.browsing_context_capture_screenshot(&p.context, None)
.await
}
}
}
pub fn engine(&self) -> Engine {
match self {
PageSession::Cdp(_) => Engine::Cdp,
PageSession::Bidi(_) => Engine::Bidi,
}
}
pub async fn close(self) {
match self {
PageSession::Cdp(p) => p.client.close().await,
PageSession::Bidi(p) => {
if p.owns_session {
let _ = p.client.session_end().await;
}
}
}
}
}
pub async fn evaluate_for_origin_with_recover_once(
endpoint: &str,
engine: Engine,
origin_url: &str,
expression: &str,
await_promise: bool,
timeout: Duration,
max_age: Duration,
) -> Result<Value> {
let first = evaluate_for_origin_once(
endpoint,
engine,
origin_url,
expression,
await_promise,
timeout,
max_age,
)
.await;
match first {
Ok(v) => Ok(v),
Err(e) if crate::errors::is_recoverable_tab_failure(&e) => {
tracing::warn!(
target = "session",
"origin-bound evaluate failed with recoverable error; re-attaching and retrying once: {e:#}"
);
evaluate_for_origin_once(
endpoint,
engine,
origin_url,
expression,
await_promise,
timeout,
max_age,
)
.await
}
Err(e) => Err(e),
}
}
async fn evaluate_for_origin_once(
endpoint: &str,
engine: Engine,
origin_url: &str,
expression: &str,
await_promise: bool,
timeout: Duration,
max_age: Duration,
) -> Result<Value> {
let session = PageSession::attach_for_origin(endpoint, engine, origin_url).await?;
let result = async {
session.ensure_fresh(max_age).await?;
session
.evaluate_with_timeout(expression, await_promise, Some(timeout))
.await
}
.await;
session.close().await;
result
}
const PICK_PROBE_TIMEOUT: Duration = Duration::from_millis(500);
async fn pick_cdp_page(client: &CdpClient, pattern: Option<&Regex>) -> Result<String> {
let targets = client.list_targets().await?;
let pages: Vec<CdpTarget> = CdpTarget::pages(&targets).collect();
let Some(re) = pattern else {
return pages
.into_iter()
.next()
.map(|t| t.id)
.ok_or_else(|| anyhow!("no page target found"));
};
let matches: Vec<CdpTarget> = pages.into_iter().filter(|t| re.is_match(&t.url)).collect();
if matches.is_empty() {
return Err(anyhow!("no CDP page target matched URL regex"));
}
let mut hung_count = 0usize;
let mut last_target: Option<String> = None;
let mut last_url: Option<String> = None;
for t in &matches {
let target_id = t.id.clone();
last_target = Some(target_id.clone());
last_url = Some(t.url.clone());
if probe_cdp_target(client, &target_id, PICK_PROBE_TIMEOUT).await {
return Ok(target_id);
}
hung_count += 1;
}
let err: anyhow::Error = SessionError::TabHung {
target_id: last_target,
url: last_url,
timeout_ms: PICK_PROBE_TIMEOUT.as_millis() as u64,
hint: "all-matches-hung",
}
.into();
Err(err.context(format!(
"URL regex matched {hung_count} page(s) but none responded to a {}ms probe",
PICK_PROBE_TIMEOUT.as_millis()
)))
}
async fn probe_cdp_target(client: &CdpClient, target_id: &str, budget: Duration) -> bool {
let attach = tokio::time::timeout(
budget,
client.send(
"Target.attachToTarget",
json!({ "targetId": target_id, "flatten": true }),
),
)
.await;
let session_id = match attach {
Ok(Ok(v)) => match v.get("sessionId").and_then(|s| s.as_str()) {
Some(s) => s.to_string(),
None => return false,
},
_ => return false,
};
let eval = client.send_with_session(
"Runtime.evaluate",
json!({
"expression": "1",
"returnByValue": true,
"awaitPromise": false,
}),
Some(&session_id),
);
let alive = matches!(tokio::time::timeout(budget, eval).await, Ok(Ok(_)));
let _ = client
.send(
"Target.detachFromTarget",
json!({ "sessionId": session_id }),
)
.await;
alive
}
async fn pick_bidi_context(client: &BidiClient, pattern: Option<&Regex>) -> Result<String> {
let tree = client.send("browsingContext.getTree", json!({})).await?;
let contexts = BidiContext::from_tree(&tree);
let Some(re) = pattern else {
return contexts
.into_iter()
.next()
.map(|c| c.context)
.ok_or_else(|| anyhow!("no top-level browsing context"));
};
let matches: Vec<BidiContext> = contexts
.into_iter()
.filter(|c| re.is_match(&c.url))
.collect();
if matches.is_empty() {
return Err(anyhow!("no BiDi context matched URL regex"));
}
let mut hung_count = 0usize;
let mut last_ctx: Option<String> = None;
let mut last_url: Option<String> = None;
for c in &matches {
let ctx = c.context.clone();
last_ctx = Some(ctx.clone());
last_url = Some(c.url.clone());
if probe_bidi_context(client, &ctx, PICK_PROBE_TIMEOUT).await {
return Ok(ctx);
}
hung_count += 1;
}
let err: anyhow::Error = SessionError::TabHung {
target_id: last_ctx,
url: last_url,
timeout_ms: PICK_PROBE_TIMEOUT.as_millis() as u64,
hint: "all-matches-hung",
}
.into();
Err(err.context(format!(
"URL regex matched {hung_count} context(s) but none responded to a {}ms probe",
PICK_PROBE_TIMEOUT.as_millis()
)))
}
async fn probe_bidi_context(client: &BidiClient, context: &str, budget: Duration) -> bool {
matches!(
tokio::time::timeout(budget, client.script_evaluate(context, "1")).await,
Ok(Ok(_))
)
}
pub(crate) fn same_origin(a: &url::Url, b: &url::Url) -> bool {
a.scheme() == b.scheme()
&& a.host_str() == b.host_str()
&& a.port_or_known_default() == b.port_or_known_default()
}
pub(crate) fn origin_root_url(u: &url::Url) -> String {
let scheme = u.scheme();
let host = u.host_str().unwrap_or("");
match (u.port(), u.port_or_known_default()) {
(Some(p), _) => format!("{scheme}://{host}:{p}/"),
(None, _) => format!("{scheme}://{host}/"),
}
}
async fn find_cdp_target_for_origin(client: &CdpClient, want: &url::Url) -> Result<Option<String>> {
let targets = client.list_targets().await?;
let found = CdpTarget::pages(&targets).find_map(|t| {
let parsed = url::Url::parse(&t.url).ok()?;
same_origin(&parsed, want).then_some(t.id)
});
Ok(found)
}
async fn create_cdp_tab(client: &CdpClient, url: &str) -> Result<String> {
let v = client
.send(
"Target.createTarget",
json!({ "url": url, "background": true }),
)
.await?;
v.get("targetId")
.and_then(|x| x.as_str())
.map(|s| s.to_string())
.ok_or_else(|| anyhow!("Target.createTarget did not return targetId"))
}
async fn find_bidi_context_for_origin(
client: &BidiClient,
want: &url::Url,
) -> Result<Option<String>> {
let tree = client.send("browsingContext.getTree", json!({})).await?;
Ok(BidiContext::from_tree(&tree).into_iter().find_map(|c| {
let parsed = url::Url::parse(&c.url).ok()?;
same_origin(&parsed, want).then_some(c.context)
}))
}
async fn create_bidi_tab(client: &BidiClient, url: &str) -> Result<String> {
let v = client
.send("browsingContext.create", json!({ "type": "tab" }))
.await?;
let ctx = v
.get("context")
.and_then(|x| x.as_str())
.ok_or_else(|| anyhow!("browsingContext.create did not return context"))?
.to_string();
client.browsing_context_navigate(&ctx, url).await?;
Ok(ctx)
}
#[cfg(test)]
mod tests {
use super::*;
use futures_util::{SinkExt, StreamExt};
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
};
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;
async fn spawn_cdp_mock(targets: Vec<Value>) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
while let Some(Ok(Message::Text(t))) = ws.next().await {
let req: Value = serde_json::from_str(&t).unwrap();
let id = req["id"].as_u64().unwrap();
let method = req["method"].as_str().unwrap_or("");
let result = match method {
"Target.getTargets" => json!({"targetInfos": targets.clone()}),
"Target.attachToTarget" => json!({"sessionId": "S1"}),
"Target.createTarget" => json!({"targetId": "NEW"}),
"Runtime.evaluate" => json!({"result": {"value": "ok"}}),
"Page.navigate" => json!({}),
"Page.captureScreenshot" => json!({"data": "PNGDATA"}),
_ => json!({}),
};
let resp = json!({"id": id, "result": result});
ws.send(Message::Text(resp.to_string())).await.unwrap();
}
});
format!("ws://{addr}")
}
async fn spawn_cdp_origin_eval_mock(
targets: Vec<Value>,
fail_first_eval: bool,
) -> (String, Arc<Mutex<Vec<Value>>>, Arc<Mutex<Vec<String>>>) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let targets = Arc::new(targets);
let created_params = Arc::new(Mutex::new(Vec::new()));
let attached_targets = Arc::new(Mutex::new(Vec::new()));
let eval_count = Arc::new(AtomicUsize::new(0));
tokio::spawn({
let targets = targets.clone();
let created_params = created_params.clone();
let attached_targets = attached_targets.clone();
let eval_count = eval_count.clone();
async move {
loop {
let Ok((stream, _)) = listener.accept().await else {
break;
};
let targets = targets.clone();
let created_params = created_params.clone();
let attached_targets = attached_targets.clone();
let eval_count = eval_count.clone();
tokio::spawn(async move {
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
while let Some(Ok(Message::Text(t))) = ws.next().await {
let req: Value = serde_json::from_str(&t).unwrap();
let id = req["id"].as_u64().unwrap();
let method = req["method"].as_str().unwrap_or("");
if method == "Runtime.evaluate"
&& fail_first_eval
&& eval_count.fetch_add(1, Ordering::SeqCst) == 0
{
let resp = json!({
"id": id,
"error": {
"code": -32000,
"message": "No target with given id",
}
});
ws.send(Message::Text(resp.to_string())).await.unwrap();
continue;
}
let result = match method {
"Target.getTargets" => {
json!({"targetInfos": targets.as_ref().clone()})
}
"Target.createTarget" => {
created_params.lock().await.push(req["params"].clone());
json!({"targetId": "NEW"})
}
"Target.attachToTarget" => {
let target_id = req
.pointer("/params/targetId")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let mut attached = attached_targets.lock().await;
attached.push(target_id);
json!({"sessionId": format!("S{}", attached.len())})
}
"Target.detachFromTarget" => json!({}),
"Inspector.enable" => json!({}),
"Runtime.evaluate" => {
let expression = req
.pointer("/params/expression")
.and_then(|v| v.as_str())
.unwrap_or("");
let value = if expression == freshness::READY_STATE_EXPR {
json!("complete")
} else if expression == freshness::PAGE_FRESHNESS_EXPR {
json!({
"href": "https://example.com/login",
"ageMs": 0.0,
"readyState": "complete"
})
} else {
json!("ok")
};
json!({"result": {"value": value}})
}
_ => json!({}),
};
let resp = json!({"id": id, "result": result});
ws.send(Message::Text(resp.to_string())).await.unwrap();
}
});
}
}
});
(format!("ws://{addr}"), created_params, attached_targets)
}
#[test]
fn same_origin_basic() {
let a = url::Url::parse("https://example.com/path?q=1").unwrap();
let b = url::Url::parse("https://example.com/other").unwrap();
let c = url::Url::parse("https://other.test/path").unwrap();
let d = url::Url::parse("http://example.com/").unwrap();
assert!(same_origin(&a, &b));
assert!(!same_origin(&a, &c));
assert!(!same_origin(&a, &d));
}
#[test]
fn origin_root_strips_path_and_default_port() {
let u = url::Url::parse("https://example.com/foo/bar?x=1#z").unwrap();
assert_eq!(origin_root_url(&u), "https://example.com/");
let u2 = url::Url::parse("http://localhost:8080/foo").unwrap();
assert_eq!(origin_root_url(&u2), "http://localhost:8080/");
}
#[tokio::test]
async fn attach_for_origin_reuses_matching_tab() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://other.test/x"}),
json!({"targetId":"b","type":"page","url":"https://example.com/login"}),
])
.await;
let s = PageSession::attach_for_origin(&url, Engine::Cdp, "https://example.com/api/v1")
.await
.unwrap();
match s {
PageSession::Cdp(p) => assert_eq!(p.target_id, "b"),
_ => panic!("expected CDP"),
}
}
#[tokio::test]
async fn attach_for_origin_creates_tab_when_no_match() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://other.test/"}),
])
.await;
let s = PageSession::attach_for_origin(&url, Engine::Cdp, "https://example.com/api")
.await
.unwrap();
match s {
PageSession::Cdp(p) => assert_eq!(p.target_id, "NEW"),
_ => panic!("expected CDP"),
}
}
#[tokio::test]
async fn evaluate_for_origin_creates_origin_tab_when_no_match() {
let (url, created_params, attached_targets) = spawn_cdp_origin_eval_mock(
vec![json!({"targetId":"a","type":"page","url":"https://other.test/"})],
false,
)
.await;
let value = evaluate_for_origin_with_recover_once(
&url,
Engine::Cdp,
"https://example.com/api",
"1+1",
true,
Duration::from_secs(1),
freshness::DEFAULT_MAX_AGE,
)
.await
.unwrap();
assert_eq!(value, json!("ok"));
let created = created_params.lock().await;
assert_eq!(created.len(), 1);
assert_eq!(created[0]["url"], "https://example.com/");
assert_eq!(created[0]["background"], true);
assert_eq!(*attached_targets.lock().await, vec!["NEW".to_string()]);
}
#[tokio::test]
async fn evaluate_for_origin_reattaches_and_retries_once() {
let (url, created_params, attached_targets) = spawn_cdp_origin_eval_mock(
vec![json!({"targetId":"A","type":"page","url":"https://example.com/login"})],
true,
)
.await;
let value = evaluate_for_origin_with_recover_once(
&url,
Engine::Cdp,
"https://example.com/api",
"1+1",
true,
Duration::from_secs(1),
freshness::DEFAULT_MAX_AGE,
)
.await
.unwrap();
assert_eq!(value, json!("ok"));
assert!(created_params.lock().await.is_empty());
assert_eq!(
*attached_targets.lock().await,
vec!["A".to_string(), "A".to_string()]
);
}
#[tokio::test]
async fn attach_cdp_picks_first_page_when_no_regex() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://example.com/"}),
json!({"targetId":"b","type":"page","url":"https://other.test/"}),
])
.await;
let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
match s {
PageSession::Cdp(p) => {
assert_eq!(p.target_id, "a");
assert_eq!(p.session_id, "S1");
}
_ => panic!("expected CDP"),
}
}
#[tokio::test]
async fn attach_cdp_url_regex_selects_matching() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://example.com/"}),
json!({"targetId":"b","type":"page","url":"https://other.test/"}),
])
.await;
let s = PageSession::attach(&url, Engine::Cdp, Some(r"other"))
.await
.unwrap();
match s {
PageSession::Cdp(p) => assert_eq!(p.target_id, "b"),
_ => panic!("expected CDP"),
}
}
#[tokio::test]
async fn attach_cdp_url_regex_no_match_errors() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://example.com/"}),
])
.await;
let err = match PageSession::attach(&url, Engine::Cdp, Some("nomatch")).await {
Ok(_) => panic!("expected error"),
Err(e) => e,
};
assert!(err.to_string().contains("no CDP page target matched"));
}
#[tokio::test]
async fn evaluate_round_trip_cdp() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://example.com/"}),
])
.await;
let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
let v = s.evaluate("1+1", false).await.unwrap();
assert_eq!(v, json!("ok"));
s.close().await;
}
#[tokio::test]
async fn screenshot_round_trip_cdp() {
let url = spawn_cdp_mock(vec![
json!({"targetId":"a","type":"page","url":"https://example.com/"}),
])
.await;
let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
let b64 = s.screenshot(false).await.unwrap();
assert_eq!(b64, "PNGDATA");
s.close().await;
}
async fn spawn_cdp_mock_eval_hangs(targets: Vec<Value>) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
while let Some(Ok(Message::Text(t))) = ws.next().await {
let req: Value = serde_json::from_str(&t).unwrap();
let id = req["id"].as_u64().unwrap();
let method = req["method"].as_str().unwrap_or("");
if method == "Runtime.evaluate" {
continue;
}
let result = match method {
"Target.getTargets" => json!({"targetInfos": targets.clone()}),
"Target.attachToTarget" => json!({"sessionId": "S1"}),
_ => json!({}),
};
let resp = json!({"id": id, "result": result});
ws.send(Message::Text(resp.to_string())).await.unwrap();
}
});
format!("ws://{addr}")
}
#[tokio::test]
async fn evaluate_with_timeout_returns_tab_hung_on_no_reply() {
let url = spawn_cdp_mock_eval_hangs(vec![
json!({"targetId":"iLO","type":"page","url":"https://192.168.2.28/"}),
])
.await;
let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
let start = std::time::Instant::now();
let err = s
.evaluate_with_timeout("1+1", false, Some(Duration::from_millis(300)))
.await
.expect_err("must return TabHung");
let elapsed = start.elapsed();
assert!(
elapsed < Duration::from_secs(1),
"did not honour 300ms bound, took {elapsed:?}"
);
let downcast = err.downcast_ref::<SessionError>().expect("typed error");
match downcast {
SessionError::TabHung {
target_id,
timeout_ms,
hint,
..
} => {
assert_eq!(target_id.as_deref(), Some("iLO"));
assert_eq!(*timeout_ms, 300);
assert_eq!(*hint, "op-timeout");
}
other => panic!("expected TabHung, got {other:?}"),
}
s.close().await;
}
#[tokio::test]
async fn stuck_eval_does_not_block_sibling_session() {
let bad = spawn_cdp_mock_eval_hangs(vec![
json!({"targetId":"BAD","type":"page","url":"https://192.168.2.28/"}),
])
.await;
let good = spawn_cdp_mock(vec![
json!({"targetId":"GOOD","type":"page","url":"https://example.com/"}),
])
.await;
let s_bad = PageSession::attach(&bad, Engine::Cdp, None).await.unwrap();
let s_good = PageSession::attach(&good, Engine::Cdp, None).await.unwrap();
let bad_fut = s_bad.evaluate_with_timeout("1+1", false, Some(Duration::from_millis(200)));
let good_fut = s_good.evaluate_with_timeout("1+1", false, Some(Duration::from_secs(5)));
let (bad_res, good_res) = tokio::join!(bad_fut, good_fut);
assert!(bad_res.is_err(), "bad session must surface TabHung");
assert_eq!(good_res.unwrap(), json!("ok"));
s_bad.close().await;
s_good.close().await;
}
async fn spawn_cdp_mock_per_target_eval(
targets: Vec<Value>,
wedged_targets: Vec<&'static str>,
) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
let mut session_wedge: std::collections::HashMap<String, bool> =
std::collections::HashMap::new();
let mut next_session: u32 = 0;
while let Some(Ok(Message::Text(t))) = ws.next().await {
let req: Value = serde_json::from_str(&t).unwrap();
let id = req["id"].as_u64().unwrap();
let method = req["method"].as_str().unwrap_or("");
if method == "Runtime.evaluate" {
if let Some(sid) = req.get("sessionId").and_then(|v| v.as_str()) {
if session_wedge.get(sid).copied().unwrap_or(false) {
continue;
}
}
}
let result = match method {
"Target.getTargets" => json!({"targetInfos": targets.clone()}),
"Target.attachToTarget" => {
let target_id = req
.get("params")
.and_then(|p| p.get("targetId"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
next_session += 1;
let sid = format!("S{next_session}");
let wedge = wedged_targets.iter().any(|w| *w == target_id);
session_wedge.insert(sid.clone(), wedge);
json!({"sessionId": sid})
}
"Target.detachFromTarget" => json!({}),
"Runtime.evaluate" => json!({"result": {"value": "ok"}}),
_ => json!({}),
};
let resp = json!({"id": id, "result": result});
ws.send(Message::Text(resp.to_string())).await.unwrap();
}
});
format!("ws://{addr}")
}
#[tokio::test]
async fn pick_cdp_iterates_past_hung_match() {
let url = spawn_cdp_mock_per_target_eval(
vec![
json!({"targetId":"DEAD","type":"page","url":"https://twitch.tv/gametechnology"}),
json!({"targetId":"LIVE","type":"page","url":"https://gametechnology.somewhere.com"}),
],
vec!["DEAD"],
)
.await;
let s = PageSession::attach(&url, Engine::Cdp, Some(r"gametechnology"))
.await
.expect("must iterate past the wedged tab and pick LIVE");
match s {
PageSession::Cdp(p) => assert_eq!(p.target_id, "LIVE"),
_ => panic!("expected CDP"),
}
}
#[tokio::test]
async fn pick_cdp_all_matches_hung_returns_tab_hung() {
let url = spawn_cdp_mock_per_target_eval(
vec![
json!({"targetId":"A","type":"page","url":"https://example.com/foo"}),
json!({"targetId":"B","type":"page","url":"https://example.com/bar"}),
],
vec!["A", "B"],
)
.await;
let start = std::time::Instant::now();
let err = match PageSession::attach(&url, Engine::Cdp, Some(r"example\.com")).await {
Ok(_) => panic!("all matches wedged → must error"),
Err(e) => e,
};
let elapsed = start.elapsed();
assert!(
elapsed < PICK_PROBE_TIMEOUT * 2 + Duration::from_millis(500),
"took too long: {elapsed:?}"
);
let typed = err.downcast_ref::<SessionError>().expect("typed error");
match typed {
SessionError::TabHung { hint, .. } => {
assert_eq!(*hint, "all-matches-hung");
}
other => panic!("expected TabHung, got {other:?}"),
}
let text = format!("{err:#}");
assert!(
text.contains("URL regex matched 2 page(s)"),
"context missing count: {text}"
);
}
#[tokio::test]
async fn pick_cdp_single_healthy_match_is_picked() {
let url = spawn_cdp_mock_per_target_eval(
vec![json!({"targetId":"OK","type":"page","url":"https://example.com/x"})],
vec![],
)
.await;
let s = PageSession::attach(&url, Engine::Cdp, Some(r"example"))
.await
.unwrap();
match s {
PageSession::Cdp(p) => assert_eq!(p.target_id, "OK"),
_ => panic!("expected CDP"),
}
}
}