feat: improve processor reloading and allow it for SSL

prosa/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "prosa"
-version = "0.5.0"
+version = "0.5.1"
 authors.workspace = true
 description = "ProSA core"
 homepage.workspace = true

prosa/examples/proc.rs

@@ -79,23 +79,25 @@ where
                             info!("Proc {} received an error: {:?}", self.get_proc_id(), err);
                         },
                         InternalMsg::Config(config) => {
-                            let settings = config.get_proc::<MyProcSettings>(self.proc.as_ref())?;
-
-                            if self.settings.service_name != settings.service_name {
-                                self.proc
-                                    .remove_service_proc(vec![self.settings.service_name.clone()])
-                                    .await?;
-                                self.proc
-                                    .add_service_proc(vec![settings.service_name.clone()])
-                                    .await?;
+                            if let Some(settings) = config
+                                .reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor)
+                            {
+                                if self.settings.service_name != settings.service_name {
+                                    self.proc
+                                        .remove_service_proc(vec![self.settings.service_name.clone()])
+                                        .await?;
+                                    self.proc
+                                        .add_service_proc(vec![settings.service_name.clone()])
+                                        .await?;
+                                }
+
+                                if self.settings.tick_secs != settings.tick_secs {
+                                    interval = settings.interval();
+                                }
+
+                                info!("Proc {} reloaded settings: {:?}", self.get_proc_id(), settings);
+                                self.settings = settings;
                             }
-
-                            if self.settings.tick_secs != settings.tick_secs {
-                                interval = settings.interval();
-                            }
-
-                            info!("Proc {} reloaded settings: {:?}", self.get_proc_id(), settings);
-                            self.settings = settings;
                         },
                         InternalMsg::Service(table) => {
                             debug!("New service table received:\n{}\n", table);

prosa/src/core/adaptor.rs

@@ -62,12 +62,17 @@ pub trait Adaptor {
     /// existing adaptors compatible by ignoring the configuration.
     ///
     /// Processor implementations should call this from their loop when they
-    /// receive [`InternalMsg::Config`](crate::core::msg::InternalMsg::Config):
+    /// receive [`InternalMsg::Config`](crate::core::msg::InternalMsg::Config), which
+    /// [`ProsaConfig::reload_proc`](crate::core::settings::ProsaConfig::reload_proc) does along with
+    /// the processor settings:
     ///
     /// ```rust,ignore
     /// InternalMsg::Config(config) => {
-    ///     self.settings = config.get_proc(self.proc.as_ref())?;
-    ///     adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref()))?;
+    ///     if let Some(settings) =
+    ///         config.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor)
+    ///     {
+    ///         self.settings = settings;
+    ///     }
     /// }
     /// ```
     fn reload_config(&self, _config: Option<&config::Config>) -> Result<(), config::ConfigError> {

prosa/src/core/proc.rs

@@ -116,8 +116,12 @@
 //!                        // TODO process the error
 //!                     },
 //!                     InternalMsg::Config(config) => {
-//!                         self.settings = config.get_proc(self.proc.as_ref())?;
-//!                         adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref()))?;
+//!                         if let Some(settings) =
+//!                             config.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor)
+//!                         {
+//!                             // TODO apply the difference between `settings` and `self.settings`
+//!                             self.settings = settings;
+//!                         }
 //!                     },
 //!                     InternalMsg::Service(table) => self.service = table,
 //!                     InternalMsg::Shutdown => {

prosa/src/core/settings.rs

@@ -20,6 +20,7 @@ use serde::Serialize;
 use serde::de::DeserializeOwned;
 use tokio::sync::mpsc;
 
+use super::adaptor::Adaptor;
 use super::proc::ProcBusParam;
 
 /// Re-export of prosa_utils for observability config
@@ -312,18 +313,61 @@ impl ProsaConfig {
     }
 
     /// Deserialize a processor configuration from its processor name.
-    pub fn get_proc<C>(&self, proc: &impl ProcBusParam) -> Result<C, config::ConfigError>
+    pub fn get_proc<C>(&self, proc: &(impl ProcBusParam + ?Sized)) -> Result<C, config::ConfigError>
     where
         C: DeserializeOwned,
     {
         self.config.get::<C>(&proc.get_proc_config_key())
     }
 
     /// Access a processor adaptor configuration from its processor name.
-    pub fn get_adaptor_config(&self, proc: &impl ProcBusParam) -> Option<&Config> {
+    pub fn get_adaptor_config(&self, proc: &(impl ProcBusParam + ?Sized)) -> Option<&Config> {
         self.adaptor_configs.get(&proc.get_proc_config_key())
     }
 
+    /// Reload a processor settings and its adaptor configuration in one step.
+    ///
+    /// Return `None` if either of them can't be reloaded, so the processor can keep running on its
+    /// current configuration. A processor that has no configuration section keeps the settings it
+    /// was created with, which is reported at debug level. Anything else is logged as a warning:
+    ///
+    /// ```rust,ignore
+    /// InternalMsg::Config(config) => {
+    ///     if let Some(settings) =
+    ///         config.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor)
+    ///     {
+    ///         // ... apply the difference between `settings` and `self.settings`
+    ///         self.settings = settings;
+    ///     }
+    /// }
+    /// ```
+    pub fn reload_proc<S>(&self, proc: &dyn ProcBusParam, adaptor: &dyn Adaptor) -> Option<S>
+    where
+        S: DeserializeOwned,
+    {
+        let settings = match self.get_proc::<S>(proc) {
+            Ok(settings) => settings,
+            Err(config::ConfigError::NotFound(_)) => {
+                log::debug!("No configuration section for processor {}", proc.name());
+                return None;
+            }
+            Err(err) => {
+                log::warn!("Can't reload settings for processor {}: {err}", proc.name());
+                return None;
+            }
+        };
+
+        if let Err(err) = adaptor.reload_config(self.get_adaptor_config(proc)) {
+            log::warn!(
+                "Can't reload adaptor configuration for processor {}: {err}",
+                proc.name()
+            );
+            return None;
+        }
+
+        Some(settings)
+    }
+
     /// Return every configuration path watched to maintain this configuration.
     pub fn watch_paths(&self, config_path: &str) -> Vec<PathBuf> {
         let mut watch_paths = config_watch_paths(Path::new(config_path))
@@ -662,6 +706,107 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_reload_proc() -> Result<(), config::ConfigError> {
+        struct TestProc(&'static str);
+        impl ProcBusParam for TestProc {
+            fn get_proc_id(&self) -> u32 {
+                1
+            }
+
+            fn name(&self) -> &str {
+                self.0
+            }
+        }
+
+        struct TestAdaptor {
+            fail: bool,
+        }
+        impl Adaptor for TestAdaptor {
+            fn reload_config(&self, _config: Option<&Config>) -> Result<(), config::ConfigError> {
+                if self.fail {
+                    Err(config::ConfigError::Message("adaptor failure".into()))
+                } else {
+                    Ok(())
+                }
+            }
+
+            fn terminate(&self) {}
+        }
+
+        #[derive(serde::Deserialize)]
+        struct TestProcSettings {
+            service_name: String,
+        }
+
+        let config = ProsaConfig::from_config(
+            Config::builder()
+                .set_override("proc_1.service_name", "PROC_TEST")?
+                .build()?,
+        )?;
+
+        let settings = config
+            .reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: false })
+            .expect("Processor settings should be reloaded");
+        assert_eq!("PROC_TEST", settings.service_name);
+
+        // The adaptor keeps the processor on its current configuration
+        assert!(
+            config
+                .reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: true })
+                .is_none()
+        );
+
+        // A processor without a configuration section keeps its settings without warning
+        assert!(matches!(
+            config.get_proc::<TestProcSettings>(&TestProc("proc-unknown")),
+            Err(config::ConfigError::NotFound(_))
+        ));
+        assert!(
+            config
+                .reload_proc::<TestProcSettings>(
+                    &TestProc("proc-unknown"),
+                    &TestAdaptor { fail: false }
+                )
+                .is_none()
+        );
+
+        // But an invalid section is a reload failure
+        let invalid_config = ProsaConfig::from_config(
+            Config::builder()
+                .set_override("proc_1.service_name", vec!["not", "a", "string"])?
+                .build()?,
+        )?;
+        assert!(matches!(
+            invalid_config.get_proc::<TestProcSettings>(&TestProc("proc-1")),
+            Err(config::ConfigError::Type { .. })
+        ));
+        assert!(
+            invalid_config
+                .reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: false })
+                .is_none()
+        );
+
+        // So is a section that exists but misses a mandatory setting, which `config` reports as
+        // `At` rather than the `NotFound` of an absent section
+        let incomplete_config = ProsaConfig::from_config(
+            Config::builder()
+                .set_override("proc_1.unrelated", "value")?
+                .build()?,
+        )?;
+        assert!(matches!(
+            incomplete_config.get_proc::<TestProcSettings>(&TestProc("proc-1")),
+            Err(config::ConfigError::At { .. })
+        ));
+        assert!(
+            incomplete_config
+                .reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: false })
+                .is_none()
+        );
+
+        Ok(())
+    }
+
     fn unique_test_dir(prefix: &str) -> PathBuf {
         let timestamp = SystemTime::now()
             .duration_since(UNIX_EPOCH)

prosa/src/inj/proc.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use crate::otel::{KeyValue, metrics::Histogram};
-use crate::tracing::{debug, warn};
+use crate::tracing::debug;
 use prosa_macros::{proc, proc_settings};
 use serde::{Deserialize, Serialize};
 
@@ -198,26 +198,12 @@ impl InjProc {
                 let _ = next_transaction.get_or_insert_with(|| adaptor.build_transaction());
             }
             InternalMsg::Config(config) => {
-                let settings = match config.get_proc::<InjSettings>(self.proc.as_ref()) {
-                    Ok(settings) => settings,
-                    Err(err) => {
-                        warn!("Can't reload settings for processor {}: {err}", self.name());
-                        return Ok(());
-                    }
-                };
-
-                if let Err(err) =
-                    adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref()))
+                if let Some(settings) =
+                    config.reload_proc::<InjSettings>(self.proc.as_ref(), adaptor)
                 {
-                    warn!(
-                        "Can't reload adaptor configuration for processor {}: {err}",
-                        self.name()
-                    );
-                    return Ok(());
+                    *regulator = settings.get_regulator();
+                    self.settings = settings;
                 }
