Repository: slawlor/ractor
Pull request #333 (2025-02-18 16:54:49)

Add ability for a factory's internal state to be updated dynamically
This allows someone to say reload configeration settings, and internally tear-down and rebuild the factory without fully shutting down and restarting.

Some settings are not dynamically settable without tearing down and rebuilding the factory, unless we want to introduce more strong-types to the message type. This includes, unfortunately, the worker builder, queue, and router.

For settings on those two, there are two choices. (1) tear down and rebuild the factory or (2) thread in dynamic settings to your router and/or queue protocols such that they can internally change their functionality.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/slawlor/ractor/pull/333?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) Report
Attention: Patch coverage is `80.88235%` with `39 lines` in your changes missing coverage. Please review.
> Project coverage is 82.30%. Comparing base [(`2f864f2`)](https://app.codecov.io/gh/slawlor/ractor/commit/2f864f22204f6f71f5b831da3a57f5de9ba849bb?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) to head [(`57329d5`)](https://app.codecov.io/gh/slawlor/ractor/commit/57329d5d3d477db7886c1942cfb458c67d905f57?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).

| [Files with missing lines](https://app.codecov.io/gh/slawlor/ractor/pull/333?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) | Patch % | Lines |
|---|---|---|
| [ractor/src/factory/factoryimpl.rs](https://app.codecov.io/gh/slawlor/ractor/pull/333?src=pr&el=tree&filepath=ractor%2Fsrc%2Ffactory%2Ffactoryimpl.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9mYWN0b3J5L2ZhY3RvcnlpbXBsLnJz) | 61.29% | [24 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/333?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |
| [ractor/src/factory/mod.rs](https://app.codecov.io/gh/slawlor/ractor/pull/333?src=pr&el=tree&filepath=ractor%2Fsrc%2Ffactory%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9mYWN0b3J5L21vZC5ycw==) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/333?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |
| [ractor/src/factory/tests/dynamic\_settings.rs](https://app.codecov.io/gh/slawlor/ractor/pull/333?src=pr&el=tree&filepath=ractor%2Fsrc%2Ffactory%2Ftests%2Fdynamic_settings.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9mYWN0b3J5L3Rlc3RzL2R5bmFtaWNfc2V0dGluZ3MucnM=) | 99.21% | [1 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/333?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #333      +/-   ##
==========================================
+ Coverage   82.18%   82.30%   +0.11%     
==========================================
  Files          64       65       +1     
  Lines       12692    12895     +203     
==========================================
+ Hits        10431    10613     +182     
- Misses       2261     2282      +21     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/slawlor/ractor/pull/333?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).


Patch:
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index d593c9d..fca52fd 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.15.1"
+version = "0.15.2"
 authors = ["Sean Lawlor", "Evan Au", "Dillon George"]
 description = "A actor framework for Rust"
 documentation = "https://docs.rs/ractor"
diff --git a/ractor/src/factory/discard.rs b/ractor/src/factory/discard.rs
index a5ffaf2..f6a6ae0 100644
--- a/ractor/src/factory/discard.rs
+++ b/ractor/src/factory/discard.rs
@@ -28,7 +28,7 @@ pub enum DiscardMode {
 /// workers. The workers "think" it's static, but the factory handles the dynamics.
 /// This way the factory can keep the [DynamicDiscardHandler] as a single, uncloned
 /// instance. It also moves NUM_WORKER calculations to 1.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub(crate) enum WorkerDiscardSettings {
     None,
     Static { limit: usize, mode: DiscardMode },
diff --git a/ractor/src/factory/factoryimpl.rs b/ractor/src/factory/factoryimpl.rs
index 100b7e6..3105ed7 100644
--- a/ractor/src/factory/factoryimpl.rs
+++ b/ractor/src/factory/factoryimpl.rs
@@ -691,6 +691,72 @@ where
         Ok(())
     }
 
+    async fn update_settings(
+        &mut self,
+        myself: &ActorRef<FactoryMessage<TKey, TMsg>>,
+        UpdateSettingsRequest {
+            discard_handler,
+            discard_settings,
+            dead_mans_switch,
+            capacity_controller,
+            lifecycle_hooks,
+            stats,
+            worker_count,
+        }: UpdateSettingsRequest<TKey, TMsg>,
+    ) -> Result<(), SpawnErr> {
+        if let Some(discard_handler) = discard_handler {
+            tracing::debug!(
+                "Updating discard handler: HashValue={}",
+                discard_handler.is_some()
+            );
+            for (_, worker) in self.pool.iter_mut() {
+                worker.discard_handler = discard_handler.clone();
+            }
+            self.discard_handler = discard_handler;
+        }
+        if let Some(discard_settings) = discard_settings {
+            tracing::debug!("Updating discard settings: {discard_settings:?}");
+            let worker_discard_settings = if self.router.is_factory_queueing() {
+                discard::WorkerDiscardSettings::None
+            } else {
+                discard_settings.get_worker_settings()
+            };
+            for (_, worker) in self.pool.iter_mut() {
+                worker.discard_settings = worker_discard_settings.clone();
+            }
+            self.discard_settings = discard_settings;
+        }
+        if let Some(dead_mans_switch) = dead_mans_switch {
+            tracing::debug!(
+                "Updating dead man's switch settings: HasValue={}",
+                dead_mans_switch.is_some()
+            );
+            self.dead_mans_switch = dead_mans_switch;
+        }
+        if let Some(capacity_controller) = capacity_controller {
+            tracing::debug!(
+                "Updating capacity controller: HasValue={}",
+                capacity_controller.is_some()
+            );
+            self.capacity_controller = capacity_controller;
+        }
+        if let Some(lifecycle_hooks) = lifecycle_hooks {
+            tracing::debug!(
+                "Updating lifecycle hooks: HasValue={}",
+                lifecycle_hooks.is_some()
+            );
+            self.lifecycle_hooks = lifecycle_hooks;
+        }
+        if let Some(stats) = stats {
+            tracing::debug!("Updating factory stats layer: HasValue={}", stats.is_some());
+            self.stats = stats;
+        }
+        if let Some(worker_count) = worker_count {
+            self.resize_pool(myself, worker_count).await?;
+        }
+        Ok(())
+    }
+
     fn reply_with_available_capacity(&self, reply: RpcReplyPort<usize>) {
         // calculate the worker's free capacity
         let worker_availability = self
@@ -701,7 +767,7 @@ where
         match self.discard_settings.get_limit_and_mode() {
             Some((limit, _)) => {
                 // get the queue space and add it to the worker availability
-                let count = (limit - self.queue.len()) + worker_availability;
+                let count = std::cmp::max(limit - self.queue.len(), 0) + worker_availability;
                 let _ = reply.send(count);
             }
             None => {
@@ -972,6 +1038,9 @@ where
             FactoryMessage::DrainRequests => {
                 state.drain_requests(&myself).await?;
             }
+            FactoryMessage::UpdateSettings(settings) => {
+                state.update_settings(&myself, settings).await?;
+            }
         }
 
         if state.is_drained() {
diff --git a/ractor/src/factory/mod.rs b/ractor/src/factory/mod.rs
index b84b323..3af9178 100644
--- a/ractor/src/factory/mod.rs
+++ b/ractor/src/factory/mod.rs
@@ -158,6 +158,8 @@
 //! }
 //! ```
 
+use std::sync::Arc;
+
 use crate::concurrency::{Duration, Instant};
 #[cfg(feature = "cluster")]
 use crate::message::BoxedDowncastErr;
@@ -191,6 +193,54 @@ pub use worker::{
     WorkerProperties, WorkerStartContext,
 };
 
+/// The settings to change for an update request to the factory at runtime.
+///
+/// Note: A value of `Some(..)` means that the internal value should be updated
+/// inside the factory's state. For values which are originally optional to the factory,
+/// we use `Option<Option<T>>`, so if you want to UNSET the value, it would be `Some(None)`.
+#[derive(bon::Builder)]
+pub struct UpdateSettingsRequest<TKey, TMsg>
+where
+    TKey: JobKey,
+    TMsg: Message,
+{
+    /// The discard handler callback
+    pub discard_handler: Option<Option<Arc<dyn DiscardHandler<TKey, TMsg>>>>,
+    /// The discard settings
+    pub discard_settings: Option<DiscardSettings>,
+    /// Dead-man's switch settings
+    pub dead_mans_switch: Option<Option<DeadMansSwitchConfiguration>>,
+    /// Capacity controller
+    pub capacity_controller: Option<Option<Box<dyn WorkerCapacityController>>>,
+    /// Lifecycle hooks
+    pub lifecycle_hooks: Option<Option<Box<dyn FactoryLifecycleHooks<TKey, TMsg>>>>,
+    /// Statistics layer
+    pub stats: Option<Option<Arc<dyn FactoryStatsLayer>>>,
+    /// The worker count
+    pub worker_count: Option<usize>,
+}
+
+impl<TKey, TMsg> std::fmt::Debug for UpdateSettingsRequest<TKey, TMsg>
+where
+    TKey: JobKey,
+    TMsg: Message,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("UpdateSettingsRequest")
+            .field("set_discard_handler", &self.discard_handler.is_some())
+            .field("set_discard_settings", &self.discard_settings.is_some())
+            .field("set_dead_mans_switch", &self.dead_mans_switch.is_some())
+            .field(
+                "set_capacity_controller",
+                &self.capacity_controller.is_some(),
+            )
+            .field("set_lifecycle_hooks", &self.lifecycle_hooks.is_some())
+            .field("set_stats", &self.stats.is_some())
+            .field("set_worker_count", &self.worker_count.is_some())
+            .finish()
+    }
+}
+
 /// Unique identifier of a disctinct worker in the factory
 pub type WorkerId = usize;
 /// Messages to a factory.
@@ -252,6 +302,18 @@ where
     /// Therefore in order to propertly drain a factory, you should use the
     /// `DrainRequests` version so the internal pending queue is properly flushed.
     DrainRequests,
+
+    /// Dynamically update the factory's settings, for those which don't require strong-type
+    /// guarantees. This allows, at runtime, changing the
+    ///
+    /// * Worker Count
+    /// * Discard Settings
+    /// * Lifecycle Hooks
+    /// * Statistics collection
+    /// * Capacity controller
+    /// * Dead-man's switch
+    /// * Discard handler
+    UpdateSettings(UpdateSettingsRequest<TKey, TMsg>),
 }
 
 #[cfg(feature = "cluster")]
diff --git a/ractor/src/factory/worker.rs b/ractor/src/factory/worker.rs
index 5d8526b..a0aebcd 100644
--- a/ractor/src/factory/worker.rs
+++ b/ractor/src/factory/worker.rs
@@ -496,10 +496,10 @@ where
     /// will cause an oldest job at the head of the queue will be dropped.
     ///
     /// Default is [WorkerDiscardSettings::None]
-    discard_settings: WorkerDiscardSettings,
+    pub(crate) discard_settings: WorkerDiscardSettings,
 
     /// A function to be called for each job to be dropped.
-    discard_handler: Option<Arc<dyn DiscardHandler<TKey, TMsg>>>,
+    pub(crate) discard_handler: Option<Arc<dyn DiscardHandler<TKey, TMsg>>>,
 
     /// Flag indicating if this worker has a ping currently pending
     is_ping_pending: bool,
diff --git a/ractor_cluster/Cargo.toml b/ractor_cluster/Cargo.toml
index 696280c..cf53627 100644
--- a/ractor_cluster/Cargo.toml
+++ b/ractor_cluster/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster"
-version = "0.15.1"
+version = "0.15.2"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Distributed cluster environment of Ractor actors"
 documentation = "https://docs.rs/ractor"
diff --git a/ractor_cluster_derive/Cargo.toml b/ractor_cluster_derive/Cargo.toml
index 5acac51..a4da697 100644
--- a/ractor_cluster_derive/Cargo.toml
+++ b/ractor_cluster_derive/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster_derive"
-version = "0.15.1"
+version = "0.15.2"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Derives for ractor_cluster"
 license = "MIT"


Test patch:
diff --git a/ractor/src/factory/tests/dynamic_settings.rs b/ractor/src/factory/tests/dynamic_settings.rs
new file mode 100644
index 0000000..7af003c
--- /dev/null
+++ b/ractor/src/factory/tests/dynamic_settings.rs
@@ -0,0 +1,197 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Tests around dynamic setting changes in factories
+
+use std::sync::{
+    atomic::{AtomicU8, Ordering},
+    Arc,
+};
+
+use crate::concurrency::sleep;
+use crate::factory::*;
+use crate::{Actor, ActorProcessingErr, ActorRef};
+
+struct TestWorker;
+
+#[cfg_attr(feature = "async-trait", crate::async_trait)]
+impl Worker for TestWorker {
+    type State = ();
+    type Key = ();
+    type Message = ();
+    type Arguments = ();
+
+    async fn pre_start(
+        &self,
+        _wid: WorkerId,
+        _factory: &ActorRef<FactoryMessage<Self::Key, Self::Message>>,
+        _args: Self::Arguments,
+    ) -> Result<Self::State, ActorProcessingErr> {
+        tracing::debug!("Worker started");
+        Ok(())
+    }
+
+    async fn handle(
+        &self,
+        _wid: WorkerId,
+        _factory: &ActorRef<FactoryMessage<Self::Key, Self::Message>>,
+        _job: Job<Self::Key, Self::Message>,
+        _state: &mut Self::State,
+    ) -> Result<Self::Key, ActorProcessingErr> {
+        tracing::debug!("Worker received dispatch");
+        sleep(Duration::from_millis(10)).await;
+        Ok(())
+    }
+}
+
+struct TestWorkerBuilder;
+
+impl WorkerBuilder<TestWorker, ()> for TestWorkerBuilder {
+    fn build(&mut self, _wid: crate::factory::WorkerId) -> (TestWorker, ()) {
+        (TestWorker, ())
+    }
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_dynamic_settings() {
+    let counter_one = Arc::new(AtomicU8::new(0));
+    let counter_two = Arc::new(AtomicU8::new(0));
+
+    struct TestDiscardHandler {
+        counter: Arc<AtomicU8>,
+    }
+
+    impl DiscardHandler<(), ()> for TestDiscardHandler {
+        fn discard(&self, _reason: DiscardReason, _job: &mut Job<(), ()>) {
+            self.counter.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    let factory_definition = Factory::<
+        (),
+        (),
+        (),
+        TestWorker,
+        routing::QueuerRouting<(), ()>,
+        queues::DefaultQueue<(), ()>,
+    >::default();
+    let args = FactoryArguments::builder()
+        .num_initial_workers(1)
+        .queue(Default::default())
+        .router(Default::default())
+        .worker_builder(Box::new(TestWorkerBuilder))
+        .discard_handler(Arc::new(TestDiscardHandler {
+            counter: counter_one.clone(),
+        }))
+        .discard_settings(DiscardSettings::Static {
+            limit: 0,
+            mode: DiscardMode::Newest,
+        })
+        .build();
+    let (factory, factory_handle) = Actor::spawn(None, factory_definition, args)
+        .await
+        .expect("Failed to spawn factory");
+
+    // check that there's 1 worker running
+    let worker_count = factory
+        .call(
+            FactoryMessage::GetAvailableCapacity,
+            Some(Duration::from_secs(1)),
+        )
+        .await
+        .expect("Failed to message factory")
+        .expect("Failed to get reply from factory");
+    assert_eq!(1, worker_count);
+
+    // send 2 messages, making sure we discard one.
+    for _ in 0..2 {
+        factory
+            .cast(FactoryMessage::Dispatch(Job {
+                accepted: None,
+                key: (),
+                msg: (),
+                options: JobOptions::default(),
+            }))
+            .expect("Failed to message factory");
+    }
+
+    // fetch the worker count, to make sure the factory has processed all
+    // the messages from above
+    _ = factory
+        .call(
+            FactoryMessage::GetAvailableCapacity,
+            Some(Duration::from_secs(1)),
+        )
+        .await
+        .expect("Failed to message factory")
+        .expect("Failed to get reply from factory");
+
+    // wait for worker to finish
+    sleep(Duration::from_millis(100)).await;
+
+    // update factory logic
+    factory
+        .cast(FactoryMessage::UpdateSettings(
+            UpdateSettingsRequest::builder()
+                .worker_count(2)
+                .discard_handler(Some(Arc::new(TestDiscardHandler {
+                    counter: counter_two.clone(),
+                })))
+                // these are the defaults, but exercise the path for sanity.
+                .capacity_controller(None)
+                .dead_mans_switch(None)
+                .discard_settings(DiscardSettings::Static {
+                    limit: 0,
+                    mode: DiscardMode::Newest,
+                })
+                .lifecycle_hooks(None)
+                .stats(None)
+                .build(),
+        ))
+        .expect("Failed to send request to update factory");
+
+    // Check updated worker count
+    let worker_count = factory
+        .call(
+            FactoryMessage::GetAvailableCapacity,
+            Some(Duration::from_secs(1)),
+        )
+        .await
+        .expect("Failed to message factory")
+        .expect("Failed to get reply from factory");
+    assert_eq!(2, worker_count);
+
+    // send 3 messages, making sure we discard one with the
+    // new handler
+    for _ in 0..3 {
+        factory
+            .cast(FactoryMessage::Dispatch(Job {
+                accepted: None,
+                key: (),
+                msg: (),
+                options: JobOptions::default(),
+            }))
+            .expect("Failed to message factory");
+    }
+
+    // Make sure messages processed
+    _ = factory
+        .call(
+            FactoryMessage::GetAvailableCapacity,
+            Some(Duration::from_secs(1)),
+        )
+        .await
+        .expect("Failed to message factory")
+        .expect("Failed to get reply from factory");
+
+    // Check both discard handler counters
+    assert_eq!(counter_one.load(Ordering::SeqCst), 1);
+    assert_eq!(counter_two.load(Ordering::SeqCst), 1);
+
+    // Cleanup
+    factory.stop(None);
+    factory_handle.await.unwrap();
+}
diff --git a/ractor/src/factory/tests/mod.rs b/ractor/src/factory/tests/mod.rs
index 5cccd46..a57650b 100644
--- a/ractor/src/factory/tests/mod.rs
+++ b/ractor/src/factory/tests/mod.rs
@@ -9,6 +9,7 @@ mod basic;
 mod draining_requests;
 mod dynamic_discarding;
 mod dynamic_pool;
+mod dynamic_settings;
 mod lifecycle;
 mod priority_queueing;
 mod ratelim;