From cefcb150c984de7ef0c4ccb64428109863905a47 Mon Sep 17 00:00:00 2001 From: wonfen <96291150+wonfen@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:58:47 +0800 Subject: [PATCH] fix: core runstate review/improve (#7617) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(core): reopen PAC when a start attempt never happens `save_proxy_ports` closes the PAC endpoint before it stages a candidate, because the endpoint resolves the Mixed Port through the draft layer and would otherwise answer with a port nothing is listening on. Most ways out of that function are a rejection, though: the candidate is refused and the Core keeps serving on its old ports, having never been stopped. PAC is re-derived only when the Running Mode changes, and that is exactly what did not happen — so the endpoint stayed shut for the rest of the session, and every browser on PAC silently went direct. Give `core_starting` the counterpart it was missing. `core_start_settled` re-derives PAC from the Running Mode rather than restoring a saved flag, which makes it correct on all three endings — a start that succeeded, one that was abandoned, one that failed and left the Core down — and therefore safe to run from a guard on every path. * fix(config): refuse overlapping draft transactions instead of losing an edit A layer holds exactly one draft, so a second writer's `edit_draft` overwrites the first writer's staged value in place. `DraftTransaction` nonetheless rolled back by calling `discard` on the shared layer, which discarded whatever was staged there — including a change another transaction was about to commit. The other transaction then found no draft, and `apply` on an absent draft does nothing at all. It saved the committed value, returned Ok, and its side effect had already happened: auto-launch registered with the OS while the config that describes it says otherwise, and no log line either way. "Roll back only my edit" cannot be implemented over a single slot, so enforce the exclusivity the type already assumed. Beginning a transaction claims every layer and refuses rather than interleaving, which is the same shape as `try_start_config_update` and `begin_operation` elsewhere. `patch_verge` is the caller that matters: it holds its draft across `process_terminated_flags`, which restarts the Core and can take seconds. The TUN reconciliation runs on every Run State transition and can now lose that race, so it treats a busy layer as "later" rather than as a failure worth telling the user about. * fix(core): reconcile TUN availability when the setting itself changes Moving the TUN availability guard from the frontend into Rust also moved what triggers it. The old effect depended on `enable_tun_mode`; the new reconciliation is spawned from `RealEnv::publish`, so it only runs on a Run State transition. Those two sets are not the same, and the setting itself falls in the gap. Switching TUN on from the global hotkey is the case that reaches it. That path is not gated on availability — unlike the tray item and the settings switch, both of which read `tun_capable` — and the flags a TUN patch raises are answered by a Core reload, which changes no Run State at all. So on a machine with no Service and no elevation the setting stayed on: tray icon, settings switch and `enable_tun_mode` in verge.yaml all reported TUN as enabled while nothing carried its traffic, until some unrelated core lifecycle transition happened to reconcile it. Reconcile after any patch that touched the setting. The reconciliation writes configuration itself, so it goes through the plain apply rather than back through the patch-and-reconcile form; the absence of a cycle is then a property of the call graph rather than of a runtime early-return. `toggle_tun_mode` now reads the setting back instead of returning what it asked for, so the hotkey cannot report TUN as on moments before the notice saying it was turned off. * fix(core): an action already under way is not a decision to put to the user The legacy status could not distinguish a requested privileged operation from one already running, because readers blocked until the operation finished and so never observed the request at all. Run State pushes every transition instead, and `service_needs_attention` treated any pending action as a question for the user. So an install started from the settings page or the home card raised the migration dialog on top of it, and an uninstall raised one offering to reinstall the Service being removed. Both offer a button that can only fail, since the privileged-operation slot is already taken. An operation in flight is a report of progress. The dialog the user is already driving keeps itself open through its own loading state. `tun_should_be_disabled` is unaffected: it already refused to act while an operation was in flight, so the term this adds is redundant there. * fix(core): let the snapshot decide whether the Run State has settled `settled` read the operation slot twice: once to decide whether to wait, then again inside the snapshot it returned. An operation claiming the slot between those two reads was described rather than waited for. `prepare_startup` reads this. A start racing an install would see the requested install, flatten it to `InstallRequired`, and decide `Wait` — leaving the core stopped where waiting for the install to finish would have told it whether the Service came back. Read the slot once, inside the snapshot, so there is no window rather than a second check closing it. The rewrite that introduced this also fixed a lost wakeup here, and that fix stands: the waiter is still registered before the state is read. * fix(ui): re-read the Run State once its listener is live Replacing the poll with a push left a gap. `listen` resolves asynchronously, so between the first `get_runtime_state` and the subscription being registered there is a window, and a transition landing in it is not queued anywhere — it is simply missed. The next event is whenever the backend happens to move again, which during startup can be a long time; the 30-second safety net was all that stood behind it, where the old code polled every two seconds precisely through startup. The Run State is the one thing here that arrives only by event, so it is the one thing a missed event leaves stale. Re-read it once every listener is live, which closes the gap without going back to polling for it. `onSubscribed` fires even when a registration failed, because a resync is most needed when a subscription is missing, and not at all after teardown. * docs(config): state what makes the PAC endpoint's draft read safe `MixedPort::desired` reads the draft layer, so the PAC endpoint can name a port the Core has not moved to yet. That is only safe because the one path staging a listener port closes PAC across the change. Nothing said so at either end, and the two are in different modules. * fix(core): ask again when staging goes silent rather than assume it failed The Service commits the generation before it answers, so a reply that never arrives is not evidence that nothing was written: config.yaml can already hold the new configuration while this side concludes the staging never happened. The Core keeps running the previous configuration, so nothing looks wrong -- until the Core dies and the Service's watchdog restarts it in the same generation from whatever config.yaml says, quietly making live a configuration the app reported as failed. Silence is the one answer that has to be resolved rather than acted on, so ask once more before falling back. It is cheap exactly where it matters: staging plans against the manifest it wrote, so a bundle that already landed copies nothing and answers in milliseconds, and the Service's lifecycle lock makes a second ask wait for a first one still in flight instead of interleaving with it. A deadline keeps that wait off the caller, and a second silence goes where the first one used to. --- crates/clash-verge-draft/src/lib.rs | 85 ++++++++- crates/clash-verge-draft/tests/test_me.rs | 77 +++++++- src-tauri/src/config/mixed_port.rs | 7 + src-tauri/src/config/port.rs | 2 +- src-tauri/src/constants.rs | 4 + src-tauri/src/core/manager/config.rs | 183 +++++++++++++++++-- src-tauri/src/core/manager/mod.rs | 8 + src-tauri/src/core/runstate/health.rs | 47 ++++- src-tauri/src/core/runstate/mod.rs | 101 +++++++++- src-tauri/src/feat/config.rs | 29 ++- src-tauri/src/feat/listener.rs | 13 +- src-tauri/src/feat/proxy.rs | 7 +- src-tauri/src/feat/tun.rs | 15 +- src-tauri/src/utils/server.rs | 4 +- src/pages/_layout/hooks/use-layout-events.ts | 55 +++--- src/services/events.test.ts | 106 +++++++++++ src/services/events.ts | 20 +- 17 files changed, 691 insertions(+), 72 deletions(-) create mode 100644 src/services/events.test.ts 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 () => {