-
-                *regulator = settings.get_regulator();
-                self.settings = settings;
             }
             InternalMsg::Service(table) => self.service = table,
             InternalMsg::Shutdown => {

prosa/src/io.rs

@@ -554,11 +554,10 @@ mod tests {
             .expect("Certificate path should exist")
             .to_string();
 
-        let mut server_ssl_config = SslConfig::new_self_cert(cert_path.clone());
-        server_ssl_config.set_alpn(vec!["prosa/1".into(), "h2".into()]);
-
-        let listener_settings =
+        let server_ssl_config = SslConfig::new_self_cert(cert_path.clone());
+        let mut listener_settings =
             listener::ListenerSetting::new(addr.clone(), Some(server_ssl_config));
+        listener_settings.set_alpn(vec!["prosa/1".into(), "h2".into()]);
         assert!(
             format!("{listener_settings:?}").contains("tls")
                 && format!("{listener_settings:?}").contains("localhost")
@@ -599,9 +598,9 @@ mod tests {
         };
 
         let mut client_ssl_config = SslConfig::default();
-        client_ssl_config.set_alpn(vec!["http/1.1".into(), "prosa/1".into()]);
         client_ssl_config.set_store(Store::File { path: cert_path });
-        let target_settings = stream::TargetSetting::new(addr, Some(client_ssl_config), None);
+        let mut target_settings = stream::TargetSetting::new(addr, Some(client_ssl_config), None);
+        target_settings.set_alpn(vec!["http/1.1".into(), "prosa/1".into()]);
         assert_eq!(addr_str, target_settings.to_string());
 
         let client = async {