[pbs-devel] [PATCH proxmox v2 1/4] rate-limiter: add crate for traffic rate limiter implementations

[Christian Ebner <c.ebner@proxmox.com>]

Factors out the traffic rate limiter implementations currently tied
to the proxmox-backup and proxmox-http crates to make them
independent and easily reusable, e.g. for the s3-client
implementation.

The shared rate limiter implementation from PBS relies on mmapping
for state sharing, the file exposed having a predefined magic number.
In order to be backwards compatible, leave the magic number as is and
only adapt the constant name to be more generic, although the string
the magic number is derived from is PBS specific.

Further, the user for file ownership and base path of the mmapped
file are now passed as parameters during shared rate limiter
instantiation.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
Changes since version 1:
- move rate limiter implementations into dedicated crate instead of
  proxmox-http
- Adapt shared state file magic constant name to be generic

 Cargo.toml                                    |   1 +
 proxmox-rate-limiter/Cargo.toml               |  29 +++
 proxmox-rate-limiter/debian/changelog         |   5 +
 proxmox-rate-limiter/debian/control           |  70 ++++++
 proxmox-rate-limiter/debian/copyright         |  18 ++
 proxmox-rate-limiter/debian/debcargo.toml     |   7 +
 proxmox-rate-limiter/src/lib.rs               |  13 ++
 proxmox-rate-limiter/src/rate_limiter.rs      | 214 ++++++++++++++++++
 .../src/shared_rate_limiter.rs                | 130 +++++++++++
 9 files changed, 487 insertions(+)
 create mode 100644 proxmox-rate-limiter/Cargo.toml
 create mode 100644 proxmox-rate-limiter/debian/changelog
 create mode 100644 proxmox-rate-limiter/debian/control
 create mode 100644 proxmox-rate-limiter/debian/copyright
 create mode 100644 proxmox-rate-limiter/debian/debcargo.toml
 create mode 100644 proxmox-rate-limiter/src/lib.rs
 create mode 100644 proxmox-rate-limiter/src/rate_limiter.rs
 create mode 100644 proxmox-rate-limiter/src/shared_rate_limiter.rs

