Add silent-retry support for factory messages + support for factories without async-trait

ractor/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.12.2"
+version = "0.12.3"
 authors = ["Sean Lawlor", "Evan Au", "Dillon George"]
 description = "A actor framework for Rust"
 documentation = "https://docs.rs/ractor"

ractor/src/actor/actor_cell.rs

@@ -103,7 +103,7 @@ impl ActorPortSet {
     /// Returns [Ok(`TState`)] when the future completes without
     /// signal interruption, [Err(Signal)] in the event the
     /// signal interrupts the async work.
-    pub async fn run_with_signal<TState>(
+    pub(crate) async fn run_with_signal<TState>(
         &mut self,
         future: impl std::future::Future<Output = TState>,
     ) -> Result<TState, Signal>
@@ -148,7 +148,9 @@ impl ActorPortSet {
     ///
     /// Returns [Ok(ActorPortMessage)] on a successful message reception, [MessagingErr]
     /// in the event any of the channels is closed.
-    pub async fn listen_in_priority(&mut self) -> Result<ActorPortMessage, MessagingErr<()>> {
+    pub(crate) async fn listen_in_priority(
+        &mut self,
+    ) -> Result<ActorPortMessage, MessagingErr<()>> {
         #[cfg(feature = "async-std")]
         {
             crate::concurrency::select! {

ractor/src/actor/actor_properties.rs

@@ -42,7 +42,7 @@ pub(crate) struct ActorProperties {
 }
 
 impl ActorProperties {
-    pub fn new<TActor>(
+    pub(crate) fn new<TActor>(
         name: Option<ActorName>,
     ) -> (
         Self,
@@ -57,7 +57,7 @@ impl ActorProperties {
         Self::new_remote::<TActor>(name, crate::actor::actor_id::get_new_local_id())
     }
 
-    pub fn new_remote<TActor>(
+    pub(crate) fn new_remote<TActor>(
         name: Option<ActorName>,
         id: ActorId,
     ) -> (
@@ -96,7 +96,7 @@ impl ActorProperties {
         )
     }
 
-    pub fn get_status(&self) -> ActorStatus {
+    pub(crate) fn get_status(&self) -> ActorStatus {
         match self.status.load(Ordering::SeqCst) {
             0u8 => ActorStatus::Unstarted,
             1u8 => ActorStatus::Starting,
@@ -108,11 +108,11 @@ impl ActorProperties {
         }
     }
 
-    pub fn set_status(&self, status: ActorStatus) {
+    pub(crate) fn set_status(&self, status: ActorStatus) {
         self.status.store(status as u8, Ordering::SeqCst);
     }
 
-    pub fn send_signal(&self, signal: Signal) -> Result<(), MessagingErr<()>> {
+    pub(crate) fn send_signal(&self, signal: Signal) -> Result<(), MessagingErr<()>> {
         self.signal
             .lock()
             .unwrap()
@@ -122,14 +122,17 @@ impl ActorProperties {
             })
     }
 
-    pub fn send_supervisor_evt(
+    pub(crate) fn send_supervisor_evt(
         &self,
         message: SupervisionEvent,
     ) -> Result<(), MessagingErr<SupervisionEvent>> {
         self.supervision.send(message).map_err(|e| e.into())
     }
 
-    pub fn send_message<TMessage>(&self, message: TMessage) -> Result<(), MessagingErr<TMessage>>
+    pub(crate) fn send_message<TMessage>(
+        &self,
+        message: TMessage,
+    ) -> Result<(), MessagingErr<TMessage>>
     where
         TMessage: Message,
     {
@@ -156,7 +159,7 @@ impl ActorProperties {
             })
     }
 
-    pub fn drain(&self) -> Result<(), MessagingErr<()>> {
+    pub(crate) fn drain(&self) -> Result<(), MessagingErr<()>> {
         let _ = self
             .status
             .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |f| {
@@ -172,7 +175,7 @@ impl ActorProperties {
     }
 
     #[cfg(feature = "cluster")]
-    pub fn send_serialized(
+    pub(crate) fn send_serialized(
         &self,
         message: SerializedMessage,
     ) -> Result<(), MessagingErr<SerializedMessage>> {
@@ -189,7 +192,10 @@ impl ActorProperties {
             })
     }
 
-    pub fn send_stop(&self, reason: Option<String>) -> Result<(), MessagingErr<StopMessage>> {
+    pub(crate) fn send_stop(
+        &self,
+        reason: Option<String>,
+    ) -> Result<(), MessagingErr<StopMessage>> {
         let msg = reason.map(StopMessage::Reason).unwrap_or(StopMessage::Stop);
         self.stop
             .lock()
@@ -201,7 +207,7 @@ impl ActorProperties {
     }
 
     /// Send the stop signal, threading in a OneShot sender which notifies when the shutdown is completed
-    pub async fn send_stop_and_wait(
+    pub(crate) async fn send_stop_and_wait(
         &self,
         reason: Option<String>,
     ) -> Result<(), MessagingErr<StopMessage>> {
@@ -212,15 +218,18 @@ impl ActorProperties {
     }
 
     /// Send the kill signal, threading in a OneShot sender which notifies when the shutdown is completed
-    pub async fn send_signal_and_wait(&self, signal: Signal) -> Result<(), MessagingErr<()>> {
+    pub(crate) async fn send_signal_and_wait(
+        &self,
+        signal: Signal,
+    ) -> Result<(), MessagingErr<()>> {
         // first bind the wait handler
         let rx = self.wait_handler.notified();
         let _ = self.send_signal(signal);
         rx.await;
         Ok(())
     }
 
-    pub fn notify_stop_listener(&self) {
+    pub(crate) fn notify_stop_listener(&self) {
         self.wait_handler.notify_waiters();
     }
 }

ractor/src/actor/messages.rs

@@ -24,6 +24,12 @@
     pub msg: Option<Box<dyn Any + Send>>,
 }
 
+impl Debug for BoxedState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "BoxedState")
+    }
+}
+
 impl BoxedState {
     /// Create a new [BoxedState] from a strongly-typed message
     pub fn new<T>(msg: T) -> Self

ractor/src/actor/mod.rs

@@ -50,6 +50,7 @@
 //! log to `stderr` for tracing. You can additionally setup a [panic hook](https://doc.rust-lang.org/std/panic/fn.set_hook.html)
 //! to do things like capturing backtraces on the unwinding panic.
 
+use std::fmt::Debug;
 #[cfg(not(feature = "async-trait"))]
 use std::future::Future;
 use std::panic::AssertUnwindSafe;
@@ -478,6 +479,16 @@
     name: Option<String>,
 }
 
+impl<TActor: Actor> Debug for ActorRuntime<TActor> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        if let Some(name) = self.name.as_ref() {
+            write!(f, "ActorRuntime('{}' - {})", name, self.id)
+        } else {
+            write!(f, "ActorRuntime({})", self.id)
+        }
+    }
+}
+
 impl<TActor> ActorRuntime<TActor>
 where
     TActor: Actor,

ractor/src/actor/supervision.rs

@@ -21,37 +21,37 @@ use super::{actor_cell::ActorCell, messages::SupervisionEvent};
 use crate::ActorId;
 
 /// A supervision tree
-#[derive(Default)]
-pub struct SupervisionTree {
+#[derive(Default, Debug)]
+pub(crate) struct SupervisionTree {
     children: Arc<Mutex<HashMap<ActorId, ActorCell>>>,
     supervisor: Arc<Mutex<Option<ActorCell>>>,
 }
 
 impl SupervisionTree {
     /// Push a child into the tere
-    pub fn insert_child(&self, child: ActorCell) {
+    pub(crate) fn insert_child(&self, child: ActorCell) {
         self.children.lock().unwrap().insert(child.get_id(), child);
     }
 
     /// Remove a specific actor from the supervision tree (e.g. actor died)
-    pub fn remove_child(&self, child: ActorId) {
+    pub(crate) fn remove_child(&self, child: ActorId) {
         self.children.lock().unwrap().remove(&child);
     }
 
     /// Push a parent into the tere
-    pub fn set_supervisor(&self, parent: ActorCell) {
+    pub(crate) fn set_supervisor(&self, parent: ActorCell) {
         *(self.supervisor.lock().unwrap()) = Some(parent);
     }
 
     /// Remove a specific actor from the supervision tree (e.g. actor died)
-    pub fn clear_supervisor(&self) {
+    pub(crate) fn clear_supervisor(&self) {
         *(self.supervisor.lock().unwrap()) = None;
     }
 
     /// Terminate all your supervised children and unlink them
     /// from the supervision tree since the supervisor is shutting down
     /// and can't deal with superivison events anyways
-    pub fn terminate_all_children(&self) {
+    pub(crate) fn terminate_all_children(&self) {
         let mut guard = self.children.lock().unwrap();
         let cells = guard.iter().map(|(_, a)| a.clone()).collect::<Vec<_>>();
         guard.clear();
@@ -64,7 +64,7 @@ impl SupervisionTree {
     }
 
     /// Determine if the specified actor is a parent of this actor
-    pub fn is_child_of(&self, id: ActorId) -> bool {
+    pub(crate) fn is_child_of(&self, id: ActorId) -> bool {
         if let Some(parent) = &*(self.supervisor.lock().unwrap()) {
             parent.get_id() == id
         } else {
@@ -73,21 +73,21 @@ impl SupervisionTree {
     }
 
     /// Send a notification to the supervisor.
-    pub fn notify_supervisor(&self, evt: SupervisionEvent) {
+    pub(crate) fn notify_supervisor(&self, evt: SupervisionEvent) {
         if let Some(parent) = &*(self.supervisor.lock().unwrap()) {
             let _ = parent.send_supervisor_evt(evt);
         }
     }
 
     /// Retrieve the number of supervised children
     #[cfg(test)]
-    pub fn get_num_children(&self) -> usize {
+    pub(crate) fn get_num_children(&self) -> usize {
         self.children.lock().unwrap().len()
     }
 
     /// Retrieve the number of supervised children
     #[cfg(test)]
-    pub fn get_num_parents(&self) -> usize {
+    pub(crate) fn get_num_parents(&self) -> usize {
         usize::from(self.supervisor.lock().unwrap().is_some())
     }
 }

ractor/src/actor/tests/mod.rs

@@ -967,7 +967,7 @@ fn returns_actor_references() {
 async fn actor_failing_in_spawn_err_doesnt_poison_registries() {
     struct Test;
 
-    #[crate::async_trait]
+    #[cfg_attr(feature = "async-trait", crate::async_trait)]
     impl Actor for Test {
         type Msg = ();
         type State = ();
@@ -980,7 +980,7 @@ async fn actor_failing_in_spawn_err_doesnt_poison_registries() {
 
     struct Test2;
 
-    #[crate::async_trait]
+    #[cfg_attr(feature = "async-trait", crate::async_trait)]
     impl Actor for Test2 {
         type Msg = ();
         type State = ();

ractor/src/factory/discard.rs

@@ -9,7 +9,7 @@
 use crate::Message;
 
 /// The discard mode of a factory
-#[derive(Eq, PartialEq, Clone, Copy)]
+#[derive(Debug, Eq, PartialEq, Clone, Copy)]
 pub enum DiscardMode {
     /// Discard oldest incoming jobs under backpressure
     Oldest,
@@ -28,6 +28,7 @@
 /// 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)]
 pub(crate) enum WorkerDiscardSettings {
     None,
     Static { limit: usize, mode: DiscardMode },
@@ -84,6 +85,26 @@
     },
 }
 
+impl std::fmt::Debug for DiscardSettings {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DiscardSettings::None => {
+                write!(f, "DiscardSettings::None")
+            }
+            DiscardSettings::Static { limit, mode } => f
+                .debug_struct("DiscardSettings::Static")
+                .field("limit", limit)
+                .field("mode", mode)
+                .finish(),
+            DiscardSettings::Dynamic { limit, mode, .. } => f
+                .debug_struct("DiscardSettings::Dynamic")
+                .field("limit", limit)
+                .field("mode", mode)
+                .finish(),
+        }
+    }
+}
+
 impl DiscardSettings {
     pub(crate) fn get_worker_settings(&self) -> WorkerDiscardSettings {
         match &self {
@@ -124,10 +145,26 @@
     /// `base_controller.factory.{FACTORY_NAME}.{STAT}`
     ///
     /// If no factory name is set, then "all" will be inserted
+    #[cfg(feature = "async-trait")]
     async fn compute(&mut self, current_threshold: usize) -> usize;
+
+    /// Compute the new threshold for discarding
+    ///
+    /// If you want to utilize metrics exposed in [crate::factory::stats] you can gather them
+    /// by utilizing `stats_facebook::service_data::get_service_data_singleton` to retrieve a
+    /// accessor to `ServiceData` which you can then resolve stats by name (either timeseries or
+    /// counters)
+    ///
+    /// The namespace of stats collected on the base controller factory are
+    /// `base_controller.factory.{FACTORY_NAME}.{STAT}`
+    ///
+    /// If no factory name is set, then "all" will be inserted
+    #[cfg(not(feature = "async-trait"))]
+    fn compute(&mut self, current_threshold: usize) -> futures::future::BoxFuture<'_, usize>;
 }
 
 /// Reason for discarding a job
+#[derive(Debug)]
 pub enum DiscardReason {
     /// The job TTLd
     TtlExpired,