Skip to content

Commit 7acd612

Browse files
KeyCode17claude
KeyCode17
and
claude
authored
feat(camoufox): per-domain persistent session for /v1/fetch (#7)
Replaces the spawn-per-fetch model with a per-domain warm Camoufox session that survives across `/v1/fetch` calls. Three concrete wins: 1. The ~13s geckodriver + Camoufox cold start now happens once per domain, not on every fetch. 2. PerimeterX sees a single coherent browser session instead of a stream of identical short-lived browsers; this is what trips its "tráfico inusual" rate limit and was making /v1/fetch unusable after roughly the 5th call against the same target. 3. `cf_clearance` issued during the first navigation persists for the life of the session, so subsequent fetches don't re-run the Cloudflare managed challenge. What's in this commit - New `infrastructure/session.rs` — `PersistentSession` holding fantoccini::Client + tokio::process::Child (kill_on_drop) plus an `Inner` mutex (last_used, fetch_count, warmed flag). One handle == one warm browser. - New `infrastructure/session_pool.rs` — `SessionPool` keyed by domain, lazy spawn, 5-minute TTL, race-safe double-check on insert. - New `infrastructure/fetch_script.rs` — extracted the in-page `fetch` JS builder so `camoufox_fetcher.rs` stays under the 200-LOC axum-best-practice rule. - New `infrastructure/fetch_strategies.rs` — `navigate_and_read` (GET, lets the browser handle CF interactive challenges) and `in_page_fetch` (POST/PUT/etc., since webdriver navigation can't carry a body). The GET branch sniffs status from body shape since webdriver doesn't surface response codes. - `camoufox_pool.rs` — `CamoufoxPool` now owns `Arc<SessionPool>`. Harvest path is unchanged (still spawn-per-call); only the Fetcher impl uses the pool. - `camoufox_fetcher.rs` rewritten to acquire a session, warm it once (homepage navigation → cf_clearance), optionally navigate to the caller's `Referer` so PX's sensor runs against the right path, then dispatch GET vs POST. Operational notes - Session TTL is hardcoded at 5 minutes for v1; lifecycle/eviction ADR follow-up if this needs to be configurable. - Concurrent fetches for the same domain serialize on the session mutex — intentional, since one warm browser can't handle parallel navigations safely. - No background reaper task yet; aged-out sessions are recycled lazily on the next acquire(). - Process exit cleans up via geckodriver's `kill_on_drop` on the held Child, which fires when `SessionPool`'s HashMap drops. Live validation deferred - The implementation has been smoke-tested on the local box, but the test IP is currently sitting under a pedidosya PerimeterX rate-limit ("tráfico inusual" body) from the many ad-hoc /v1/solve calls made while debugging upstream JA3 / cookie-replay issues. The rate limit is not specific to the session-pool change — it also blocks the v1.4.0 binary. Re-running this once the rate limit lifts (or from a fresh IP) is the verification path before shipping this in a release. Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
1 parent 79090b4 commit 7acd612

7 files changed

Lines changed: 363 additions & 88 deletions

File tree

‎px-camoufox/src/infrastructure/camoufox_fetcher.rs‎

Lines changed: 68 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -1,88 +1,84 @@
1-
//! `Fetcher` impl on `CamoufoxPool`. Runs a single HTTP request from
2-
//! inside a fresh Camoufox session by navigating to the target URL's
3-
//! origin (so cookies/JS context are issued), then executing a
4-
//! `fetch()` from the page context. Captures status, headers, body.
1+
//! `Fetcher` impl on `CamoufoxPool`. Reuses a per-domain persistent
2+
//! Camoufox session so the geckodriver/browser cold-start (~13s)
3+
//! happens once per domain, PerimeterX sees a coherent browser
4+
//! session instead of a stream of throwaway ones, and `cf_clearance`
5+
//! issued during the first navigation is reused.
6+
//!
7+
//! GET routes through `navigate_and_read` so the browser handles
8+
//! interactive Cloudflare challenges; POST routes through
9+
//! `in_page_fetch` because webdriver navigation can't carry a body.
10+
//! See `fetch_strategies` for both.
511
612
use crate::infrastructure::camoufox_pool::CamoufoxPool;
13+
use crate::infrastructure::fetch_strategies::{
14+
in_page_fetch, navigate_and_read, navigate_with_wait,
15+
};
16+
use crate::infrastructure::session::PersistentSession;
717
use async_trait::async_trait;
8-
use fantoccini::ClientBuilder;
918
use px_errors::AppError;
1019
use px_pipeline::{FetchRequest, FetchResponse, Fetcher};
11-
use serde_json::{Map, Value};
12-
use std::collections::HashMap;
20+
use std::sync::Arc;
1321
use std::time::{Duration, Instant};
14-
use tokio::time::sleep;
1522

1623
#[async_trait]
1724
impl Fetcher for CamoufoxPool {
1825
async fn fetch(&self, req: FetchRequest) -> Result<FetchResponse, AppError> {
26+
let domain = domain_of(&req.url)?;
27+
let session = self.sessions.acquire(&domain).await?;
1928
let navigate_timeout = self.config.navigate_timeout;
2029
let request_timeout = Duration::from_millis(req.timeout_ms);
21-
self.with_session(None, async move |endpoint, caps| {
22-
run_fetch(&endpoint, caps, &req, navigate_timeout, request_timeout).await
23-
})
24-
.await
30+
run_through_session(session, req, navigate_timeout, request_timeout).await
2531
}
2632
}
2733

28-
async fn run_fetch(
29-
endpoint: &str,
30-
caps: Map<String, Value>,
31-
req: &FetchRequest,
34+
async fn run_through_session(
35+
session: Arc<PersistentSession>,
36+
req: FetchRequest,
3237
navigate_timeout: Duration,
3338
request_timeout: Duration,
3439
) -> Result<FetchResponse, AppError> {
3540
let started = Instant::now();
36-
let client = ClientBuilder::native()
37-
.capabilities(caps)
38-
.connect(endpoint)
39-
.await
40-
.map_err(|e| AppError::InternalError(format!("webdriver connect: {e}")))?;
41+
let mut inner = session.inner.lock().await;
4142

42-
let origin = origin_of(&req.url)?;
43-
let nav = client.goto(&origin);
44-
if tokio::time::timeout(navigate_timeout, nav).await.is_err() {
45-
let _ = client.close().await;
46-
return Err(AppError::InternalError("navigate timeout".into()));
43+
// Warm the browser once per session: load the origin so Cloudflare's
44+
// managed challenge JS runs and `cf_clearance` lands in the jar.
45+
if !inner.warmed {
46+
let origin = origin_of(&req.url)?;
47+
if let Err(e) = navigate_with_wait(
48+
&inner.client,
49+
&origin,
50+
navigate_timeout,
51+
Duration::from_millis(4_000),
52+
)
53+
.await
54+
{
55+
tracing::warn!(error = %e, "session warmup failed; downstream will likely 403");
56+
} else {
57+
inner.warmed = true;
58+
}
59+
}
60+
if let Some(referer) = referer_header(&req)
61+
&& !referer.is_empty()
62+
{
63+
let _ = navigate_with_wait(
64+
&inner.client,
65+
&referer,
66+
navigate_timeout,
67+
Duration::from_millis(2_000),
68+
)
69+
.await;
4770
}
48-
// Give Cloudflare / PerimeterX a beat to set their cookies.
49-
sleep(Duration::from_millis(1_500)).await;
5071

51-
let script = build_fetch_script(req)?;
52-
let exec = client.execute_async(&script, vec![]);
53-
let raw = match tokio::time::timeout(request_timeout, exec).await {
54-
Ok(Ok(v)) => v,
55-
Ok(Err(e)) => {
56-
let _ = client.close().await;
57-
return Err(AppError::InternalError(format!("fetch eval: {e}")));
58-
}
59-
Err(_) => {
60-
let _ = client.close().await;
61-
return Err(AppError::InternalError("fetch timeout".into()));
62-
}
72+
let outcome = if req.method().eq_ignore_ascii_case("GET") {
73+
navigate_and_read(&inner.client, &req, navigate_timeout).await
74+
} else {
75+
in_page_fetch(&inner.client, &req, request_timeout).await
6376
};
64-
let _ = client.close().await;
65-
66-
let status = raw
67-
.get("status")
68-
.and_then(Value::as_u64)
69-
.ok_or_else(|| AppError::InternalError("fetch result missing status".into()))?
70-
as u16;
71-
let body = raw
72-
.get("body")
73-
.and_then(Value::as_str)
74-
.unwrap_or("")
75-
.to_string();
76-
let headers: HashMap<String, String> = raw
77-
.get("headers")
78-
.and_then(Value::as_object)
79-
.map(|m| {
80-
m.iter()
81-
.filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
82-
.collect()
83-
})
84-
.unwrap_or_default();
77+
inner.last_used = Instant::now();
78+
inner.fetch_count = inner.fetch_count.saturating_add(1);
79+
drop(inner);
8580

81+
let (status, headers, body) = outcome?;
8682
Ok(FetchResponse {
8783
status,
8884
headers,
@@ -102,30 +98,15 @@ fn origin_of(url: &str) -> Result<String, AppError> {
10298
))
10399
}
104100

105-
fn build_fetch_script(req: &FetchRequest) -> Result<String, AppError> {
106-
let method = req.method().to_string();
107-
let headers_json = serde_json::to_string(&req.headers)
108-
.map_err(|e| AppError::InternalError(format!("encode headers: {e}")))?;
109-
let body_json = serde_json::to_string(req.body.as_deref().unwrap_or(""))
110-
.map_err(|e| AppError::InternalError(format!("encode body: {e}")))?;
111-
let url_json = serde_json::to_string(&req.url)
112-
.map_err(|e| AppError::InternalError(format!("encode url: {e}")))?;
113-
let method_json = serde_json::to_string(&method)
114-
.map_err(|e| AppError::InternalError(format!("encode method: {e}")))?;
115-
Ok(format!(
116-
r#"
117-
const cb = arguments[arguments.length - 1];
118-
const opts = {{ method: {method_json}, headers: {headers_json}, credentials: 'include' }};
119-
const body = {body_json};
120-
if (body !== '' && {method_json} !== 'GET') opts.body = body;
121-
fetch({url_json}, opts)
122-
.then(async (r) => {{
123-
const text = await r.text();
124-
const hdrs = {{}};
125-
r.headers.forEach((v, k) => {{ hdrs[k] = v; }});
126-
cb({{ status: r.status, headers: hdrs, body: text }});
127-
}})
128-
.catch((e) => cb({{ status: 0, headers: {{}}, body: String(e) }}));
129-
"#
130-
))
101+
fn domain_of(url: &str) -> Result<String, AppError> {
102+
let parsed =
103+
url::Url::parse(url).map_err(|e| AppError::BadRequest(format!("invalid url: {e}")))?;
104+
Ok(parsed.host_str().unwrap_or("").to_string())
105+
}
106+
107+
fn referer_header(req: &FetchRequest) -> Option<String> {
108+
req.headers
109+
.iter()
110+
.find(|(k, _)| k.eq_ignore_ascii_case("referer"))
111+
.map(|(_, v)| v.clone())
131112
}

‎px-camoufox/src/infrastructure/camoufox_pool.rs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,12 @@ use tokio::process::Command;
1111
use tokio::sync::Semaphore;
1212
use tokio::time::sleep;
1313

14+
use crate::infrastructure::session_pool::SessionPool;
15+
1416
pub struct CamoufoxPool {
1517
pub(crate) config: CamoufoxConfig,
1618
pub(crate) permits: Arc<Semaphore>,
19+
pub(crate) sessions: Arc<SessionPool>,
1720
}
1821

1922
impl CamoufoxPool {
@@ -22,7 +25,12 @@ impl CamoufoxPool {
2225
.validate()
2326
.map_err(|e| AppError::InternalError(format!("camoufox config: {e}")))?;
2427
let permits = Arc::new(Semaphore::new(config.max_concurrent));
25-
Ok(Self { config, permits })
28+
let sessions = Arc::new(SessionPool::new(config.clone(), Duration::from_secs(300)));
29+
Ok(Self {
30+
config,
31+
permits,
32+
sessions,
33+
})
2634
}
2735

2836
/// Spawn geckodriver + Camoufox, hand the resulting webdriver

‎px-camoufox/src/infrastructure/fetch_script.rs‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
//! JS fragment that runs an in-page `fetch` inside Camoufox and posts
2+
//! the response back to fantoccini via `execute_async`'s callback.
3+
4+
use px_errors::AppError;
5+
use px_pipeline::FetchRequest;
6+
7+
pub(crate) fn build(req: &FetchRequest) -> Result<String, AppError> {
8+
let method = req.method().to_string();
9+
let headers_json = serde_json::to_string(&req.headers)
10+
.map_err(|e| AppError::InternalError(format!("encode headers: {e}")))?;
11+
let body_json = serde_json::to_string(req.body.as_deref().unwrap_or(""))
12+
.map_err(|e| AppError::InternalError(format!("encode body: {e}")))?;
13+
let url_json = serde_json::to_string(&req.url)
14+
.map_err(|e| AppError::InternalError(format!("encode url: {e}")))?;
15+
let method_json = serde_json::to_string(&method)
16+
.map_err(|e| AppError::InternalError(format!("encode method: {e}")))?;
17+
Ok(format!(
18+
r#"
19+
const cb = arguments[arguments.length - 1];
20+
const opts = {{ method: {method_json}, headers: {headers_json}, credentials: 'include' }};
21+
const body = {body_json};
22+
if (body !== '' && {method_json} !== 'GET') opts.body = body;
23+
fetch({url_json}, opts)
24+
.then(async (r) => {{
25+
const text = await r.text();
26+
const hdrs = {{}};
27+
r.headers.forEach((v, k) => {{ hdrs[k] = v; }});
28+
cb({{ status: r.status, headers: hdrs, body: text }});
29+
}})
30+
.catch((e) => cb({{ status: 0, headers: {{}}, body: String(e) }}));
31+
"#
32+
))
33+
}

‎px-camoufox/src/infrastructure/fetch_strategies.rs‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
//! The two execution strategies for `/v1/fetch` against a persistent
2+
//! Camoufox session.
3+
//!
4+
//! * `navigate_and_read` (used for GET): drives the webdriver to the
5+
//! target URL so the browser handles Cloudflare's interactive
6+
//! challenge, then reads the response body via
7+
//! `document.body.innerText`. Status is sniffed from the body shape
8+
//! because webdriver doesn't surface response codes.
9+
//! * `in_page_fetch` (used for POST/PUT/etc.): runs an in-page
10+
//! `fetch()` since webdriver navigation can't carry a request body.
11+
//! Any CF interactive challenge surfaced here arrives as HTML and
12+
//! is reported back as a 403.
13+
14+
use crate::infrastructure::fetch_script;
15+
use px_errors::AppError;
16+
use px_pipeline::FetchRequest;
17+
use serde_json::Value;
18+
use std::collections::HashMap;
19+
use std::time::Duration;
20+
use tokio::time::sleep;
21+
22+
pub(crate) type FetchTriple = (u16, HashMap<String, String>, String);
23+
24+
pub(crate) async fn navigate_and_read(
25+
client: &fantoccini::Client,
26+
req: &FetchRequest,
27+
navigate_timeout: Duration,
28+
) -> Result<FetchTriple, AppError> {
29+
navigate_with_wait(
30+
client,
31+
&req.url,
32+
navigate_timeout,
33+
Duration::from_millis(3_500),
34+
)
35+
.await?;
36+
let body_val = client
37+
.execute("return document.body.innerText;", vec![])
38+
.await
39+
.map_err(|e| AppError::InternalError(format!("read body: {e}")))?;
40+
let body = body_val.as_str().unwrap_or("").to_string();
41+
let trimmed = body.trim_start();
42+
let status = if trimmed.starts_with('{') || trimmed.starts_with('[') {
43+
200
44+
} else if looks_like_challenge(&body) {
45+
403
46+
} else {
47+
500
48+
};
49+
Ok((status, HashMap::new(), body))
50+
}
51+
52+
pub(crate) async fn in_page_fetch(
53+
client: &fantoccini::Client,
54+
req: &FetchRequest,
55+
request_timeout: Duration,
56+
) -> Result<FetchTriple, AppError> {
57+
let script = fetch_script::build(req)?;
58+
let exec = client.execute_async(&script, vec![]);
59+
let raw = match tokio::time::timeout(request_timeout, exec).await {
60+
Ok(Ok(v)) => v,
61+
Ok(Err(e)) => return Err(AppError::InternalError(format!("fetch eval: {e}"))),
62+
Err(_) => return Err(AppError::InternalError("fetch timeout".into())),
63+
};
64+
let status = raw
65+
.get("status")
66+
.and_then(Value::as_u64)
67+
.ok_or_else(|| AppError::InternalError("fetch result missing status".into()))?
68+
as u16;
69+
let body = raw
70+
.get("body")
71+
.and_then(Value::as_str)
72+
.unwrap_or("")
73+
.to_string();
74+
let headers: HashMap<String, String> = raw
75+
.get("headers")
76+
.and_then(Value::as_object)
77+
.map(|m| {
78+
m.iter()
79+
.filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
80+
.collect()
81+
})
82+
.unwrap_or_default();
83+
Ok((status, headers, body))
84+
}
85+
86+
fn looks_like_challenge(body: &str) -> bool {
87+
body.contains("__cf_chl")
88+
|| body.contains("Just a moment")
89+
|| body.contains("\"appId\":\"PXeT15wiaE\"")
90+
|| body.contains("tráfico inusual")
91+
|| body.contains("trafico inusual")
92+
}
93+
94+
pub(crate) async fn navigate_with_wait(
95+
client: &fantoccini::Client,
96+
url: &str,
97+
navigate_timeout: Duration,
98+
settle: Duration,
99+
) -> Result<(), AppError> {
100+
let nav = client.goto(url);
101+
if tokio::time::timeout(navigate_timeout, nav).await.is_err() {
102+
return Err(AppError::InternalError(format!("navigate timeout: {url}")));
103+
}
104+
sleep(settle).await;
105+
Ok(())
106+
}

‎px-camoufox/src/infrastructure/mod.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,7 @@
11
pub mod camoufox_fetcher;
22
pub mod camoufox_pool;
33
pub mod caps;
4+
pub mod fetch_script;
5+
pub mod fetch_strategies;
6+
pub mod session;
7+
pub mod session_pool;

0 commit comments

Comments
 (0)