mirror of https://github.com/telemt/telemt.git
New Hot-Reload method
This commit is contained in:
parent
f7e3ddcdb6
commit
8b0b47145d
|
|
@ -21,9 +21,11 @@
|
||||||
//! `network.*`, `use_middle_proxy`) are **not** applied; a warning is emitted.
|
//! `network.*`, `use_middle_proxy`) are **not** applied; a warning is emitted.
|
||||||
//! Non-hot changes are never mixed into the runtime config snapshot.
|
//! Non-hot changes are never mixed into the runtime config snapshot.
|
||||||
|
|
||||||
|
use std::collections::BTreeSet;
|
||||||
use std::net::IpAddr;
|
use std::net::IpAddr;
|
||||||
use std::path::PathBuf;
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::Arc;
|
use std::sync::{Arc, RwLock as StdRwLock};
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use notify::{EventKind, RecursiveMode, Watcher, recommended_watcher};
|
use notify::{EventKind, RecursiveMode, Watcher, recommended_watcher};
|
||||||
use tokio::sync::{mpsc, watch};
|
use tokio::sync::{mpsc, watch};
|
||||||
|
|
@ -33,7 +35,10 @@ use crate::config::{
|
||||||
LogLevel, MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy, MeTelemetryLevel,
|
LogLevel, MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy, MeTelemetryLevel,
|
||||||
MeWriterPickMode,
|
MeWriterPickMode,
|
||||||
};
|
};
|
||||||
use super::load::ProxyConfig;
|
use super::load::{LoadedConfig, ProxyConfig};
|
||||||
|
|
||||||
|
const HOT_RELOAD_STABLE_SNAPSHOTS: u8 = 2;
|
||||||
|
const HOT_RELOAD_DEBOUNCE: Duration = Duration::from_millis(50);
|
||||||
|
|
||||||
// ── Hot fields ────────────────────────────────────────────────────────────────
|
// ── Hot fields ────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
@ -287,6 +292,149 @@ fn listeners_equal(
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||||
|
struct WatchManifest {
|
||||||
|
files: BTreeSet<PathBuf>,
|
||||||
|
dirs: BTreeSet<PathBuf>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WatchManifest {
|
||||||
|
fn from_source_files(source_files: &[PathBuf]) -> Self {
|
||||||
|
let mut files = BTreeSet::new();
|
||||||
|
let mut dirs = BTreeSet::new();
|
||||||
|
|
||||||
|
for path in source_files {
|
||||||
|
let normalized = normalize_watch_path(path);
|
||||||
|
files.insert(normalized.clone());
|
||||||
|
if let Some(parent) = normalized.parent() {
|
||||||
|
dirs.insert(parent.to_path_buf());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Self { files, dirs }
|
||||||
|
}
|
||||||
|
|
||||||
|
fn matches_event_paths(&self, event_paths: &[PathBuf]) -> bool {
|
||||||
|
event_paths
|
||||||
|
.iter()
|
||||||
|
.map(|path| normalize_watch_path(path))
|
||||||
|
.any(|path| self.files.contains(&path))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Default)]
|
||||||
|
struct ReloadState {
|
||||||
|
applied_snapshot_hash: Option<u64>,
|
||||||
|
candidate_snapshot_hash: Option<u64>,
|
||||||
|
candidate_hits: u8,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ReloadState {
|
||||||
|
fn new(applied_snapshot_hash: Option<u64>) -> Self {
|
||||||
|
Self {
|
||||||
|
applied_snapshot_hash,
|
||||||
|
candidate_snapshot_hash: None,
|
||||||
|
candidate_hits: 0,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_applied(&self, hash: u64) -> bool {
|
||||||
|
self.applied_snapshot_hash == Some(hash)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn observe_candidate(&mut self, hash: u64) -> u8 {
|
||||||
|
if self.candidate_snapshot_hash == Some(hash) {
|
||||||
|
self.candidate_hits = self.candidate_hits.saturating_add(1);
|
||||||
|
} else {
|
||||||
|
self.candidate_snapshot_hash = Some(hash);
|
||||||
|
self.candidate_hits = 1;
|
||||||
|
}
|
||||||
|
self.candidate_hits
|
||||||
|
}
|
||||||
|
|
||||||
|
fn reset_candidate(&mut self) {
|
||||||
|
self.candidate_snapshot_hash = None;
|
||||||
|
self.candidate_hits = 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
fn mark_applied(&mut self, hash: u64) {
|
||||||
|
self.applied_snapshot_hash = Some(hash);
|
||||||
|
self.reset_candidate();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn normalize_watch_path(path: &Path) -> PathBuf {
|
||||||
|
path.canonicalize().unwrap_or_else(|_| {
|
||||||
|
if path.is_absolute() {
|
||||||
|
path.to_path_buf()
|
||||||
|
} else {
|
||||||
|
std::env::current_dir()
|
||||||
|
.map(|cwd| cwd.join(path))
|
||||||
|
.unwrap_or_else(|_| path.to_path_buf())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn sync_watch_paths<W: Watcher>(
|
||||||
|
watcher: &mut W,
|
||||||
|
current: &BTreeSet<PathBuf>,
|
||||||
|
next: &BTreeSet<PathBuf>,
|
||||||
|
recursive_mode: RecursiveMode,
|
||||||
|
kind: &str,
|
||||||
|
) {
|
||||||
|
for path in current.difference(next) {
|
||||||
|
if let Err(e) = watcher.unwatch(path) {
|
||||||
|
warn!(path = %path.display(), error = %e, "config watcher: failed to unwatch {kind}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for path in next.difference(current) {
|
||||||
|
if let Err(e) = watcher.watch(path, recursive_mode) {
|
||||||
|
warn!(path = %path.display(), error = %e, "config watcher: failed to watch {kind}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn apply_watch_manifest<W1: Watcher, W2: Watcher>(
|
||||||
|
notify_watcher: Option<&mut W1>,
|
||||||
|
poll_watcher: Option<&mut W2>,
|
||||||
|
manifest_state: &Arc<StdRwLock<WatchManifest>>,
|
||||||
|
next_manifest: WatchManifest,
|
||||||
|
) {
|
||||||
|
let current_manifest = manifest_state
|
||||||
|
.read()
|
||||||
|
.map(|manifest| manifest.clone())
|
||||||
|
.unwrap_or_default();
|
||||||
|
|
||||||
|
if current_manifest == next_manifest {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(watcher) = notify_watcher {
|
||||||
|
sync_watch_paths(
|
||||||
|
watcher,
|
||||||
|
¤t_manifest.dirs,
|
||||||
|
&next_manifest.dirs,
|
||||||
|
RecursiveMode::NonRecursive,
|
||||||
|
"config directory",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(watcher) = poll_watcher {
|
||||||
|
sync_watch_paths(
|
||||||
|
watcher,
|
||||||
|
¤t_manifest.files,
|
||||||
|
&next_manifest.files,
|
||||||
|
RecursiveMode::NonRecursive,
|
||||||
|
"config file",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Ok(mut manifest) = manifest_state.write() {
|
||||||
|
*manifest = next_manifest;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig {
|
fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig {
|
||||||
let mut cfg = old.clone();
|
let mut cfg = old.clone();
|
||||||
|
|
||||||
|
|
@ -970,18 +1118,42 @@ fn reload_config(
|
||||||
log_tx: &watch::Sender<LogLevel>,
|
log_tx: &watch::Sender<LogLevel>,
|
||||||
detected_ip_v4: Option<IpAddr>,
|
detected_ip_v4: Option<IpAddr>,
|
||||||
detected_ip_v6: Option<IpAddr>,
|
detected_ip_v6: Option<IpAddr>,
|
||||||
) {
|
reload_state: &mut ReloadState,
|
||||||
let new_cfg = match ProxyConfig::load(config_path) {
|
) -> Option<WatchManifest> {
|
||||||
Ok(c) => c,
|
let loaded = match ProxyConfig::load_with_metadata(config_path) {
|
||||||
|
Ok(loaded) => loaded,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
reload_state.reset_candidate();
|
||||||
error!("config reload: failed to parse {:?}: {}", config_path, e);
|
error!("config reload: failed to parse {:?}: {}", config_path, e);
|
||||||
return;
|
return None;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
let LoadedConfig {
|
||||||
|
config: new_cfg,
|
||||||
|
source_files,
|
||||||
|
rendered_hash,
|
||||||
|
} = loaded;
|
||||||
|
let next_manifest = WatchManifest::from_source_files(&source_files);
|
||||||
|
|
||||||
if let Err(e) = new_cfg.validate() {
|
if let Err(e) = new_cfg.validate() {
|
||||||
|
reload_state.reset_candidate();
|
||||||
error!("config reload: validation failed: {}; keeping old config", e);
|
error!("config reload: validation failed: {}; keeping old config", e);
|
||||||
return;
|
return Some(next_manifest);
|
||||||
|
}
|
||||||
|
|
||||||
|
if reload_state.is_applied(rendered_hash) {
|
||||||
|
return Some(next_manifest);
|
||||||
|
}
|
||||||
|
|
||||||
|
let candidate_hits = reload_state.observe_candidate(rendered_hash);
|
||||||
|
if candidate_hits < HOT_RELOAD_STABLE_SNAPSHOTS {
|
||||||
|
info!(
|
||||||
|
snapshot_hash = rendered_hash,
|
||||||
|
candidate_hits,
|
||||||
|
required_hits = HOT_RELOAD_STABLE_SNAPSHOTS,
|
||||||
|
"config reload: candidate snapshot observed but not stable yet"
|
||||||
|
);
|
||||||
|
return Some(next_manifest);
|
||||||
}
|
}
|
||||||
|
|
||||||
let old_cfg = config_tx.borrow().clone();
|
let old_cfg = config_tx.borrow().clone();
|
||||||
|
|
@ -996,17 +1168,19 @@ fn reload_config(
|
||||||
}
|
}
|
||||||
|
|
||||||
if !hot_changed {
|
if !hot_changed {
|
||||||
return;
|
reload_state.mark_applied(rendered_hash);
|
||||||
|
return Some(next_manifest);
|
||||||
}
|
}
|
||||||
|
|
||||||
if old_hot.dns_overrides != applied_hot.dns_overrides
|
if old_hot.dns_overrides != applied_hot.dns_overrides
|
||||||
&& let Err(e) = crate::network::dns_overrides::install_entries(&applied_hot.dns_overrides)
|
&& let Err(e) = crate::network::dns_overrides::install_entries(&applied_hot.dns_overrides)
|
||||||
{
|
{
|
||||||
|
reload_state.reset_candidate();
|
||||||
error!(
|
error!(
|
||||||
"config reload: invalid network.dns_overrides: {}; keeping old config",
|
"config reload: invalid network.dns_overrides: {}; keeping old config",
|
||||||
e
|
e
|
||||||
);
|
);
|
||||||
return;
|
return Some(next_manifest);
|
||||||
}
|
}
|
||||||
|
|
||||||
log_changes(
|
log_changes(
|
||||||
|
|
@ -1018,6 +1192,8 @@ fn reload_config(
|
||||||
detected_ip_v6,
|
detected_ip_v6,
|
||||||
);
|
);
|
||||||
config_tx.send(Arc::new(applied_cfg)).ok();
|
config_tx.send(Arc::new(applied_cfg)).ok();
|
||||||
|
reload_state.mark_applied(rendered_hash);
|
||||||
|
Some(next_manifest)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── Public API ────────────────────────────────────────────────────────────────
|
// ── Public API ────────────────────────────────────────────────────────────────
|
||||||
|
|
@ -1040,80 +1216,86 @@ pub fn spawn_config_watcher(
|
||||||
let (config_tx, config_rx) = watch::channel(initial);
|
let (config_tx, config_rx) = watch::channel(initial);
|
||||||
let (log_tx, log_rx) = watch::channel(initial_level);
|
let (log_tx, log_rx) = watch::channel(initial_level);
|
||||||
|
|
||||||
// Bridge: sync notify callbacks → async task via mpsc.
|
let config_path = normalize_watch_path(&config_path);
|
||||||
let (notify_tx, mut notify_rx) = mpsc::channel::<()>(4);
|
let initial_loaded = ProxyConfig::load_with_metadata(&config_path).ok();
|
||||||
|
let initial_manifest = initial_loaded
|
||||||
|
.as_ref()
|
||||||
|
.map(|loaded| WatchManifest::from_source_files(&loaded.source_files))
|
||||||
|
.unwrap_or_else(|| WatchManifest::from_source_files(std::slice::from_ref(&config_path)));
|
||||||
|
let initial_snapshot_hash = initial_loaded.as_ref().map(|loaded| loaded.rendered_hash);
|
||||||
|
|
||||||
// Canonicalize so path matches what notify returns (absolute) in events.
|
|
||||||
let config_path = match config_path.canonicalize() {
|
|
||||||
Ok(p) => p,
|
|
||||||
Err(_) => config_path.to_path_buf(),
|
|
||||||
};
|
|
||||||
|
|
||||||
// Watch the parent directory rather than the file itself, because many
|
|
||||||
// editors (vim, nano) and systemd write via rename, which would cause
|
|
||||||
// inotify to lose track of the original inode.
|
|
||||||
let watch_dir = config_path
|
|
||||||
.parent()
|
|
||||||
.unwrap_or_else(|| std::path::Path::new("."))
|
|
||||||
.to_path_buf();
|
|
||||||
|
|
||||||
// ── inotify watcher (instant on local fs) ────────────────────────────
|
|
||||||
let config_file = config_path.clone();
|
|
||||||
let tx_inotify = notify_tx.clone();
|
|
||||||
let inotify_ok = match recommended_watcher(move |res: notify::Result<notify::Event>| {
|
|
||||||
let Ok(event) = res else { return };
|
|
||||||
let is_our_file = event.paths.iter().any(|p| p == &config_file);
|
|
||||||
if !is_our_file { return; }
|
|
||||||
if matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)) {
|
|
||||||
let _ = tx_inotify.try_send(());
|
|
||||||
}
|
|
||||||
}) {
|
|
||||||
Ok(mut w) => match w.watch(&watch_dir, RecursiveMode::NonRecursive) {
|
|
||||||
Ok(()) => {
|
|
||||||
info!("config watcher: inotify active on {:?}", config_path);
|
|
||||||
Box::leak(Box::new(w));
|
|
||||||
true
|
|
||||||
}
|
|
||||||
Err(e) => { warn!("config watcher: inotify watch failed: {}", e); false }
|
|
||||||
},
|
|
||||||
Err(e) => { warn!("config watcher: inotify unavailable: {}", e); false }
|
|
||||||
};
|
|
||||||
|
|
||||||
// ── poll watcher (always active, fixes Docker bind mounts / NFS) ─────
|
|
||||||
// inotify does not receive events for files mounted from the host into
|
|
||||||
// a container. PollWatcher compares file contents every 3 s and fires
|
|
||||||
// on any change regardless of the underlying fs.
|
|
||||||
let config_file2 = config_path.clone();
|
|
||||||
let tx_poll = notify_tx.clone();
|
|
||||||
match notify::poll::PollWatcher::new(
|
|
||||||
move |res: notify::Result<notify::Event>| {
|
|
||||||
let Ok(event) = res else { return };
|
|
||||||
let is_our_file = event.paths.iter().any(|p| p == &config_file2);
|
|
||||||
if !is_our_file { return; }
|
|
||||||
if matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)) {
|
|
||||||
let _ = tx_poll.try_send(());
|
|
||||||
}
|
|
||||||
},
|
|
||||||
notify::Config::default()
|
|
||||||
.with_poll_interval(std::time::Duration::from_secs(3))
|
|
||||||
.with_compare_contents(true),
|
|
||||||
) {
|
|
||||||
Ok(mut w) => match w.watch(&config_path, RecursiveMode::NonRecursive) {
|
|
||||||
Ok(()) => {
|
|
||||||
if inotify_ok {
|
|
||||||
info!("config watcher: poll watcher also active (Docker/NFS safe)");
|
|
||||||
} else {
|
|
||||||
info!("config watcher: poll watcher active on {:?} (3s interval)", config_path);
|
|
||||||
}
|
|
||||||
Box::leak(Box::new(w));
|
|
||||||
}
|
|
||||||
Err(e) => warn!("config watcher: poll watch failed: {}", e),
|
|
||||||
},
|
|
||||||
Err(e) => warn!("config watcher: poll watcher unavailable: {}", e),
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── event loop ───────────────────────────────────────────────────────
|
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
|
let (notify_tx, mut notify_rx) = mpsc::channel::<()>(4);
|
||||||
|
let manifest_state = Arc::new(StdRwLock::new(WatchManifest::default()));
|
||||||
|
let mut reload_state = ReloadState::new(initial_snapshot_hash);
|
||||||
|
|
||||||
|
let tx_inotify = notify_tx.clone();
|
||||||
|
let manifest_for_inotify = manifest_state.clone();
|
||||||
|
let mut inotify_watcher = match recommended_watcher(move |res: notify::Result<notify::Event>| {
|
||||||
|
let Ok(event) = res else { return };
|
||||||
|
if !matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let is_our_file = manifest_for_inotify
|
||||||
|
.read()
|
||||||
|
.map(|manifest| manifest.matches_event_paths(&event.paths))
|
||||||
|
.unwrap_or(false);
|
||||||
|
if is_our_file {
|
||||||
|
let _ = tx_inotify.try_send(());
|
||||||
|
}
|
||||||
|
}) {
|
||||||
|
Ok(watcher) => Some(watcher),
|
||||||
|
Err(e) => {
|
||||||
|
warn!("config watcher: inotify unavailable: {}", e);
|
||||||
|
None
|
||||||
|
}
|
||||||
|
};
|
||||||
|
apply_watch_manifest(
|
||||||
|
inotify_watcher.as_mut(),
|
||||||
|
Option::<&mut notify::poll::PollWatcher>::None,
|
||||||
|
&manifest_state,
|
||||||
|
initial_manifest.clone(),
|
||||||
|
);
|
||||||
|
if inotify_watcher.is_some() {
|
||||||
|
info!("config watcher: inotify active on {:?}", config_path);
|
||||||
|
}
|
||||||
|
|
||||||
|
let tx_poll = notify_tx.clone();
|
||||||
|
let manifest_for_poll = manifest_state.clone();
|
||||||
|
let mut poll_watcher = match notify::poll::PollWatcher::new(
|
||||||
|
move |res: notify::Result<notify::Event>| {
|
||||||
|
let Ok(event) = res else { return };
|
||||||
|
if !matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let is_our_file = manifest_for_poll
|
||||||
|
.read()
|
||||||
|
.map(|manifest| manifest.matches_event_paths(&event.paths))
|
||||||
|
.unwrap_or(false);
|
||||||
|
if is_our_file {
|
||||||
|
let _ = tx_poll.try_send(());
|
||||||
|
}
|
||||||
|
},
|
||||||
|
notify::Config::default()
|
||||||
|
.with_poll_interval(Duration::from_secs(3))
|
||||||
|
.with_compare_contents(true),
|
||||||
|
) {
|
||||||
|
Ok(watcher) => Some(watcher),
|
||||||
|
Err(e) => {
|
||||||
|
warn!("config watcher: poll watcher unavailable: {}", e);
|
||||||
|
None
|
||||||
|
}
|
||||||
|
};
|
||||||
|
apply_watch_manifest(
|
||||||
|
Option::<&mut notify::RecommendedWatcher>::None,
|
||||||
|
poll_watcher.as_mut(),
|
||||||
|
&manifest_state,
|
||||||
|
initial_manifest.clone(),
|
||||||
|
);
|
||||||
|
if poll_watcher.is_some() {
|
||||||
|
info!("config watcher: poll watcher active (Docker/NFS safe)");
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
let mut sighup = {
|
let mut sighup = {
|
||||||
use tokio::signal::unix::{SignalKind, signal};
|
use tokio::signal::unix::{SignalKind, signal};
|
||||||
|
|
@ -1133,11 +1315,25 @@ pub fn spawn_config_watcher(
|
||||||
#[cfg(not(unix))]
|
#[cfg(not(unix))]
|
||||||
if notify_rx.recv().await.is_none() { break; }
|
if notify_rx.recv().await.is_none() { break; }
|
||||||
|
|
||||||
// Debounce: drain extra events that arrive within 50 ms.
|
// Debounce: drain extra events that arrive within a short quiet window.
|
||||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
tokio::time::sleep(HOT_RELOAD_DEBOUNCE).await;
|
||||||
while notify_rx.try_recv().is_ok() {}
|
while notify_rx.try_recv().is_ok() {}
|
||||||
|
|
||||||
reload_config(&config_path, &config_tx, &log_tx, detected_ip_v4, detected_ip_v6);
|
if let Some(next_manifest) = reload_config(
|
||||||
|
&config_path,
|
||||||
|
&config_tx,
|
||||||
|
&log_tx,
|
||||||
|
detected_ip_v4,
|
||||||
|
detected_ip_v6,
|
||||||
|
&mut reload_state,
|
||||||
|
) {
|
||||||
|
apply_watch_manifest(
|
||||||
|
inotify_watcher.as_mut(),
|
||||||
|
poll_watcher.as_mut(),
|
||||||
|
&manifest_state,
|
||||||
|
next_manifest,
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|
@ -1152,6 +1348,40 @@ mod tests {
|
||||||
ProxyConfig::default()
|
ProxyConfig::default()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn write_reload_config(path: &Path, ad_tag: Option<&str>, server_port: Option<u16>) {
|
||||||
|
let mut config = String::from(
|
||||||
|
r#"
|
||||||
|
[censorship]
|
||||||
|
tls_domain = "example.com"
|
||||||
|
|
||||||
|
[access.users]
|
||||||
|
user = "00000000000000000000000000000000"
|
||||||
|
"#,
|
||||||
|
);
|
||||||
|
|
||||||
|
if ad_tag.is_some() {
|
||||||
|
config.push_str("\n[general]\n");
|
||||||
|
if let Some(tag) = ad_tag {
|
||||||
|
config.push_str(&format!("ad_tag = \"{tag}\"\n"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(port) = server_port {
|
||||||
|
config.push_str("\n[server]\n");
|
||||||
|
config.push_str(&format!("port = {port}\n"));
|
||||||
|
}
|
||||||
|
|
||||||
|
std::fs::write(path, config).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
fn temp_config_path(prefix: &str) -> PathBuf {
|
||||||
|
let nonce = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.unwrap()
|
||||||
|
.as_nanos();
|
||||||
|
std::env::temp_dir().join(format!("{prefix}_{nonce}.toml"))
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn overlay_applies_hot_and_preserves_non_hot() {
|
fn overlay_applies_hot_and_preserves_non_hot() {
|
||||||
let old = sample_config();
|
let old = sample_config();
|
||||||
|
|
@ -1219,4 +1449,61 @@ mod tests {
|
||||||
assert_eq!(applied.general.use_middle_proxy, old.general.use_middle_proxy);
|
assert_eq!(applied.general.use_middle_proxy, old.general.use_middle_proxy);
|
||||||
assert!(!config_equal(&applied, &new));
|
assert!(!config_equal(&applied, &new));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reload_requires_stable_snapshot_before_hot_apply() {
|
||||||
|
let initial_tag = "11111111111111111111111111111111";
|
||||||
|
let final_tag = "22222222222222222222222222222222";
|
||||||
|
let path = temp_config_path("telemt_hot_reload_stable");
|
||||||
|
|
||||||
|
write_reload_config(&path, Some(initial_tag), None);
|
||||||
|
let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap());
|
||||||
|
let initial_hash = ProxyConfig::load_with_metadata(&path).unwrap().rendered_hash;
|
||||||
|
let (config_tx, _config_rx) = watch::channel(initial_cfg.clone());
|
||||||
|
let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone());
|
||||||
|
let mut reload_state = ReloadState::new(Some(initial_hash));
|
||||||
|
|
||||||
|
write_reload_config(&path, None, None);
|
||||||
|
reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
config_tx.borrow().general.ad_tag.as_deref(),
|
||||||
|
Some(initial_tag)
|
||||||
|
);
|
||||||
|
|
||||||
|
write_reload_config(&path, Some(final_tag), None);
|
||||||
|
reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
config_tx.borrow().general.ad_tag.as_deref(),
|
||||||
|
Some(initial_tag)
|
||||||
|
);
|
||||||
|
|
||||||
|
reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap();
|
||||||
|
assert_eq!(config_tx.borrow().general.ad_tag.as_deref(), Some(final_tag));
|
||||||
|
|
||||||
|
let _ = std::fs::remove_file(path);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reload_keeps_hot_apply_when_non_hot_fields_change() {
|
||||||
|
let initial_tag = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
|
||||||
|
let final_tag = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
|
||||||
|
let path = temp_config_path("telemt_hot_reload_mixed");
|
||||||
|
|
||||||
|
write_reload_config(&path, Some(initial_tag), None);
|
||||||
|
let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap());
|
||||||
|
let initial_hash = ProxyConfig::load_with_metadata(&path).unwrap().rendered_hash;
|
||||||
|
let (config_tx, _config_rx) = watch::channel(initial_cfg.clone());
|
||||||
|
let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone());
|
||||||
|
let mut reload_state = ReloadState::new(Some(initial_hash));
|
||||||
|
|
||||||
|
write_reload_config(&path, Some(final_tag), Some(initial_cfg.server.port + 1));
|
||||||
|
reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap();
|
||||||
|
reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap();
|
||||||
|
|
||||||
|
let applied = config_tx.borrow().clone();
|
||||||
|
assert_eq!(applied.general.ad_tag.as_deref(), Some(final_tag));
|
||||||
|
assert_eq!(applied.server.port, initial_cfg.server.port);
|
||||||
|
|
||||||
|
let _ = std::fs::remove_file(path);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,9 @@
|
||||||
#![allow(deprecated)]
|
#![allow(deprecated)]
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::{BTreeSet, HashMap};
|
||||||
|
use std::hash::{DefaultHasher, Hash, Hasher};
|
||||||
use std::net::{IpAddr, SocketAddr};
|
use std::net::{IpAddr, SocketAddr};
|
||||||
use std::path::Path;
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use rand::Rng;
|
use rand::Rng;
|
||||||
use tracing::warn;
|
use tracing::warn;
|
||||||
|
|
@ -13,7 +14,37 @@ use crate::error::{ProxyError, Result};
|
||||||
use super::defaults::*;
|
use super::defaults::*;
|
||||||
use super::types::*;
|
use super::types::*;
|
||||||
|
|
||||||
fn preprocess_includes(content: &str, base_dir: &Path, depth: u8) -> Result<String> {
|
#[derive(Debug, Clone)]
|
||||||
|
pub(crate) struct LoadedConfig {
|
||||||
|
pub(crate) config: ProxyConfig,
|
||||||
|
pub(crate) source_files: Vec<PathBuf>,
|
||||||
|
pub(crate) rendered_hash: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
fn normalize_config_path(path: &Path) -> PathBuf {
|
||||||
|
path.canonicalize().unwrap_or_else(|_| {
|
||||||
|
if path.is_absolute() {
|
||||||
|
path.to_path_buf()
|
||||||
|
} else {
|
||||||
|
std::env::current_dir()
|
||||||
|
.map(|cwd| cwd.join(path))
|
||||||
|
.unwrap_or_else(|_| path.to_path_buf())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn hash_rendered_snapshot(rendered: &str) -> u64 {
|
||||||
|
let mut hasher = DefaultHasher::new();
|
||||||
|
rendered.hash(&mut hasher);
|
||||||
|
hasher.finish()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn preprocess_includes(
|
||||||
|
content: &str,
|
||||||
|
base_dir: &Path,
|
||||||
|
depth: u8,
|
||||||
|
source_files: &mut BTreeSet<PathBuf>,
|
||||||
|
) -> Result<String> {
|
||||||
if depth > 10 {
|
if depth > 10 {
|
||||||
return Err(ProxyError::Config("Include depth > 10".into()));
|
return Err(ProxyError::Config("Include depth > 10".into()));
|
||||||
}
|
}
|
||||||
|
|
@ -25,10 +56,16 @@ fn preprocess_includes(content: &str, base_dir: &Path, depth: u8) -> Result<Stri
|
||||||
if let Some(rest) = rest.strip_prefix('=') {
|
if let Some(rest) = rest.strip_prefix('=') {
|
||||||
let path_str = rest.trim().trim_matches('"');
|
let path_str = rest.trim().trim_matches('"');
|
||||||
let resolved = base_dir.join(path_str);
|
let resolved = base_dir.join(path_str);
|
||||||
|
source_files.insert(normalize_config_path(&resolved));
|
||||||
let included = std::fs::read_to_string(&resolved)
|
let included = std::fs::read_to_string(&resolved)
|
||||||
.map_err(|e| ProxyError::Config(e.to_string()))?;
|
.map_err(|e| ProxyError::Config(e.to_string()))?;
|
||||||
let included_dir = resolved.parent().unwrap_or(base_dir);
|
let included_dir = resolved.parent().unwrap_or(base_dir);
|
||||||
output.push_str(&preprocess_includes(&included, included_dir, depth + 1)?);
|
output.push_str(&preprocess_includes(
|
||||||
|
&included,
|
||||||
|
included_dir,
|
||||||
|
depth + 1,
|
||||||
|
source_files,
|
||||||
|
)?);
|
||||||
output.push('\n');
|
output.push('\n');
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
@ -138,10 +175,16 @@ pub struct ProxyConfig {
|
||||||
|
|
||||||
impl ProxyConfig {
|
impl ProxyConfig {
|
||||||
pub fn load<P: AsRef<Path>>(path: P) -> Result<Self> {
|
pub fn load<P: AsRef<Path>>(path: P) -> Result<Self> {
|
||||||
let content =
|
Self::load_with_metadata(path).map(|loaded| loaded.config)
|
||||||
std::fs::read_to_string(&path).map_err(|e| ProxyError::Config(e.to_string()))?;
|
}
|
||||||
let base_dir = path.as_ref().parent().unwrap_or(Path::new("."));
|
|
||||||
let processed = preprocess_includes(&content, base_dir, 0)?;
|
pub(crate) fn load_with_metadata<P: AsRef<Path>>(path: P) -> Result<LoadedConfig> {
|
||||||
|
let path = path.as_ref();
|
||||||
|
let content = std::fs::read_to_string(path).map_err(|e| ProxyError::Config(e.to_string()))?;
|
||||||
|
let base_dir = path.parent().unwrap_or(Path::new("."));
|
||||||
|
let mut source_files = BTreeSet::new();
|
||||||
|
source_files.insert(normalize_config_path(path));
|
||||||
|
let processed = preprocess_includes(&content, base_dir, 0, &mut source_files)?;
|
||||||
|
|
||||||
let parsed_toml: toml::Value =
|
let parsed_toml: toml::Value =
|
||||||
toml::from_str(&processed).map_err(|e| ProxyError::Config(e.to_string()))?;
|
toml::from_str(&processed).map_err(|e| ProxyError::Config(e.to_string()))?;
|
||||||
|
|
@ -786,7 +829,11 @@ impl ProxyConfig {
|
||||||
.entry("203".to_string())
|
.entry("203".to_string())
|
||||||
.or_insert_with(|| vec!["91.105.192.100:443".to_string()]);
|
.or_insert_with(|| vec!["91.105.192.100:443".to_string()]);
|
||||||
|
|
||||||
Ok(config)
|
Ok(LoadedConfig {
|
||||||
|
config,
|
||||||
|
source_files: source_files.into_iter().collect(),
|
||||||
|
rendered_hash: hash_rendered_snapshot(&processed),
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn validate(&self) -> Result<()> {
|
pub fn validate(&self) -> Result<()> {
|
||||||
|
|
@ -1111,6 +1158,48 @@ mod tests {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn load_with_metadata_collects_include_files() {
|
||||||
|
let nonce = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.unwrap()
|
||||||
|
.as_nanos();
|
||||||
|
let dir = std::env::temp_dir().join(format!("telemt_load_metadata_{nonce}"));
|
||||||
|
std::fs::create_dir_all(&dir).unwrap();
|
||||||
|
let main_path = dir.join("config.toml");
|
||||||
|
let include_path = dir.join("included.toml");
|
||||||
|
|
||||||
|
std::fs::write(
|
||||||
|
&include_path,
|
||||||
|
r#"
|
||||||
|
[access.users]
|
||||||
|
user = "00000000000000000000000000000000"
|
||||||
|
"#,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
std::fs::write(
|
||||||
|
&main_path,
|
||||||
|
r#"
|
||||||
|
include = "included.toml"
|
||||||
|
|
||||||
|
[censorship]
|
||||||
|
tls_domain = "example.com"
|
||||||
|
"#,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let loaded = ProxyConfig::load_with_metadata(&main_path).unwrap();
|
||||||
|
let main_normalized = normalize_config_path(&main_path);
|
||||||
|
let include_normalized = normalize_config_path(&include_path);
|
||||||
|
|
||||||
|
assert!(loaded.source_files.contains(&main_normalized));
|
||||||
|
assert!(loaded.source_files.contains(&include_normalized));
|
||||||
|
|
||||||
|
let _ = std::fs::remove_file(main_path);
|
||||||
|
let _ = std::fs::remove_file(include_path);
|
||||||
|
let _ = std::fs::remove_dir(dir);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn dc_overrides_inject_dc203_default() {
|
fn dc_overrides_inject_dc203_default() {
|
||||||
let toml = r#"
|
let toml = r#"
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue