Repository: slawlor/ractor
Pull request #353 (2025-05-07 14:15:52)

Fix bug in draining non-factory-queued routing
When the queueing occurs on workers directly, the factory wasn't tracking the in-flight message count correctly, such that the worker might still have work in-flight when requesting a draining operation.

This change fixes 2 things

1. The evaluation of if a worker is "empty" of all jobs, i.e. drained, by checking the current & queued jobs
2. The tracking of the current job on the worker, when a worker has locally dequeued the next job.

Test coverage added for this scenario. Prior to the fix, the added test fails with

```
thread 'factory::tests::draining_requests::test_worker_drains_when_worker_queueing' panicked at ractor/src/factory/tests/draining_requests.rs:221:5:
assertion `left == right` failed
  left: 999
 right: 1
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
```

since the factory was marked "drained" prior to all messages completing. Now it passes with this patch.

Also docs updated.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/slawlor/ractor/pull/353?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 `92.00000%` with `4 lines` in your changes missing coverage. Please review.
> Project coverage is 80.99%. Comparing base [(`642bdcd`)](https://app.codecov.io/gh/slawlor/ractor/commit/642bdcd472afdb3f7c80bf254d963d2c5892b081?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) to head [(`ae4c55e`)](https://app.codecov.io/gh/slawlor/ractor/commit/ae4c55e3821753a255cc6aa26e42af1d2d075a19?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/353?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/tests/draining\_requests.rs](https://app.codecov.io/gh/slawlor/ractor/pull/353?src=pr&el=tree&filepath=ractor%2Fsrc%2Ffactory%2Ftests%2Fdraining_requests.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9mYWN0b3J5L3Rlc3RzL2RyYWluaW5nX3JlcXVlc3RzLnJz) | 91.48% | [4 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/353?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     #353      +/-   ##
==========================================
+ Coverage   80.95%   80.99%   +0.03%     
==========================================
  Files          70       70              
  Lines       14367    14415      +48     
==========================================
+ Hits        11631    11675      +44     
- Misses       2736     2740       +4     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/slawlor/ractor/pull/353?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).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>

Patch:
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index 0edb86e..9a5c76d 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.15.5"
+version = "0.15.6"
 authors = ["Sean Lawlor", "Evan Au", "Dillon George"]
 description = "A actor framework for Rust"
 documentation = "https://docs.rs/ractor"
diff --git a/ractor/src/factory.rs b/ractor/src/factory.rs
index 3af9178..4e36b2f 100644
--- a/ractor/src/factory.rs
+++ b/ractor/src/factory.rs
@@ -83,7 +83,7 @@
 //!         &self,
 //!         wid: WorkerId,
 //!         factory: &ActorRef<FactoryMessage<(), ExampleMessage>>,
-//!         Job {msg, key, ..}: Job<(), ExampleMessage>,
+//!         Job { msg, key, .. }: Job<(), ExampleMessage>,
 //!         _state: &mut Self::State,
 //!     ) -> Result<(), ActorProcessingErr> {
 //!         // Actual business logic that we want to parallelize
@@ -115,7 +115,7 @@
 //!         (),
 //!         ExampleWorker,
 //!         routing::QueuerRouting<(), ExampleMessage>,
-//!         queues::DefaultQueue<(), ExampleMessage>
+//!         queues::DefaultQueue<(), ExampleMessage>,
 //!     >::default();
 //!     let factory_args = FactoryArguments::builder()
 //!         .worker_builder(Box::new(ExampleWorkerBuilder))
@@ -123,7 +123,7 @@
 //!         .router(Default::default())
 //!         .num_initial_workers(5)
 //!         .build();
-//!     
+//!
 //!     let (factory, handle) = Actor::spawn(None, factory_def, factory_args)
 //!         .await
 //!         .expect("Failed to startup factory");
diff --git a/ractor/src/factory/worker.rs b/ractor/src/factory/worker.rs
index fe6b5be..54712ab 100644
--- a/ractor/src/factory/worker.rs
+++ b/ractor/src/factory/worker.rs
@@ -616,12 +616,12 @@ where
 
     /// Identify if the worker is available for enqueueing work
     pub fn is_available(&self) -> bool {
-        self.curr_jobs.is_empty()
+        self.curr_jobs.is_empty() && self.message_queue.is_empty()
     }
 
     /// Identify if the worker is currently processing any requests
     pub fn is_working(&self) -> bool {
-        !self.curr_jobs.is_empty()
+        !self.is_available()
     }
 
     /// Denotes if the worker is stuck (i.e. unable to complete it's current job)
