mirror of
https://github.com/clash-verge-rev/clash-verge-rev.git
synced 2026-11-08 19:22:44 +08:00
refactor(service): simplify DNS override sync and runtime asset staging
This commit is contained in:
@@ -5,7 +5,7 @@ use crate::core::manager::CLASH_LOGGER;
|
||||
use crate::core::proxy_control::{self, SysproxyFailure};
|
||||
use crate::core::service::{SERVICE_MANAGER, ServiceStatus};
|
||||
use anyhow::Result;
|
||||
use clash_verge_logging::{Type, logging, logging_error};
|
||||
use clash_verge_logging::{Type, logging};
|
||||
use scopeguard::defer;
|
||||
use smartstring::alias::String;
|
||||
use std::path::Path;
|
||||
@@ -428,7 +428,6 @@ impl CoreManager {
|
||||
}
|
||||
// Drop the guard last so an earlier failure cannot leave a running Sidecar unguarded.
|
||||
proxy_control::stop_guard().await;
|
||||
let starts_core = matches!(*self.get_running_mode(), RunningMode::NotRunning);
|
||||
let result = async {
|
||||
self.start_core_inner().await?;
|
||||
if !matches!(*self.get_running_mode(), RunningMode::Sidecar)
|
||||
@@ -437,7 +436,7 @@ impl CoreManager {
|
||||
anyhow::bail!("Sidecar did not become ready");
|
||||
}
|
||||
// A Sidecar that was already running never loaded this generation.
|
||||
if starts_core {
|
||||
if matches!(*mode, RunningMode::NotRunning) {
|
||||
Config::runtime().await.apply();
|
||||
}
|
||||
tracing::Span::current().record("readiness_generation", self.current_core_readiness_generation());
|
||||
@@ -456,12 +455,7 @@ impl CoreManager {
|
||||
Ok(())
|
||||
}
|
||||
.await;
|
||||
let result = result.map_err(|error| report_sidecar_failure(error, service_rejection));
|
||||
// Lock order is config then lifecycle.
|
||||
drop(_life);
|
||||
// A failure after the commit still leaves that runtime for the Sidecar or the next start.
|
||||
logging_error!(Type::Config, Config::sync_dns_override().await);
|
||||
result
|
||||
result.map_err(|error| report_sidecar_failure(error, service_rejection))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, level = "info", fields(readiness_generation = tracing::field::Empty))]
|
||||
@@ -473,7 +467,7 @@ impl CoreManager {
|
||||
self.finish_config_update();
|
||||
}
|
||||
// Lock order is config then lifecycle, for the unclaimed draft write below.
|
||||
let config_write = Config::lock_config_write().await;
|
||||
let _config_write = Config::lock_config_write().await;
|
||||
let _life = self.lifecycle_lock.lock().await;
|
||||
|
||||
self.controlled_stop_core_inner().await?;
|
||||
@@ -509,15 +503,8 @@ impl CoreManager {
|
||||
)
|
||||
.await;
|
||||
|
||||
let cleanup = if result.is_err() {
|
||||
self.rollback_failed_sidecar_transition().await
|
||||
} else {
|
||||
Ok(())
|
||||
};
|
||||
// A failure after the commit still leaves that runtime for the Sidecar or the next start.
|
||||
logging_error!(Type::Config, Config::sync_dns_override_locked(&config_write).await);
|
||||
if let Err(error) = result {
|
||||
if let Err(cleanup_error) = cleanup {
|
||||
if let Err(cleanup_error) = self.rollback_failed_sidecar_transition().await {
|
||||
return Err(proxy_control::rollback_failure(error, cleanup_error)
|
||||
.context("failed to clear the proxy before rolling back"));
|
||||
}
|
||||
|
||||
@@ -274,6 +274,10 @@ impl CoreManager {
|
||||
|
||||
pub(crate) fn finish_config_update(&self) {
|
||||
self.config_update_in_progress.store(false, Ordering::Release);
|
||||
// Spawned because the caller may still hold the config write lock this waits for.
|
||||
crate::process::AsyncHandler::spawn(|| async {
|
||||
logging_error!(Type::Config, crate::config::Config::sync_dns_override().await);
|
||||
});
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, level = "info", fields(retries = 0))]
|
||||
@@ -303,7 +307,6 @@ impl CoreManager {
|
||||
}
|
||||
match crate::config::Config::resolve_startup_mixed_port().await {
|
||||
Ok(true) => {
|
||||
logging_error!(Type::Config, crate::config::Config::sync_dns_override().await);
|
||||
let retries = retries + 1;
|
||||
tracing::Span::current().record("retries", retries);
|
||||
logging!(
|
||||
|
||||
@@ -3,7 +3,7 @@ use clash_verge_logging::{Type, logging};
|
||||
use clash_verge_service_ipc::{RemoteProvider, RuntimeAsset, RuntimeBundle};
|
||||
use serde_yaml_ng::{Mapping, Value};
|
||||
use sha2::{Digest as _, Sha256};
|
||||
use std::collections::{BTreeMap, HashMap, HashSet};
|
||||
use std::collections::{BTreeMap, HashSet};
|
||||
use std::path::{Component, Path, PathBuf, Prefix};
|
||||
|
||||
pub(crate) const GEO_ASSETS: &[&str] = &[
|
||||
@@ -297,7 +297,6 @@ fn collect_ssh_private_keys(
|
||||
}
|
||||
}
|
||||
|
||||
let mut staged = HashMap::new();
|
||||
for proxy in proxy_lists.into_iter().flatten().filter_map(Value::as_mapping_mut) {
|
||||
if proxy.get("type").and_then(Value::as_str) != Some("ssh") {
|
||||
continue;
|
||||
@@ -323,17 +322,12 @@ fn collect_ssh_private_keys(
|
||||
continue;
|
||||
};
|
||||
// A root-level alias keeps the user's path away from service name rules and mihomo's inline-key test.
|
||||
let destination = staged
|
||||
.entry(source.to_string_lossy().into_owned())
|
||||
.or_insert_with_key(|source| {
|
||||
let destination = allocate_provider_destination("ssh-private-key", source, destinations);
|
||||
assets.push(RuntimeAsset {
|
||||
source: source.clone(),
|
||||
destination: destination.clone(),
|
||||
});
|
||||
destination
|
||||
})
|
||||
.clone();
|
||||
let source = source.to_string_lossy().into_owned();
|
||||
let destination = allocate_provider_destination("ssh-private-key", &source, destinations);
|
||||
assets.push(RuntimeAsset {
|
||||
source,
|
||||
destination: destination.clone(),
|
||||
});
|
||||
proxy.insert(Value::String("private-key".to_owned()), Value::String(destination));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1248,36 +1248,29 @@ async fn sync_runtime_geo(
|
||||
let Some((Some(_), _, Some(runtime_mtime_ns))) = read_chunk(credentials, session, asset, u64::MAX).await? else {
|
||||
return Ok(());
|
||||
};
|
||||
let local_mtime_ns = local
|
||||
.modified()?
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map_or(0, |since_epoch| since_epoch.as_nanos());
|
||||
// A newer local file is a change the next staging still has to copy into the runtime.
|
||||
if u128::from(runtime_mtime_ns) <= local_mtime_ns {
|
||||
if UNIX_EPOCH + Duration::from_nanos(runtime_mtime_ns) <= local.modified()? {
|
||||
return Ok(());
|
||||
}
|
||||
let (temp, file) = create_sync_temp(target).await?;
|
||||
let published = async {
|
||||
let Fetch::Complete {
|
||||
let result = async {
|
||||
if let Fetch::Complete {
|
||||
mtime_ns: Some(mtime_ns),
|
||||
..
|
||||
} = fetch_runtime_file(credentials, session, asset, file).await?
|
||||
else {
|
||||
return Ok(false);
|
||||
};
|
||||
// Matching the runtime copy's time lets the service skip copying these bytes straight back.
|
||||
std::fs::File::options()
|
||||
.write(true)
|
||||
.open(&temp)?
|
||||
.set_modified(UNIX_EPOCH + Duration::from_nanos(mtime_ns))?;
|
||||
tokio::fs::rename(&temp, target).await?;
|
||||
anyhow::Ok(true)
|
||||
{
|
||||
// Matching the runtime copy's time lets the service skip copying these bytes straight back.
|
||||
std::fs::File::options()
|
||||
.write(true)
|
||||
.open(&temp)?
|
||||
.set_modified(UNIX_EPOCH + Duration::from_nanos(mtime_ns))?;
|
||||
tokio::fs::rename(&temp, target).await?;
|
||||
}
|
||||
anyhow::Ok(())
|
||||
}
|
||||
.await;
|
||||
if !matches!(published, Ok(true)) {
|
||||
remove_temp(&temp).await;
|
||||
}
|
||||
published.map(|_| ())
|
||||
remove_temp(&temp).await;
|
||||
result
|
||||
}
|
||||
|
||||
// The on-disk YAML may describe a rejected apply; use the committed runtime.
|
||||
|
||||
Reference in New Issue
Block a user