fix(build): support non-Linux Unix targets (illumos/Solaris/*BSD) (#5853)

* fix(build): support non-Linux Unix targets (illumos/Solaris/*BSD)

Two independent build-infrastructure blockers kept RustFS from building on
non-Linux Unix platforms. Neither touches runtime logic.

1. pulsar regenerates its protobuf bindings in build.rs on every build, which
   needs `protoc`. Platforms without a packaged protoc (illumos/Solaris/*BSD)
   now enable pulsar's `protobuf-src` feature via a cfg-gated dependency, which
   builds a vendored protoc from C++ sources. Mainstream targets keep the lean
   dependency and their existing system/CI protoc.

2. clocksource 0.8.3 (pulled in transitively by ratelimit 0.10) used the
   Linux-only `CLOCK_MONOTONIC_COARSE`. ratelimit 2.0 dropped the clocksource
   dependency entirely, so upgrading removes the portability problem at the
   root rather than patching clocksource. The bandwidth throttle's bulk
   `consume()` is rewritten onto ratelimit 2.0's `try_wait_n`, preserving the
   best-effort partial-consumption semantics.

Verified: cargo check + bandwidth monitor unit tests pass; cargo tree confirms
protobuf-src is enabled only for illumos/Solaris/*BSD and clocksource is gone
from the graph. The final illumos build must be confirmed on-platform.

Closes #3195

* fix(ecstore): guard ratelimit v2 capacity overflow

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): avoid slow bandwidth reader timeout

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(targets): drop vendored pulsar protobuf build

Co-Authored-By: heihutu <heihutu@gmail.com>

---------
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>

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",
 ]
 

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"

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<C: Clock>(limiter: &Ratelimiter<C>, 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<Mutex<Ratelimiter>>,
@@ -34,9 +61,9 @@ impl BucketThrottle {
     fn new(node_bandwidth_per_sec: i64) -> Result<Self, RatelimitError> {
         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))
+        // ratelimit 2.0's builder takes a per-second rate; the refill period
-            .max_tokens(amount)
+        // defaults to one second, so `amount` tokens accrue per second.
-            .build()?;
+        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.
+    /// Best-effort bulk token consumption: consume up to `n` tokens and report
-    /// try_wait() first to consume 1 token AND trigger the internal refill
+    /// how many were taken plus any shortfall.
-    /// mechanism (tokens are only refilled during try_wait/wait calls).
+    ///
-    /// directly adjust available tokens via set_available() to consume the remaining amount.
+    /// `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 {
+        consume_tokens(&guard, n)
-            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)
     }
 }
 
@@ -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<AtomicU64>,
+    }
+
+    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");

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);

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 }