diff --git a/crates/clash-verge-draft/src/lib.rs b/crates/clash-verge-draft/src/lib.rs index 95fa02d48..e547a9ead 100644 --- a/crates/clash-verge-draft/src/lib.rs +++ b/crates/clash-verge-draft/src/lib.rs @@ -17,6 +17,8 @@ struct DraftInner { data: Mutex>, data_modifying: AtomicBool, data_modify_notify: Notify, + /// Held by the one [`DraftTransaction`] currently allowed to stage a change. + claimed: AtomicBool, } struct DataModifyPermit<'a>(&'a AtomicBool, &'a Notify); @@ -42,6 +44,7 @@ impl Draft { data: Mutex::new((Arc::new(Box::new(data)), None)), data_modifying: AtomicBool::new(false), data_modify_notify: Notify::new(), + claimed: AtomicBool::new(false), }), } } @@ -146,15 +149,33 @@ impl Draft { } } -/// A draft layer that can be committed or rolled back without knowing what it holds. +/// A draft layer that can be claimed, committed or rolled back without knowing what it holds. /// /// Exists so a change spanning several differently-typed layers can be treated as one unit. pub trait DraftLayer: Send + Sync { + /// Take exclusive staging rights, or report that someone else holds them. + fn try_claim(&self) -> bool; + fn release(&self); fn apply_draft(&self); fn discard_draft(&self); + /// Names the layer in a [`DraftBusy`] error. Only ever read by humans. + fn layer_name(&self) -> &'static str; } impl DraftLayer for Draft { + #[inline] + fn try_claim(&self) -> bool { + self.inner + .claimed + .compare_exchange(false, true, Acquire, Relaxed) + .is_ok() + } + + #[inline] + fn release(&self) { + self.inner.claimed.store(false, Release); + } + #[inline] fn apply_draft(&self) { self.apply(); @@ -164,14 +185,46 @@ impl DraftLayer for Draft { fn discard_draft(&self) { self.discard(); } + + #[inline] + fn layer_name(&self) -> &'static str { + std::any::type_name::() + } } +/// Another transaction is already staging a change to one of the layers asked for. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DraftBusy { + /// The layer that was already claimed. + pub layer: &'static str, +} + +impl std::fmt::Display for DraftBusy { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "configuration layer {} is being changed elsewhere", + self.layer + ) + } +} + +impl std::error::Error for DraftBusy {} + /// Commits a set of draft layers together, or rolls them all back. /// /// Rolls back when dropped, so an early return — a `?`, a `bail!`, a panic — cannot leave a /// change half-applied across layers. Committing is the explicit act, because forgetting to /// commit costs a retry while forgetting to roll back leaves the app describing a state it is /// not in. +/// +/// Beginning one takes exclusive staging rights over every layer, and overlapping with another +/// transaction is refused rather than allowed to interleave. That is not caution, it is the +/// only honest option: a layer holds exactly one draft, so a second writer's `edit_draft` +/// overwrites the first writer's staged value *in place*. There is no version of "roll back +/// only my edit" that could be implemented — rolling back would discard a change the other +/// writer is about to commit, and it did: the other writer's `apply` then found no draft, did +/// nothing at all, and still reported success while its side effect had already happened. #[must_use = "a transaction that is dropped without committing rolls back"] pub struct DraftTransaction<'a> { layers: Vec<&'a dyn DraftLayer>, @@ -179,12 +232,26 @@ pub struct DraftTransaction<'a> { } impl<'a> DraftTransaction<'a> { - /// Begin a transaction over `layers`, which are committed in the order given. - pub fn new(layers: Vec<&'a dyn DraftLayer>) -> Self { - Self { + /// Claim `layers` and begin a transaction over them; they are committed in the order given. + /// + /// Fails if any layer is already claimed, having released the ones taken so far, so a + /// refused attempt leaves nothing held. + pub fn begin(layers: Vec<&'a dyn DraftLayer>) -> Result { + for (index, layer) in layers.iter().enumerate() { + if layer.try_claim() { + continue; + } + for claimed in &layers[..index] { + claimed.release(); + } + return Err(DraftBusy { + layer: layer.layer_name(), + }); + } + Ok(Self { layers, committed: false, - } + }) } /// Commit every layer. Consumes the transaction so it cannot be committed twice. @@ -207,11 +274,11 @@ impl<'a> DraftTransaction<'a> { impl Drop for DraftTransaction<'_> { fn drop(&mut self) { - if self.committed { - return; - } for layer in &self.layers { - layer.discard_draft(); + if !self.committed { + layer.discard_draft(); + } + layer.release(); } } } diff --git a/crates/clash-verge-draft/tests/test_me.rs b/crates/clash-verge-draft/tests/test_me.rs index cc17fbaf9..c4df846ca 100644 --- a/crates/clash-verge-draft/tests/test_me.rs +++ b/crates/clash-verge-draft/tests/test_me.rs @@ -307,12 +307,16 @@ mod tests { mod transactions { use clash_verge_draft::{Draft, DraftTransaction}; + fn begin<'a>(layers: Vec<&'a dyn clash_verge_draft::DraftLayer>) -> DraftTransaction<'a> { + DraftTransaction::begin(layers).expect("no other transaction should hold these layers") + } + #[test] fn committing_applies_every_layer() { let first = Draft::new(1_u8); let second = Draft::new("a".to_owned()); - let transaction = DraftTransaction::new(vec![&first, &second]); + let transaction = begin(vec![&first, &second]); first.edit_draft(|value| *value = 2); second.edit_draft(|value| *value = "b".to_owned()); transaction.commit(); @@ -327,7 +331,7 @@ mod transactions { let second = Draft::new("a".to_owned()); { - let _transaction = DraftTransaction::new(vec![&first, &second]); + let _transaction = begin(vec![&first, &second]); first.edit_draft(|value| *value = 2); second.edit_draft(|value| *value = "b".to_owned()); } @@ -344,7 +348,7 @@ mod transactions { #[test] fn an_early_return_rolls_back_without_the_caller_saying_so() { fn attempt(draft: &Draft, succeed: bool) -> Result<(), &'static str> { - let transaction = DraftTransaction::new(vec![draft]); + let transaction = begin(vec![draft]); draft.edit_draft(|value| *value = 9); if !succeed { return Err("failed after editing"); @@ -367,14 +371,77 @@ mod transactions { let edited = Draft::new(1_u8); let untouched = Draft::new(5_u8); - drop(DraftTransaction::new(vec![&edited, &untouched])); + drop(begin(vec![&edited, &untouched])); edited.edit_draft(|value| *value = 2); { - let _transaction = DraftTransaction::new(vec![&edited, &untouched]); + let _transaction = begin(vec![&edited, &untouched]); } assert_eq!(**edited.latest_arc(), 1, "the earlier stray draft is cleared too"); assert_eq!(**untouched.latest_arc(), 5); } } + +mod transaction_exclusivity { + use clash_verge_draft::{Draft, DraftTransaction}; + + #[test] + fn a_second_transaction_over_a_claimed_layer_is_refused() { + let draft = Draft::new(1_u8); + let _held = DraftTransaction::begin(vec![&draft]).expect("the first claim should succeed"); + + let refused = DraftTransaction::begin(vec![&draft]); + + assert!(refused.is_err(), "a layer may only be staged by one transaction"); + } + + #[test] + fn a_refused_transaction_leaves_the_layers_it_did_claim_free() { + // Claiming is all-or-nothing: a caller that failed on the third layer must not keep + // the first two, or the next attempt would be refused for a transaction that no + // longer exists. + let first = Draft::new(1_u8); + let second = Draft::new(2_u8); + let _contended = DraftTransaction::begin(vec![&second]).expect("the first claim should succeed"); + + assert!(DraftTransaction::begin(vec![&first, &second]).is_err()); + + drop(DraftTransaction::begin(vec![&first]).expect("the layer that was rolled off must be free")); + } + + #[test] + fn a_layer_is_free_again_once_its_transaction_ends() { + let draft = Draft::new(1_u8); + + drop(DraftTransaction::begin(vec![&draft]).expect("first")); + let committed = DraftTransaction::begin(vec![&draft]).expect("free after a rollback"); + committed.commit(); + + drop(DraftTransaction::begin(vec![&draft]).expect("free after a commit")); + } + + #[test] + fn an_overlapping_writer_can_no_longer_discard_a_staged_edit() { + // The regression this pins. A holds a transaction across a slow side effect; B starts, + // stages its own edit over the one draft slot, fails, and rolls back. Before layers + // were claimed, B's rollback cleared the single draft, A's commit then found nothing to + // apply and did nothing at all — while A's caller was told it had succeeded and A's own + // side effect had already happened. + let verge = Draft::new(String::from("committed")); + + let transaction_a = DraftTransaction::begin(vec![&verge]).expect("A claims the layer"); + verge.edit_draft(|value| *value = String::from("A-wants-this")); + + let transaction_b = DraftTransaction::begin(vec![&verge]); + assert!(transaction_b.is_err(), "B must be refused rather than share the slot"); + + transaction_a.commit(); + + assert_eq!( + **verge.data_arc(), + "A-wants-this", + "A's edit must survive an overlapping writer" + ); + } +} diff --git a/src-tauri/src/config/mixed_port.rs b/src-tauri/src/config/mixed_port.rs index df6b6f95e..a8da3da35 100644 --- a/src-tauri/src/config/mixed_port.rs +++ b/src-tauri/src/config/mixed_port.rs @@ -28,6 +28,13 @@ impl MixedPort { /// /// Reads the draft layer, so an edit in progress is visible to the code applying it. Safe /// to call before the Core is running, unlike [`Self::effective`]. + /// + /// Reading the draft means a caller that hands this to something *outside* the app — the + /// PAC endpoint is the one that does — can name a port the Core has not moved to yet. + /// What makes that safe is that the only path staging a listener port, + /// `feat::listener::save_proxy_ports`, closes PAC across the staging with `core_starting` + /// and reopens it with `core_start_settled`. A second path that stages a listener port + /// while the Core is serving would have to do the same. pub async fn desired() -> u16 { let selected = Config::verge().await.latest_arc().verge_mixed_port; // `get_mixed_port` already falls back to the default when the Merge Config is silent. diff --git a/src-tauri/src/config/port.rs b/src-tauri/src/config/port.rs index 73f5d22fc..68d2bd23a 100644 --- a/src-tauri/src/config/port.rs +++ b/src-tauri/src/config/port.rs @@ -90,7 +90,7 @@ impl Config { let verge = Self::verge().await; let runtime = Self::runtime().await; // Every failure below leaves the three layers as they were, without saying so. - let transaction = DraftTransaction::new(vec![&clash, &verge, &runtime]); + let transaction = DraftTransaction::begin(vec![&clash, &verge, &runtime])?; clash.edit_draft(|draft| { draft.0.insert(MIXED_PORT_KEY.into(), new_port.into()); diff --git a/src-tauri/src/constants.rs b/src-tauri/src/constants.rs index bd0451935..4326afe22 100644 --- a/src-tauri/src/constants.rs +++ b/src-tauri/src/constants.rs @@ -20,6 +20,10 @@ pub mod timing { pub const CONFIG_UPDATE_DEBOUNCE: Duration = Duration::from_millis(300); pub const STARTUP_ERROR_DELAY: Duration = Duration::from_secs(2); + // How long a re-asked staging request is worth waiting for. One that already landed answers in + // milliseconds; anything slower means the Service is still working, so ReplaceCore beats waiting. + pub const STAGE_CONFIRM_TIMEOUT: Duration = Duration::from_secs(5); + // Windows 服务冷启动较慢,避免过早回退 sidecar。 #[cfg(target_os = "windows")] pub const SERVICE_WAIT_MAX: Duration = Duration::from_millis(30000); diff --git a/src-tauri/src/core/manager/config.rs b/src-tauri/src/core/manager/config.rs index f19c27bfc..898123a15 100644 --- a/src-tauri/src/core/manager/config.rs +++ b/src-tauri/src/core/manager/config.rs @@ -15,7 +15,11 @@ use clash_verge_logging::{Type, logging}; use clash_verge_service_ipc::StageRuntimeOutcome; use scopeguard::defer; use smartstring::alias::String; -use std::{collections::HashSet, path::PathBuf, time::Instant}; +use std::{ + collections::HashSet, + path::PathBuf, + time::{Duration, Instant}, +}; use tauri_plugin_mihomo::Error as MihomoError; /// What came back from asking the Service to stage a runtime. @@ -23,7 +27,7 @@ use tauri_plugin_mihomo::Error as MihomoError; /// Separated from the decision below so the decision stays a pure function of it. The variants are /// the four ways an attempt can end, and what distinguishes them is how much they let us conclude: /// a refusal is the Service's verdict on this bundle, while silence says nothing at all. -#[derive(Debug, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] enum StageAttempt { /// The Service that owns the running Core predates staging. Unsupported, @@ -100,6 +104,60 @@ fn plan_config_application(attempt: &StageAttempt) -> ConfigApplication { } } +/// One round trip to the Service, classified but not judged. +async fn ask_to_stage(path: &std::path::Path) -> StageAttempt { + match crate::core::service::stage_runtime_by_service(path).await { + Ok(StageRequest::Answered(outcome)) => StageAttempt::Answered(outcome), + Ok(StageRequest::Refused { code, message }) => { + let message = message.to_string().into(); + if StageRequest::is_about_the_bundle(code) { + StageAttempt::RefusedTheBundle(message) + } else { + StageAttempt::RefusedForAnotherReason(message) + } + } + Err(err) => StageAttempt::Unanswered(format!("{err:#}").into()), + } +} + +/// Ask once, and ask again only when the first attempt came back silent. +/// +/// Silence is the one answer that has to be resolved rather than acted on. The Service commits the +/// generation before it replies, so a reply that never arrives can leave `config.yaml` already +/// replaced while this side concludes nothing happened — and the Service restarts a Core that dies +/// later from exactly that file. Treating silence as "nothing happened" is therefore a guess, and +/// the expensive half of it is silent: the Core keeps running the previous configuration until +/// something restarts it, and then it does not. +/// +/// Asking again is cheap precisely where it matters. Staging plans against the manifest it wrote, +/// so a bundle that already landed copies nothing and answers in milliseconds; and because the +/// Service holds its lifecycle lock for the whole operation, a second ask that overtakes a first +/// one still in flight waits for it rather than interleaving with it. `confirm_within` is what +/// keeps that wait from being the caller's problem — past it, the answer has stopped being worth +/// more than the fallback, which is where a second silence goes anyway. +async fn stage_with_confirmation(confirm_within: Duration, ask: Ask) -> StageAttempt +where + Ask: Fn() -> Fut, + Fut: std::future::Future, +{ + let first = match ask().await { + StageAttempt::Unanswered(first) => first, + answered => return answered, + }; + logging!( + warn, + Type::Core, + "Staging did not answer ({first}); asking once more before replacing the core" + ); + match tokio::time::timeout(confirm_within, ask()).await { + Ok(StageAttempt::Unanswered(again)) => { + StageAttempt::Unanswered(format!("{first}; asked again: {again}").into()) + } + Ok(answered) => answered, + Err(_) => StageAttempt::Unanswered(format!("{first}; the second ask did not answer either").into()), + } +} + impl CoreManager { pub async fn use_default_config(&self, error_key: &str, error_msg: &str) -> Result<()> { use crate::constants::files::RUNTIME_CONFIG; @@ -178,7 +236,7 @@ impl CoreManager { async fn perform_config_update(&self) -> Result { let runtime = Config::runtime().await; - let transaction = DraftTransaction::new(vec![&runtime]); + let transaction = DraftTransaction::begin(vec![&runtime])?; if let Err(err) = Config::generate().await { let message: String = err.to_string().into(); @@ -201,7 +259,7 @@ impl CoreManager { } let runtime = Config::runtime().await; - let transaction = DraftTransaction::new(vec![&runtime]); + let transaction = DraftTransaction::begin(vec![&runtime])?; runtime.edit_draft(f); self.validate_and_apply(transaction).await } @@ -287,18 +345,7 @@ impl CoreManager { if !crate::core::service::active_service_supports_runtime_staging() { return StageAttempt::Unsupported; } - match crate::core::service::stage_runtime_by_service(path).await { - Ok(StageRequest::Answered(outcome)) => StageAttempt::Answered(outcome), - Ok(StageRequest::Refused { code, message }) => { - let message = message.to_string().into(); - if StageRequest::is_about_the_bundle(code) { - StageAttempt::RefusedTheBundle(message) - } else { - StageAttempt::RefusedForAnotherReason(message) - } - } - Err(err) => StageAttempt::Unanswered(format!("{err:#}").into()), - } + stage_with_confirmation(timing::STAGE_CONFIRM_TIMEOUT, || ask_to_stage(path)).await } /// Reload the Core from `path`, and replace the Core if it will not take it. @@ -332,8 +379,17 @@ impl CoreManager { #[cfg(test)] mod tests { - use super::{ConfigApplication, StageAttempt, StageRequest, plan_config_application}; + use super::{ConfigApplication, StageAttempt, StageRequest, plan_config_application, stage_with_confirmation}; use clash_verge_service_ipc::{StageRejection, StageRuntimeOutcome}; + use std::{cell::Cell, time::Duration}; + + const CONFIRM_WITHIN: Duration = Duration::from_secs(5); + + fn staged() -> StageAttempt { + StageAttempt::Answered(StageRuntimeOutcome::Staged { + config_path: "/service/runtime.generation-1/config.yaml".to_owned(), + }) + } #[test] fn a_staged_runtime_is_reloaded_from_where_the_service_put_it() { @@ -430,4 +486,97 @@ mod tests { ConfigApplication::ReplaceCore ); } + + #[tokio::test] + async fn an_answer_is_taken_at_its_word_and_not_asked_for_twice() { + // Staging is only idempotent in what it writes, not in what it costs: a second ask after a + // perfectly good answer would re-send the whole bundle for nothing. + for answer in [ + staged(), + StageAttempt::Answered(StageRuntimeOutcome::RestartRequired { + reason: StageRejection::CoreNotRunning, + }), + StageAttempt::RefusedTheBundle("runtime asset is unavailable".into()), + StageAttempt::RefusedForAnotherReason("owner session is stale".into()), + ] { + let asks = Cell::new(0); + let outcome = stage_with_confirmation(CONFIRM_WITHIN, || { + asks.set(asks.get() + 1); + std::future::ready(answer.clone()) + }) + .await; + + assert_eq!(outcome, answer); + assert_eq!(asks.get(), 1, "{answer:?} was answered, so asking again buys nothing"); + } + } + + #[tokio::test] + async fn silence_is_resolved_by_asking_again_rather_than_assumed() { + // The case this exists for: the service committed the generation and the reply was lost. + // Asking again reaches a service that has already done the work, so the second answer is + // the truth the first one failed to deliver — and it keeps the core off the replace path. + let asks = Cell::new(0); + let outcome = stage_with_confirmation(CONFIRM_WITHIN, || { + asks.set(asks.get() + 1); + std::future::ready(match asks.get() { + 1 => StageAttempt::Unanswered("connection reset".into()), + _ => staged(), + }) + }) + .await; + + assert_eq!( + outcome, + staged(), + "the second answer must decide, not the first silence" + ); + assert_eq!(asks.get(), 2); + assert_eq!( + plan_config_application(&outcome), + ConfigApplication::ReloadFrom("/service/runtime.generation-1/config.yaml".into()), + "resolving the silence is only worth doing if it changes what happens next" + ); + } + + #[tokio::test] + async fn a_second_silence_still_replaces_the_core() { + let asks = Cell::new(0); + let outcome = stage_with_confirmation(CONFIRM_WITHIN, || { + asks.set(asks.get() + 1); + std::future::ready(StageAttempt::Unanswered("connection reset".into())) + }) + .await; + + assert_eq!(asks.get(), 2, "asking a third time would just be a retry loop"); + assert_eq!(plan_config_application(&outcome), ConfigApplication::ReplaceCore); + assert!( + matches!(&outcome, StageAttempt::Unanswered(detail) + if detail.contains("connection reset") && detail.contains("asked again")), + "two silences must stay unanswered and record that both were tried, got {outcome:?}" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_second_ask_that_keeps_working_is_not_waited_out() { + // A service still copying assets is why the deadline exists: the answer would be worth + // having, but not at the cost of leaving the user's configuration unapplied indefinitely. + let asks = Cell::new(0); + let started = tokio::time::Instant::now(); + let outcome = stage_with_confirmation(CONFIRM_WITHIN, || async { + asks.set(asks.get() + 1); + match asks.get() { + 1 => StageAttempt::Unanswered("deadline has elapsed".into()), + _ => std::future::pending().await, + } + }) + .await; + + assert_eq!(plan_config_application(&outcome), ConfigApplication::ReplaceCore); + assert_eq!( + started.elapsed(), + CONFIRM_WITHIN, + "the caller must be released at the deadline, not at the service's convenience" + ); + } } diff --git a/src-tauri/src/core/manager/mod.rs b/src-tauri/src/core/manager/mod.rs index a91167c1a..4d3515674 100644 --- a/src-tauri/src/core/manager/mod.rs +++ b/src-tauri/src/core/manager/mod.rs @@ -170,10 +170,18 @@ impl CoreManager { } /// A start attempt is under way and the Core is not serving yet. + /// + /// Must be paired with [`Self::core_start_settled`] on every path out, including the ones + /// where the start never happened. pub fn core_starting(&self) { self.run_state.core_starting(); } + /// The start attempt is over: PAC goes back to following the Running Mode. + pub fn core_start_settled(&self) { + self.run_state.core_start_settled(); + } + pub fn set_running_child_sidecar(&self, child: CommandChild) { let state = self.state.load(); state.child_sidecar.store(Some(Arc::new(child))); diff --git a/src-tauri/src/core/runstate/health.rs b/src-tauri/src/core/runstate/health.rs index 4d6958065..bb0e2b1c0 100644 --- a/src-tauri/src/core/runstate/health.rs +++ b/src-tauri/src/core/runstate/health.rs @@ -97,11 +97,21 @@ impl RunState { /// The Service cannot be used and the user has to choose what to do about it. /// - /// A requested action always needs an answer. Otherwise it is only the states where the - /// Service is present but unusable: a Service that is simply absent, or a session that has - /// already settled on Sidecar, needs nothing from anyone. + /// A requested action always needs an answer — unless it is already being carried out, + /// which is a report of progress rather than a question. The legacy status could not tell + /// the two apart because readers blocked until the operation finished and so never saw the + /// request at all; now that every transition is pushed, an install started from the + /// settings page would otherwise raise the migration dialog on top of it, and an uninstall + /// would raise one offering to reinstall the very Service being removed. + /// + /// Otherwise it is only the states where the Service is present but unusable: a Service + /// that is simply absent, or a session that has already settled on Sidecar, needs nothing + /// from anyone. #[must_use] pub const fn service_needs_attention(&self) -> bool { + if self.op_in_flight { + return false; + } if self.pending.is_some() { return true; } @@ -247,7 +257,7 @@ mod tests { } #[test] - fn a_requested_action_always_needs_an_answer() { + fn a_requested_action_needs_an_answer_until_it_is_being_carried_out() { for action in [ PendingAction::Install, PendingAction::Uninstall, @@ -256,7 +266,34 @@ mod tests { ] { let mut run_state = state(ServiceHealth::NotInstalled, false, false); run_state.pending = Some(action); - assert!(run_state.service_needs_attention(), "{action:?}"); + assert!(run_state.service_needs_attention(), "{action:?} is still a question"); + + run_state.op_in_flight = true; + assert!( + !run_state.service_needs_attention(), + "{action:?} is under way, which is progress rather than a question" + ); + } + } + + #[test] + fn an_operation_under_way_is_never_a_decision_to_put_to_the_user() { + // The regression this pins: an install or uninstall started from the settings page + // pushes its pending action to the frontend the moment it is requested. Readers used + // to block until the operation finished and so never saw it; now that every transition + // is pushed, treating it as a decision raises the migration dialog on top of an + // operation already running — and for an uninstall, one offering to reinstall. + for health in [ + ServiceHealth::Unknown, + ServiceHealth::Ready, + ServiceHealth::NotInstalled, + ServiceHealth::VersionMismatch, + ServiceHealth::Unavailable("boom".into()), + ] { + let mut run_state = state(health.clone(), false, true); + run_state.pending = Some(PendingAction::Uninstall); + + assert!(!run_state.service_needs_attention(), "{health:?}"); } } diff --git a/src-tauri/src/core/runstate/mod.rs b/src-tauri/src/core/runstate/mod.rs index 7c69ed0e6..bbcbd40cd 100644 --- a/src-tauri/src/core/runstate/mod.rs +++ b/src-tauri/src/core/runstate/mod.rs @@ -114,19 +114,26 @@ impl RunStateStore { } /// The Run State once any in-flight privileged operation has finished. + /// + /// The snapshot is the decision, not a separate check before it. Reading the slot twice — + /// once to decide whether to wait, once to build the snapshot — leaves a window in which an + /// operation claims it, and the snapshot then *describes* that operation instead of waiting + /// for it. `prepare_startup` reads this: it would see a requested install as a reason not to + /// start, where waiting would have told it whether the install worked. pub async fn settled(&self) -> RunState { loop { let notified = self.operation_done.notified(); tokio::pin!(notified); - // Register as a waiter *before* checking the flag. `notify_waiters` wakes only + // Register as a waiter *before* reading the state. `notify_waiters` wakes only // those already registered and leaves no permit behind, so without this an - // operation finishing between the check and the await is a wakeup lost for good + // operation finishing between the read and the await is a wakeup lost for good // — and this future would then wait for some unrelated later operation, or // forever. notified.as_mut().enable(); - if !self.operation_running.load(Ordering::Acquire) { - return self.state(); + let state = self.state(); + if !state.op_in_flight { + return state; } notified.await; } @@ -350,10 +357,27 @@ impl RunStateStore { /// /// Closes the PAC endpoint without disturbing the Running Mode, so a handover cannot /// hand out a PAC script for a proxy port that is between owners. + /// + /// Every caller must pair this with [`Self::core_start_settled`]. A start attempt that + /// never happens — a candidate rejected before anything was stopped — otherwise leaves PAC + /// closed for a Core that is still serving, and nothing else would ever reopen it: PAC is + /// re-derived only when the Running Mode changes, and that is exactly what did not happen. pub fn core_starting(&self) { self.env.set_pac_available(false); } + /// The start attempt is over, however it ended: PAC goes back to following the Running Mode. + /// + /// Idempotent and safe on every path, because it re-derives rather than restores. After a + /// start that succeeded the mode is already running and this confirms PAC open; after one + /// that was abandoned the mode never moved and this reopens PAC for the Core that kept + /// serving; after one that failed and stopped the Core the mode says NotRunning and PAC + /// stays shut. + pub fn core_start_settled(&self) { + let running = !matches!(**self.mode.load(), RunningMode::NotRunning); + self.env.set_pac_available(running); + } + /// Move to `mode` and re-derive everything that follows from it. /// /// PAC availability and the outward mirror are derived here and nowhere else, so they @@ -628,6 +652,33 @@ mod tests { assert!(!settled.op_in_flight); } + // Multi-threaded on purpose, so operations really do start and finish underneath the reader. + // This is a guard, not a reproducer: the window it protects against was narrow enough that + // reverting the fix did not fail this test in six runs. What makes the invariant hold is + // that `settled` reads the slot once, inside the snapshot it returns; this fails if someone + // reintroduces a separate check before it. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_settled_snapshot_never_describes_an_operation_in_flight() { + let store = Arc::new(with_env(FakeEnv::new())); + store.observe(ServiceHealth::Ready); + + for _ in 0..64 { + let claimer = { + let store = Arc::clone(&store); + tokio::spawn(async move { + let guard = store.begin_operation(); + tokio::task::yield_now().await; + drop(guard); + }) + }; + + let settled = store.settled().await; + assert!(!settled.op_in_flight, "a settled snapshot describes a settled state"); + + claimer.await.expect("the claimer should not panic"); + } + } + #[tokio::test] async fn probing_a_ready_service_records_it() { let store = with_env(FakeEnv::new().service_ready()); @@ -831,6 +882,48 @@ mod tests { ); } + #[test] + fn an_abandoned_start_attempt_reopens_pac_for_the_core_that_kept_running() { + // The regression this pins: a proxy-port candidate rejected before anything was + // stopped. `core_starting` closed PAC, the Running Mode never moved, and so nothing + // would ever re-derive PAC — the Core kept serving while its PAC endpoint stayed shut. + for mode in [RunningMode::Service, RunningMode::Sidecar] { + let store = with_env(FakeEnv::new()); + store.core_started(mode); + store.core_starting(); + assert_eq!(store.env.pac_available(), Some(false)); + + store.core_start_settled(); + + assert_eq!(store.env.pac_available(), Some(true), "{mode} kept serving"); + assert_eq!(store.state().mode, mode, "an abandoned start is not a mode change"); + } + } + + #[test] + fn a_settled_start_leaves_pac_shut_when_no_core_is_running() { + let store = with_env(FakeEnv::new()); + store.core_starting(); + + store.core_start_settled(); + + assert_eq!(store.env.pac_available(), Some(false)); + } + + #[test] + fn settling_a_start_is_idempotent_and_never_contradicts_the_mode() { + // Callers pair this with `core_starting` via a guard that runs on every path, so it + // also runs after a start that already opened PAC itself. + let store = with_env(FakeEnv::new()); + store.core_starting(); + store.core_started(RunningMode::Service); + + store.core_start_settled(); + store.core_start_settled(); + + assert_eq!(store.env.pac_available(), Some(true)); + } + #[test] fn pac_availability_can_never_disagree_with_the_running_mode() { let store = with_env(FakeEnv::new()); diff --git a/src-tauri/src/feat/config.rs b/src-tauri/src/feat/config.rs index 172406be3..66befa2dc 100644 --- a/src-tauri/src/feat/config.rs +++ b/src-tauri/src/feat/config.rs @@ -268,13 +268,40 @@ async fn process_terminated_flags(update_flags: UpdateFlags, patch: &IVerge) -> Ok(()) } +/// Apply a patch to the app's configuration, then re-check anything it can invalidate. +/// +/// Today that is TUN, which is a question about the *setting* rather than about the Run State: +/// no Run State transition follows a TUN patch, so the reconciliation that reacts to those +/// would never see it. Switching TUN on from the global hotkey is the case that reaches here — +/// unlike the tray item and the settings switch it is not gated on availability, and the flags +/// the patch raises are answered by a Core reload that changes no Run State at all. Left alone +/// the setting stays on and every surface reports TUN as enabled while nothing carries its +/// traffic. +/// +/// The reconciliation writes configuration of its own, so it goes through +/// [`apply_verge_patch`] rather than back through here. That makes the absence of a cycle a +/// property of the call graph rather than of a runtime early-return. pub async fn patch_verge(patch: &IVerge, not_save_file: bool) -> Result<()> { + apply_verge_patch(patch, not_save_file).await?; + if patch.enable_tun_mode.is_some() { + super::reconcile_tun_availability().await; + } + Ok(()) +} + +/// Apply a patch and nothing else. For callers that are themselves a reconciliation. +pub(super) async fn apply_verge_patch(patch: &IVerge, not_save_file: bool) -> Result<()> { let verge = Config::verge().await; // Applying the flags can fail, and until now that `?` returned straight out of here past // an `if let Err(..) { discard() }` the compiler could never reach — leaving the failed // edit sitting in the draft, where every later reader saw a value that was never applied // and never written to disk. - let transaction = DraftTransaction::new(vec![&verge]); + // + // Claiming the layer is what makes the rollback safe. This function holds its draft across + // `process_terminated_flags`, which restarts the Core and can take seconds; a second patch + // arriving in that window used to share the one draft slot, and whichever of the two failed + // first discarded the other's staged edit too. + let transaction = DraftTransaction::begin(vec![&verge])?; verge.edit_draft(|d| d.patch_config(patch)); let update_flags = determine_update_flags(patch); diff --git a/src-tauri/src/feat/listener.rs b/src-tauri/src/feat/listener.rs index 1a9ab80fb..39b40d963 100644 --- a/src-tauri/src/feat/listener.rs +++ b/src-tauri/src/feat/listener.rs @@ -46,12 +46,21 @@ pub async fn save_proxy_ports(settings: ProxyPortSettings) -> Result) -> bool { { Ok(_) => { handle::Handle::refresh_verge(); - enable + // Read back rather than returning what was asked for: patching TUN reconciles it + // afterwards, and where TUN cannot work that reconciliation turns it straight off + // again. This path is not gated on availability — it is the global hotkey — so + // reporting the request would tell the caller TUN is on moments before the notice + // saying it was disabled. + Config::verge().await.latest_arc().enable_tun_mode.unwrap_or(false) } Err(err) => { logging!(error, Type::ProxyMode, "{err}"); diff --git a/src-tauri/src/feat/tun.rs b/src-tauri/src/feat/tun.rs index 90e81dc9b..4ec8baf9c 100644 --- a/src-tauri/src/feat/tun.rs +++ b/src-tauri/src/feat/tun.rs @@ -6,6 +6,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; +use clash_verge_draft::DraftBusy; use clash_verge_logging::{Type, logging}; use scopeguard::defer; @@ -70,8 +71,20 @@ pub async fn reconcile_tun_availability() { ..IVerge::default() }; - match super::patch_verge(&patch, false).await { + // The patch-and-reconcile form would call straight back into here; this is the plain apply. + match super::apply_verge_patch(&patch, false).await { Ok(()) => Handle::notice_message("tun_mode::auto_disabled", ""), + // Losing the race for the config layer is not a failure the user can act on: this runs + // on every Run State transition and again whenever the setting itself is patched, so + // the writer that won will be followed by another reconciliation. Telling the user that + // turning TUN off "failed" would be both alarming and wrong. + Err(error) if error.downcast_ref::().is_some() => { + logging!( + debug, + Type::Core, + "deferring the TUN reconciliation, configuration is busy: {error:#}" + ); + } Err(error) => { logging!(error, Type::Core, "failed to turn TUN mode off: {error:#}"); Handle::notice_message("tun_mode::auto_disable_failed", ""); diff --git a/src-tauri/src/utils/server.rs b/src-tauri/src/utils/server.rs index f00b494b2..dc71ae0a9 100644 --- a/src-tauri/src/utils/server.rs +++ b/src-tauri/src/utils/server.rs @@ -215,7 +215,9 @@ fn start_embedded_server(listener: tokio::net::TcpListener, token: String) { let verge_data = verge_config.data_arc(); let pac_content = verge_data.pac_file_content.as_deref().unwrap_or(DEFAULT_PAC); // Served per browser request, so this stays a configuration read rather than a - // round-trip to the Core. PAC is closed while the Core is between ports anyway. + // round-trip to the Core. It reads the draft layer, which is only correct because + // whoever stages a listener port closes this endpoint across the change — see + // `MixedPort::desired`. Reaching here at all means the Core is serving. let pac_port = MixedPort::desired().await; let processed_content = pac_content.replace("%mixed-port%", &format!("{pac_port}")); Ok::<_, warp::Rejection>( diff --git a/src/pages/_layout/hooks/use-layout-events.ts b/src/pages/_layout/hooks/use-layout-events.ts index fc584231c..46394b4fa 100644 --- a/src/pages/_layout/hooks/use-layout-events.ts +++ b/src/pages/_layout/hooks/use-layout-events.ts @@ -13,28 +13,39 @@ export const useLayoutEvents = ( void revalidateQueries(keys.map((key) => [key])) } - return subscribeVergeEvents({ - 'verge://refresh-clash-config': () => { - revalidateKeys([ - 'getProxyView', - 'getVersion', - 'getClashConfig', - 'getClashInfo', - 'getClashMode', - 'getRuntimeConfig', - 'getRules', - 'getRuleProviders', - ]) + return subscribeVergeEvents( + { + 'verge://refresh-clash-config': () => { + revalidateKeys([ + 'getProxyView', + 'getVersion', + 'getClashConfig', + 'getClashInfo', + 'getClashMode', + 'getRuntimeConfig', + 'getRules', + 'getRuleProviders', + ]) + }, + 'verge://refresh-verge-config': () => { + revalidateKeys([ + 'getVergeConfig', + 'getSystemProxy', + 'getAutotemProxy', + ]) + }, + // The Run State is pushed, not polled: every transition carries the new snapshot, so it + // is written straight into the cache instead of triggering a fetch. + 'verge://run-state-changed': (payload) => { + void setCacheDataAsync(runStateQueryKey, payload) + }, + 'verge://notice-message': handleNotice, }, - 'verge://refresh-verge-config': () => { - revalidateKeys(['getVergeConfig', 'getSystemProxy', 'getAutotemProxy']) - }, - // The Run State is pushed, not polled: every transition carries the new snapshot, so it - // is written straight into the cache instead of triggering a fetch. - 'verge://run-state-changed': (payload) => { - void setCacheDataAsync(runStateQueryKey, payload) - }, - 'verge://notice-message': handleNotice, - }) + // The Run State is the one thing here that arrives *only* by event, so it is the one + // thing a missed event leaves stale — until the next transition, which during startup can + // be a long time. Re-read once the listener is live, so the gap between the first read + // and the subscription cannot outlive the subscription. + () => revalidateKeys(['getRuntimeState']), + ) }, [handleNotice]) } diff --git a/src/services/events.test.ts b/src/services/events.test.ts new file mode 100644 index 000000000..efaa4ccae --- /dev/null +++ b/src/services/events.test.ts @@ -0,0 +1,106 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { subscribeVergeEvents } from './events' + +const listen = vi.hoisted(() => vi.fn()) + +vi.mock('@tauri-apps/api/event', () => ({ listen })) + +/** A `listen` whose registrations only resolve when the test says so. */ +const deferredListen = () => { + const resolvers: Array<() => void> = [] + listen.mockImplementation( + () => + new Promise((resolve) => { + resolvers.push(() => resolve(() => {})) + }), + ) + return { + registrations: resolvers, + settleAll: () => resolvers.forEach((resolve) => resolve()), + } +} + +/** Let every already-resolved promise in the chain run. */ +const flush = () => new Promise((resolve) => setTimeout(resolve, 0)) + +beforeEach(() => { + listen.mockReset() +}) + +describe('subscribeVergeEvents', () => { + it('reports subscribed only once every listener is live', async () => { + const { registrations, settleAll } = deferredListen() + const onSubscribed = vi.fn() + + subscribeVergeEvents( + { + 'verge://run-state-changed': () => {}, + 'verge://notice-message': () => {}, + }, + onSubscribed, + ) + + expect(registrations).toHaveLength(2) + registrations[0]?.() + await flush() + expect(onSubscribed).not.toHaveBeenCalled() + + settleAll() + await flush() + expect(onSubscribed).toHaveBeenCalledTimes(1) + }) + + it('does not report subscribed after teardown', async () => { + // Registration outliving the caller is the whole reason this is tracked here; resyncing + // for a subscription that no longer exists would fetch into a cache nobody reads. + const { settleAll } = deferredListen() + const onSubscribed = vi.fn() + + const teardown = subscribeVergeEvents( + { 'verge://run-state-changed': () => {} }, + onSubscribed, + ) + teardown() + + settleAll() + await flush() + expect(onSubscribed).not.toHaveBeenCalled() + }) + + it('still reports subscribed when a listener fails to register', async () => { + // A resync is most needed when a subscription is missing, not least. + listen.mockRejectedValue(new Error('no such event')) + vi.spyOn(console, 'error').mockImplementation(() => {}) + const onSubscribed = vi.fn() + + subscribeVergeEvents( + { 'verge://run-state-changed': () => {} }, + onSubscribed, + ) + + await vi.waitFor(() => expect(onSubscribed).toHaveBeenCalledTimes(1)) + }) + + it('delivers event payloads to the matching handler', async () => { + const handlers = new Map void>() + listen.mockImplementation( + (name: string, handler: (event: unknown) => void) => { + handlers.set(name, (payload) => handler({ payload })) + return Promise.resolve(() => {}) + }, + ) + const onRunState = vi.fn() + const onNotice = vi.fn() + + subscribeVergeEvents({ + 'verge://run-state-changed': onRunState, + 'verge://notice-message': onNotice, + }) + + handlers.get('verge://run-state-changed')?.({ mode: 'Sidecar' }) + + expect(onRunState).toHaveBeenCalledWith({ mode: 'Sidecar' }) + expect(onNotice).not.toHaveBeenCalled() + }) +}) diff --git a/src/services/events.ts b/src/services/events.ts index 731b08304..59d3afdc7 100644 --- a/src/services/events.ts +++ b/src/services/events.ts @@ -45,17 +45,24 @@ type VergeEventHandlers = { * Registration is asynchronous, so a subscription can resolve after the caller has already * torn down; that is tracked here rather than at each call site, where it was implemented * four different ways and omitted in one of them. + * + * `onSubscribed` runs once every listener is live. Anything whose state arrives only by event + * needs it: between the first read and the listener being registered there is a gap, and a + * transition landing in it is not queued anywhere — it is simply missed, and the next event is + * whenever the backend happens to move again. Re-reading at that point closes the gap without + * going back to polling for it. */ export const subscribeVergeEvents = ( handlers: VergeEventHandlers, + onSubscribed?: () => void, ): (() => void) => { let disposed = false const unlisteners: UnlistenFn[] = [] - for (const [name, handler] of Object.entries(handlers)) { - if (!handler) continue + const registrations = Object.entries(handlers).map(([name, handler]) => { + if (!handler) return Promise.resolve() - listen(name, ({ payload }) => { + return listen(name, ({ payload }) => { ;(handler as (payload: unknown) => void)(payload) }) .then((unlisten) => { @@ -69,6 +76,13 @@ export const subscribeVergeEvents = ( .catch((error) => { console.error(`[events] failed to subscribe to ${name}:`, error) }) + }) + + if (onSubscribed) { + void Promise.all(registrations).then(() => { + if (disposed) return + onSubscribed() + }) } return () => {