diff --git a/Cargo.toml b/Cargo.toml
index f149af65..bde32b17 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,6 +29,7 @@ members = [
     "proxmox-notify",
     "proxmox-openid",
     "proxmox-product-config",
+    "proxmox-rate-limiter",
     "proxmox-resource-scheduling",
     "proxmox-rest-server",
     "proxmox-router",
diff --git a/proxmox-rate-limiter/Cargo.toml b/proxmox-rate-limiter/Cargo.toml
new file mode 100644
index 00000000..6d8d96cc
--- /dev/null
+++ b/proxmox-rate-limiter/Cargo.toml
@@ -0,0 +1,29 @@
+[package]
+name = "proxmox-rate-limiter"
+description = "Token bucket based traffic rate limiter implementation"
+version = "1.0.0"
+
+authors.workspace = true
+edition.workspace = true
+exclude.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+hyper = { workspace = true, optional = true }
+nix = { workspace = true, optional = true }
+
+proxmox-shared-memory = { workspace = true, optional = true }
+proxmox-sys = { workspace = true, optional = true }
+
+[features]
+default = []
+rate-limiter = ["dep:hyper"]
+shared-rate-limiter = [
+    "dep:nix",
+    "dep:proxmox-shared-memory",
+    "dep:proxmox-sys",
+]
diff --git a/proxmox-rate-limiter/debian/changelog b/proxmox-rate-limiter/debian/changelog
new file mode 100644
index 00000000..0bffa551
--- /dev/null
+++ b/proxmox-rate-limiter/debian/changelog
@@ -0,0 +1,5 @@
+rust-proxmox-rate-limiter (1.0.0-1) bookworm; urgency=medium
+
+  * initial packaging
+
+ -- Proxmox Support Team <support@proxmox.com>  Tue, 16 Sep 2025 11:06:23 +0200
diff --git a/proxmox-rate-limiter/debian/control b/proxmox-rate-limiter/debian/control
new file mode 100644
index 00000000..689fe02e
--- /dev/null
+++ b/proxmox-rate-limiter/debian/control
@@ -0,0 +1,70 @@
+Source: rust-proxmox-rate-limiter
+Section: rust
+Priority: optional
+Build-Depends: debhelper-compat (= 13),
+ dh-sequence-cargo
+Build-Depends-Arch: cargo:native <!nocheck>,
+ rustc:native (>= 1.82) <!nocheck>,
+ libstd-rust-dev <!nocheck>,
+ librust-anyhow-1+default-dev <!nocheck>
+Maintainer: Proxmox Support Team <support@proxmox.com>
+Standards-Version: 4.7.0
+Vcs-Git: git://git.proxmox.com/git/proxmox.git
+Vcs-Browser: https://git.proxmox.com/?p=proxmox.git
+Homepage: https://proxmox.com
+X-Cargo-Crate: proxmox-rate-limiter
+Rules-Requires-Root: no
+
+Package: librust-proxmox-rate-limiter-dev
+Architecture: any
+Multi-Arch: same
+Depends:
+ ${misc:Depends},
+ librust-anyhow-1+default-dev
+Suggests:
+ librust-proxmox-rate-limiter+rate-limiter-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter+shared-rate-limiter-dev (= ${binary:Version})
+Provides:
+ librust-proxmox-rate-limiter+default-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1+default-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0+default-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0.0-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0.0+default-dev (= ${binary:Version})
+Description: Token bucket based traffic rate limiter implementation - Rust source code
+ Source code for Debianized Rust crate "proxmox-rate-limiter"
+
+Package: librust-proxmox-rate-limiter+rate-limiter-dev
+Architecture: any
+Multi-Arch: same
+Depends:
+ ${misc:Depends},
+ librust-proxmox-rate-limiter-dev (= ${binary:Version}),
+ librust-hyper-1+default-dev
+Provides:
+ librust-proxmox-rate-limiter-1+rate-limiter-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0+rate-limiter-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0.0+rate-limiter-dev (= ${binary:Version})
+Description: Token bucket based traffic rate limiter implementation - feature "rate-limiter"
+ This metapackage enables feature "rate-limiter" for the Rust proxmox-rate-
+ limiter crate, by pulling in any additional dependencies needed by that
+ feature.
+
+Package: librust-proxmox-rate-limiter+shared-rate-limiter-dev
+Architecture: any
+Multi-Arch: same
+Depends:
+ ${misc:Depends},
+ librust-proxmox-rate-limiter-dev (= ${binary:Version}),
+ librust-nix-0.29+default-dev,
+ librust-proxmox-shared-memory-1+default-dev,
+ librust-proxmox-sys-1+default-dev
+Provides:
+ librust-proxmox-rate-limiter-1+shared-rate-limiter-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0+shared-rate-limiter-dev (= ${binary:Version}),
+ librust-proxmox-rate-limiter-1.0.0+shared-rate-limiter-dev (= ${binary:Version})
+Description: Token bucket based traffic rate limiter implementation - feature "shared-rate-limiter"
+ This metapackage enables feature "shared-rate-limiter" for the Rust proxmox-
+ rate-limiter crate, by pulling in any additional dependencies needed by that
+ feature.
diff --git a/proxmox-rate-limiter/debian/copyright b/proxmox-rate-limiter/debian/copyright
new file mode 100644
index 00000000..d6e3c304
--- /dev/null
+++ b/proxmox-rate-limiter/debian/copyright
@@ -0,0 +1,18 @@
+Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
+
+Files:
+ *
+Copyright: 2025 Proxmox Server Solutions GmbH <support@proxmox.com>
+License: AGPL-3.0-or-later
+ This program is free software: you can redistribute it and/or modify it under
+ the terms of the GNU Affero General Public License as published by the Free
+ Software Foundation, either version 3 of the License, or (at your option) any
+ later version.
+ .
+ This program is distributed in the hope that it will be useful, but WITHOUT
+ ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+ details.
+ .
+ You should have received a copy of the GNU Affero General Public License along
+ with this program. If not, see <https://www.gnu.org/licenses/>.
diff --git a/proxmox-rate-limiter/debian/debcargo.toml b/proxmox-rate-limiter/debian/debcargo.toml
new file mode 100644
index 00000000..b7864cdb
--- /dev/null
+++ b/proxmox-rate-limiter/debian/debcargo.toml
@@ -0,0 +1,7 @@
+overlay = "."
+crate_src_path = ".."
+maintainer = "Proxmox Support Team <support@proxmox.com>"
+
+[source]
+vcs_git = "git://git.proxmox.com/git/proxmox.git"
+vcs_browser = "https://git.proxmox.com/?p=proxmox.git"
diff --git a/proxmox-rate-limiter/src/lib.rs b/proxmox-rate-limiter/src/lib.rs
new file mode 100644
index 00000000..a8d7cfdd
--- /dev/null
+++ b/proxmox-rate-limiter/src/lib.rs
@@ -0,0 +1,13 @@
+//! Token bucket based traffic rate limiter implementations.
+
+#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
+
+#[cfg(feature = "rate-limiter")]
+mod rate_limiter;
+#[cfg(feature = "rate-limiter")]
+pub use rate_limiter::{RateLimit, RateLimiter, RateLimiterVec, ShareableRateLimit};
+
+#[cfg(feature = "shared-rate-limiter")]
+mod shared_rate_limiter;
+#[cfg(feature = "shared-rate-limiter")]
+pub use shared_rate_limiter::SharedRateLimiter;
diff --git a/proxmox-rate-limiter/src/rate_limiter.rs b/proxmox-rate-limiter/src/rate_limiter.rs
new file mode 100644
index 00000000..945c77a6
--- /dev/null
+++ b/proxmox-rate-limiter/src/rate_limiter.rs
@@ -0,0 +1,214 @@
+use std::convert::TryInto;
+use std::time::{Duration, Instant};
+
+use anyhow::{bail, Error};
+
+/// Rate limiter interface.
+pub trait RateLimit {
+    /// Update rate and bucket size
+    fn update_rate(&mut self, rate: u64, bucket_size: u64);
+
+    /// Returns the overall traffic (since started)
+    fn traffic(&self) -> u64;
+
+    /// Register traffic, returning a proposed delay to reach the
+    /// expected rate.
+    fn register_traffic(&mut self, current_time: Instant, data_len: u64) -> Duration;
+}
+
+/// Like [`RateLimit`], but does not require self to be mutable.
+///
+/// This is useful for types providing internal mutability (Mutex).
+pub trait ShareableRateLimit: Send + Sync {
+    fn update_rate(&self, rate: u64, bucket_size: u64);
+    fn traffic(&self) -> u64;
+    fn register_traffic(&self, current_time: Instant, data_len: u64) -> Duration;
+}
+
+/// IMPORTANT: We use this struct in shared memory, so please do not
+/// change/modify the layout (do not add fields)
+#[derive(Clone)]
+#[repr(C)]
+struct TbfState {
+    traffic: u64, // overall traffic
+    last_update: Instant,
+    consumed_tokens: u64,
+}
+
+impl TbfState {
+    const NO_DELAY: Duration = Duration::from_millis(0);
+
+    fn refill_bucket(&mut self, rate: u64, current_time: Instant) {
+        let time_diff = match current_time.checked_duration_since(self.last_update) {
+            Some(duration) => duration.as_nanos(),
+            None => return,
+        };
+
+        if time_diff == 0 {
+            return;
+        }
+
+        self.last_update = current_time;
+
+        let allowed_traffic = ((time_diff.saturating_mul(rate as u128)) / 1_000_000_000)
+            .try_into()
+            .unwrap_or(u64::MAX);
+
+        self.consumed_tokens = self.consumed_tokens.saturating_sub(allowed_traffic);
+    }
+
+    fn register_traffic(
+        &mut self,
+        rate: u64,
+        bucket_size: u64,
+        current_time: Instant,
+        data_len: u64,
+    ) -> Duration {
+        self.refill_bucket(rate, current_time);
+
+        self.traffic += data_len;
+        self.consumed_tokens += data_len;
+
+        if self.consumed_tokens <= bucket_size {
+            return Self::NO_DELAY;
+        }
+        Duration::from_nanos(
+            (self.consumed_tokens - bucket_size).saturating_mul(1_000_000_000) / rate,
+        )
+    }
+}
+
+/// Token bucket based rate limiter
+///
+/// IMPORTANT: We use this struct in shared memory, so please do not
+/// change/modify the layout (do not add fields)
+#[repr(C)]
+pub struct RateLimiter {
+    rate: u64,        // tokens/second
+    bucket_size: u64, // TBF bucket size
+    state: TbfState,
+}
+
+impl RateLimiter {
+    /// Creates a new instance, using [Instant::now] as start time.
+    pub fn new(rate: u64, bucket_size: u64) -> Self {
+        let start_time = Instant::now();
+        Self::with_start_time(rate, bucket_size, start_time)
+    }
+
+    /// Creates a new instance with specified `rate`, `bucket_size` and `start_time`.
+    pub fn with_start_time(rate: u64, bucket_size: u64, start_time: Instant) -> Self {
+        Self {
+            rate,
+            bucket_size,
+            state: TbfState {
+                traffic: 0,
+                last_update: start_time,
+                // start with empty bucket (all tokens consumed)
+                consumed_tokens: bucket_size,
+            },
+        }
+    }
+}
+
+impl RateLimit for RateLimiter {
+    fn update_rate(&mut self, rate: u64, bucket_size: u64) {
+        self.rate = rate;
+
+        if bucket_size < self.bucket_size && self.state.consumed_tokens > bucket_size {
+            self.state.consumed_tokens = bucket_size; // start again
+        }
+
+        self.bucket_size = bucket_size;
+    }
+
+    fn traffic(&self) -> u64 {
+        self.state.traffic
+    }
+
+    fn register_traffic(&mut self, current_time: Instant, data_len: u64) -> Duration {
+        self.state
+            .register_traffic(self.rate, self.bucket_size, current_time, data_len)
+    }
+}
+
+impl<R: RateLimit + Send> ShareableRateLimit for std::sync::Mutex<R> {
+    fn update_rate(&self, rate: u64, bucket_size: u64) {
+        self.lock().unwrap().update_rate(rate, bucket_size);
+    }
+
+    fn traffic(&self) -> u64 {
+        self.lock().unwrap().traffic()
+    }
+
+    fn register_traffic(&self, current_time: Instant, data_len: u64) -> Duration {
+        self.lock()
+            .unwrap()
+            .register_traffic(current_time, data_len)
+    }
+}
+
+/// Array of rate limiters.
+///
+/// A group of rate limiters with same configuration.
+pub struct RateLimiterVec {
+    rate: u64,        // tokens/second
+    bucket_size: u64, // TBF bucket size
+    state: Vec<TbfState>,
+}
+
+impl RateLimiterVec {
+    /// Creates a new instance, using [Instant::now] as start time.
+    pub fn new(group_size: usize, rate: u64, bucket_size: u64) -> Self {
+        let start_time = Instant::now();
+        Self::with_start_time(group_size, rate, bucket_size, start_time)
+    }
+
+    /// Creates a new instance with specified `rate`, `bucket_size` and `start_time`.
+    pub fn with_start_time(
+        group_size: usize,
+        rate: u64,
+        bucket_size: u64,
+        start_time: Instant,
+    ) -> Self {
+        let state = TbfState {
+            traffic: 0,
+            last_update: start_time,
+            // start with empty bucket (all tokens consumed)
+            consumed_tokens: bucket_size,
+        };
+        Self {
+            rate,
+            bucket_size,
+            state: vec![state; group_size],
+        }
+    }
+
+    #[allow(clippy::len_without_is_empty)]
+    /// Return the number of TBF entries (group_size)
+    pub fn len(&self) -> usize {
+        self.state.len()
+    }
+
+    /// Traffic for the specified index
+    pub fn traffic(&self, index: usize) -> Result<u64, Error> {
+        if index >= self.state.len() {
+            bail!("RateLimiterVec::traffic - index out of range");
+        }
+        Ok(self.state[index].traffic)
+    }
+
+    /// Register traffic at the specified index
+    pub fn register_traffic(
+        &mut self,
+        index: usize,
+        current_time: Instant,
+        data_len: u64,
+    ) -> Result<Duration, Error> {
+        if index >= self.state.len() {
+            bail!("RateLimiterVec::register_traffic - index out of range");
+        }
+
+        Ok(self.state[index].register_traffic(self.rate, self.bucket_size, current_time, data_len))
+    }
+}
diff --git a/proxmox-rate-limiter/src/shared_rate_limiter.rs b/proxmox-rate-limiter/src/shared_rate_limiter.rs
new file mode 100644
index 00000000..2822e7ea
--- /dev/null
+++ b/proxmox-rate-limiter/src/shared_rate_limiter.rs
@@ -0,0 +1,130 @@
+//! Rate limiter designed for shared memory
+
+use std::mem::MaybeUninit;
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+use anyhow::{bail, Error};
+use nix::sys::stat::Mode;
+use nix::unistd::User;
+
+use proxmox_shared_memory::{check_subtype, initialize_subtype};
+use proxmox_shared_memory::{Init, SharedMemory, SharedMutex};
+use proxmox_sys::fs::{create_path, CreateOptions};
+
+use crate::{RateLimit, RateLimiter, ShareableRateLimit};
+
+/// Magic number for shared rate limiter exposed file mappings
+///
+/// Generated by `openssl::sha::sha256(b"Proxmox Backup SharedRateLimiter v1.0")[0..8];`
+/// Original magic number kept when factored out from the initial
+/// PBS implementation for full backwards compatibility.
+pub const PROXMOX_SHARED_RATE_LIMITER_MAGIC_1_0: [u8; 8] = [6, 58, 213, 96, 161, 122, 130, 117];
+
+// Wrap RateLimiter, so that we can provide an Init impl
+#[repr(C)]
+struct WrapLimiter(RateLimiter);
+
+impl Init for WrapLimiter {
+    fn initialize(this: &mut MaybeUninit<Self>) {
+        // default does not matter here, because we override later
+        this.write(WrapLimiter(RateLimiter::new(1_000_000, 1_000_000)));
+    }
+}
+
+#[repr(C)]
+struct SharedRateLimiterData {
+    magic: [u8; 8],
+    tbf: SharedMutex<WrapLimiter>,
+    padding: [u8; 4096 - 104],
+}
+
+impl Init for SharedRateLimiterData {
+    fn initialize(this: &mut MaybeUninit<Self>) {
+        unsafe {
+            let me = &mut *this.as_mut_ptr();
+            me.magic = PROXMOX_SHARED_RATE_LIMITER_MAGIC_1_0;
+            initialize_subtype(&mut me.tbf);
+        }
+    }
+
+    fn check_type_magic(this: &MaybeUninit<Self>) -> Result<(), Error> {
+        unsafe {
+            let me = &*this.as_ptr();
+            if me.magic != PROXMOX_SHARED_RATE_LIMITER_MAGIC_1_0 {
+                bail!("SharedRateLimiterData: wrong magic number");
+            }
+            check_subtype(&me.tbf)?;
+            Ok(())
+        }
+    }
+}
+
+/// Rate limiter designed for shared memory ([SharedMemory])
+///
+/// The actual [RateLimiter] is protected by a [SharedMutex] and
+/// implements [Init]. This way we can share the limiter between
+/// different processes.
+pub struct SharedRateLimiter {
+    shmem: SharedMemory<SharedRateLimiterData>,
+}
+
+impl SharedRateLimiter {
+    /// Creates a new mmap'ed instance.
+    ///
+    /// Data is mapped in `<base_path>/<name>` using
+    /// `TMPFS`.
+    pub fn mmap_shmem<P: AsRef<Path>>(
+        name: &str,
+        rate: u64,
+        burst: u64,
+        user: User,
+        base_path: P,
+    ) -> Result<Self, Error> {
+        let mut path = base_path.as_ref().to_path_buf();
+
+        let dir_opts = CreateOptions::new()
+            .perm(Mode::from_bits_truncate(0o770))
+            .owner(user.uid)
+            .group(user.gid);
+
+        create_path(&path, Some(dir_opts), Some(dir_opts))?;
+
+        path.push(name);
+
+        let file_opts = CreateOptions::new()
+            .perm(Mode::from_bits_truncate(0o660))
+            .owner(user.uid)
+            .group(user.gid);
+
+        let shmem: SharedMemory<SharedRateLimiterData> = SharedMemory::open(&path, file_opts)?;
+
+        shmem.data().tbf.lock().0.update_rate(rate, burst);
+
+        Ok(Self { shmem })
+    }
+}
+
+impl ShareableRateLimit for SharedRateLimiter {
+    fn update_rate(&self, rate: u64, bucket_size: u64) {
+        self.shmem
+            .data()
+            .tbf
+            .lock()
+            .0
+            .update_rate(rate, bucket_size);
+    }
+
+    fn traffic(&self) -> u64 {
+        self.shmem.data().tbf.lock().0.traffic()
+    }
+
+    fn register_traffic(&self, current_time: Instant, data_len: u64) -> Duration {
+        self.shmem
+            .data()
+            .tbf
+            .lock()
+            .0
+            .register_traffic(current_time, data_len)
+    }
+}
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel