Skip to content

feat: improve processor reloading and allow it for SSL #87

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
reneca wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat: improve processor reloading and allow it for SSL #87

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c1da3e0
feat: improve processor reloading and allow it for SSL
reneca Aug 6, 2026
630a54b
feat: listener don't rebind if ssl change
reneca Aug 18, 2026
648d380
fix: certificate reload edge cases
reneca Aug 19, 2026
db0c3be
fix: error with reload_proc and fix according to code review
reneca Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion prosa/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "prosa"
version = "0.5.0"
version = "0.5.1"
authors.workspace = true
description = "ProSA core"
homepage.workspace = true
Expand Down
42 changes: 26 additions & 16 deletions prosa/examples/proc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,23 +79,33 @@ 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?;
match config
.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor)
{
Ok(settings) => {
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;
},
Err(err) => {
warn!(
"Failed to reload configuration for processor {}: {err}",
self.name()
);
}
}

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);
Expand Down
13 changes: 10 additions & 3 deletions prosa/src/core/adaptor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,12 +62,19 @@ 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()))?;
/// match config.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor) {
/// Ok(settings) => self.settings = settings,
/// Err(err) => prosa::tracing::warn!(
/// "Failed to reload configuration for processor {}: {err}",
/// self.name()
/// ),
/// }
/// }
/// ```
fn reload_config(&self, _config: Option<&config::Config>) -> Result<(), config::ConfigError> {
Expand Down
12 changes: 10 additions & 2 deletions prosa/src/core/proc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,8 +116,16 @@
//! // 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()))?;
//! match config.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor) {
//! Ok(settings) => {
//! // TODO apply the difference between `settings` and `self.settings`
//! self.settings = settings;
//! }
//! Err(err) => prosa::tracing::warn!(
//! "Failed to reload configuration for processor {}: {err}",
//! self.name()
//! ),
//! }
//! },
//! InternalMsg::Service(table) => self.service = table,
//! InternalMsg::Shutdown => {
Expand Down
123 changes: 121 additions & 2 deletions prosa/src/core/settings.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -312,18 +313,49 @@ 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's settings and adaptor configuration in one step.
///
/// Returns an error if either cannot be reloaded:
///
/// ```rust,ignore
/// InternalMsg::Config(config) => {
/// match config.reload_proc::<MyProcSettings>(self.proc.as_ref(), &adaptor) {
/// Ok(settings) => {
/// // ... apply the difference between `settings` and `self.settings`
/// self.settings = settings;
/// }
/// Err(err) => prosa::tracing::warn!(
/// "Failed to reload configuration for processor {}: {err}",
/// self.name()
/// ),
/// }
/// }
/// ```
pub fn reload_proc<S>(
&self,
proc: &dyn ProcBusParam,
adaptor: &dyn Adaptor,
) -> Result<S, config::ConfigError>
where
S: DeserializeOwned,
{
let settings = self.get_proc::<S>(proc)?;
adaptor.reload_config(self.get_adaptor_config(proc))?;
Ok(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))
Expand Down Expand Up @@ -662,6 +694,93 @@ 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);

assert!(matches!(
config
.reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: true }),
Err(config::ConfigError::Message(message)) if message == "adaptor failure"
));

// A processor without a configuration section returns the deserialization error
assert!(matches!(
config.reload_proc::<TestProcSettings>(
&TestProc("proc-unknown"),
&TestAdaptor { fail: false }
),
Err(config::ConfigError::NotFound(_))
));

// An invalid section returns the deserialization error
let invalid_config = ProsaConfig::from_config(
Config::builder()
.set_override("proc_1.service_name", vec!["not", "a", "string"])?
.build()?,
)?;
assert!(matches!(
invalid_config
.reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: false }),
Err(config::ConfigError::Type { .. })
));

// A section that misses a mandatory setting returns `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
.reload_proc::<TestProcSettings>(&TestProc("proc-1"), &TestAdaptor { fail: false }),
Err(config::ConfigError::At { .. })
));

Ok(())
}

fn unique_test_dir(prefix: &str) -> PathBuf {
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
Expand Down
26 changes: 9 additions & 17 deletions prosa/src/inj/proc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,26 +198,18 @@ 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,
match config.reload_proc::<InjSettings>(self.proc.as_ref(), adaptor) {
Ok(settings) => {
*regulator = settings.get_regulator();
self.settings = settings;
}
Err(err) => {
warn!("Can't reload settings for processor {}: {err}", self.name());
return Ok(());
warn!(
"Failed to reload configuration for processor {}: {err}",
self.name()
);
}
};

if let Err(err) =
adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref()))
{
warn!(
"Can't reload adaptor configuration for processor {}: {err}",
self.name()
);
return Ok(());
}

*regulator = settings.get_regulator();
self.settings = settings;
}
InternalMsg::Service(table) => self.service = table,
InternalMsg::Shutdown => {
Expand Down
11 changes: 5 additions & 6 deletions prosa/src/io.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading