diff --git a/Cargo.lock b/Cargo.lock index 74b31f4a3..cb8ff239b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2006,17 +2006,6 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" -[[package]] -name = "clocksource" -version = "0.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "46a4f8c23584e9dc6e40de1406e8c776ae727c49f7cb85c0bb23fb8c2096f7e0" -dependencies = [ - "libc", - "time", - "winapi", -] - [[package]] name = "cmake" version = "0.1.58" @@ -8394,12 +8383,10 @@ dependencies = [ [[package]] name = "ratelimit" -version = "0.10.1" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5dc94ed8e3de45f6d8d052869d48c0dbeebcaa7a6c345ec7f0f917e10347428e" +checksum = "e78b08065c51c82ff8c4a0d88e3dce3edfce39c375e946c3464210fff3433fb8" dependencies = [ - "clocksource", - "parking_lot", "thiserror 2.0.20", ] diff --git a/Cargo.toml b/Cargo.toml index e39182f4a..944db98b4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -278,7 +278,7 @@ percent-encoding = "2.3.2" pin-project-lite = "0.2.17" pretty_assertions = "1.4.1" rand = { version = "0.10.2" } -ratelimit = "0.10.1" +ratelimit = "2.0.0" rayon = "1.12.0" reed-solomon-erasure = { package = "rustfs-erasure-codec", version = "8.0.2" } reed-solomon-simd = "3.1.0" diff --git a/crates/ecstore/src/bucket/bandwidth/monitor.rs b/crates/ecstore/src/bucket/bandwidth/monitor.rs index 2da030b66..e99c81385 100644 --- a/crates/ecstore/src/bucket/bandwidth/monitor.rs +++ b/crates/ecstore/src/bucket/bandwidth/monitor.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::bucket::bandwidth::reader::BucketOptions; -use ratelimit::{Error as RatelimitError, Ratelimiter}; +use ratelimit::{Clock, Error as RatelimitError, Ratelimiter}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; @@ -24,6 +24,33 @@ use tracing::warn; /// BETA_BUCKET is the weight used to calculate exponential moving average const BETA_BUCKET: f64 = 0.1; +// ratelimit 2.0 stores tokens at six decimal places. Above this limit its +// scaled capacity and token-cost calculations saturate instead of preserving +// the configured bandwidth. +const MAX_RATELIMIT_TOKENS: i64 = 18_446_744_073_709; + +fn consume_tokens(limiter: &Ratelimiter, n: u64) -> (u64, f64, u64) { + if n == 0 { + return (0, limiter.rate() as f64, 0); + } + + let mut consumed = 0u64; + // Consuming one token also refills the bucket based on elapsed time, so + // the subsequent `available()` read reflects freshly accrued tokens. + if limiter.try_wait().is_ok() { + consumed = 1; + } + let available = limiter.available(); + let to_consume = n - consumed; + let batch = to_consume.min(available); + if batch > 0 && limiter.try_wait_n(batch).is_ok() { + consumed += batch; + } + let deficit = n.saturating_sub(consumed); + let rate = limiter.rate() as f64; + (deficit, rate, consumed) +} + #[derive(Clone)] pub struct BucketThrottle { limiter: Arc>, @@ -34,9 +61,9 @@ impl BucketThrottle { fn new(node_bandwidth_per_sec: i64) -> Result { let node_bandwidth_per_sec = node_bandwidth_per_sec.max(1); let amount = node_bandwidth_per_sec as u64; - let limiter_inner = Ratelimiter::builder(amount, Duration::from_secs(1)) - .max_tokens(amount) - .build()?; + // ratelimit 2.0's builder takes a per-second rate; the refill period + // defaults to one second, so `amount` tokens accrue per second. + let limiter_inner = Ratelimiter::builder(amount).max_tokens(amount).build()?; Ok(Self { limiter: Arc::new(Mutex::new(limiter_inner)), node_bandwidth_per_sec, @@ -47,32 +74,21 @@ impl BucketThrottle { self.limiter.lock().unwrap_or_else(|e| e.into_inner()).max_tokens() } - /// The ratelimit crate (0.10.0) does not provide a bulk token consumption API. - /// try_wait() first to consume 1 token AND trigger the internal refill - /// mechanism (tokens are only refilled during try_wait/wait calls). - /// directly adjust available tokens via set_available() to consume the remaining amount. + /// Best-effort bulk token consumption: consume up to `n` tokens and report + /// how many were taken plus any shortfall. + /// + /// `try_wait_n` on the ratelimit crate is all-or-nothing, so we cannot ask + /// for `n` directly and still consume a partial amount. Instead we take one + /// token first (which also triggers the internal time-based refill), read + /// the now-current available count, and consume `min(remaining, available)` + /// in a single `try_wait_n` call — that batch never exceeds `available`, so + /// it always succeeds. pub(crate) fn consume(&self, n: u64) -> (u64, f64, u64) { let guard = self.limiter.lock().unwrap_or_else(|e| { warn!("bucket throttle mutex poisoned, recovering"); e.into_inner() }); - if n == 0 { - return (0, guard.rate(), 0); - } - let mut consumed = 0u64; - if guard.try_wait().is_ok() { - consumed = 1; - } - let available = guard.available(); - let to_consume = n - consumed; - let batch = to_consume.min(available); - if batch > 0 { - let _ = guard.set_available(available - batch); - consumed += batch; - } - let deficit = n.saturating_sub(consumed); - let rate = guard.rate(); - (deficit, rate, consumed) + consume_tokens(&guard, n) } } @@ -329,6 +345,16 @@ impl Monitor { "bandwidth limit too small for cluster size, per-node limit will clamp to 1 byte/s" ); } + if limit_bytes > MAX_RATELIMIT_TOKENS { + warn!( + bucket = bucket, + arn = arn, + limit_bytes = limit_bytes, + max_limit_bytes = MAX_RATELIMIT_TOKENS, + "bandwidth limit exceeds ratelimiter capacity, throttling disabled for this target" + ); + return; + } let opts = BucketOptions { name: bucket.to_string(), replication_arn: arn.to_string(), @@ -375,6 +401,30 @@ mod tests { use super::*; use std::panic::{AssertUnwindSafe, catch_unwind}; + #[derive(Clone)] + struct TestClock { + elapsed_ns: Arc, + } + + impl TestClock { + fn new() -> Self { + Self { + elapsed_ns: Arc::new(AtomicU64::new(0)), + } + } + + fn advance(&self, duration: Duration) { + let elapsed_ns = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX); + self.elapsed_ns.fetch_add(elapsed_ns, Ordering::Relaxed); + } + } + + impl Clock for TestClock { + fn elapsed(&self) -> Duration { + Duration::from_nanos(self.elapsed_ns.load(Ordering::Relaxed)) + } + } + #[test] fn test_set_and_get_throttle_with_node_split() { let monitor = Monitor::new(4); @@ -426,6 +476,15 @@ mod tests { assert!(!monitor.is_throttled("b1", "arn1")); } + #[test] + fn test_set_bandwidth_limit_rejects_unrepresentable_rate() { + let monitor = Monitor::new(1); + + monitor.set_bandwidth_limit("b1", "arn1", MAX_RATELIMIT_TOKENS + 1); + + assert!(!monitor.is_throttled("b1", "arn1")); + } + #[test] fn test_consume_returns_deficit_when_tokens_exhausted() { let throttle = BucketThrottle::new(100).expect("test"); @@ -436,6 +495,23 @@ mod tests { assert!(rate > 0.0); } + #[test] + fn test_consume_refills_continuously() { + let clock = TestClock::new(); + let limiter = Ratelimiter::with_clock(100, clock.clone()); + + assert_eq!(consume_tokens(&limiter, 100), (100, 100.0, 0)); + + clock.advance(Duration::from_millis(250)); + assert_eq!(consume_tokens(&limiter, 100), (75, 100.0, 25)); + + clock.advance(Duration::from_millis(250)); + assert_eq!(consume_tokens(&limiter, 100), (75, 100.0, 25)); + + clock.advance(Duration::from_millis(500)); + assert_eq!(consume_tokens(&limiter, 100), (50, 100.0, 50)); + } + #[test] fn test_consume_no_deficit_when_tokens_sufficient() { let throttle = BucketThrottle::new(10000).expect("test"); diff --git a/crates/ecstore/src/bucket/bandwidth/reader.rs b/crates/ecstore/src/bucket/bandwidth/reader.rs index d9c5fc89c..9bf27bfda 100644 --- a/crates/ecstore/src/bucket/bandwidth/reader.rs +++ b/crates/ecstore/src/bucket/bandwidth/reader.rs @@ -306,7 +306,7 @@ mod tests { #[tokio::test] async fn test_monitored_reader_header_size_accounting() { let monitor = Monitor::new(1); - monitor.set_bandwidth_limit("b1", "arn1", 100); + monitor.set_bandwidth_limit("b1", "arn1", 1_000_000_000); let data = vec![0u8; 200]; let inner = TestAsyncReader::new(&data); diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index a4ab1d2d5..29b179e12 100644 --- a/crates/targets/Cargo.toml +++ b/crates/targets/Cargo.toml @@ -59,6 +59,10 @@ hyper = { workspace = true, features = ["http2", "http1", "server"] } hyper-rustls = { workspace = true, default-features = false, features = ["native-tokio", "tls12", "logging", "aws-lc-rs"] } lapin = { workspace = true, default-features = false, features = ["tokio", "rustls", "rustls--aws_lc_rs"] } libc = { workspace = true } +# pulsar regenerates its protobuf bindings from `PulsarApi.proto` in build.rs on every +# build, which needs `protoc`. Keep this dependency lean and install a system +# protobuf compiler on platforms where it is not provided by the base image. +# See https://github.com/rustfs/rustfs/issues/3195. pulsar = { workspace = true, default-features = false, features = ["tokio-rustls-runtime", "telemetry"] } regex = { workspace = true } rayon = { workspace = true }