aboutsummaryrefslogtreecommitdiff
path: root/crates/arti
diff options
context:
space:
mode:
Diffstat (limited to 'crates/arti')
-rw-r--r--crates/arti/src/reload_cfg.rs196
-rw-r--r--crates/arti/src/rpc.rs15
-rw-r--r--crates/arti/src/rpc/session.rs4
-rw-r--r--crates/arti/src/rpc/superuser.rs8
-rw-r--r--crates/arti/src/subcommands/proxy.rs26
5 files changed, 197 insertions, 52 deletions
diff --git a/crates/arti/src/reload_cfg.rs b/crates/arti/src/reload_cfg.rs
index f5406329b..a55afa0cc 100644
--- a/crates/arti/src/reload_cfg.rs
+++ b/crates/arti/src/reload_cfg.rs
@@ -1,5 +1,6 @@
//! Code to watch configuration files for any changes.
+use std::collections::HashSet;
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;
@@ -7,10 +8,14 @@ use anyhow::Context;
use arti_client::TorClient;
use arti_client::config::Reconfigure;
use futures::StreamExt;
+use futures::stream::BoxStream;
use futures::{FutureExt as _, Stream, select_biased};
+#[cfg(feature = "rpc")]
+use tor_config::ConfigurationTree;
use tor_config::file_watcher::{
self, FileEventReceiver, FileEventSender, FileWatcher, FileWatcherBuilder,
};
+use tor_config::load::{ConfigResolveOptions, DisfavouredKey};
use tor_config::{ConfigurationSource, ConfigurationSources, sources::FoundConfigFiles};
use tor_rtcompat::Runtime;
use tor_rtcompat::SpawnExt;
@@ -58,18 +63,52 @@ pub(crate) struct CfgMgr<R> {
/// A sender to use when constructing new [`FileWatcher`]s.
tx: FileEventSender,
- /// A list of modules to alert whenever the configuration has changed.
- modules: Vec<Weak<dyn ReconfigurableModule>>,
-
/// Mutable state.
inner: Mutex<CfgMgrInner>,
}
/// Mutable part of a CfgMgr.
+#[derive(Default)]
struct CfgMgrInner {
+ /// A list of modules to alert whenever the configuration has changed.
+ modules: Vec<Weak<dyn ReconfigurableModule>>,
+
/// If present, a [`FileWatcher`] that is currently watching for changes
/// in the configuration files and directories.
watcher: Option<FileWatcher>,
+
+ /// RPC only: a fully populated, normalized configuration tree, based on the most recent time
+ /// that we called [`CfgMgr::reload_configuration`].
+ #[cfg(feature = "rpc")]
+ normalized_cfg: ConfigurationTree,
+
+ /// RPC only: a set of unrecognized options from the configuration.
+ #[cfg(feature = "rpc")]
+ unrecognized_keys: HashSet<DisfavouredKey>,
+
+ /// RPC only: a set of deprecated options from the configuration
+ #[cfg(feature = "rpc")]
+ deprecated_keys: HashSet<DisfavouredKey>,
+}
+
+/// A watcher process that we have not yet launched.
+#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
+#[must_use = "UnlaunchedWatcher does nothing unless you launch it."]
+pub(crate) struct UnlaunchedWatcher<R> {
+ /// The related [`CfgMgr`] that we should tell about reconfiguration events.
+ weak_mgr: Weak<CfgMgr<R>>,
+
+ /// A stream on which we will get alerts about SIGHUP events.
+ sighup_stream: BoxStream<'static, ()>,
+
+ /// A stream that will tell us when our files are changed.
+ watcher_rx: FileEventReceiver,
+
+ /// An interval that we wait to debounce events from watcher_rx or sighup_stream.
+ debounce_interval: Option<Duration>,
+
+ /// If true, we start watching for file changes immediately at launch.
+ watch_files_at_start: bool,
}
impl<R: Runtime> CfgMgr<R> {
@@ -86,19 +125,21 @@ impl<R: Runtime> CfgMgr<R> {
/// See the [`FileWatcher`](FileWatcher#Limitations) docs for limitations.
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
#[instrument(level = "trace", skip_all)]
- pub(crate) fn launch(
+ pub(crate) fn new(
runtime: R,
sources: ConfigurationSources,
config: &ArtiConfig,
modules: Vec<Weak<dyn ReconfigurableModule>>,
- ) -> anyhow::Result<Arc<Self>> {
+ ) -> anyhow::Result<(Arc<Self>, UnlaunchedWatcher<R>)> {
let (tx, rx) = file_watcher::channel();
let mgr = Arc::new(CfgMgr {
runtime,
sources,
tx,
- modules,
- inner: Mutex::new(CfgMgrInner { watcher: None }),
+ inner: Mutex::new(CfgMgrInner {
+ modules,
+ ..Default::default()
+ }),
});
cfg_if::cfg_if! {
@@ -108,27 +149,17 @@ impl<R: Runtime> CfgMgr<R> {
let sighup_stream = stream::pending();
}
}
+ let sighup_stream = sighup_stream.boxed();
+
+ let watcher = UnlaunchedWatcher {
+ weak_mgr: Arc::downgrade(&mgr),
+ sighup_stream,
+ watcher_rx: rx,
+ debounce_interval: Some(DEBOUNCE_INTERVAL),
+ watch_files_at_start: config.application().watch_configuration,
+ };
- let rt = mgr.runtime.clone();
- let weak_mgr = Arc::downgrade(&mgr);
- mgr.runtime
- .spawn(async move {
- let res: anyhow::Result<()> =
- run_watcher(rt, rx, sighup_stream, weak_mgr, Some(DEBOUNCE_INTERVAL)).await;
- match res {
- Ok(()) => debug!("Config watcher task exiting"),
- // TODO: warn_report does not work on anyhow::Error.
- Err(e) => error!("Config watcher task exiting: {}", tor_error::Report(e)),
- }
- })
- .context("failed to spawn task")?;
-
- if config.application().watch_configuration {
- let (watcher, _files) = mgr.launch_file_watcher()?;
- mgr.inner.lock().expect("lock poisoned").watcher = Some(watcher);
- }
-
- Ok(mgr)
+ Ok((mgr, watcher))
}
/// Create a new [`FileWatcher`] for the files in this configuration.
@@ -146,10 +177,14 @@ impl<R: Runtime> CfgMgr<R> {
/// Reload the configuration.
#[instrument(level = "trace", skip_all)]
- fn reload_configuration(&self) -> anyhow::Result<()> {
+ #[cfg_attr(feature = "experimental-api", visibility::make(pub))]
+ pub(crate) fn reload_configuration(&self) -> anyhow::Result<()> {
let mut inner = self.inner.lock().expect("Lock poisoned");
// TODO RPC: Take 'how' as an argument.
+ //
+ // Question: I do not understand why we are making a new file watcher unconditionally
+ // at this point. -nm
let found_files = if inner.watcher.is_some() {
let (watcher, files) = self
.launch_file_watcher()
@@ -162,7 +197,7 @@ impl<R: Runtime> CfgMgr<R> {
.context("FS watch: failed to rescan config")?
};
- match reconfigure(found_files, &self.modules) {
+ match reconfigure(found_files, &mut inner) {
Ok(watch) => {
info!("Successfully reloaded configuration.");
if watch && inner.watcher.is_none() {
@@ -184,9 +219,75 @@ impl<R: Runtime> CfgMgr<R> {
}
}
+impl<R: Runtime> UnlaunchedWatcher<R> {
+ /// Begin running the file watcher task for a given configuration manager.
+ #[cfg_attr(feature = "experimental-api", visibility::make(pub))]
+ #[instrument(level = "trace", skip_all)]
+ pub(crate) fn launch(self) -> anyhow::Result<()> {
+ let UnlaunchedWatcher {
+ weak_mgr,
+ sighup_stream,
+ watcher_rx,
+ debounce_interval,
+ watch_files_at_start,
+ } = self;
+ let Some(mgr) = weak_mgr.upgrade() else {
+ return Err(anyhow::anyhow!(
+ "CfgMgr disappeared before we could launch the monitor task"
+ ));
+ };
+
+ let rt = mgr.runtime.clone();
+ let weak_mgr = Arc::downgrade(&mgr);
+ mgr.runtime
+ .spawn(async move {
+ let res: anyhow::Result<()> =
+ run_watcher(rt, watcher_rx, sighup_stream, weak_mgr, debounce_interval).await;
+ match res {
+ Ok(()) => debug!("Config watcher task exiting"),
+ // TODO: warn_report does not work on anyhow::Error.
+ Err(e) => error!("Config watcher task exiting: {}", tor_error::Report(e)),
+ }
+ })
+ .context("failed to spawn task")?;
+
+ if watch_files_at_start {
+ // Note: You might think that there was a race condition here, where launching the
+ // watcher _now_ would fail to catch any file changes that had happened between
+ // reading the configuration initially and now.
+ //
+ // You'd be right, except that the [`FileWatcher`] code starts every new FileWatcher
+ // with a pending `rescan` event.
+ let (watcher, _files) = mgr.launch_file_watcher()?;
+ mgr.inner.lock().expect("lock poisoned").watcher = Some(watcher);
+ }
+
+ Ok(())
+ }
+
+ /// Add `module` to the set of modules that need to be reconfigured when the configuration changes.
+ ///
+ /// This method is on the [`UnlaunchedWatcher`] because is not (yet) meant to be called after
+ /// the watcher task is launched.
+ #[cfg_attr(feature = "experimental-api", visibility::make(pub))]
+ pub(crate) fn add_module(&self, module: &Arc<dyn ReconfigurableModule>) -> anyhow::Result<()> {
+ let weak_module = Arc::downgrade(module);
+
+ let Some(mgr) = self.weak_mgr.upgrade() else {
+ return Err(anyhow::anyhow!(
+ "CfgMgr disappeared before launching watcher task."
+ ));
+ };
+
+ let mut inner = mgr.inner.lock().expect("poisoned lock");
+ inner.modules.push(weak_module);
+ Ok(())
+ }
+}
+
/// Start watching for configuration changes.
///
-/// Spawned from [`CfgMgr::launch`].
+/// Spawned from [`UnlaunchedWatcher::launch`].
#[instrument(level = "trace", skip_all)]
async fn run_watcher<R: Runtime>(
runtime: R,
@@ -383,14 +484,29 @@ fn prepare<'a, R: Runtime>(
#[instrument(level = "trace", skip_all)]
fn reconfigure(
found_files: FoundConfigFiles<'_>,
- reconfigurable: &[Weak<dyn ReconfigurableModule>],
+ mgr_inner: &mut CfgMgrInner,
) -> anyhow::Result<bool> {
- let _ = reconfigurable;
let config = found_files.load()?;
- let config = tor_config::resolve::<ArtiCombinedConfig>(config)?;
+ #[allow(unused_mut)]
+ let mut resolve_options = ConfigResolveOptions::default();
+ #[cfg(feature = "rpc")]
+ {
+ resolve_options.want_output_tree = true;
+ }
+
+ let rs = tor_config::resolve_return_results::<ArtiCombinedConfig>(config, &resolve_options)?;
+ let config = rs.value;
+ #[cfg(feature = "rpc")]
+ {
+ mgr_inner.normalized_cfg = rs
+ .output_tree
+ .expect("normalized cfg not exposed as expected!?");
+ mgr_inner.deprecated_keys = rs.deprecated.into_iter().collect();
+ mgr_inner.unrecognized_keys = rs.unrecognized.into_iter().collect();
+ }
// Filter out the modules that have been dropped
- let reconfigurable = reconfigurable.iter().flat_map(Weak::upgrade);
+ let reconfigurable = mgr_inner.modules.iter().flat_map(Weak::upgrade);
// If there are no more modules, we should exit.
let mut has_modules = false;
@@ -509,8 +625,10 @@ mod test {
runtime: rt.clone(),
sources: cfg_sources,
tx: fw_tx,
- modules: vec![Arc::downgrade(&module)],
- inner: Mutex::new(CfgMgrInner { watcher: None }),
+ inner: Mutex::new(CfgMgrInner {
+ modules: vec![Arc::downgrade(&module)],
+ ..Default::default()
+ }),
});
let (watcher, _) = mgr.launch_file_watcher().unwrap();
@@ -567,8 +685,10 @@ mod test {
runtime: rt.clone(),
sources: cfg_sources,
tx: fw_tx,
- modules: vec![Arc::downgrade(&module)],
- inner: Mutex::new(CfgMgrInner { watcher: None }),
+ inner: Mutex::new(CfgMgrInner {
+ modules: vec![Arc::downgrade(&module)],
+ ..Default::default()
+ }),
});
let (watcher, _) = mgr.launch_file_watcher().unwrap();
diff --git a/crates/arti/src/rpc.rs b/crates/arti/src/rpc.rs
index 466224174..a40d5800c 100644
--- a/crates/arti/src/rpc.rs
+++ b/crates/arti/src/rpc.rs
@@ -24,7 +24,7 @@ mod superuser;
use listener::RpcListenerSetConfig;
pub(crate) use session::{RpcStateSender, RpcVisibleArtiState};
-use crate::reload_cfg::LaunchableTorClient;
+use crate::reload_cfg::{CfgMgr, LaunchableTorClient};
use crate::rpc::superuser::RpcSuperuser;
/// Configuration for Arti's RPC subsystem.
@@ -135,6 +135,7 @@ pub(crate) async fn launch_rpc_mgr<R: Runtime>(
mistrust: &Mistrust,
client: Arc<TorClient<R>>,
launchable: Arc<LaunchableTorClient<R>>,
+ cfg_mgr: Arc<CfgMgr<R>>,
) -> Result<Option<RpcProxySupport>> {
if !cfg.enable {
return Ok(None);
@@ -164,6 +165,7 @@ pub(crate) async fn launch_rpc_mgr<R: Runtime>(
rpc_mgr_clone,
client,
launchable,
+ cfg_mgr,
rpc_state,
)
.await;
@@ -185,6 +187,7 @@ async fn run_rpc_listener<R: Runtime>(
rpc_mgr: Arc<RpcMgr>,
client: Arc<TorClient<R>>,
launchable: Arc<LaunchableTorClient<R>>,
+ cfg_mgr: Arc<CfgMgr<R>>,
rpc_state: Arc<RpcVisibleArtiState>,
) -> Result<()> {
while let Some((stream, _addr, info)) = incoming.next().await.transpose()? {
@@ -193,8 +196,16 @@ async fn run_rpc_listener<R: Runtime>(
let client_clone = client.clone();
let rpc_state_clone = rpc_state.clone();
let launchable = launchable.clone();
+ let cfg_mgr_clone = cfg_mgr.clone();
let connection = rpc_mgr.new_connection(info.auth.clone(), move |auth| {
- ArtiRpcSession::new(auth, &client_clone, &launchable, &rpc_state_clone, &info) as _
+ ArtiRpcSession::new(
+ auth,
+ &client_clone,
+ &launchable,
+ &rpc_state_clone,
+ &cfg_mgr_clone,
+ &info,
+ ) as _
});
let (input, output) = stream.split();
diff --git a/crates/arti/src/rpc/session.rs b/crates/arti/src/rpc/session.rs
index 853e6ae21..be1bca55b 100644
--- a/crates/arti/src/rpc/session.rs
+++ b/crates/arti/src/rpc/session.rs
@@ -12,7 +12,7 @@ use tor_rtcompat::Runtime;
use crate::{
proxy::port_info,
- reload_cfg::LaunchableTorClient,
+ reload_cfg::{CfgMgr, LaunchableTorClient},
rpc::{listener::RpcConnInfo, superuser::RpcSuperuser},
};
@@ -82,6 +82,7 @@ impl ArtiRpcSession {
client_root: &Arc<TorClient<R>>,
launchable_client: &Arc<LaunchableTorClient<R>>,
arti_state: &Arc<RpcVisibleArtiState>,
+ cfg_mgr: &Arc<CfgMgr<R>>,
listener_info: &RpcConnInfo,
) -> Arc<Self> {
let _ = auth; // This is currently unused; any authentication gives the same result.
@@ -91,6 +92,7 @@ impl ArtiRpcSession {
session.provide_superuser_permission(Arc::new(RpcSuperuser::new(
client_root.clone(),
launchable_client.clone(),
+ cfg_mgr.clone(),
)) as _);
}
Arc::new(ArtiRpcSession {
diff --git a/crates/arti/src/rpc/superuser.rs b/crates/arti/src/rpc/superuser.rs
index c8a3a5cb8..9e36e608f 100644
--- a/crates/arti/src/rpc/superuser.rs
+++ b/crates/arti/src/rpc/superuser.rs
@@ -13,7 +13,7 @@ use std::sync::Arc;
use tor_rpcbase::{self as rpc};
use tor_rtcompat::Runtime;
-use crate::reload_cfg::LaunchableTorClient;
+use crate::reload_cfg::{CfgMgr, LaunchableTorClient};
/// An object representing superuser access to Arti over an RPC session.
///
@@ -27,6 +27,10 @@ pub(super) struct RpcSuperuser<R: Runtime> {
/// A wrapper around `tor_client` with the ability to launch a deferred-bootstrap client.
launchable: Arc<LaunchableTorClient<R>>,
+
+ /// A handle to the manager for configuration information.
+ #[allow(unused)] // TODO(rpc) remove
+ cfg_mgr: Arc<CfgMgr<R>>,
}
impl<R: Runtime> RpcSuperuser<R> {
@@ -34,10 +38,12 @@ impl<R: Runtime> RpcSuperuser<R> {
pub(super) fn new(
tor_client: Arc<TorClient<R>>,
launchable: Arc<LaunchableTorClient<R>>,
+ cfg_mgr: Arc<CfgMgr<R>>,
) -> Self {
RpcSuperuser {
tor_client,
launchable,
+ cfg_mgr,
}
}
diff --git a/crates/arti/src/subcommands/proxy.rs b/crates/arti/src/subcommands/proxy.rs
index 95a10cdd4..17f480ae4 100644
--- a/crates/arti/src/subcommands/proxy.rs
+++ b/crates/arti/src/subcommands/proxy.rs
@@ -138,6 +138,9 @@ async fn run_proxy<R: ToplevelRuntime>(
false => BootstrapBehavior::OnDemand,
};
+ let (cfg_mgr, cfg_watcher_task) =
+ reload_cfg::CfgMgr::new(runtime.clone(), config_sources, &arti_config, vec![])?;
+
let client_builder = TorClient::with_runtime(runtime.clone())
.config(client_config)
.bootstrap_behavior(bootstrap_behavior);
@@ -172,18 +175,17 @@ async fn run_proxy<R: ToplevelRuntime>(
}
};
- // We weak references here to prevent the thread spawned by watch_for_config_changes from
+ // The add_module function will use references here
+ // to prevent the task spawned by watch_for_config_changes from
// keeping these modules alive after this function exits.
//
// NOTE: reconfigurable_modules stores the only strong references to these modules,
- // so we must keep the variable alive until the end of the function
- let weak_modules = reconfigurable_modules.iter().map(Arc::downgrade).collect();
- let _cfg_mgr = reload_cfg::CfgMgr::launch(
- client.runtime().clone(),
- config_sources,
- &arti_config,
- weak_modules,
- )?;
+ // so we must keep that variable alive until the end of the function
+ reconfigurable_modules
+ .iter()
+ .try_for_each(|m| cfg_watcher_task.add_module(m))?;
+
+ cfg_watcher_task.launch()?;
cfg_if::cfg_if! {
if #[cfg(feature = "rpc")] {
@@ -194,6 +196,7 @@ async fn run_proxy<R: ToplevelRuntime>(
&fs_mistrust,
client.clone(),
launchable_client.clone(),
+ cfg_mgr.clone(),
)
.await?;
let (rpc_mgr, mut rpc_state_sender) = rpc_data
@@ -322,8 +325,11 @@ async fn run_proxy<R: ToplevelRuntime>(
=> r.context("bootstrap"),
)?;
- // The modules can be dropped now, because we are exiting.
+ // The modules and CfgMgr can be dropped now, because we are exiting.
+ // (We drop them explicitly to make sure that they were not dropped
+ // accidentally before.)
drop(reconfigurable_modules);
+ drop(cfg_mgr);
Ok(())
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-08-27 12:21:07 +0000