Use governor for rate limiter (#2547)

* use governor rate limiter

* removed naive rate limter implementations

* clippy issue

* missing cargo.lock changes for programs crate

* panic on u64 to u32 overflow error

* safe check on u64 to u32 conversion

Cargo.toml

@@ -237,6 +237,7 @@ generic-array = { version = "0.14.7", default-features = false }
 gethostname = "0.2.3"
 getrandom = "0.2.10"
 goauth = "0.13.1"
+governor = "0.6.3"
 hex = "0.4.3"
 hidapi = { version = "2.6.3", default-features = false }
 histogram = "0.6.9"

streamer/Cargo.toml

@@ -16,6 +16,7 @@ crossbeam-channel = { workspace = true }
 dashmap = { workspace = true }
 futures  = { workspace = true }
 futures-util = { workspace = true }
+governor = { workspace = true }
 histogram = { workspace = true }
 indexmap = { workspace = true }
 itertools = { workspace = true }

streamer/src/nonblocking/connection_rate_limiter.rs

@@ -1,25 +1,27 @@
 use {
-    crate::nonblocking::{keyed_rate_limiter::KeyedRateLimiter, rate_limiter::RateLimiter},
-    std::{net::IpAddr, time::Duration},
+    governor::{DefaultDirectRateLimiter, DefaultKeyedRateLimiter, Quota, RateLimiter},
+    std::{net::IpAddr, num::NonZeroU32},
 };
 
 pub struct ConnectionRateLimiter {
-    limiter: KeyedRateLimiter<IpAddr>,
+    limiter: DefaultKeyedRateLimiter<IpAddr>,
 }
 
 impl ConnectionRateLimiter {
     /// Create a new rate limiter per IpAddr. The rate is specified as the count per minute to allow for
     /// less frequent connections.
     pub fn new(limit_per_minute: u64) -> Self {
+        let quota =
+            Quota::per_minute(NonZeroU32::new(u32::try_from(limit_per_minute).unwrap()).unwrap());
         Self {
-            limiter: KeyedRateLimiter::new(limit_per_minute, Duration::from_secs(60)),
+            limiter: DefaultKeyedRateLimiter::keyed(quota),
         }
     }
 
     /// Check if the connection from the said `ip` is allowed.
     pub fn is_allowed(&self, ip: &IpAddr) -> bool {
         // Acquire a permit from the rate limiter for the given IP address
-        if self.limiter.check_and_update(*ip) {
+        if self.limiter.check_key(ip).is_ok() {
             debug!("Request from IP {:?} allowed", ip);
             true // Request allowed
         } else {
@@ -48,20 +50,26 @@ impl ConnectionRateLimiter {
 /// Connection rate limiter for enforcing connection rates from
 /// all clients.
 pub struct TotalConnectionRateLimiter {
-    limiter: RateLimiter,
+    limiter: DefaultDirectRateLimiter,
 }
 
 impl TotalConnectionRateLimiter {
     /// Create a new rate limiter. The rate is specified as the count per second.
     pub fn new(limit_per_second: u64) -> Self {
+        let quota =
+            Quota::per_second(NonZeroU32::new(u32::try_from(limit_per_second).unwrap()).unwrap());
         Self {
-            limiter: RateLimiter::new(limit_per_second, Duration::from_secs(1)),
+            limiter: RateLimiter::direct(quota),
         }
     }
 
     /// Check if a connection is allowed.
-    pub fn is_allowed(&mut self) -> bool {
-        self.limiter.check_and_update()
+    pub fn is_allowed(&self) -> bool {
+        if self.limiter.check().is_ok() {
+            true // Request allowed
+        } else {
+            false // Request blocked
+        }
     }
 }
 
@@ -71,7 +79,7 @@ pub mod test {
 
     #[tokio::test]
     async fn test_total_connection_rate_limiter() {
-        let mut limiter = TotalConnectionRateLimiter::new(2);
+        let limiter = TotalConnectionRateLimiter::new(2);
         assert!(limiter.is_allowed());
         assert!(limiter.is_allowed());
         assert!(!limiter.is_allowed());