chore: sync backend repairs into frontend contract flow

This commit is contained in:
Tunglies
2026-10-06 14:41:21 +08:00
6 changed files with 80 additions and 144 deletions
+20 -36
View File
@@ -2,39 +2,23 @@
These rules supplement the repository's root `AGENTS.md`.
1. **Keep configuration ownership in `config`.** Configuration writes must use
its draft and persistence APIs. Patch orchestration must use
`feat/effects.rs` and `feat/executor.rs`; refresh notifications must use
`core::notify::announce` after successful persistence and commit. This keeps
rolled-back transactions from publishing state. If runtime and memory state
are already committed and cannot be rolled back, announce that state even
when persistence fails, while preserving the persistence error. Independent
transactions retain their boundaries and announce their own committed
results; runtime-only profile activation announces after successful
application. Cross-layer operations must preserve these ownership and
transaction boundaries.
2. **Register every new `IVerge` field.** Register the field in the
exhaustive effect registry, including an explicit `NoEffect` where
appropriate, and preserve its platform attributes. Manual field-to-flag
chains hide missing effects and are forbidden.
3. **Use `utils::retry` for bounded retries.** Do not introduce handwritten retry
loops, including `loop` plus `sleep`; preserve attempt budgets, delays and
terminal errors. Permanent watchers are the exception and must document
their exit conditions. Traversal of distinct data or port candidates is not
a retry policy.
4. **Keep notifications typed.** Send notice statuses through `NoticeStatus` and
use `Arc<str>` for notice message payloads. Preserve the event names and
status strings as a shared wire contract. Contract changes must keep
frontend consumers and backend producers aligned.
5. **Preserve lock order.** The order is `config_write` before
`config_update_in_progress` before `lifecycle_lock`. Patch callers must use
the executor's guarded claim instead of independently combining these locks;
this preserves Busy rejection and prevents inverted acquisition. Existing
lifecycle and independent port transactions keep their established locking
boundaries. A new lock requires registering its position in this rule before
introducing acquisition sites.
6. **Prefer reconciliation.** Express new effects as idempotent `ensure`
operations in `EFFECT_ORDER` so repeated application converges on the same
state. A trigger chain is an exception only with a code comment explaining
why the operation cannot be idempotent. Configuration rollback does not
authorize reverse OS effects.
- **Configuration ownership.** `config` owns source configuration, drafts and
persistence. The effects layer owns patch orchestration. Rollbackable
transactions publish state after persistence and commit; already applied
runtime and independently committed state remain authoritative even when a
later step fails. Preserve that state, its notifications and the error.
- **Effect ownership.** The declarative registry classifies every Verge patch field,
including fields without effects and platform-specific fields. Effects
reconcile toward the requested state with explicit dependencies and ordering.
Configuration rollback does not imply reversal of OS or lifecycle effects.
- **Shared wire contract.** Notifications are typed. Event names, status values
and payload shapes stay compatible with frontend producers and consumers.
- **Synchronization ownership.** Configuration-write serialization precedes
configuration-update ownership, which precedes Core lifecycle ownership.
When both are held, the lifecycle lock precedes the DNS lock. Independent
transactions retain their ownership boundaries; asynchronous work does not
hold synchronous data guards across suspension.
- **Retry ownership.** Bounded retries share backend policy while preserving
attempts, delays and terminal errors. Background monitors own their lifetime
and cancellation separately; waiting for a condition or traversing distinct
candidates is not a retry policy.
+45 -2
View File
@@ -322,7 +322,7 @@ impl CoreManager {
self.validate_and_apply(transaction).await
}
/// Validates and applies the caller's transaction, committing only on success.
/// Commits the applied runtime even if restoring the system proxy subsequently fails.
async fn validate_and_apply(&self, transaction: DraftTransaction<'_>) -> Result<ValidationOutcome> {
let outcome = self.validate_and_apply_draft().await?;
if outcome.is_valid() {
@@ -342,7 +342,10 @@ impl CoreManager {
}
let run_path = Config::write_runtime_file(&yaml).await?;
self.apply_config(run_path).await?;
let previous = self.current_core_readiness_generation();
let result = self.apply_config(run_path).await;
self.retain_applied_runtime(&Config::runtime().await, previous, &result);
result?;
// Under the lifecycle lock, so a stop for exit waits for this write instead of outrunning it.
#[cfg(target_os = "macos")]
{
@@ -352,6 +355,21 @@ impl CoreManager {
Ok(ValidationOutcome::Valid)
}
fn retain_applied_runtime(
&self,
runtime: &clash_verge_draft::Draft<IRuntime>,
previous: Option<u64>,
result: &Result<()>,
) {
if result.is_ok()
|| self
.current_core_readiness_generation()
.is_some_and(|current| Some(current) != previous)
{
runtime.apply();
}
}
/// Applies a generated configuration through the active core owner.
async fn apply_config(&self, path: PathBuf) -> Result<()> {
if matches!(*self.get_running_mode(), RunningMode::Service) {
@@ -443,6 +461,31 @@ mod tests {
use clash_verge_service_ipc::{StageRejection, StageRuntimeOutcome};
use std::{cell::Cell, time::Duration};
#[test]
fn a_ready_replacement_keeps_the_applied_runtime_when_proxy_restore_fails() -> anyhow::Result<()> {
use crate::{config::runtime::IRuntime, core::CoreManager};
use clash_verge_draft::{Draft, DraftTransaction};
for replaced in [false, true] {
let manager = CoreManager::default();
manager.mark_core_ready();
let previous = manager.current_core_readiness_generation();
let runtime = Draft::new(IRuntime::default());
let transaction = DraftTransaction::begin(vec![&runtime])?;
runtime.edit_draft(|draft| {
draft.exists_keys.insert("replacement".into());
});
if replaced {
manager.mark_core_ready();
}
let result = Err(anyhow::anyhow!("system proxy restore failed"));
manager.retain_applied_runtime(&runtime, previous, &result);
transaction.rollback();
assert_eq!(runtime.data_arc().exists_keys.contains("replacement"), replaced);
}
Ok(())
}
const CONFIRM_WITHIN: Duration = Duration::from_secs(5);
fn staged() -> StageAttempt {
+3 -50
View File
@@ -1,7 +1,6 @@
use super::{handle::Handle, notification::FrontendEvent};
use anyhow::Result;
use parking_lot::Mutex;
use std::{cell::RefCell, future::Future, time::Duration};
use std::time::Duration;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Refresh {
@@ -11,10 +10,6 @@ pub enum Refresh {
Proxies,
}
tokio::task_local! {
static PENDING: RefCell<Vec<Refresh>>;
}
const REFRESH_WINDOW: Duration = Duration::from_millis(20);
static BATCH: RefreshBatch = RefreshBatch(Mutex::new(Vec::new()));
@@ -42,7 +37,7 @@ impl RefreshBatch {
}
pub fn announce(refresh: Refresh) {
if PENDING.try_with(|pending| pending.borrow_mut().push(refresh)).is_err() && BATCH.enqueue(refresh) {
if BATCH.enqueue(refresh) {
crate::AsyncHandler::spawn(|| BATCH.flush(emit));
}
}
@@ -56,54 +51,12 @@ fn emit(refresh: Refresh) {
});
}
async fn collect_committed<T: Send>(operation: impl Future<Output = Result<T>> + Send) -> Result<(T, Vec<Refresh>)> {
PENDING
.scope(RefCell::new(Vec::new()), async {
let value = operation.await?;
let refreshes = PENDING.with(|pending| pending.take());
Ok((value, refreshes))
})
.await
}
pub async fn after_commit<T: Send>(operation: impl Future<Output = Result<T>> + Send) -> Result<T> {
let (value, refreshes) = collect_committed(operation).await?;
// Nested transactions hand their notifications to the outer transaction.
for refresh in refreshes {
announce(refresh);
}
Ok(value)
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn only_successful_transactions_publish_refreshes() -> Result<()> {
let (_, refreshes) = collect_committed(async {
announce(Refresh::Clash);
after_commit(async {
announce(Refresh::Verge);
Ok(())
})
.await?;
let failed = after_commit(async {
announce(Refresh::Profiles);
Err::<(), _>(anyhow::anyhow!("persistence failed"))
})
.await;
assert!(failed.is_err());
announce(Refresh::Clash);
Ok(())
})
.await?;
assert_eq!(refreshes, [Refresh::Clash, Refresh::Verge, Refresh::Clash]);
Ok(())
}
#[tokio::test]
async fn bursts_keep_each_refresh_kind_and_rearm_during_delivery() -> Result<()> {
async fn bursts_keep_each_refresh_kind_and_rearm_during_delivery() -> anyhow::Result<()> {
let batch = RefreshBatch::default();
let delivered = Mutex::new(Vec::new());
assert!(batch.enqueue(Refresh::Clash));
+1 -1
View File
@@ -1,5 +1,5 @@
mod announce;
pub use announce::{Refresh, after_commit, announce};
pub use announce::{Refresh, announce};
pub mod desktop;
pub mod handle;
pub mod notification;
+5 -13
View File
@@ -63,8 +63,6 @@ async fn ensure_effect(
}
Effect::ClashConfig => {
manager.update_config_in_patch(update).await?;
Config::runtime().await.apply();
announce(Refresh::Clash);
}
Effect::Autostart => autostart::update_launch().await?,
Effect::Language => {
@@ -123,7 +121,7 @@ async fn ensure_effect(
Ok(())
}
async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> {
pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> {
let manager = CoreManager::global();
let update = manager.claim_config_update(config_write)?;
let clash = Config::clash().await;
@@ -167,6 +165,7 @@ async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effect
if let Err(error) = result {
transaction.rollback();
if !Arc::ptr_eq(&original_runtime, &runtime.data_arc()) {
announce(Refresh::Clash);
// The Core already loaded this runtime; restoring its file would only hide that state.
let runtime_path = crate::utils::dirs::app_home_dir()?.join(crate::constants::files::RUNTIME_CONFIG);
snapshots.retain(|snapshot| snapshot.path != runtime_path);
@@ -179,6 +178,9 @@ async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effect
transaction.commit();
match patch {
Patch::Verge { .. } => {
if effects.contains(&Effect::ClashConfig) {
announce(Refresh::Clash);
}
logging_error!(Type::Backup, AutoBackupManager::global().refresh_settings().await);
announce(Refresh::Verge);
}
@@ -187,16 +189,6 @@ async fn apply_inner(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effect
Ok(())
}
pub(super) async fn apply(config_write: &MutexGuard<'_, ()>, patch: Patch<'_>, effects: Effects) -> Result<()> {
let runtime = Config::runtime().await;
let original = runtime.data_arc();
let result = crate::core::notify::after_commit(apply_inner(config_write, patch, effects)).await;
if result.is_err() && !Arc::ptr_eq(&original, &runtime.data_arc()) {
announce(Refresh::Clash);
}
result
}
#[cfg(test)]
mod tests {
use super::*;
+6 -42
View File
@@ -1,46 +1,18 @@
use backon::{BackoffBuilder as _, ExponentialBackoff, ExponentialBuilder};
use std::{future::Future, num::NonZeroUsize, time::Duration};
#[derive(Clone, Copy, Debug)]
pub enum Backoff {
Fixed(Duration),
#[allow(dead_code)]
Exponential {
base: Duration,
cap: Duration,
},
}
#[derive(Clone, Copy, Debug)]
pub struct RetryPolicy {
pub attempts: NonZeroUsize,
pub backoff: Backoff,
pub jitter: bool,
delay: Duration,
}
impl RetryPolicy {
pub const fn fixed(attempts: NonZeroUsize, delay: Duration) -> Self {
Self {
attempts,
backoff: Backoff::Fixed(delay),
jitter: false,
}
Self { attempts, delay }
}
fn delays(self) -> ExponentialBackoff {
let (base, cap, factor) = match self.backoff {
Backoff::Fixed(delay) => (delay, delay, 1.0),
Backoff::Exponential { base, cap } => (base, cap, 2.0),
};
let mut builder = ExponentialBuilder::default()
.with_min_delay(base)
.with_max_delay(cap)
.with_factor(factor)
.with_max_times(self.attempts.get() - 1);
if self.jitter {
builder = builder.with_jitter();
}
builder.build()
fn delays(self) -> impl Iterator<Item = Duration> {
std::iter::repeat_n(self.delay, self.attempts.get() - 1)
}
}
@@ -82,11 +54,7 @@ pub async fn try_strategies<
mut operation: impl FnMut(S) -> Fut + Send,
) -> Result<(S, T), E> {
retry(
RetryPolicy {
attempts: NonZeroUsize::MIN.saturating_add(N),
backoff: Backoff::Fixed(Duration::ZERO),
jitter: false,
},
RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(N), Duration::ZERO),
move |attempt| {
let strategy = if attempt == 0 { first } else { rest[attempt - 1] };
let future = operation(strategy);
@@ -145,11 +113,7 @@ mod tests {
#[tokio::test]
async fn retries_stop_at_success_terminal_error_or_attempt_limit() {
let policy = RetryPolicy {
attempts: NonZeroUsize::MIN.saturating_add(2),
backoff: Backoff::Fixed(Duration::ZERO),
jitter: false,
};
let policy = RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(2), Duration::ZERO);
let mut calls = 0;
let success = retry(policy, |attempt| {
calls += 1;