mirror of
https://github.com/zhom/donutbrowser.git
synced 2026-10-19 14:07:19 +02:00
refactor: proxy lifecycle management
This commit is contained in:
1 parent
62e37c884d
commit
2dd75a3d3b
5 files changed
+465
-41
No files matched your search
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdir, readdir, readFile, stat, writeFile } from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import test from "node:test";
|
||||
@@ -1038,6 +1039,49 @@ async function waitForCondition(check, description, timeoutMs = 30_000) {
|
||||
throw new Error(`Timed out after ${timeoutMs}ms waiting for ${description}`);
|
||||
}
|
||||
|
||||
async function assertWorkerRoute(port) {
|
||||
const target = new URL(fixtureUrl);
|
||||
const socket = net.createConnection({ host: "127.0.0.1", port });
|
||||
socket.setTimeout(5_000, () =>
|
||||
socket.destroy(new Error("Proxy request timed out")),
|
||||
);
|
||||
const chunks = socket[Symbol.asyncIterator]();
|
||||
let buffered = Buffer.alloc(0);
|
||||
const read = async (length) => {
|
||||
while (buffered.length < length) {
|
||||
const { value, done } = await chunks.next();
|
||||
assert.equal(done, false, "Proxy closed before its reply");
|
||||
buffered = Buffer.concat([buffered, value]);
|
||||
}
|
||||
const value = buffered.subarray(0, length);
|
||||
buffered = buffered.subarray(length);
|
||||
return value;
|
||||
};
|
||||
try {
|
||||
socket.write(Buffer.from([5, 1, 0]));
|
||||
assert.deepEqual(await read(2), Buffer.from([5, 0]));
|
||||
const host = Buffer.from(target.hostname);
|
||||
const targetPort = Number(target.port || 80);
|
||||
socket.write(
|
||||
Buffer.concat([
|
||||
Buffer.from([5, 1, 0, 3, host.length]),
|
||||
host,
|
||||
Buffer.from([targetPort >> 8, targetPort & 255]),
|
||||
]),
|
||||
);
|
||||
assert.deepEqual(await read(4), Buffer.from([5, 0, 0, 1]));
|
||||
await read(6);
|
||||
socket.write(
|
||||
`GET /worker-recovery HTTP/1.1\r\nHost: ${target.host}\r\nConnection: close\r\n\r\n`,
|
||||
);
|
||||
const response = [buffered];
|
||||
for await (const chunk of chunks) response.push(chunk);
|
||||
assert.match(Buffer.concat(response).toString(), /^HTTP\/1\.1 200 /);
|
||||
} finally {
|
||||
socket.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
/// Launch a real profile and return it together with the worker serving it,
|
||||
/// asserting the worker is pinned to that exact browser process.
|
||||
async function launchWithWorker(app, version, name) {
|
||||
@@ -1077,7 +1121,7 @@ async function launchWithWorker(app, version, name) {
|
||||
// recorded a verified owner identity it just kept running and users killed it
|
||||
// by hand. Phase one covers closing the browser; phase two covers the reported
|
||||
// order (app closed first, so nothing is left to reap anything).
|
||||
test("a proxy worker dies with its browser, with and without the app running", async () => {
|
||||
test("a proxy worker recovers for a live browser and dies with its browser", async () => {
|
||||
assert.ok(process.env.WAYFERN_TEST_TOKEN, "WAYFERN_TEST_TOKEN is required");
|
||||
const localWayfernVersion = cachedFixtureVersion(
|
||||
process.env.DONUT_E2E_PROJECT_ROOT,
|
||||
@@ -1106,6 +1150,41 @@ test("a proxy worker dies with its browser, with and without the app running", a
|
||||
// Phase one: the app is up, the browser goes away.
|
||||
const first = await launchWithWorker(app, prepared.version, "Worker Reap");
|
||||
strays.add(first.browserPid).add(first.workerPid);
|
||||
let worker = (await readWorkerConfigs(app)).find(
|
||||
(config) => config.id === first.workerId,
|
||||
);
|
||||
await assertWorkerRoute(worker.local_port);
|
||||
for (const afterRestart of [false, true]) {
|
||||
if (afterRestart) await app.restart();
|
||||
process.kill(worker.pid, "SIGKILL");
|
||||
await waitForCondition(
|
||||
() => !processExists(worker.pid),
|
||||
"failed worker to exit",
|
||||
);
|
||||
const recovered = await app.waitFor(
|
||||
async () => {
|
||||
const current = (await readWorkerConfigs(app)).find(
|
||||
(config) => config.id === worker.id,
|
||||
);
|
||||
return current?.pid !== worker.pid && processExists(current?.pid)
|
||||
? current
|
||||
: false;
|
||||
},
|
||||
{
|
||||
timeoutMs: 45_000,
|
||||
description: "automatic proxy recovery on the original port",
|
||||
},
|
||||
);
|
||||
strays.add(recovered.pid);
|
||||
assert.deepEqual(recovered, { ...worker, pid: recovered.pid });
|
||||
assert.ok(
|
||||
processExists(first.browserPid),
|
||||
"recovery must keep the browser running",
|
||||
);
|
||||
await assertWorkerRoute(worker.local_port);
|
||||
worker = recovered;
|
||||
}
|
||||
first.workerPid = worker.pid;
|
||||
process.kill(first.browserPid, "SIGKILL");
|
||||
await waitForCondition(
|
||||
() => !processExists(first.browserPid),
|
||||
|
||||
@@ -13,6 +13,9 @@ use crate::browser::ProxySettings;
|
||||
use crate::events;
|
||||
use crate::ip_utils;
|
||||
|
||||
static PROXY_RECOVERY: crate::log_streak::KeyedStreak =
|
||||
crate::log_streak::KeyedStreak::new(module_path!(), "Proxy worker recovery");
|
||||
|
||||
// Export data format for JSON export
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ProxyExportData {
|
||||
@@ -358,6 +361,8 @@ pub struct ProxyManager {
|
||||
// We only reap a worker after it has been missed in N consecutive scans —
|
||||
// a single sysinfo blip under load shouldn't kill a still-running worker.
|
||||
dead_browser_misses: Mutex<HashMap<u32, u8>>,
|
||||
// A requested stop must not race a recovery that starts the worker again.
|
||||
worker_lifecycle: tokio::sync::Mutex<()>,
|
||||
}
|
||||
|
||||
impl ProxyManager {
|
||||
@@ -368,6 +373,7 @@ impl ProxyManager {
|
||||
profile_active_proxy_ids: Mutex::new(HashMap::new()),
|
||||
stored_proxies: Mutex::new(HashMap::new()),
|
||||
dead_browser_misses: Mutex::new(HashMap::new()),
|
||||
worker_lifecycle: tokio::sync::Mutex::new(()),
|
||||
};
|
||||
|
||||
// Load stored proxies on initialization
|
||||
@@ -1952,6 +1958,7 @@ impl ProxyManager {
|
||||
// the returned ProxySettings.proxy_type so the caller formats the right local proxy URL scheme.
|
||||
local_protocol: &str,
|
||||
) -> Result<ProxySettings, String> {
|
||||
let _lifecycle = self.worker_lifecycle.lock().await;
|
||||
let started = std::time::Instant::now();
|
||||
if let Some(name) = profile_id {
|
||||
// Check if we have an active proxy recorded for this profile
|
||||
@@ -2224,6 +2231,15 @@ impl ProxyManager {
|
||||
&self,
|
||||
app_handle: tauri::AppHandle,
|
||||
browser_pid: u32,
|
||||
) -> Result<(), String> {
|
||||
let _lifecycle = self.worker_lifecycle.lock().await;
|
||||
self.stop_proxy_inner(app_handle, browser_pid).await
|
||||
}
|
||||
|
||||
async fn stop_proxy_inner(
|
||||
&self,
|
||||
app_handle: tauri::AppHandle,
|
||||
browser_pid: u32,
|
||||
) -> Result<(), String> {
|
||||
let (proxy_id, profile_id): (String, Option<String>) = {
|
||||
let mut proxies = self.active_proxies.lock().unwrap();
|
||||
@@ -2283,11 +2299,8 @@ impl ProxyManager {
|
||||
app_handle: tauri::AppHandle,
|
||||
profile_id: &str,
|
||||
) -> Result<(), String> {
|
||||
// Find the proxy ID for this profile
|
||||
let proxy_id = {
|
||||
let map = self.profile_active_proxy_ids.lock().unwrap();
|
||||
map.get(profile_id).cloned()
|
||||
};
|
||||
let _lifecycle = self.worker_lifecycle.lock().await;
|
||||
let proxy_id = self.resolve_proxy_id_for_profile(profile_id);
|
||||
|
||||
if let Some(proxy_id) = proxy_id {
|
||||
// Find the PID for this proxy
|
||||
@@ -2304,7 +2317,7 @@ impl ProxyManager {
|
||||
|
||||
if let Some(pid) = pid {
|
||||
// Use the existing stop_proxy method
|
||||
self.stop_proxy(app_handle, pid).await
|
||||
self.stop_proxy_inner(app_handle, pid).await
|
||||
} else {
|
||||
// Proxy not found in active_proxies, try to stop it directly by ID
|
||||
let worker = WorkerSnapshot::take(&proxy_id);
|
||||
@@ -2420,21 +2433,42 @@ impl ProxyManager {
|
||||
.map(|config| config.id)
|
||||
}
|
||||
|
||||
// Clean up proxies for dead browser processes
|
||||
// Only clean up orphaned config files where the proxy process itself is dead
|
||||
// Restore routes for live browsers before removing dead worker records.
|
||||
pub async fn cleanup_dead_proxies(
|
||||
&self,
|
||||
_app_handle: tauri::AppHandle,
|
||||
) -> Result<Vec<u32>, String> {
|
||||
// Don't stop proxies for dead browser processes - let them run indefinitely
|
||||
// The proxy processes are idle and don't consume CPU when not in use
|
||||
// Only clean up config files where the proxy process itself is dead (see below)
|
||||
let _lifecycle = self.worker_lifecycle.lock().await;
|
||||
let dead_pids: Vec<u32> = Vec::new();
|
||||
|
||||
// Clean up orphaned proxy configs (only where proxy process is definitely dead)
|
||||
// IMPORTANT: Only clean up configs where the proxy process itself is dead
|
||||
// If the proxy process is running (even if idle), leave it alone
|
||||
// The user doesn't care if proxy processes run indefinitely as long as they're not consuming CPU
|
||||
// Read disk as well as the in-memory map. A browser and its route can
|
||||
// outlive the GUI, so a fresh GUI has no active_proxies entry for them.
|
||||
let mut recovering_ids = std::collections::HashSet::new();
|
||||
for config in crate::proxy_storage::list_proxy_configs() {
|
||||
if !crate::proxy_runner::proxy_needs_recovery(&config) {
|
||||
continue;
|
||||
}
|
||||
recovering_ids.insert(config.id.clone());
|
||||
let key = format!(
|
||||
"profile={} proxy_worker={}",
|
||||
config.profile_id.as_deref().unwrap_or("none"),
|
||||
config.id
|
||||
);
|
||||
match crate::proxy_runner::recover_proxy_process(&config.id).await {
|
||||
Ok(Some(recovered)) => {
|
||||
PROXY_RECOVERY.succeeded(&key);
|
||||
log::info!(
|
||||
"Proxy worker recovered {key} old_pid={} pid={} port={}",
|
||||
config.pid.unwrap_or_default(),
|
||||
recovered.pid.unwrap_or_default(),
|
||||
recovered.local_port.unwrap_or_default()
|
||||
);
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(error) => PROXY_RECOVERY.failed(&key, error),
|
||||
}
|
||||
}
|
||||
|
||||
let orphaned_configs = {
|
||||
use crate::proxy_storage::{is_process_running, list_proxy_configs};
|
||||
|
||||
@@ -2447,6 +2481,10 @@ impl ProxyManager {
|
||||
all_configs
|
||||
.into_iter()
|
||||
.filter(|config| {
|
||||
// Keep a failed recovery's route so the next pass can retry it.
|
||||
if recovering_ids.contains(&config.id) {
|
||||
return false;
|
||||
}
|
||||
// If proxy is tracked in active_proxies, it's definitely not orphaned
|
||||
if tracked_proxy_ids.contains(&config.id) {
|
||||
return false;
|
||||
|
||||
@@ -289,10 +289,8 @@ pub async fn start_proxy_process_with_profile(
|
||||
) -> Result<ProxyConfig, Box<dyn std::error::Error>> {
|
||||
ensure_sidecar_version().await?;
|
||||
|
||||
let started = std::time::Instant::now();
|
||||
let id = generate_proxy_id();
|
||||
let upstream = upstream_url.unwrap_or_else(|| "DIRECT".to_string());
|
||||
let upstream_kind = upstream_scheme(&upstream).to_string();
|
||||
|
||||
// Get available port if not specified
|
||||
let local_port = port.unwrap_or_else(|| {
|
||||
@@ -310,12 +308,51 @@ pub async fn start_proxy_process_with_profile(
|
||||
.with_record_domains(record_domains);
|
||||
save_proxy_config(&config)?;
|
||||
|
||||
spawn_proxy_worker(&config, false).await
|
||||
}
|
||||
|
||||
/// Restart only a dead worker whose browser is still alive. Keep its full
|
||||
/// route and listening port: the running browser cannot use a new port.
|
||||
pub async fn recover_proxy_process(
|
||||
id: &str,
|
||||
) -> Result<Option<ProxyConfig>, Box<dyn std::error::Error>> {
|
||||
let Some(config) = get_proxy_config(id) else {
|
||||
return Ok(None);
|
||||
};
|
||||
if !proxy_needs_recovery(&config) {
|
||||
return Ok(None);
|
||||
}
|
||||
ensure_sidecar_version().await?;
|
||||
spawn_proxy_worker(&config, true).await.map(Some)
|
||||
}
|
||||
|
||||
pub(crate) fn proxy_needs_recovery(config: &ProxyConfig) -> bool {
|
||||
config.profile_id.is_some()
|
||||
&& config.local_port.is_some_and(|port| port != 0)
|
||||
&& config.pid.is_some_and(|pid| !is_process_running(pid))
|
||||
&& crate::proxy_storage::browser_owner_is_alive(config)
|
||||
}
|
||||
|
||||
async fn spawn_proxy_worker(
|
||||
config: &ProxyConfig,
|
||||
recovering: bool,
|
||||
) -> Result<ProxyConfig, Box<dyn std::error::Error>> {
|
||||
let started = std::time::Instant::now();
|
||||
let id = config.id.clone();
|
||||
let profile_id = config.profile_id.as_deref().unwrap_or("none");
|
||||
let upstream_kind = upstream_scheme(&config.upstream_url);
|
||||
|
||||
// Spawn proxy worker process in the background using std::process::Command
|
||||
// This ensures proper process detachment on Unix systems
|
||||
let exe = find_sidecar_executable("donut-proxy")?;
|
||||
let temp_dir = std::env::temp_dir();
|
||||
let log_path = temp_dir.join(format!("donut-proxy-{id}.log"));
|
||||
let log_file = crate::app_dirs::create_owner_only(&log_path);
|
||||
let log_file = if recovering && log_path.exists() {
|
||||
crate::app_dirs::restrict_to_owner(&log_path);
|
||||
std::fs::OpenOptions::new().append(true).open(&log_path)
|
||||
} else {
|
||||
crate::app_dirs::create_owner_only(&log_path)
|
||||
};
|
||||
prune_stale_proxy_logs(&temp_dir, RETAINED_PROXY_LOGS);
|
||||
let spawned_pid: u32;
|
||||
|
||||
@@ -369,8 +406,7 @@ pub async fn start_proxy_process_with_profile(
|
||||
processes.insert(id.clone(), pid);
|
||||
}
|
||||
|
||||
// Don't wait for the child - it's detached
|
||||
drop(child);
|
||||
watch_proxy_exit(child, id.clone());
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
@@ -438,7 +474,7 @@ pub async fn start_proxy_process_with_profile(
|
||||
processes.insert(id.clone(), pid);
|
||||
}
|
||||
|
||||
drop(child);
|
||||
watch_proxy_exit(child, id.clone());
|
||||
}
|
||||
|
||||
// Give the process a moment to start up before checking
|
||||
@@ -454,7 +490,7 @@ pub async fn start_proxy_process_with_profile(
|
||||
// Use shorter sleep for faster startup
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||
|
||||
if let Some(updated_config) = get_proxy_config(&id) {
|
||||
if let Some(updated_config) = get_proxy_config(&id).filter(|c| c.pid == Some(spawned_pid)) {
|
||||
// Check if local_url is set (worker has bound and updated config)
|
||||
if let Some(ref local_url) = updated_config.local_url {
|
||||
if !local_url.is_empty() {
|
||||
@@ -469,7 +505,7 @@ pub async fn start_proxy_process_with_profile(
|
||||
Ok(Ok(_stream)) => {
|
||||
log::info!(
|
||||
"Proxy worker started id={id} pid={spawned_pid} port={port} profile={} upstream={upstream_kind} elapsed_ms={}",
|
||||
profile_id.as_deref().unwrap_or("none"),
|
||||
profile_id,
|
||||
started.elapsed().as_millis()
|
||||
);
|
||||
return Ok(updated_config);
|
||||
@@ -499,12 +535,22 @@ pub async fn start_proxy_process_with_profile(
|
||||
// The detached worker (if it did spawn) would otherwise outlive this
|
||||
// failed start with nothing tracking it — callers only get an error
|
||||
// string, so this is the last place that can still kill it.
|
||||
let _ = stop_proxy_process_because(&id, "start_timeout").await;
|
||||
let _ = stop_proxy_process_inner(&id, "start_timeout", !recovering).await;
|
||||
return Err(format!("Proxy worker failed to start in time. {detail}").into());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn watch_proxy_exit(mut child: std::process::Child, id: String) {
|
||||
let pid = child.id();
|
||||
// A recovered worker is a child of the long-lived GUI. Reap it off the
|
||||
// runtime so repeated failures do not leave zombies until the GUI exits.
|
||||
std::thread::spawn(move || match child.wait() {
|
||||
Ok(status) => log::info!("Proxy worker exited id={id} pid={pid} status=\"{status}\""),
|
||||
Err(error) => log::warn!("Proxy worker wait failed id={id} pid={pid} err=\"{error}\""),
|
||||
});
|
||||
}
|
||||
|
||||
pub async fn stop_proxy_process(id: &str) -> Result<bool, Box<dyn std::error::Error>> {
|
||||
stop_proxy_process_because(id, "requested").await
|
||||
}
|
||||
@@ -513,6 +559,14 @@ pub async fn stop_proxy_process(id: &str) -> Result<bool, Box<dyn std::error::Er
|
||||
pub(crate) async fn stop_proxy_process_because(
|
||||
id: &str,
|
||||
reason: &str,
|
||||
) -> Result<bool, Box<dyn std::error::Error>> {
|
||||
stop_proxy_process_inner(id, reason, true).await
|
||||
}
|
||||
|
||||
async fn stop_proxy_process_inner(
|
||||
id: &str,
|
||||
reason: &str,
|
||||
remove_config: bool,
|
||||
) -> Result<bool, Box<dyn std::error::Error>> {
|
||||
let config = get_proxy_config(id);
|
||||
let pid = PROXY_PROCESSES
|
||||
@@ -554,7 +608,9 @@ pub(crate) async fn stop_proxy_process_because(
|
||||
}
|
||||
|
||||
// Delete the config file
|
||||
delete_proxy_config(id);
|
||||
if remove_config {
|
||||
delete_proxy_config(id);
|
||||
}
|
||||
|
||||
let how = if !was_running {
|
||||
"already_gone"
|
||||
|
||||
@@ -1612,6 +1612,7 @@ pub fn redacted_upstream(upstream: &str) -> String {
|
||||
|
||||
pub async fn run_proxy_server(config: ProxyConfig) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let started = std::time::Instant::now();
|
||||
let connection_limit = worker_connection_limit();
|
||||
// Load the config from disk to get the latest state
|
||||
let config = match crate::proxy_storage::get_proxy_config(&config.id) {
|
||||
Some(c) => c,
|
||||
@@ -1865,7 +1866,7 @@ pub async fn run_proxy_server(config: ProxyConfig) -> Result<(), Box<dyn std::er
|
||||
};
|
||||
|
||||
log::info!(
|
||||
"Worker started kind=proxy id={} pid={} listen=127.0.0.1:{actual_port} serve={local_protocol} upstream={} profile={} blocklist={} bypass_rules={} record_domains={} elapsed_ms={}",
|
||||
"Worker started kind=proxy id={} pid={} listen=127.0.0.1:{actual_port} serve={local_protocol} upstream={} profile={} blocklist={} bypass_rules={} record_domains={} connections={connection_limit} elapsed_ms={}",
|
||||
config.id,
|
||||
std::process::id(),
|
||||
upstream_url
|
||||
@@ -1885,26 +1886,19 @@ pub async fn run_proxy_server(config: ProxyConfig) -> Result<(), Box<dyn std::er
|
||||
started.elapsed().as_millis()
|
||||
);
|
||||
|
||||
// Bound concurrent connection handlers. A client retry-storm (e.g. a browser
|
||||
// hammering CONNECT requests while DNS is failing) must not spawn unbounded
|
||||
// tasks,
|
||||
// each of which parks a Tokio blocking thread inside getaddrinfo — that is
|
||||
// what exhausted the resolver pool and pegged the CPU on long-lived workers.
|
||||
// A real browser never approaches this ceiling; waiting for a permit
|
||||
// backpressures a storm instead of amplifying it.
|
||||
let conn_semaphore = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_CONNECTIONS));
|
||||
let conn_semaphore = Arc::new(tokio::sync::Semaphore::new(connection_limit));
|
||||
|
||||
// Keep the runtime alive with an infinite loop
|
||||
// This ensures the process doesn't exit even if there are no active connections
|
||||
loop {
|
||||
// Reserve the full connection budget before opening another descriptor.
|
||||
let permit = conn_semaphore
|
||||
.clone()
|
||||
.acquire_owned()
|
||||
.await
|
||||
.expect("connection semaphore is never closed");
|
||||
match listener.accept().await {
|
||||
Ok((stream, _peer_addr)) => {
|
||||
// The semaphore is never closed, so acquire cannot fail.
|
||||
let permit = conn_semaphore
|
||||
.clone()
|
||||
.acquire_owned()
|
||||
.await
|
||||
.expect("connection semaphore is never closed");
|
||||
let upstream = upstream_url.clone();
|
||||
let matcher = bypass_matcher.clone();
|
||||
let blocker = blocklist_matcher.clone();
|
||||
@@ -2036,6 +2030,46 @@ fn watchdog_poll_interval() -> std::time::Duration {
|
||||
/// Tokio blocking thread inside getaddrinfo).
|
||||
const MAX_CONCURRENT_CONNECTIONS: usize = 512;
|
||||
|
||||
#[cfg(unix)]
|
||||
fn connections_for_descriptor_limit(limit: u64) -> usize {
|
||||
// A SOCKS5 UDP relay uses four sockets, plus temporary resolver sockets.
|
||||
// Keep room for the runtime, logs, traffic files, and owner checks.
|
||||
const RESERVED: u64 = 64;
|
||||
const PER_CONNECTION: u64 = 6;
|
||||
((limit.saturating_sub(RESERVED) / PER_CONNECTION).clamp(1, MAX_CONCURRENT_CONNECTIONS as u64))
|
||||
as usize
|
||||
}
|
||||
|
||||
fn worker_connection_limit() -> usize {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let mut limit = libc::rlimit {
|
||||
rlim_cur: 0,
|
||||
rlim_max: 0,
|
||||
};
|
||||
if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) } != 0 {
|
||||
return connections_for_descriptor_limit(256);
|
||||
}
|
||||
// macOS GUI children normally inherit a soft limit of 256. Request room
|
||||
// for the connection cap without changing the hard limit or the host.
|
||||
let requested = limit.rlim_max.min(4096);
|
||||
if limit.rlim_cur < requested {
|
||||
let raised = libc::rlimit {
|
||||
rlim_cur: requested,
|
||||
rlim_max: limit.rlim_max,
|
||||
};
|
||||
if unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &raised) } == 0 {
|
||||
limit = raised;
|
||||
}
|
||||
}
|
||||
connections_for_descriptor_limit(limit.rlim_cur)
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
MAX_CONCURRENT_CONNECTIONS
|
||||
}
|
||||
}
|
||||
|
||||
/// Connect timeout for the direct (no-upstream) dial path. Bounds a wedged
|
||||
/// `getaddrinfo` so a broken resolver can't park a blocking thread for the
|
||||
/// full OS timeout.
|
||||
@@ -2548,6 +2582,22 @@ mod tests {
|
||||
use super::*;
|
||||
use std::io::Write;
|
||||
|
||||
#[test]
|
||||
#[cfg(unix)]
|
||||
fn connection_budget_leaves_room_for_dns_and_worker_files() {
|
||||
assert_eq!(connections_for_descriptor_limit(128), 10);
|
||||
assert_eq!(connections_for_descriptor_limit(256), 32);
|
||||
assert_eq!(
|
||||
connections_for_descriptor_limit(4096),
|
||||
MAX_CONCURRENT_CONNECTIONS
|
||||
);
|
||||
assert_eq!(
|
||||
connections_for_descriptor_limit(u64::MAX),
|
||||
MAX_CONCURRENT_CONNECTIONS
|
||||
);
|
||||
assert_eq!(connections_for_descriptor_limit(32), 1);
|
||||
}
|
||||
|
||||
/// Build an upstream URL with `urlencoding::encode`-d user/pass,
|
||||
/// mirroring what `proxy_manager::build_proxy_url` actually emits
|
||||
fn parse_encoded_upstream(scheme: &str, user: &str, pass: &str) -> Url {
|
||||
|
||||
@@ -109,6 +109,207 @@ struct ProxyWorkerProcess {
|
||||
config_id: String,
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
#[cfg(unix)]
|
||||
async fn test_worker_connection_pressure_respects_its_descriptor_limit(
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
use donutbrowser_lib::proxy_storage::{
|
||||
generate_proxy_id, get_proxy_config, save_proxy_config, ProxyConfig,
|
||||
};
|
||||
use std::os::unix::process::CommandExt;
|
||||
|
||||
let isolated_root = tempfile::tempdir()?;
|
||||
let _data_root = EnvironmentRestore::set_path("DONUTBROWSER_DATA_ROOT", isolated_root.path());
|
||||
let binary_path = setup_test().await?;
|
||||
let id = generate_proxy_id();
|
||||
let config = ProxyConfig::new(id.clone(), "DIRECT".to_string(), Some(0))
|
||||
.with_local_protocol(Some("socks5".to_string()));
|
||||
save_proxy_config(&config).map_err(|error| error.to_string())?;
|
||||
let log_path = isolated_root.path().join("worker.log");
|
||||
let mut command = std::process::Command::new(binary_path);
|
||||
command
|
||||
.args(["proxy-worker", "start", "--id", &id])
|
||||
.stdout(std::process::Stdio::null())
|
||||
.stderr(std::fs::File::create(&log_path)?);
|
||||
unsafe {
|
||||
command.pre_exec(|| {
|
||||
let limit = libc::rlimit {
|
||||
rlim_cur: 128,
|
||||
rlim_max: 128,
|
||||
};
|
||||
if libc::setrlimit(libc::RLIMIT_NOFILE, &limit) != 0 {
|
||||
return Err(std::io::Error::last_os_error());
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
let _worker = ProxyWorkerProcess {
|
||||
child: command.spawn()?,
|
||||
config_id: id.clone(),
|
||||
};
|
||||
let ready = tokio::time::timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
if let Some(ready) = get_proxy_config(&id).filter(|config| config.local_url.is_some()) {
|
||||
break ready;
|
||||
}
|
||||
sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
let port = ready.local_port.unwrap();
|
||||
let mut clients = Vec::new();
|
||||
// Fill the descriptor limit without exceeding macOS's 128-entry listen
|
||||
// backlog: queued connections must wait outside the worker's descriptor set.
|
||||
for index in 0..128 {
|
||||
let mut stream = tokio::time::timeout(
|
||||
Duration::from_secs(5),
|
||||
TcpStream::connect(("127.0.0.1", port)),
|
||||
)
|
||||
.await
|
||||
.map_err(|error| format!("queued connection {index}: {error}"))??;
|
||||
stream.write_all(&[5, 1, 0]).await?;
|
||||
clients.push(stream);
|
||||
}
|
||||
let mut greeting = [0; 2];
|
||||
tokio::time::timeout(Duration::from_secs(5), clients[0].read_exact(&mut greeting)).await??;
|
||||
assert_eq!(greeting, [5, 0]);
|
||||
sleep(Duration::from_millis(500)).await;
|
||||
drop(clients);
|
||||
let log = std::fs::read_to_string(log_path)?;
|
||||
assert!(
|
||||
!log.contains("Too many open files"),
|
||||
"worker exhausted its descriptors: {log}"
|
||||
);
|
||||
assert!(
|
||||
log.contains("connections=10"),
|
||||
"worker must use its actual hard limit"
|
||||
);
|
||||
|
||||
let (target_port, target) = start_mock_http_server("AFTER-CONNECTION-PRESSURE").await;
|
||||
let response = reqwest::Client::builder()
|
||||
.proxy(reqwest::Proxy::all(ready.local_url.as_ref().unwrap())?)
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()?
|
||||
.get(format!("http://127.0.0.1:{target_port}/"))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(response.text().await?, "AFTER-CONNECTION-PRESSURE");
|
||||
target.abort();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_dead_worker_recovers_its_route_and_retries_an_occupied_port(
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
use donutbrowser_lib::proxy_runner::recover_proxy_process;
|
||||
use donutbrowser_lib::proxy_storage::{
|
||||
generate_proxy_id, get_proxy_config, process_start_time, save_proxy_config, ProxyConfig,
|
||||
};
|
||||
|
||||
let isolated_root = tempfile::tempdir()?;
|
||||
let _data_root = EnvironmentRestore::set_path("DONUTBROWSER_DATA_ROOT", isolated_root.path());
|
||||
let binary_path = setup_test().await?;
|
||||
let mut tracker = ProxyTestTracker::new(binary_path.clone());
|
||||
let (upstream_port, upstream) = start_mock_http_server("RECOVERED-UPSTREAM").await;
|
||||
let blocklist = isolated_root.path().join("allowlist.txt");
|
||||
std::fs::write(&blocklist, "recovery.invalid\n")?;
|
||||
let id = generate_proxy_id();
|
||||
let mut config = ProxyConfig::new(
|
||||
id.clone(),
|
||||
format!("http://user:password@127.0.0.1:{upstream_port}"),
|
||||
Some(0),
|
||||
)
|
||||
.with_profile_id(Some("recovery-test".to_string()))
|
||||
.with_bypass_rules(vec!["bypass.invalid".to_string()])
|
||||
.with_blocklist_file(Some(blocklist.to_string_lossy().to_string()))
|
||||
.with_dns_allowlist_mode(true)
|
||||
.with_local_protocol(Some("http".to_string()))
|
||||
.with_record_domains(false);
|
||||
config.browser_pid = Some(std::process::id());
|
||||
config.browser_pid_start_time = process_start_time(std::process::id());
|
||||
config.ignore_proxy_certificate = Some(true);
|
||||
save_proxy_config(&config).map_err(|error| error.to_string())?;
|
||||
tracker.track_proxy(id.clone());
|
||||
let mut worker = ProxyWorkerProcess {
|
||||
child: std::process::Command::new(&binary_path)
|
||||
.args(["proxy-worker", "start", "--id", &id])
|
||||
.stdout(std::process::Stdio::null())
|
||||
.stderr(std::process::Stdio::null())
|
||||
.spawn()?,
|
||||
config_id: id.clone(),
|
||||
};
|
||||
let original = tokio::time::timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
if let Some(ready) = get_proxy_config(&id).filter(|config| config.local_url.is_some()) {
|
||||
break ready;
|
||||
}
|
||||
sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
assert!(recover_proxy_process(&id)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?
|
||||
.is_none());
|
||||
worker.child.kill()?;
|
||||
worker.child.wait()?;
|
||||
|
||||
let mut wrong_owner = original.clone();
|
||||
wrong_owner.browser_pid_start_time = Some(1);
|
||||
save_proxy_config(&wrong_owner).map_err(|error| error.to_string())?;
|
||||
assert!(recover_proxy_process(&id)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?
|
||||
.is_none());
|
||||
save_proxy_config(&original).map_err(|error| error.to_string())?;
|
||||
|
||||
let port = original.local_port.unwrap();
|
||||
let occupied = tokio::net::TcpListener::bind(("127.0.0.1", port)).await?;
|
||||
assert!(recover_proxy_process(&id).await.is_err());
|
||||
assert!(
|
||||
get_proxy_config(&id).is_some(),
|
||||
"a failed restart must keep its route"
|
||||
);
|
||||
drop(occupied);
|
||||
|
||||
let recovered = recover_proxy_process(&id)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?
|
||||
.expect("worker recovered");
|
||||
assert_ne!(recovered.pid, original.pid);
|
||||
let mut expected = original.clone();
|
||||
expected.pid = recovered.pid;
|
||||
assert_eq!(
|
||||
serde_json::to_value(&recovered)?,
|
||||
serde_json::to_value(expected)?
|
||||
);
|
||||
|
||||
let client = reqwest::Client::builder()
|
||||
.proxy(reqwest::Proxy::all(recovered.local_url.as_ref().unwrap())?)
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()?;
|
||||
let response = client.get("http://recovery.invalid/").send().await?;
|
||||
assert_eq!(response.text().await?, "RECOVERED-UPSTREAM");
|
||||
assert_eq!(
|
||||
client.get("http://blocked.invalid/").send().await?.status(),
|
||||
reqwest::StatusCode::FORBIDDEN
|
||||
);
|
||||
|
||||
let stopped = TestUtils::execute_command(&binary_path, &["proxy", "stop", "--id", &id]).await?;
|
||||
assert!(stopped.status.success());
|
||||
assert!(
|
||||
recover_proxy_process(&id)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?
|
||||
.is_none(),
|
||||
"a requested stop must stay stopped"
|
||||
);
|
||||
upstream.abort();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl Drop for ProxyWorkerProcess {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.child.kill();
|
||||
|
||||
Reference in new issue
Block a user