@@ -721,6 +721,7 @@ where
         let options = self.curr_jobs.remove(&key);
         // maybe queue up the next job
         if let Some(mut job) = self.get_next_non_expired_job() {
+            self.curr_jobs.insert(job.key.clone(), job.options.clone());
             job.set_worker_time();
             self.actor.cast(WorkerMessage::Dispatch(job))?;
         }
diff --git a/ractor/src/thread_local.rs b/ractor/src/thread_local.rs
index ad74471..75ed245 100644
--- a/ractor/src/thread_local.rs
+++ b/ractor/src/thread_local.rs
@@ -47,6 +47,63 @@ mod tests;
 /// patterns. Panics are also captured from the inner functions and wrapped into an Error
 /// type, however should an [Err(_)] result from any of these functions the **actor will
 /// terminate** and cleanup.
+///
+/// # Example
+///
+/// ```
+/// use ractor::thread_local::ThreadLocalActor;
+/// use ractor::thread_local::ThreadLocalActorSpawner;
+/// use ractor::ActorProcessingErr;
+/// use ractor::ActorRef;
+///
+/// #[derive(Default)]
+/// struct TheActor;
+///
+/// impl ThreadLocalActor for TheActor {
+///     type Msg = ();
+///     type Arguments = String;
+///     type State = String;
+///
+///     async fn pre_start(
+///         &self,
+///         _myself: ActorRef<Self::Msg>,
+///         args: Self::Arguments,
+///     ) -> Result<Self::State, ActorProcessingErr> {
+///         Ok(args)
+///     }
+///
+///     async fn handle(
+///         &self,
+///         _myself: ActorRef<Self::Msg>,
+///         _msg: (),
+///         state: &mut Self::State,
+///     ) -> Result<(), ActorProcessingErr> {
+///         println!("Message! {state}");
+///         Ok(())
+///     }
+/// }
+///
+/// #[tokio::main]
+/// async fn main() {
+///     // Create the thread-local spawner
+///     let spawner = ThreadLocalActorSpawner::new();
+///     // spawn the actor
+///     let (who, handle) =
+///         ractor::spawn_local::<TheActor>("Something".to_string(), spawner.clone())
+///             .await
+///             .expect("Failed to spawn thread-local actor!");
+///
+///     // send messages to the actor
+///     who.cast(()).expect("Failed to send");
+///     who.cast(()).expect("Failed to send");
+///
+///     // Tell the actor to drain then stop
+///     who.drain();
+///
+///     // wait for the termination
+///     handle.await.unwrap();
+/// }
+/// ```
 pub trait ThreadLocalActor: Default + Sized + 'static {
     /// The message type for this actor
     type Msg: Message;
@@ -179,9 +236,10 @@ pub trait ThreadLocalActor: Default + Sized + 'static {
     /// Spawn an actor of this type, which is unsupervised, automatically starting
     ///
     /// * `name`: A name to give the actor. Useful for global referencing or debug printing
-    /// * `handler` The implementation of Self
     /// * `startup_args`: Arguments passed to the `pre_start` call of the [ThreadLocalActor] to facilitate startup and
     ///   initial state creation
+    /// * `spawner`: The [ThreadLocalActorSpawner] which will control starting this actor on a specific
+    ///   thread
     ///
     /// Returns a [Ok((ActorRef, JoinHandle<()>))] upon successful start, denoting the actor reference
     /// along with the join handle which will complete when the actor terminates. Returns [Err(SpawnErr)] if
@@ -196,10 +254,11 @@ pub trait ThreadLocalActor: Default + Sized + 'static {
     /// Spawn an actor of this type with a supervisor, automatically starting the actor
     ///
     /// * `name`: A name to give the actor. Useful for global referencing or debug printing
-    /// * `handler` The implementation of Self
     /// * `startup_args`: Arguments passed to the `pre_start` call of the [ThreadLocalActor] to
     ///   facilitate startup and initial state creation
     /// * `supervisor`: The [ActorCell] which is to become the supervisor (parent) of this actor
+    /// * `spawner`: The [ThreadLocalActorSpawner] which will control starting this actor on a specific
+    ///   thread
     ///
     /// Returns a [Ok((ActorRef, JoinHandle<()>))] upon successful start, denoting the actor reference
     /// along with the join handle which will complete when the actor terminates. Returns [Err(SpawnErr)] if
@@ -386,7 +445,7 @@ impl ActorCell {
     ///
     /// * `name`: A name to give the actor. Useful for global referencing or debug printing
     /// * `handler` The implementation of Self
-    /// * `startup_args`: Arguments passed to the `pre_start` call of the [Actor] to facilitate startup and
+    /// * `startup_args`: Arguments passed to the `pre_start` call of the [ThreadLocalActor] to facilitate startup and
     ///   initial state creation
     ///
     /// Returns a [Ok((ActorRef, JoinHandle<()>))] upon successful start, denoting the actor reference
diff --git a/ractor_cluster/Cargo.toml b/ractor_cluster/Cargo.toml
index 1a61390..30ce186 100644
--- a/ractor_cluster/Cargo.toml
+++ b/ractor_cluster/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster"
-version = "0.15.5"
+version = "0.15.6"
 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 ac2301e..0d251c9 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.5"
+version = "0.15.6"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Derives for ractor_cluster"
 license = "MIT"


Test patch:
diff --git a/ractor/src/factory/tests/draining_requests.rs b/ractor/src/factory/tests/draining_requests.rs
index 58a8020..104636a 100644
--- a/ractor/src/factory/tests/draining_requests.rs
+++ b/ractor/src/factory/tests/draining_requests.rs
@@ -168,3 +168,55 @@ async fn test_request_draining() {
     // check the counter
     assert_eq!(999, counter.load(Ordering::Relaxed));
 }
+
+#[crate::concurrency::test]
+#[cfg_attr(
+    not(all(target_arch = "wasm32", target_os = "unknown")),
+    tracing_test::traced_test
+)]
+async fn test_worker_drains_when_worker_queueing() {
+    let counter = Arc::new(AtomicU16::new(0));
+
+    let worker_builder = SlowWorkerBuilder {
+        counter: counter.clone(),
+    };
+    let factory_definition = Factory::<
+        TestKey,
+        TestMessage,
+        (),
+        TestWorker,
+        routing::KeyPersistentRouting<TestKey, TestMessage>,
+        queues::DefaultQueue<TestKey, TestMessage>,
+    >::default();
+    let args = FactoryArguments::builder()
+        .num_initial_workers(1)
+        .queue(Default::default())
+        .router(Default::default())
+        .worker_builder(Box::new(worker_builder))
+        .build();
+    let (factory, factory_handle) = Actor::spawn(None, factory_definition, args)
+        .await
+        .expect("Failed to spawn factory");
+
+    for id in 0..999 {
+        factory
+            .cast(FactoryMessage::Dispatch(Job {
+                key: TestKey { id },
+                msg: TestMessage::Ok,
+                options: JobOptions::default(),
+                accepted: None,
+            }))
+            .expect("Failed to send to factory");
+    }
+
+    // start draining requests
+    factory
+        .cast(FactoryMessage::DrainRequests)
+        .expect("Failed to contact factory");
+
+    // wait for factory to exit (it should once drained)
+    factory_handle.await.unwrap();
+
+    // check the counter, make sure all messages processed
+    assert_eq!(999, counter.load(Ordering::Relaxed));
+}
diff --git a/ractor_cluster_integration_tests/src/tests/encryption.rs b/ractor_cluster_integration_tests/src/tests/encryption.rs
index d18221a..24ec3e4 100644
--- a/ractor_cluster_integration_tests/src/tests/encryption.rs
+++ b/ractor_cluster_integration_tests/src/tests/encryption.rs
@@ -5,7 +5,7 @@
 
 //! A clone of the [super::auth_handshake] test but with encryped communications
 //!
-//! Encryption certificates used are the same as [rustls]'s examples: <https://github.com/rustls/rustls>
+//! Encryption certificates used are the same as [tokio_rustls::rustls]'s examples: <https://github.com/rustls/rustls>
 
 use std::convert::TryFrom;
 use std::fs::File;