Repository: slawlor/ractor
Pull request #331 (2025-02-13 02:55:27)

Generalize the factory routing product & add rate limiting support
To be used in all routing decisions so we can centralize future handling logic related to rate limiting.

The goal is to add a general rate limiting offering which can be used at the routing layer, right before sending the job to a worker.

Additionally this adds a new result from routing operations, which denotes that a job isn't accepted but rater rate limited and should be loadshed.

This is a major version break as the API for `Router` will change with added fields and an additional return type.

Lastly this adds a basic leaky-bucket rate limiter which can manage the throughput of the factory based on the leaky bucket algo.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/slawlor/ractor/pull/331?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 `67.05882%` with `28 lines` in your changes missing coverage. Please review.
> Project coverage is 81.65%. Comparing base [(`9e0c8a7`)](https://app.codecov.io/gh/slawlor/ractor/commit/9e0c8a7024e9caaad3c1e4086ce20b0378d32c14?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) to head [(`dc40790`)](https://app.codecov.io/gh/slawlor/ractor/commit/dc40790c081bf6be39c6f4aa7aaa30ecc8e3420c?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/331?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/331?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) | 65.78% | [13 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/331?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |
| [ractor/src/factory/routing.rs](https://app.codecov.io/gh/slawlor/ractor/pull/331?src=pr&el=tree&filepath=ractor%2Fsrc%2Ffactory%2Frouting.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9mYWN0b3J5L3JvdXRpbmcucnM=) | 76.19% | [10 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/331?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |
| [ractor/src/factory/stats.rs](https://app.codecov.io/gh/slawlor/ractor/pull/331?src=pr&el=tree&filepath=ractor%2Fsrc%2Ffactory%2Fstats.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9mYWN0b3J5L3N0YXRzLnJz) | 0.00% | [5 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/331?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     #331      +/-   ##
==========================================
+ Coverage   81.64%   81.65%   +0.01%     
==========================================
  Files          62       62              
  Lines       12319    12365      +46     
==========================================
+ Hits        10058    10097      +39     
- Misses       2261     2268       +7     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/slawlor/ractor/pull/331?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/README.md b/README.md
index ac8e361..db73465 100644
--- a/README.md
+++ b/README.md
@@ -66,7 +66,7 @@ Install `ractor` by adding the following to your Cargo.toml dependencies.
 
 ```toml
 [dependencies]
-ractor = "0.14"
+ractor = "0.15"
 ```
 
 The minimum supported Rust version (MSRV) of `ractor` is `1.64`. However to utilize the native `async fn` support in traits and not rely on the `async-trait` crate's desugaring functionliaty, you need to be on Rust version `>= 1.75`. The stabilization of `async fn` in traits [was recently added](https://blog.rust-lang.org/2023/12/21/async-fn-rpit-in-traits.html).
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index 7911dcd..88c11e1 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.14.7"
+version = "0.15.0"
 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 79fb3ec..a5ffaf2 100644
--- a/ractor/src/factory/discard.rs
+++ b/ractor/src/factory/discard.rs
@@ -164,7 +164,7 @@ pub trait DynamicDiscardController: Send + Sync + 'static {
 }
 
 /// Reason for discarding a job
-#[derive(Debug)]
+#[derive(Debug, Eq, PartialEq, Clone)]
 pub enum DiscardReason {
     /// The job TTLd
     TtlExpired,
@@ -172,6 +172,8 @@ pub enum DiscardReason {
     Loadshed,
     /// The job was dropped due to factory shutting down
     Shutdown,
+    /// The job was rejected due to rate limits being exceeded
+    RateLimited,
 }
 
 /// Trait defining the discard handler for a factory.
diff --git a/ractor/src/factory/factoryimpl.rs b/ractor/src/factory/factoryimpl.rs
index 602983d..d975e65 100644
--- a/ractor/src/factory/factoryimpl.rs
+++ b/ractor/src/factory/factoryimpl.rs
@@ -18,7 +18,6 @@ use crate::Actor;
 use crate::ActorProcessingErr;
 use crate::ActorRef;
 use crate::Message;
-use crate::MessagingErr;
 use crate::SpawnErr;
 use crate::SupervisionEvent;
 
@@ -257,6 +256,7 @@ where
             .field("has_lifecycle_hooks", &self.lifecycle_hooks.is_some())
             .field("has_stats", &self.stats.is_some())
             .field("has_discard_handler", &self.discard_handler.is_some())
+            .field("processing_messages", &self.processing_messages)
             .finish()
     }
 }
@@ -284,7 +284,7 @@ where
     fn try_route_next_active_job(
         &mut self,
         worker_hint: WorkerId,
-    ) -> Result<(), MessagingErr<WorkerMessage<TKey, TMsg>>> {
+    ) -> Result<(), ActorProcessingErr> {
         // cleanup expired messages at the head of the queue
         while let Some(true) = self.queue.peek().map(|m| m.is_expired()) {
             // remove the job from the queue
@@ -299,28 +299,42 @@ where
             }
         }
 
-        if let Some(worker) = self.pool.get_mut(&worker_hint).filter(|f| f.is_available()) {
-            if let Some(mut job) = self.queue.pop_front() {
-                job.accept();
-                self.processing_messages += 1;
-                worker.enqueue_job(job)?;
-            }
-        } else {
-            // target the next available worker
-            let target_worker = self
-                .queue
-                .peek()
-                .and_then(|job| {
-                    self.router
-                        .choose_target_worker(job, self.pool_size, &self.pool)
-                })
-                .and_then(|wid| self.pool.get_mut(&wid));
-            if let (Some(mut job), Some(worker)) = (self.queue.pop_front(), target_worker) {
-                job.accept();
-                self.processing_messages += 1;
-                worker.enqueue_job(job)?;
+        let target_worker = self.queue.peek().and_then(|job| {
+            self.router
+                .choose_target_worker(job, self.pool_size, Some(worker_hint), &self.pool)
+        });
+        if let Some(worker) = target_worker {
+            while let Some(job) = self.queue.pop_front() {
+                match self.router.route_message(
+                    job,
+                    self.pool_size,
+                    Some(worker),
+                    &mut self.pool,
+                )? {
+                    RouteResult::Handled => {
+                        // routed a job, we're done routing.
+                        return Ok(());
+                    }
+                    RouteResult::RateLimited(mut job) => {
+                        // rate limit hit, keep flushing work until we're back under the limit or queue empty.
+                        tracing::trace!("Job rate limited to {worker}");
+                        self.stats.job_rate_limited(&self.factory_name);
+                        if let Some(handler) = &self.discard_handler {
+                            handler.discard(DiscardReason::RateLimited, &mut job);
+                        }
+                        job.reject();
+                    }
+                    RouteResult::Backlog(_) => {
+                        tracing::error!(
+                            "Error routing job to {worker}. Invariant violated with backlog when we have a targeted worker"
+                        );
+
+                        panic!("Received invalid variant of `RouteResult::Backlog` because a worker is unavailable, but we already targeted a worker.");
+                    }
+                }
             }
         }
+
         Ok(())
     }
 
@@ -502,15 +516,25 @@ where
             }
             job.reject();
         } else if self.drain_state == DrainState::NotDraining {
-            if let RouteResult::Backlog(busy_job) =
-                self.router
-                    .route_message(job, self.pool_size, &mut self.pool)?
+            match self
+                .router
+                .route_message(job, self.pool_size, None, &mut self.pool)?
             {
-                // workers are busy, we need to queue a job
-                self.maybe_enqueue(busy_job);
-            } else {
-                // message was routed
-                self.processing_messages += 1;
+                RouteResult::Handled => {
+                    // message was routed
+                    self.processing_messages += 1;
+                }
+                RouteResult::RateLimited(mut job) => {
+                    self.stats.job_rate_limited(&self.factory_name);
+                    if let Some(handler) = &self.discard_handler {
+                        handler.discard(DiscardReason::RateLimited, &mut job);
+                    }
+                    job.reject();
+                }
+                RouteResult::Backlog(busy_job) => {
+                    // workers are busy, we need to queue a job
+                    self.maybe_enqueue(busy_job);
+                }
             }
         } else {
             tracing::debug!("Factory is draining but a job was received");
diff --git a/ractor/src/factory/mod.rs b/ractor/src/factory/mod.rs
index b40fd2b..b84b323 100644
--- a/ractor/src/factory/mod.rs
+++ b/ractor/src/factory/mod.rs
@@ -169,6 +169,7 @@ pub mod hash;
 pub mod job;
 pub mod lifecycle;
 pub mod queues;
+pub mod ratelim;
 pub mod routing;
 pub mod stats;
 pub mod worker;
@@ -184,6 +185,7 @@ pub use discard::{
 pub use factoryimpl::{Factory, FactoryArguments, FactoryArgumentsBuilder};
 pub use job::{Job, JobKey, JobOptions, MessageRetryStrategy, RetriableMessage};
 pub use lifecycle::FactoryLifecycleHooks;
+pub use ratelim::{LeakyBucketRateLimiter, RateLimitedRouter, RateLimiter};
 pub use worker::{
     DeadMansSwitchConfiguration, Worker, WorkerBuilder, WorkerCapacityController, WorkerMessage,
     WorkerProperties, WorkerStartContext,
diff --git a/ractor/src/factory/ratelim.rs b/ractor/src/factory/ratelim.rs
new file mode 100644
index 0000000..734cf2a
--- /dev/null
+++ b/ractor/src/factory/ratelim.rs
@@ -0,0 +1,220 @@
+// 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.
+
+//! Rate limiting protocols for factory routers
+
+use std::collections::HashMap;
+
+use crate::concurrency::{Duration, Instant};
+use crate::factory::routing::RouteResult;
+use crate::factory::routing::Router;
+use crate::factory::Job;
+use crate::factory::JobKey;
+use crate::factory::WorkerId;
+use crate::factory::WorkerProperties;
+use crate::ActorProcessingErr;
+use crate::Message;
+use crate::State;
+
+/// The maximum supported balance for leaky bucket rate limiting.
+pub const MAX_LB_BALANCE: usize = isize::MAX as usize;
+
+/// A basic trait which allows controlling rate limiting of message routing
+pub trait RateLimiter: State {
+    /// Check if we have not violated the rate limiter
+    ///
+    /// Returns [false] if we're in violation and should start rate-limiting traffic
+    /// [true] otherwise
+    fn check(&mut self) -> bool;
+
+    /// Bump the rate limit internal counter, as we've routed a message
+    /// to a worker
+    fn bump(&mut self);
+}
+
+/// A generic struct which wraps the message router and adds support for a rate-limiting implementation to rate limit
+/// jobs processed by the factory. This handles the plubming around wrapping a rate limited message router
+#[derive(Debug, bon::Builder)]
+pub struct RateLimitedRouter<TRouter, TRateLimit> {
+    /// The underlying message router which does NOT implement rate limiting
+    pub router: TRouter,
+    /// The rate limiter to apply to the message routing
+    pub rate_limiter: TRateLimit,
+}
+
+impl<TKey, TMsg, TRouter, TRateLimit> Router<TKey, TMsg> for RateLimitedRouter<TRouter, TRateLimit>
+where
+    TKey: JobKey,
+    TMsg: Message,
+    TRouter: Router<TKey, TMsg>,
+    TRateLimit: RateLimiter,
+{
+    fn route_message(
+        &mut self,
+        job: Job<TKey, TMsg>,
+        pool_size: usize,
+        worker_hint: Option<WorkerId>,
+        worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
+    ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr> {
+        if !self.rate_limiter.check() {
+            Ok(RouteResult::RateLimited(job))
+        } else {
+            let result = self
+                .router
+                .route_message(job, pool_size, worker_hint, worker_pool);
+            if matches!(result, Ok(RouteResult::Handled)) {
+                // only bump the internal state if we successfully routed a message
+                self.rate_limiter.bump();
+            }
+            result
+        }
+    }
+
+    fn choose_target_worker(
+        &mut self,
+        job: &Job<TKey, TMsg>,
+        pool_size: usize,
+        worker_hint: Option<WorkerId>,
+        worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
+    ) -> Option<WorkerId> {
+        self.router
+            .choose_target_worker(job, pool_size, worker_hint, worker_pool)
+    }
+
+    fn is_factory_queueing(&self) -> bool {
+        self.router.is_factory_queueing()
+    }
+}
+
+/// A basic leaky-bucket rate limiter. This is a synchronous implementation
+/// with no interior locking since it's only used by the [RateLimitedRouter]
+/// uniquely and doesn't share its state
+#[derive(Debug)]
+pub struct LeakyBucketRateLimiter {
+    /// Tokens to add every `per` duration.
+    pub refill: usize,
+    /// Interval in milliseconds to add tokens.
+    pub interval: Duration,
+    /// Max number of tokens associated with the rate limiter.
+    pub max: usize,
+    /// The "balance" of the rate limiter, i.e. the number of tokens still available
+    pub balance: usize,
+    /// The deadline to perform another refill
+    deadline: Instant,
+}
+
+#[bon::bon]
+impl LeakyBucketRateLimiter {
+    /// Create a new [LeakyBucketRateLimiter] instance
+    ///
+    /// * `refill` - Tokens to add every `per` duration.
+    /// * `interval` - Interval to add tokens.
+    /// * `max` - The maximum number of tokens associated with the rate limiter. Default = [MAX_LB_BALANCE]
+    /// * `initial` - The initial starting balance. If [None] will be = to max
+    ///
+    /// Returns a new [LeakyBucketRateLimiter] instance
+    #[builder]
+    pub fn new(
+        refill: usize,
+        interval: Duration,
+        #[builder(default = MAX_LB_BALANCE)] max: usize,
+        initial: Option<usize>,
+    ) -> LeakyBucketRateLimiter {
+        LeakyBucketRateLimiter {
+            refill,
+            interval,
+            max,
+            balance: initial.unwrap_or(max),
+            deadline: Instant::now() + interval,
+        }
+    }
+
+    fn refresh(&mut self, now: Instant) {
+        if now < self.deadline {
+            return;
+        }
+
+        // Time elapsed in milliseconds since the last deadline.
+        let millis = self.interval.as_millis();
+        let since = now.saturating_duration_since(self.deadline).as_millis();
+
+        let periods = usize::try_from(since / millis + 1).unwrap_or(usize::MAX);
+
+        let tokens = periods
+            .checked_mul(self.refill)
+            .unwrap_or(MAX_LB_BALANCE)
+            .min(MAX_LB_BALANCE);
+
+        let remaining_millis_until_next_deadline =
+            u64::try_from(since % millis).unwrap_or(u64::MAX);
+        self.deadline = now
+            + self
+                .interval
+                .saturating_sub(Duration::from_millis(remaining_millis_until_next_deadline));
+        self.balance = (self.balance + tokens).min(self.max);
+    }
+}
+
+impl RateLimiter for LeakyBucketRateLimiter {
+    fn check(&mut self) -> bool {
+        let now = Instant::now();
+        self.refresh(now);
+        self.balance > 0
+    }
+
+    fn bump(&mut self) {
+        if self.balance > 0 {
+            self.balance -= 1;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::concurrency::sleep;
+
+    use super::*;
+
+    #[crate::concurrency::test]
+    async fn test_basic_leaky_bucket() {
+        let mut limiter = LeakyBucketRateLimiter::builder()
+            .refill(1)
+            .initial(1)
+            .interval(Duration::from_millis(100))
+            .build();
+
+        assert!(limiter.check());
+        limiter.bump();
+        assert!(!limiter.check());
+
+        sleep(limiter.interval * 2).await;
+
+        assert!(limiter.check());
+        limiter.bump();
+        assert!(limiter.check());
+        limiter.bump();
+        assert!(!limiter.check());
+    }
+
+    #[crate::concurrency::test]
+    async fn test_leaky_bucket_max() {
+        let mut limiter = LeakyBucketRateLimiter::builder()
+            .refill(1)
+            .initial(1)
+            .max(1)
+            .interval(Duration::from_millis(100))
+            .build();
+
+        assert!(limiter.check());
+        limiter.bump();
+        assert!(!limiter.check());
+
+        sleep(limiter.interval * 2).await;
+
+        assert!(limiter.check());
+        limiter.bump();
+        assert!(!limiter.check());
+    }
+}
diff --git a/ractor/src/factory/routing.rs b/ractor/src/factory/routing.rs
index 19842cc..e8a4adc 100644
--- a/ractor/src/factory/routing.rs
+++ b/ractor/src/factory/routing.rs
@@ -14,6 +14,7 @@ use crate::factory::JobKey;
 use crate::factory::WorkerId;
 use crate::ActorProcessingErr;
 use crate::Message;
+use crate::State;
 
 /// Custom hashing behavior for factory routing to workers
 pub trait CustomHashFunction<TKey>: Send + Sync
@@ -36,11 +37,18 @@ where
     /// The job needs to be backlogged into the internal factory's queue (if
     /// configured)
     Backlog(Job<TKey, TMsg>),
+    /// The job has exceeded the internal rate limit specification of the router.
+    /// This would be returned as a route operation in the event that the router is
+    /// tracking the jobs-per-unit-time and has decided that routing this next job
+    /// would exceed that limit.
+    ///
+    /// Returns the job that was rejected
+    RateLimited(Job<TKey, TMsg>),
 }
 
 /// A routing mode controls how a request is routed from the factory to a
 /// designated worker
-pub trait Router<TKey, TMsg>: Send + Sync + 'static
+pub trait Router<TKey, TMsg>: State
 where
     TKey: JobKey,
     TMsg: Message,
@@ -49,6 +57,8 @@ where
     ///
     /// * `job` - The job to be routed
     /// * `pool_size` - The size of the ACTIVE worker pool (excluding draining workers)
+    /// * `worker_hint` - If provided, this is a "hint" at which worker should receive the job,
+    ///   if available.
     /// * `worker_pool` - The current worker pool, which may contain draining workers
     ///
     /// Returns [RouteResult::Handled] if the job was routed successfully, otherwise
@@ -58,6 +68,7 @@ where
         &mut self,
         job: Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr>;
 
@@ -73,6 +84,8 @@ where
     ///
     ///  * `job` - A reference to the job to be routed
     /// * `pool_size` - The size of the ACTIVE worker pool (excluding draining workers)
+    /// * `worker_hint` - If provided, this is a "hint" at which worker should receive the job,
+    ///   if available.
     /// * `worker_pool` - The current worker pool, which may contain draining workers
     ///
     /// Returns [None] if no worker can be identified or no worker is avaialble to accept
@@ -81,6 +94,7 @@ where
         &mut self,
         job: &Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Option<WorkerId>;
 
@@ -131,10 +145,13 @@ where
         &mut self,
         job: Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr> {
-        let key = crate::factory::hash::hash_with_max(&job.key, pool_size);
-        if let Some(worker) = worker_pool.get_mut(&key) {
+        if let Some(worker) = self
+            .choose_target_worker(&job, pool_size, worker_hint, worker_pool)
+            .and_then(|wid| worker_pool.get_mut(&wid))
+        {
             worker.enqueue_job(job)?;
         }
         Ok(RouteResult::Handled)
@@ -144,9 +161,11 @@ where
         &mut self,
         job: &Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         _worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Option<WorkerId> {
-        let key = crate::factory::hash::hash_with_max(&job.key, pool_size);
+        let key =
+            worker_hint.unwrap_or_else(|| crate::factory::hash::hash_with_max(&job.key, pool_size));
         Some(key)
     }
 
@@ -168,10 +187,11 @@ where
         &mut self,
         job: Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr> {
         if let Some(worker) = self
-            .choose_target_worker(&job, pool_size, worker_pool)
+            .choose_target_worker(&job, pool_size, worker_hint, worker_pool)
             .and_then(|wid| worker_pool.get_mut(&wid))
         {
             worker.enqueue_job(job)?;
@@ -185,8 +205,14 @@ where
         &mut self,
         _job: &Job<TKey, TMsg>,
         _pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Option<WorkerId> {
+        if let Some(worker) = worker_hint.and_then(|worker| worker_pool.get(&worker)) {
+            if worker.is_available() {
+                return worker_hint;
+            }
+        }
         worker_pool
             .iter()
             .find(|(_, worker)| worker.is_available())
@@ -215,10 +241,11 @@ where
         &mut self,
         job: Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr> {
         if let Some(worker) = self
-            .choose_target_worker(&job, pool_size, worker_pool)
+            .choose_target_worker(&job, pool_size, worker_hint, worker_pool)
             .and_then(|wid| worker_pool.get_mut(&wid))
         {
             worker.enqueue_job(job)?;
@@ -232,8 +259,16 @@ where
         &mut self,
         job: &Job<TKey, TMsg>,
         _pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Option<WorkerId> {
+        // check sticky first
+        if let Some(worker) = worker_hint.and_then(|worker| worker_pool.get(&worker)) {
+            if worker.is_processing_key(&job.key) {
+                return worker_hint;
+            }
+        }
+
         let maybe_worker = worker_pool
             .iter()
             .find(|(_, worker)| worker.is_processing_key(&job.key))
@@ -242,6 +277,13 @@ where
             return maybe_worker;
         }
 
+        // now take first available, based on hint then brute-search
+        if let Some(worker) = worker_hint.and_then(|worker| worker_pool.get(&worker)) {
+            if worker.is_available() {
+                return worker_hint;
+            }
+        }
+
         // fallback to first free worker as there's no sticky worker
         worker_pool
             .iter()
@@ -292,10 +334,11 @@ where
         &mut self,
         job: Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr> {
         if let Some(worker) = self
-            .choose_target_worker(&job, pool_size, worker_pool)
+            .choose_target_worker(&job, pool_size, worker_hint, worker_pool)
             .and_then(|wid| worker_pool.get_mut(&wid))
         {
             worker.enqueue_job(job)?;
@@ -307,8 +350,15 @@ where
         &mut self,
         _job: &Job<TKey, TMsg>,
         pool_size: usize,
-        _worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
+        worker_hint: Option<WorkerId>,
+        worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Option<WorkerId> {
+        if let Some(worker) = worker_hint.and_then(|worker| worker_pool.get(&worker)) {
+            if worker.is_available() {
+                return worker_hint;
+            }
+        }
+
         let mut key = self.last_worker + 1;
         if key >= pool_size {
             key = 0;
@@ -365,10 +415,13 @@ where
         &mut self,
         job: Job<TKey, TMsg>,
         pool_size: usize,
+        worker_hint: Option<WorkerId>,
         worker_pool: &mut HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Result<RouteResult<TKey, TMsg>, ActorProcessingErr> {
-        let key = self.hasher.hash(&job.key, pool_size);
-        if let Some(worker) = worker_pool.get_mut(&key) {
+        if let Some(worker) = self
+            .choose_target_worker(&job, pool_size, worker_hint, worker_pool)
+            .and_then(|wid| worker_pool.get_mut(&wid))
+        {
             worker.enqueue_job(job)?;
         }
         Ok(RouteResult::Handled)
@@ -378,6 +431,7 @@ where
         &mut self,
         job: &Job<TKey, TMsg>,
         pool_size: usize,
+        _worker_hint: Option<WorkerId>,
         _worker_pool: &HashMap<WorkerId, WorkerProperties<TKey, TMsg>>,
     ) -> Option<WorkerId> {
         let key = self.hasher.hash(&job.key, pool_size);
diff --git a/ractor/src/factory/stats.rs b/ractor/src/factory/stats.rs
index 456b801..6563325 100644
--- a/ractor/src/factory/stats.rs
+++ b/ractor/src/factory/stats.rs
@@ -42,6 +42,9 @@ pub trait FactoryStatsLayer: Send + Sync + 'static {
     /// Called when a job is discarded
     fn job_discarded(&self, factory: &str);
 
+    /// Called when a job is rate limited
+    fn job_rate_limited(&self, factory: &str);
+
     /// Called when jobs TTL timeout in the factory's queue
     fn job_ttl_expired(&self, factory: &str, num_removed: usize);
 
@@ -114,6 +117,13 @@ impl FactoryStatsLayer for Option<Arc<dyn FactoryStatsLayer>> {
         }
     }
 
+    /// Called when a job is rate limited
+    fn job_rate_limited(&self, factory: &str) {
+        if let Some(s) = self {
+            s.job_rate_limited(factory);
+        }
+    }
+
     /// Called when a job TTLs in the factory's queue
     fn job_ttl_expired(&self, factory: &str, num_removed: usize) {
         if let Some(s) = self {
diff --git a/ractor/src/lib.rs b/ractor/src/lib.rs
index 9d5670b..72df9a6 100644
--- a/ractor/src/lib.rs
+++ b/ractor/src/lib.rs
@@ -12,7 +12,7 @@
 //!
 //! ```toml
 //! [dependencies]
-//! ractor = "0.14"
+//! ractor = "0.15"
 //! ```
 //!
 //! The minimum supported Rust version (MSRV) is 1.64. However if you disable the `async-trait` feature, then you need Rust >= 1.75 due to the native
diff --git a/ractor_cluster/Cargo.toml b/ractor_cluster/Cargo.toml
index 68889a3..794650a 100644
--- a/ractor_cluster/Cargo.toml
+++ b/ractor_cluster/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster"
-version = "0.14.7"
+version = "0.15.0"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Distributed cluster environment of Ractor actors"
 documentation = "https://docs.rs/ractor"
@@ -30,8 +30,8 @@ prost-build = { version = "0.13" }
 bytes = { version = "1" }
 prost = { version = "0.13" }
 prost-types = { version = "0.13" }
-ractor = { version = "0.14.0", default-features = false, features = ["tokio_runtime", "message_span_propogation", "cluster"], path = "../ractor" }
-ractor_cluster_derive = { version = "0.14.0", path = "../ractor_cluster_derive" }
+ractor = { version = "0.15.0", default-features = false, features = ["tokio_runtime", "message_span_propogation", "cluster"], path = "../ractor" }
+ractor_cluster_derive = { version = "0.15.0", path = "../ractor_cluster_derive" }
 rand = "0.8"
 sha2 = "0.10"
 tokio = { version = "1.30", features = ["rt", "time", "sync", "macros", "net", "io-util", "tracing"]}
diff --git a/ractor_cluster_derive/Cargo.toml b/ractor_cluster_derive/Cargo.toml
index 1acae86..7d04428 100644
--- a/ractor_cluster_derive/Cargo.toml
+++ b/ractor_cluster_derive/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster_derive"
-version = "0.14.7"
+version = "0.15.0"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Derives for ractor_cluster"
 license = "MIT"


Test patch:
diff --git a/ractor/src/factory/tests/mod.rs b/ractor/src/factory/tests/mod.rs
index d5defdc..5cccd46 100644
--- a/ractor/src/factory/tests/mod.rs
+++ b/ractor/src/factory/tests/mod.rs
@@ -11,4 +11,5 @@ mod dynamic_discarding;
 mod dynamic_pool;
 mod lifecycle;
 mod priority_queueing;
+mod ratelim;
 mod worker_lifecycle;
diff --git a/ractor/src/factory/tests/ratelim.rs b/ractor/src/factory/tests/ratelim.rs
new file mode 100644
index 0000000..b5ccc86
--- /dev/null
+++ b/ractor/src/factory/tests/ratelim.rs
@@ -0,0 +1,294 @@
+// 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.
+
+//! Rate limiting tests for factories
+
+use std::sync::atomic::AtomicU16;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+use crate::concurrency::{sleep, Duration};
+use crate::factory::routing::*;
+use crate::factory::*;
+use crate::*;
+
+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, ())
+    }
+}
+
+struct BasicRateLimiter {
+    hard_cap: usize,
+    count: usize,
+}
+
+impl RateLimiter for BasicRateLimiter {
+    fn bump(&mut self) {
+        self.count += 1;
+    }
+
+    fn check(&mut self) -> bool {
+        self.count < self.hard_cap
+    }
+}
+
+async fn test_factory_rate_limiting_common<TRouter: Router<(), ()>>(router: TRouter) {
+    // Setup
+    let discard_counter = Arc::new(AtomicU16::new(0));
+
+    struct TestDiscarder {
+        counter: Arc<AtomicU16>,
+    }
+    impl DiscardHandler<(), ()> for TestDiscarder {
+        fn discard(&self, reason: DiscardReason, _job: &mut Job<(), ()>) {
+            tracing::debug!("Discarding job, reason {reason:?}");
+            if reason == DiscardReason::RateLimited {
+                let _ = self.counter.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+    }
+
+    let worker_builder = TestWorkerBuilder;
+    let arguments = FactoryArguments::builder()
+        .num_initial_workers(1)
+        .queue(Default::default())
+        .router(
+            // create a rate-limited message router which routes up to a hard cap, then
+            // "rate limits" everything else
+            RateLimitedRouter::builder()
+                .router(router)
+                .rate_limiter(BasicRateLimiter {
+                    count: 0,
+                    hard_cap: 5,
+                })
+                .build(),
+        )
+        .worker_builder(Box::new(worker_builder))
+        .discard_handler(Arc::new(TestDiscarder {
+            counter: discard_counter.clone(),
+        }))
+        .build();
+
+    let factory_definition = Factory::<
+        (),
+        (),
+        (),
+        TestWorker,
+        RateLimitedRouter<TRouter, BasicRateLimiter>,
+        queues::DefaultQueue<(), ()>,
+    >::default();
+    let (factory, factory_handle) = Actor::spawn(None, factory_definition, arguments)
+        .await
+        .expect("Failed to spawn factory");
+
+    // Test
+    for _ in 0..9 {
+        factory
+            .cast(FactoryMessage::Dispatch(Job {
+                accepted: None,
+                key: (),
+                msg: (),
+                options: JobOptions::default(),
+            }))
+            .expect("Failed to send message to factory");
+    }
+
+    // sleep a little to let the worker be in the "ratelim" state
+    // as it'll have routed the max allowable number of requests
+    sleep(Duration::from_millis(100)).await;
+    // send an additional request, which should be marked ratelimiting before
+    // even being queued
+    factory
+        .cast(FactoryMessage::Dispatch(Job {
+            accepted: None,
+            key: (),
+            msg: (),
+            options: JobOptions::default(),
+        }))
+        .expect("Failed to send message to factory");
+
+    // Drain factory
+    factory
+        .cast(FactoryMessage::DrainRequests)
+        .expect("Failed to message factory");
+
+    // once the factory is stopped, the shutdown handler should have been called
+    crate::concurrency::timeout(Duration::from_secs(1), factory_handle)
+        .await
+        .expect("Failed to drain requests in 1s")
+        .expect("Failed to join factory handle");
+
+    // Check that we rate-limited 5 messages, but allowed 5 through
+    assert_eq!(5, discard_counter.load(Ordering::SeqCst));
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_factory_rate_limiting_queuer() {
+    test_factory_rate_limiting_common::<QueuerRouting<(), ()>>(Default::default()).await
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_factory_rate_limiting_sticky_queuer() {
+    test_factory_rate_limiting_common::<StickyQueuerRouting<(), ()>>(Default::default()).await
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_factory_rate_limiting_key_persistent() {
+    test_factory_rate_limiting_common::<KeyPersistentRouting<(), ()>>(Default::default()).await
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_factory_rate_limiting_round_robin() {
+    test_factory_rate_limiting_common::<RoundRobinRouting<(), ()>>(Default::default()).await
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_factory_rate_limiting_custom_hash() {
+    struct MyHasher;
+
+    impl CustomHashFunction<()> for MyHasher {
+        fn hash(&self, _key: &(), _worker_count: usize) -> usize {
+            0
+        }
+    }
+
+    let router = CustomRouting::new(MyHasher);
+
+    test_factory_rate_limiting_common(router).await
+}
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn test_leaky_bucket_rate_limiting() {
+    // Setup
+
+    use crate::concurrency::sleep;
+    let discard_counter = Arc::new(AtomicU16::new(0));
+
+    struct TestDiscarder {
+        counter: Arc<AtomicU16>,
+    }
+    impl DiscardHandler<(), ()> for TestDiscarder {
+        fn discard(&self, reason: DiscardReason, _job: &mut Job<(), ()>) {
+            tracing::debug!("Discarding job, reason {reason:?}");
+            if reason == DiscardReason::RateLimited {
+                let _ = self.counter.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+    }
+
+    let worker_builder = TestWorkerBuilder;
+
+    // Setup rate limited router
+    let limiter = LeakyBucketRateLimiter::builder()
+        .max(5)
+        .initial(5)
+        .refill(1)
+        .interval(Duration::from_millis(100))
+        .build();
+
+    let router = RateLimitedRouter::builder()
+        .router(routing::QueuerRouting::<(), ()>::default())
+        .rate_limiter(limiter)
+        .build();
+    let arguments = FactoryArguments::builder()
+        .num_initial_workers(1)
+        .queue(Default::default())
+        .router(router)
+        .worker_builder(Box::new(worker_builder))
+        .discard_handler(Arc::new(TestDiscarder {
+            counter: discard_counter.clone(),
+        }))
+        .build();
+
+    let factory_definition = Factory::<
+        (),
+        (),
+        (),
+        TestWorker,
+        RateLimitedRouter<routing::QueuerRouting<(), ()>, _>,
+        queues::DefaultQueue<(), ()>,
+    >::default();
+    let (factory, factory_handle) = Actor::spawn(None, factory_definition, arguments)
+        .await
+        .expect("Failed to spawn factory");
+
+    // Test
+    for _ in 0..6 {
+        factory
+            .cast(FactoryMessage::Dispatch(Job {
+                accepted: None,
+                key: (),
+                msg: (),
+                options: JobOptions::default(),
+            }))
+            .expect("Failed to send message to factory");
+    }
+
+    // sleep >100ms and we should be able to push another job
+    sleep(crate::concurrency::Duration::from_millis(200)).await;
+    factory
+        .cast(FactoryMessage::Dispatch(Job {
+            accepted: None,
+            key: (),
+            msg: (),
+            options: JobOptions::default(),
+        }))
+        .expect("Failed to send message to factory");
+
+    // Drain factory
+    factory
+        .cast(FactoryMessage::DrainRequests)
+        .expect("Failed to message factory");
+
+    // once the factory is stopped, the shutdown handler should have been called
+    crate::concurrency::timeout(Duration::from_secs(1), factory_handle)
+        .await
+        .expect("Failed to drain requests in 1s")
+        .expect("Failed to join factory handle");
+
+    // Check that we rate-limited only 1 message, from the first batch
+    assert_eq!(1, discard_counter.load(Ordering::SeqCst));
+}