diff --git a/src-tauri/src/utils/server.rs b/src-tauri/src/utils/server.rs index 993896179..6f8f429bd 100644 --- a/src-tauri/src/utils/server.rs +++ b/src-tauri/src/utils/server.rs @@ -39,8 +39,6 @@ static INSTANCE_LOCK: OnceCell = OnceCell::new(); const PAC_INITIAL_AVAILABLE: bool = false; static PAC_AVAILABLE: AtomicBool = AtomicBool::new(PAC_INITIAL_AVAILABLE); static COMMANDS_READY: AtomicBool = AtomicBool::new(false); -#[cfg(feature = "verge-dev")] -static DEV_QUIT_REQUESTED: AtomicBool = AtomicBool::new(false); #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SingletonDisposition { @@ -144,55 +142,12 @@ fn instance_auth(token: String) -> impl warp::Filter( - token: String, - commands_ready: &'static AtomicBool, - quit_requested: &'static AtomicBool, - trigger: F, -) -> impl warp::Filter + Clone -where - F: Fn() + Clone + Send + Sync + 'static, -{ - instance_auth(token) - .and(warp::path!("commands" / "dev" / "quit")) - .and(warp::post()) - .and_then(move || { - let trigger = trigger.clone(); - async move { - if !commands_ready.load(Ordering::Acquire) { - return Ok::<_, warp::Rejection>(warp::reply::with_status( - String::from("starting"), - warp::http::StatusCode::SERVICE_UNAVAILABLE, - )); - } - if quit_requested - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_ok() - { - trigger(); - } - Ok(warp::reply::with_status( - String::from("accepted"), - warp::http::StatusCode::ACCEPTED, - )) - } - }) -} - -#[cfg(feature = "verge-dev")] -fn release_dev_quit_latch(quit_requested: &AtomicBool, outcome: clash_verge_signal::ShutdownOutcome) { - if matches!(outcome, clash_verge_signal::ShutdownOutcome::Canceled) { - quit_requested.store(false, Ordering::Release); - } -} - fn start_embedded_server(listener: tokio::net::TcpListener, token: String) { let (shutdown_tx, shutdown_rx) = oneshot::channel(); let _ = SHUTDOWN_SENDER.set(Mutex::new(Some(shutdown_tx))); #[cfg(feature = "verge-dev")] - let auth = instance_auth(token.clone()); + let auth = instance_auth(token); #[cfg(not(feature = "verge-dev"))] let auth = instance_auth(token); @@ -256,12 +211,6 @@ fn start_embedded_server(listener: tokio::net::TcpListener, token: String) { }); let commands = auth.clone().and(visible).or(auth.and(scheme)).or(pac); - #[cfg(feature = "verge-dev")] - let commands = commands.or(dev_quit_route(token, &COMMANDS_READY, &DEV_QUIT_REQUESTED, || { - AsyncHandler::spawn(|| async { - release_dev_quit_latch(&DEV_QUIT_REQUESTED, crate::feat::quit().await); - }); - })); AsyncHandler::spawn(move || async move { warp::serve(commands) .incoming(listener) @@ -432,113 +381,10 @@ fn write_instance_record(path: &Path, record: &InstanceRecord) -> Result<()> { #[cfg(test)] mod tests { - use super::{ - InstanceRecord, PAC_INITIAL_AVAILABLE, bind_primary_listener, read_instance_record, write_instance_record, - }; + use super::{InstanceRecord, bind_primary_listener, read_instance_record, write_instance_record}; #[cfg(unix)] use super::{open_instance_lock, try_lock_instance}; - #[cfg(feature = "verge-dev")] - use super::{INSTANCE_TOKEN_HEADER, dev_quit_route, release_dev_quit_latch}; - #[cfg(feature = "verge-dev")] - use std::sync::atomic::Ordering; - #[cfg(feature = "verge-dev")] - use std::sync::{ - Arc, - atomic::{AtomicBool, AtomicUsize}, - }; - #[test] - fn pac_is_fail_closed_before_core_readiness() { - // Asserts the default rather than the live flag: PAC availability is now derived from - // the Running Mode by `core::runstate`, so any test that drives a Core transition - // legitimately moves the live flag. That the derivation itself can never disagree with - // the Running Mode is covered in `core::runstate` against a fake environment. - const { assert!(!PAC_INITIAL_AVAILABLE) } - } - - #[cfg(feature = "verge-dev")] - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn dev_quit_route_requires_private_auth_ready_state_and_dispatches_once() -> anyhow::Result<()> { - static READY: AtomicBool = AtomicBool::new(false); - static REQUESTED: AtomicBool = AtomicBool::new(false); - READY.store(false, Ordering::Release); - REQUESTED.store(false, Ordering::Release); - let dispatches = Arc::new(AtomicUsize::new(0)); - let dispatches_for_route = Arc::clone(&dispatches); - let token = "ab".repeat(32); - let route = dev_quit_route(token.clone(), &READY, &REQUESTED, move || { - dispatches_for_route.fetch_add(1, Ordering::AcqRel); - }); - - let missing = warp::test::request() - .method("POST") - .path("/commands/dev/quit") - .reply(&route) - .await; - assert_eq!(missing.status(), warp::http::StatusCode::NOT_FOUND); - - let wrong_token = warp::test::request() - .method("POST") - .path("/commands/dev/quit") - .header(INSTANCE_TOKEN_HEADER, "cd".repeat(32)) - .reply(&route) - .await; - assert_eq!(wrong_token.status(), warp::http::StatusCode::NOT_FOUND); - - let not_ready = warp::test::request() - .method("POST") - .path("/commands/dev/quit") - .header(INSTANCE_TOKEN_HEADER, &token) - .reply(&route) - .await; - assert_eq!(not_ready.status(), warp::http::StatusCode::SERVICE_UNAVAILABLE); - assert!(!REQUESTED.load(Ordering::Acquire)); - - let wrong_method = warp::test::request() - .method("GET") - .path("/commands/dev/quit") - .header(INSTANCE_TOKEN_HEADER, &token) - .reply(&route) - .await; - assert_eq!(wrong_method.status(), warp::http::StatusCode::METHOD_NOT_ALLOWED); - assert!(!REQUESTED.load(Ordering::Acquire)); - - READY.store(true, Ordering::Release); - let barrier = Arc::new(tokio::sync::Barrier::new(8)); - let accepted = futures::future::join_all((0..8).map(|_| { - let barrier = Arc::clone(&barrier); - let route = route.clone(); - let token = token.clone(); - tokio::spawn(async move { - barrier.wait().await; - warp::test::request() - .method("POST") - .path("/commands/dev/quit") - .header(INSTANCE_TOKEN_HEADER, token) - .reply(&route) - .await - }) - })) - .await; - for accepted in accepted { - let accepted = accepted?; - assert_eq!(accepted.status(), warp::http::StatusCode::ACCEPTED); - } - assert_eq!(dispatches.load(Ordering::Acquire), 1); - - release_dev_quit_latch(&REQUESTED, clash_verge_signal::ShutdownOutcome::Canceled); - assert!(!REQUESTED.load(Ordering::Acquire)); - let retry = warp::test::request() - .method("POST") - .path("/commands/dev/quit") - .header(INSTANCE_TOKEN_HEADER, &token) - .reply(&route) - .await; - assert_eq!(retry.status(), warp::http::StatusCode::ACCEPTED); - assert_eq!(dispatches.load(Ordering::Acquire), 2); - Ok(()) - } - #[tokio::test] async fn retained_record_supplies_the_next_primarys_preferred_port() -> anyhow::Result<()> { let root = std::env::temp_dir().join(format!("singleton-sticky-port-{}", std::process::id()));