use std::path::Path;
use std::sync::Arc;
use crate::dispatch::ConnectionLimiter;
use anyhow::Context;
use decdn_common::cli::common::{LogFormat, LogLevel};
use decdn_common::cli::run::ObservabilityArgs;
use decdn_common::config::{
ConfigDiagnostics, ConfigNotice, ConfigNoticeLevel, FileConfig, ResolvedLoadShed,
ResolvedObservability, ResolvedSecurity, load_file_config, parse_pinned_hashes,
resolve_load_shed_into, resolve_observability_into, resolve_security_into,
};
#[derive(Debug, Clone, Copy)]
pub struct ReloadSnapshot {
pub log_level: Option<decdn_common::cli::common::LogLevel>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogLevelApply {
Installed,
KeptRustLog,
}
pub type LogLevelSetter =
Box<dyn Fn(LogLevel) -> anyhow::Result<LogLevelApply> + Send + Sync + 'static>;
pub(crate) trait ReloadableSection: Send + Sync {
fn name(&self) -> &'static str;
fn clear_buffer(&self);
fn resolve(&self, file: &FileConfig, bag: &mut ConfigDiagnostics);
fn fallible_commit(&self) -> anyhow::Result<()> {
Ok(())
}
fn infallible_swap(&self);
}
pub(crate) fn emit_config_notices(notices: &[ConfigNotice]) {
for notice in notices {
match notice.level {
ConfigNoticeLevel::Warn => {
tracing::warn!(field = %notice.field, "{}", notice.message);
}
ConfigNoticeLevel::Info => {
tracing::info!(field = %notice.field, "{}", notice.message);
}
}
}
}
fn attach_engine_to_section(
slot: &std::sync::Mutex<Option<decdn_cache::CacheEngine>>,
engine: Option<decdn_cache::CacheEngine>,
section: &'static str,
) {
match slot.lock() {
Ok(mut guard) => *guard = engine,
Err(poisoned) => {
tracing::error!(
section,
"runtime reload cache mutex poisoned during attach; recovering inner state"
);
*poisoned.into_inner() = engine;
}
}
}
fn drain_or_log<T>(slot: &std::sync::Mutex<Option<T>>, section: &'static str) -> Option<T> {
match slot.lock() {
Ok(mut g) => g.take(),
Err(poisoned) => {
tracing::error!(
section,
"section buffer mutex poisoned during infallible_swap; \
swap skipped (previous value retained on the live path)"
);
let _ = poisoned.into_inner().take();
None
}
}
}
struct LogLevelSection {
cli: ObservabilityArgs,
setter: LogLevelSetter,
startup: StartupObservability,
current: std::sync::Mutex<Option<LogLevel>>,
buf: std::sync::Mutex<Option<ResolvedObservability>>,
swap_applied: std::sync::Mutex<bool>,
}
impl ReloadableSection for LogLevelSection {
fn name(&self) -> &'static str {
"log_level"
}
fn clear_buffer(&self) {
if let Ok(mut g) = self.buf.lock() {
*g = None;
}
if let Ok(mut g) = self.swap_applied.lock() {
*g = false;
}
}
fn resolve(&self, file: &FileConfig, bag: &mut ConfigDiagnostics) {
let resolved = resolve_observability_into(&self.cli, file.observability.as_ref(), bag);
if let Ok(mut g) = self.buf.lock() {
*g = Some(resolved);
}
}
fn fallible_commit(&self) -> anyhow::Result<()> {
let new_level = {
let g = self
.buf
.lock()
.map_err(|_| anyhow::anyhow!("log-level buffer mutex poisoned"))?;
match g.as_ref() {
Some(r) => r.log_level,
None => return Err(anyhow::anyhow!("log-level buffer empty in fallible_commit")),
}
};
let mut current = self
.current
.lock()
.map_err(|_| anyhow::anyhow!("log-level mutex poisoned"))?;
let log_level_changed = match *current {
None => true,
Some(prev) => prev != new_level,
};
if !log_level_changed {
return Ok(());
}
match (self.setter)(new_level) {
Err(err) => {
tracing::warn!(error = %err, ?new_level, "failed to apply new log level; previous level retained");
return Err(err);
}
Ok(LogLevelApply::KeptRustLog) => return Ok(()),
Ok(LogLevelApply::Installed) => {}
}
{
*current = Some(new_level);
if let Ok(mut g) = self.swap_applied.lock() {
*g = true;
}
}
Ok(())
}
fn infallible_swap(&self) {
let Some(resolved) = drain_or_log(&self.buf, self.name()) else {
return;
};
let log_level_changed = self.swap_applied.lock().is_ok_and(|g| *g);
tracing::info!(
section = self.name(),
log_level = %resolved.log_level,
log_level_changed,
"config reload section applied"
);
}
}
impl LogLevelSection {
fn warn_restart_required_changes(&self) {
let Ok(guard) = self.buf.lock() else {
return;
};
let Some(resolved) = guard.as_ref() else {
return;
};
for field in observability_restart_required_changes(&self.startup, resolved) {
tracing::warn!(
field,
"config reload: ignoring change to observability.{field} (requires restart)"
);
}
}
}
#[derive(Debug)]
struct StartupObservability {
log_format: LogFormat,
metrics_port: u16,
metrics_bind: std::net::IpAddr,
admin_port: Option<u16>,
otlp_endpoint: Option<String>,
}
impl StartupObservability {
fn from_resolved(o: &ResolvedObservability) -> Self {
Self {
log_format: o.log_format,
metrics_port: o.metrics_port,
metrics_bind: o.metrics_bind,
admin_port: o.admin_port,
otlp_endpoint: o.otlp_endpoint.clone(),
}
}
}
fn observability_restart_required_changes(
startup: &StartupObservability,
reloaded: &ResolvedObservability,
) -> Vec<&'static str> {
let ResolvedObservability {
log_level: _, log_format,
metrics_port,
metrics_bind,
admin_port,
otlp_endpoint,
} = reloaded;
[
("log_format", startup.log_format != *log_format),
("metrics_port", startup.metrics_port != *metrics_port),
("metrics_bind", startup.metrics_bind != *metrics_bind),
("admin_port", startup.admin_port != *admin_port),
("otlp_endpoint", startup.otlp_endpoint != *otlp_endpoint),
]
.into_iter()
.filter_map(|(field, changed)| changed.then_some(field))
.collect()
}
struct PinnedHashesSection {
engine: std::sync::Mutex<Option<decdn_cache::CacheEngine>>,
buf: std::sync::Mutex<Option<decdn_cache::PinnedHashes>>,
cache_size_mb: u64,
}
impl ReloadableSection for PinnedHashesSection {
fn name(&self) -> &'static str {
"pinned_hashes"
}
fn clear_buffer(&self) {
if let Ok(mut g) = self.buf.lock() {
*g = None;
}
}
fn resolve(&self, file: &FileConfig, bag: &mut ConfigDiagnostics) {
let resolved = bag
.try_with(
"cache.pinned_hashes",
parse_pinned_hashes(file.cache.as_ref().and_then(|c| c.pinned_hashes.as_deref()))
.context("invalid cache.pinned_hashes"),
)
.unwrap_or_else(decdn_cache::PinnedHashes::empty);
if let Ok(mut g) = self.buf.lock() {
*g = Some(resolved);
}
}
fn infallible_swap(&self) {
let Some(resolved) = drain_or_log(&self.buf, self.name()) else {
return;
};
let pinned_count = resolved.len();
let pin_diff = if let Ok(g) = self.engine.lock() {
g.as_ref().map(|engine| {
let diff = engine.set_pinned(&resolved);
let engine = engine.clone();
let cache_size_mb = self.cache_size_mb;
tokio::spawn(async move {
engine.rescan_origins().await;
super::warn_if_pins_exceed_cache(&engine, cache_size_mb);
});
diff
})
} else {
tracing::error!(
section = self.name(),
"cache engine mutex poisoned in infallible_swap; pinning skipped"
);
None
};
let pinned_skipped_no_cache_attached = pin_diff.is_none();
tracing::info!(
section = self.name(),
pinned_hashes = pinned_count,
pinned_added = pin_diff.map(|d| d.added),
pinned_removed = pin_diff.map(|d| d.removed),
pinned_skipped_no_cache_attached,
cache_attached = pin_diff.is_some(),
"config reload section applied"
);
}
}
struct ContentSection {
deny: Arc<crate::content_deny::ContentDenylist>,
engine: std::sync::Mutex<Option<decdn_cache::CacheEngine>>,
buf: std::sync::Mutex<Option<decdn_common::config::ResolvedContent>>,
}
impl ReloadableSection for ContentSection {
fn name(&self) -> &'static str {
"content"
}
fn clear_buffer(&self) {
if let Ok(mut g) = self.buf.lock() {
*g = None;
}
}
fn resolve(&self, file: &FileConfig, bag: &mut ConfigDiagnostics) {
let denied_hashes = bag
.try_with(
"content.denied_hashes",
decdn_common::config::parse_denied_hashes(
file.content
.as_ref()
.and_then(|c| c.denied_hashes.as_deref()),
)
.context("invalid content.denied_hashes"),
)
.unwrap_or_default();
let denied_origins = bag
.try_with(
"content.denied_origins",
decdn_common::config::parse_denied_origins(
file.content
.as_ref()
.and_then(|c| c.denied_origins.as_deref()),
)
.context("invalid content.denied_origins"),
)
.unwrap_or_default();
if let Ok(mut g) = self.buf.lock() {
*g = Some(decdn_common::config::ResolvedContent {
denied_hashes,
denied_origins,
});
}
}
fn infallible_swap(&self) {
let Some(resolved) = drain_or_log(&self.buf, self.name()) else {
return;
};
let denied_origins = self.deny.set_local_origins(&resolved);
let Ok(engine_guard) = self.engine.lock() else {
tracing::error!(
section = self.name(),
"cache engine mutex poisoned; the denied-hash set was NOT updated and any \
takedown in this reload is undischarged"
);
return;
};
let hash_diff = engine_guard
.as_ref()
.map(|engine| engine.set_denied(&resolved.denied_hashes));
tracing::info!(
section = self.name(),
denied_hashes = resolved.denied_hashes.len(),
denied_hashes_added = hash_diff.map(|d| d.added),
denied_hashes_removed = hash_diff.map(|d| d.removed),
denied_hashes_skipped_no_cache_attached = hash_diff.is_none(),
denied_origins,
"config reload section applied"
);
}
}
struct SecuritySection {
limiter: std::sync::Mutex<Option<Arc<ConnectionLimiter>>>,
buf: std::sync::Mutex<Option<ResolvedSecurity>>,
}
impl ReloadableSection for SecuritySection {
fn name(&self) -> &'static str {
"security"
}
fn clear_buffer(&self) {
if let Ok(mut g) = self.buf.lock() {
*g = None;
}
}
fn resolve(&self, file: &FileConfig, bag: &mut ConfigDiagnostics) {
let resolved = resolve_security_into(file.security.as_ref(), bag);
if let Ok(mut g) = self.buf.lock() {
*g = Some(resolved);
}
}
fn infallible_swap(&self) {
let Some(resolved) = drain_or_log(&self.buf, self.name()) else {
return;
};
let security_attached = if let Ok(g) = self.limiter.lock() {
if let Some(lim) = g.as_ref() {
lim.reload(&resolved);
true
} else {
false
}
} else {
tracing::error!(
section = self.name(),
"limiter mutex poisoned in infallible_swap; security swap skipped"
);
false
};
tracing::info!(
section = self.name(),
security_attached,
max_concurrent_handlers = resolved.max_concurrent_handlers,
per_source_rate_per_sec = resolved.per_source_rate_per_sec,
max_tracked_sources = resolved.max_tracked_sources,
"config reload section applied"
);
}
}
struct LoadShedSection {
controller: std::sync::Mutex<Option<Arc<crate::load_shed::LoadShedController>>>,
buf: std::sync::Mutex<Option<ResolvedLoadShed>>,
}
impl ReloadableSection for LoadShedSection {
fn name(&self) -> &'static str {
"load_shed"
}
fn clear_buffer(&self) {
if let Ok(mut g) = self.buf.lock() {
*g = None;
}
}
fn resolve(&self, file: &FileConfig, bag: &mut ConfigDiagnostics) {
let resolved = resolve_load_shed_into(file.load_shed.as_ref(), bag);
if let Ok(mut g) = self.buf.lock() {
*g = Some(resolved);
}
}
fn infallible_swap(&self) {
let Some(resolved) = drain_or_log(&self.buf, self.name()) else {
return;
};
let load_shed_attached = if let Ok(g) = self.controller.lock() {
if let Some(ctrl) = g.as_ref() {
ctrl.reload(&resolved);
true
} else {
false
}
} else {
tracing::error!(
section = self.name(),
"controller mutex poisoned in infallible_swap; load-shed swap skipped"
);
false
};
tracing::info!(
section = self.name(),
load_shed_attached,
policy = ?resolved.policy,
egress_budget_mbps = resolved.egress_budget_mbps,
max_concurrent_serves_high = resolved.max_concurrent_serves_high,
max_concurrent_serves_low = resolved.max_concurrent_serves_low,
per_client_serve_cap = resolved.per_client_serve_cap,
"config reload section applied"
);
}
}
pub struct RuntimeReloadState {
log_level: Arc<LogLevelSection>,
pinned: Arc<PinnedHashesSection>,
security: Arc<SecuritySection>,
content: Arc<ContentSection>,
load_shed: Arc<LoadShedSection>,
sections: Vec<Arc<dyn ReloadableSection>>,
reload_lock: std::sync::Mutex<()>,
metrics: std::sync::OnceLock<Arc<crate::metrics::Metrics>>,
}
impl std::fmt::Debug for RuntimeReloadState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RuntimeReloadState")
.field("current_log_level", &self.log_level.current)
.finish_non_exhaustive()
}
}
impl RuntimeReloadState {
pub fn new(
observability_cli: ObservabilityArgs,
initial: &decdn_common::config::ResolvedConfig,
log_level_setter: LogLevelSetter,
) -> Self {
let log_level = Arc::new(LogLevelSection {
cli: observability_cli,
setter: log_level_setter,
startup: StartupObservability::from_resolved(&initial.observability),
current: std::sync::Mutex::new(None),
buf: std::sync::Mutex::new(None),
swap_applied: std::sync::Mutex::new(false),
});
let pinned = Arc::new(PinnedHashesSection {
engine: std::sync::Mutex::new(None),
buf: std::sync::Mutex::new(None),
cache_size_mb: initial.cache.cache_size_mb,
});
let security = Arc::new(SecuritySection {
limiter: std::sync::Mutex::new(None),
buf: std::sync::Mutex::new(None),
});
let content = Arc::new(ContentSection {
deny: Arc::new(crate::content_deny::ContentDenylist::new(&initial.content)),
engine: std::sync::Mutex::new(None),
buf: std::sync::Mutex::new(None),
});
let load_shed = Arc::new(LoadShedSection {
controller: std::sync::Mutex::new(None),
buf: std::sync::Mutex::new(None),
});
let sections: Vec<Arc<dyn ReloadableSection>> = vec![
Arc::clone(&log_level) as _,
Arc::clone(&pinned) as _,
Arc::clone(&security) as _,
Arc::clone(&content) as _,
Arc::clone(&load_shed) as _,
];
Self {
log_level,
pinned,
security,
content,
load_shed,
sections,
reload_lock: std::sync::Mutex::new(()),
metrics: std::sync::OnceLock::new(),
}
}
pub fn attach_metrics(&self, metrics: Arc<crate::metrics::Metrics>) {
let _already_attached = self.metrics.set(metrics);
}
pub fn attach_cache(&self, engine: Option<decdn_cache::CacheEngine>) {
attach_engine_to_section(&self.content.engine, engine.clone(), "content");
attach_engine_to_section(&self.pinned.engine, engine, "pinned_hashes");
}
pub fn content_denylist(&self) -> Arc<crate::content_deny::ContentDenylist> {
Arc::clone(&self.content.deny)
}
pub fn attach_limiter(&self, limiter: Option<Arc<ConnectionLimiter>>) {
match self.security.limiter.lock() {
Ok(mut guard) => *guard = limiter,
Err(poisoned) => {
tracing::error!(
"runtime reload limiter mutex poisoned during attach; recovering inner state"
);
*poisoned.into_inner() = limiter;
}
}
}
pub fn attach_load_shed(&self, controller: Option<Arc<crate::load_shed::LoadShedController>>) {
match self.load_shed.controller.lock() {
Ok(mut guard) => *guard = controller,
Err(poisoned) => {
tracing::error!(
"runtime reload load-shed controller mutex poisoned during attach; \
recovering inner state"
);
*poisoned.into_inner() = controller;
}
}
}
#[cfg(test)]
#[allow(clippy::too_many_lines)]
pub(crate) fn for_test_with_setter(
level: decdn_common::cli::common::LogLevel,
log_level_setter: LogLevelSetter,
) -> Self {
use std::path::PathBuf;
use decdn_common::config::{
ResolvedBlockchain, ResolvedCache, ResolvedConfig, ResolvedIdentity, ResolvedNetwork,
ResolvedObservability, ResolvedPayment, ResolvedSecurity,
};
let cfg = ResolvedConfig {
identity: ResolvedIdentity {
data_dir: PathBuf::from("/tmp/decdn-test"),
region: None,
},
network: ResolvedNetwork {
bind_port: 4433,
relay_urls: Vec::new(),
discovery: decdn_common::config::ResolvedDiscovery::default(),
},
blockchain: ResolvedBlockchain {
origin_assignment_address: None,
origin_directory_positive_ttl_sec:
decdn_common::config::DEFAULT_ORIGIN_DIRECTORY_POSITIVE_TTL_SEC,
origin_directory_negative_ttl_sec:
decdn_common::config::DEFAULT_ORIGIN_DIRECTORY_NEGATIVE_TTL_SEC,
origin_directory_cache_capacity:
decdn_common::config::DEFAULT_ORIGIN_DIRECTORY_CACHE_CAPACITY,
publisher_registry_address: None,
rpc_url: "http://localhost:8545".into(),
eth_keystore: PathBuf::from("/tmp/keystore.json"),
keystore_password_file: None,
payment_pool_address: "0x0000000000000000000000000000000000000001".into(),
capacity_bond_address: "0x0000000000000000000000000000000000000002".into(),
rpc_watchdog_interval_sec: 30,
event_poll_interval_ms: 7000,
get_logs_max_block_span: decdn_common::config::DEFAULT_GET_LOGS_MAX_BLOCK_SPAN,
fee_shares_poll_interval_sec: 3600,
redeem_threshold_micro_usdc: 1_000_000,
redeem_max_vouchers_per_tx: 300,
redeem_interval_secs: 300,
buyer_working_deposit_micro_usdc: 10_000_000,
buyer_max_approve: true,
pool_min_remaining_deposit_micro_usdc: 1_000_000,
pool_floor_signer_live_windows: 8,
slash_judge_address: "0x0000000000000000000000000000000000000003".to_string(),
content_blacklist_address: None,
content_blacklist_poll_interval_sec: 600,
chain_staleness_grace_sec: 1800,
chain_id: decdn_common::config::DEFAULT_CHAIN_ID,
},
cache: ResolvedCache {
cache_dir: PathBuf::from("/tmp/cache"),
cache_size_mb: 1024,
disk_headroom_mb: 8192,
max_blob_size_mb: 128,
max_rate_per_mb: 0,
origins: Vec::new(),
pinned_hashes: decdn_cache::PinnedHashes::empty(),
origin_retry: decdn_cache::RetryPolicy::default(),
circuit_breaker: decdn_cache::CircuitBreakerPolicy::default(),
user_agent: decdn_cache::DEFAULT_USER_AGENT.to_string(),
gc_interval_sec: 0,
fs_rescan_interval_sec: 0,
origin_probe_ttl_sec: decdn_common::config::DEFAULT_ORIGIN_PROBE_TTL_SEC,
origin_probe_negative_ttl_sec:
decdn_common::config::DEFAULT_ORIGIN_PROBE_NEGATIVE_TTL_SEC,
origin_probe_fault_ttl_sec:
decdn_common::config::DEFAULT_ORIGIN_PROBE_FAULT_TTL_SEC,
origin_probe_timeout_ms: decdn_common::config::DEFAULT_ORIGIN_PROBE_TIMEOUT_MS,
origin_probe_memo_capacity:
decdn_common::config::DEFAULT_ORIGIN_PROBE_MEMO_CAPACITY,
eviction_high_water_pct: 90,
eviction_target_pct: 80,
eviction_per_sweep_budget: 16,
eviction_tick_secs: 1,
max_probe_holds: decdn_common::config::DEFAULT_MAX_PROBE_HOLDS,
stake_lane_reserved_holds: decdn_common::config::DEFAULT_STAKE_LANE_RESERVED_HOLDS,
node_to_node_pull_through_enabled: false,
relay_foreign_namespaces: decdn_common::config::DEFAULT_RELAY_FOREIGN_NAMESPACES,
node_pull_probe_fanout: decdn_common::config::DEFAULT_NODE_PULL_PROBE_FANOUT,
node_pull_timeout_sec: decdn_common::config::DEFAULT_NODE_PULL_TIMEOUT_SEC,
node_pull_stall_window_sec:
decdn_common::config::DEFAULT_NODE_PULL_STALL_WINDOW_SEC,
node_pull_min_throughput_bps:
decdn_common::config::DEFAULT_NODE_PULL_MIN_THROUGHPUT_BPS,
eviction_policy: decdn_common::config::DEFAULT_EVICTION_POLICY.to_string(),
admission_policy: decdn_common::config::DEFAULT_ADMISSION_POLICY.to_string(),
tinylfu: decdn_common::config::ResolvedTinyLfu {
sketch_bytes: decdn_common::config::DEFAULT_TINYLFU_SKETCH_BYTES,
promotion_threshold: decdn_common::config::DEFAULT_TINYLFU_PROMOTION_THRESHOLD,
probation_target_pct:
decdn_common::config::DEFAULT_TINYLFU_PROBATION_TARGET_PCT,
aging_halflife_sec: decdn_common::config::DEFAULT_TINYLFU_AGING_HALFLIFE_SEC,
},
serve_economics: decdn_common::config::ResolvedServeEconomics {
policy: decdn_common::config::DEFAULT_SERVE_ECONOMICS_POLICY.to_string(),
discount_bps: decdn_common::config::DEFAULT_SERVE_ECONOMICS_DISCOUNT_BPS,
n_max: decdn_common::config::DEFAULT_SERVE_ECONOMICS_N_MAX,
warming_budget: decdn_common::config::DEFAULT_SERVE_ECONOMICS_WARMING_BUDGET,
warming_refill: decdn_common::config::DEFAULT_SERVE_ECONOMICS_WARMING_REFILL,
},
},
payment: ResolvedPayment {
rate_per_mb: 1,
credit_max: decdn_common::config::DEFAULT_CREDIT_MAX,
frame_target_bytes: decdn_common::config::DEFAULT_FRAME_TARGET_BYTES,
credit_ramp_divisor: decdn_common::config::DEFAULT_CREDIT_RAMP_DIVISOR,
voucher_commit_interval_ms:
decdn_common::config::DEFAULT_VOUCHER_COMMIT_INTERVAL_MS,
},
observability: ResolvedObservability {
log_level: level,
log_format: decdn_common::cli::common::LogFormat::Pretty,
metrics_port: 9090,
metrics_bind: std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
admin_port: Some(9191),
otlp_endpoint: None,
},
security: ResolvedSecurity {
max_concurrent_handlers: 256,
per_source_rate_per_sec: 100.0,
per_source_burst: 200,
max_tracked_sources: 4096,
},
load_shed: decdn_common::config::ResolvedLoadShed::default(),
dht: decdn_common::config::ResolvedDht::default(),
probe: decdn_common::config::ResolvedProbe::default(),
receipts: decdn_common::config::ResolvedReceipts::default(),
content: decdn_common::config::ResolvedContent::default(),
};
Self::new(
decdn_common::cli::run::ObservabilityArgs {
log_level: None,
log_format: None,
metrics_port: None,
metrics_bind: None,
admin_port: None,
otlp_endpoint: None,
},
&cfg,
log_level_setter,
)
}
pub fn current(&self) -> ReloadSnapshot {
let log_level = self.log_level.current.lock().ok().and_then(|guard| *guard);
ReloadSnapshot { log_level }
}
#[allow(clippy::unused_async)]
pub async fn reload(&self, path: &Path) -> anyhow::Result<()> {
let result = self.reload_sections(path);
if result.is_err()
&& let Some(metrics) = self.metrics.get()
{
metrics.config_reload_failure();
}
result
}
#[allow(clippy::cognitive_complexity)] fn reload_sections(&self, path: &Path) -> anyhow::Result<()> {
let file = match load_file_config(Some(path)) {
Ok(f) => f,
Err(err) => {
tracing::warn!(error = %err, path = %path.display(), "config reload aborted (file load failed); previous values retained for every section");
return Err(err);
}
};
let _reload_guard = match self.reload_lock.lock() {
Ok(guard) => guard,
Err(poisoned) => {
tracing::error!(
"runtime reload_lock poisoned by a prior panic-mid-reload; \
recovering and proceeding (an earlier reload may have \
partially applied before panicking)"
);
poisoned.into_inner()
}
};
for section in &self.sections {
section.clear_buffer();
}
let mut bag = ConfigDiagnostics::new();
for section in &self.sections {
section.resolve(&file, &mut bag);
}
let notices = bag.take_notices();
let problem_count = bag.problem_count();
if let Err(err) = bag.into_result() {
tracing::warn!(
error = %err,
problem_count,
"config reload aborted; entire reload rolled back \
(all-or-nothing) — all reloadable sections retained at \
their previous values",
);
return Err(err);
}
for section in &self.sections {
if let Err(err) = section.fallible_commit() {
tracing::warn!(
error = %err,
section = section.name(),
"config reload fallible_commit failed; later sections skipped"
);
return Err(err);
}
}
emit_config_notices(¬ices);
warn_restart_required_sections(&file);
self.log_level.warn_restart_required_changes();
for section in &self.sections {
section.infallible_swap();
}
tracing::info!("config reload applied");
Ok(())
}
}
fn warn_ignored(field: &'static str) {
tracing::info!(
field,
"config reload: ignoring change to {field} (requires restart)"
);
}
fn warn_restart_required_sections(file: &decdn_common::config::FileConfig) {
if file.identity.is_some() {
warn_ignored("identity.* (data_dir, region)");
}
if file.network.is_some() {
warn_ignored("network.* (bind_port, relay_urls, relay_url, discovery)");
}
if file.blockchain.is_some() {
warn_ignored(
"blockchain.* (rpc_url, eth_keystore, contract addresses, \
pool_floor_signer_live_windows)",
);
}
if file
.cache
.as_ref()
.is_some_and(cache_has_restart_required_field)
{
warn_ignored("cache.* (cache_dir, sizes, origin, decompress, max_probe_holds)");
}
if file.payment.is_some() {
warn_ignored(
"payment.* (rate_per_mb, credit_max, \
credit_ramp_divisor, frame_target_bytes, voucher_commit_interval_ms)",
);
}
if file.dht.is_some() {
warn_ignored("dht.* (rate-limit)");
}
if file.probe.is_some() {
warn_ignored("probe.* (rate_limit)");
}
if file.receipts.is_some() {
warn_ignored("receipts.* (max_file_bytes, retained_files)");
}
}
const fn cache_has_restart_required_field(c: &decdn_common::config::types::CacheConfig) -> bool {
let decdn_common::config::types::CacheConfig {
pinned_hashes: _, cache_dir,
cache_size_mb,
disk_headroom_mb,
max_blob_size_mb,
max_rate_per_mb,
origin,
origins,
origin_retry,
circuit_breaker,
user_agent,
gc_interval_sec,
fs_rescan_interval_sec,
origin_probe_ttl_sec,
origin_probe_negative_ttl_sec,
origin_probe_fault_ttl_sec,
origin_probe_timeout_ms,
origin_probe_memo_capacity,
eviction_high_water_pct,
eviction_target_pct,
eviction_per_sweep_budget,
eviction_tick_secs,
max_probe_holds,
stake_lane_reserved_holds,
node_to_node_pull_through_enabled,
relay_foreign_namespaces,
node_pull_probe_fanout,
node_pull_timeout_sec,
node_pull_stall_window_sec,
node_pull_min_throughput_bps,
eviction_policy,
admission_policy,
tinylfu,
serve_economics,
} = c;
cache_dir.is_some()
|| cache_size_mb.is_some()
|| disk_headroom_mb.is_some()
|| max_blob_size_mb.is_some()
|| max_rate_per_mb.is_some()
|| origin.is_some()
|| origins.is_some()
|| origin_retry.is_some()
|| circuit_breaker.is_some()
|| user_agent.is_some()
|| gc_interval_sec.is_some()
|| fs_rescan_interval_sec.is_some()
|| origin_probe_ttl_sec.is_some()
|| origin_probe_negative_ttl_sec.is_some()
|| origin_probe_fault_ttl_sec.is_some()
|| origin_probe_timeout_ms.is_some()
|| origin_probe_memo_capacity.is_some()
|| eviction_high_water_pct.is_some()
|| eviction_target_pct.is_some()
|| eviction_per_sweep_budget.is_some()
|| eviction_tick_secs.is_some()
|| max_probe_holds.is_some()
|| stake_lane_reserved_holds.is_some()
|| node_to_node_pull_through_enabled.is_some()
|| relay_foreign_namespaces.is_some()
|| node_pull_probe_fanout.is_some()
|| node_pull_timeout_sec.is_some()
|| node_pull_stall_window_sec.is_some()
|| node_pull_min_throughput_bps.is_some()
|| eviction_policy.is_some()
|| admission_policy.is_some()
|| tinylfu.is_some()
|| serve_economics.is_some()
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests;