From 2dd75a3d3b1611c0f590bec6bee8882979c521ba Mon Sep 17 00:00:00 2001 From: zhom <2717306+zhom@users.noreply.github.com> Date: Sat, 10 Oct 2026 13:10:46 +0400 Subject: [PATCH] refactor: proxy lifecycle management --- e2e/tests/browser.test.mjs | 81 ++++++++- src-tauri/src/proxy_manager.rs | 68 +++++-- src-tauri/src/proxy_runner.rs | 76 +++++++- src-tauri/src/proxy_server.rs | 80 ++++++-- src-tauri/tests/donut_proxy_integration.rs | 201 +++++++++++++++++++++ 5 files changed, 465 insertions(+), 41 deletions(-) diff --git a/e2e/tests/browser.test.mjs b/e2e/tests/browser.test.mjs index 6bad0959..ef437dba 100644 --- a/e2e/tests/browser.test.mjs +++ b/e2e/tests/browser.test.mjs @@ -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), diff --git a/src-tauri/src/proxy_manager.rs b/src-tauri/src/proxy_manager.rs index f95ca5e4..0183971b 100644 --- a/src-tauri/src/proxy_manager.rs +++ b/src-tauri/src/proxy_manager.rs @@ -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>, + // 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 { + 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) = { 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, 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 = 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; diff --git a/src-tauri/src/proxy_runner.rs b/src-tauri/src/proxy_runner.rs index 69c38dda..cd56b539 100644 --- a/src-tauri/src/proxy_runner.rs +++ b/src-tauri/src/proxy_runner.rs @@ -289,10 +289,8 @@ pub async fn start_proxy_process_with_profile( ) -> Result> { 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, Box> { + 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> { + 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> { stop_proxy_process_because(id, "requested").await } @@ -513,6 +559,14 @@ pub async fn stop_proxy_process(id: &str) -> Result Result> { + stop_proxy_process_inner(id, reason, true).await +} + +async fn stop_proxy_process_inner( + id: &str, + reason: &str, + remove_config: bool, ) -> Result> { 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" diff --git a/src-tauri/src/proxy_server.rs b/src-tauri/src/proxy_server.rs index 17da2806..aef9f3ca 100644 --- a/src-tauri/src/proxy_server.rs +++ b/src-tauri/src/proxy_server.rs @@ -1612,6 +1612,7 @@ pub fn redacted_upstream(upstream: &str) -> String { pub async fn run_proxy_server(config: ProxyConfig) -> Result<(), Box> { 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 Result<(), Box { - // 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 { diff --git a/src-tauri/tests/donut_proxy_integration.rs b/src-tauri/tests/donut_proxy_integration.rs index 39ec37b8..cfc16443 100644 --- a/src-tauri/tests/donut_proxy_integration.rs +++ b/src-tauri/tests/donut_proxy_integration.rs @@ -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> { + 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> { + 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();