[pbs-devel] [PATCH proxmox v2 2/3] http: track user tag updates on rate-limited streams
Christian Ebner
c.ebner at proxmox.com
Mon Nov 10 09:53:10 CET 2025
On 11/7/25 2:23 PM, Hannes Laimer wrote:
> Introduce rate-limit tags with a user variant and let rate-limited
> streams hold a shared handle so callbacks can refresh limits
> whenever the tag set changes.
>
> Signed-off-by: Hannes Laimer <h.laimer at proxmox.com>
> ---
> proxmox-http/src/lib.rs | 2 +-
> proxmox-http/src/rate_limited_stream.rs | 35 ++++++++++++++++++++++++-
> 2 files changed, 35 insertions(+), 2 deletions(-)
>
> diff --git a/proxmox-http/src/lib.rs b/proxmox-http/src/lib.rs
> index 8b6953b0..990f8f36 100644
> --- a/proxmox-http/src/lib.rs
> +++ b/proxmox-http/src/lib.rs
> @@ -34,7 +34,7 @@ pub use rate_limiter::{RateLimit, RateLimiter, RateLimiterVec, ShareableRateLimi
> #[cfg(feature = "rate-limited-stream")]
> mod rate_limited_stream;
> #[cfg(feature = "rate-limited-stream")]
> -pub use rate_limited_stream::RateLimitedStream;
> +pub use rate_limited_stream::{RateLimitedStream, RateLimiterTag, RateLimiterTags};
>
> #[cfg(feature = "body")]
> mod body;
> diff --git a/proxmox-http/src/rate_limited_stream.rs b/proxmox-http/src/rate_limited_stream.rs
> index e24df7af..4bc5c414 100644
> --- a/proxmox-http/src/rate_limited_stream.rs
> +++ b/proxmox-http/src/rate_limited_stream.rs
> @@ -15,6 +15,13 @@ use super::{RateLimiter, ShareableRateLimit};
>
> type SharedRateLimit = Arc<dyn ShareableRateLimit>;
>
> +#[derive(Clone, Debug, PartialEq, Eq)]
> +pub enum RateLimiterTag {
> + User(String),
> +}
note: just to add a justification for the enum here, which could also be
part of the commit message. I see potential here to add group based
filter if there will be a user group implementation as requested in [0].
[0] https://bugzilla.proxmox.com/show_bug.cgi?id=5867
> +
> +pub type RateLimiterTags = Vec<RateLimiterTag>;
> +
> pub type RateLimiterCallback =
> dyn Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send;
I still see potential to improve the code here as already mentioned in
v1. If the list of RateLimiterTag's is passed as paramter via the
callback, that will simplify the logic for the patches on the
rest-server implementation. See the suggested diff below an on the next
patch.
>
> @@ -26,6 +33,8 @@ pub struct RateLimitedStream<S> {
> write_delay: Option<Pin<Box<Sleep>>>,
> update_limiter_cb: Option<Box<RateLimiterCallback>>,
> last_limiter_update: Instant,
> + tag_handle: Option<Arc<Mutex<RateLimiterTags>>>,
> + last_tags: Option<RateLimiterTags>,
> stream: S,
> }
>
> @@ -53,6 +62,8 @@ impl<S> RateLimitedStream<S> {
> write_delay: None,
> update_limiter_cb: None,
> last_limiter_update: Instant::now(),
> + tag_handle: None,
> + last_tags: None,
> stream,
> }
> }
> @@ -77,13 +88,26 @@ impl<S> RateLimitedStream<S> {
> write_delay: None,
> update_limiter_cb: Some(Box::new(update_limiter_cb)),
> last_limiter_update: Instant::now(),
> + tag_handle: None,
> + last_tags: None,
> stream,
> }
> }
>
> fn update_limiters(&mut self) {
> if let Some(ref update_limiter_cb) = self.update_limiter_cb {
> - if self.last_limiter_update.elapsed().as_secs() >= 5 {
> + let mut force_update = false;
> + let current_tags = self
> + .tag_handle
> + .as_ref()
> + .map(|handle| handle.lock().unwrap().clone());
> +
> + if self.last_tags != current_tags {
> + self.last_tags = current_tags;
> + force_update = true;
> + }
> +
> + if force_update || self.last_limiter_update.elapsed().as_secs() >= 5 {
> self.last_limiter_update = Instant::now();
> let (read_limiter, write_limiter) = update_limiter_cb();
> self.read_limiter = read_limiter;
> @@ -99,6 +123,15 @@ impl<S> RateLimitedStream<S> {
> pub fn inner_mut(&mut self) -> &mut S {
> &mut self.stream
> }
> +
> + pub fn tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> + self.tag_handle.as_ref().map(Arc::clone)
> + }
> +
> + pub fn set_tag_handle(&mut self, handle: Arc<Mutex<RateLimiterTags>>) {
> + self.tag_handle = Some(handle);
> + self.last_tags = None;
> + }
> }
>
> fn register_traffic(limiter: &dyn ShareableRateLimit, count: usize) -> Option<Pin<Box<Sleep>>> {
diff --git a/proxmox-http/src/rate_limited_stream.rs
b/proxmox-http/src/rate_limited_stream.rs
index 4bc5c414..bbb6cd6d 100644
--- a/proxmox-http/src/rate_limited_stream.rs
+++ b/proxmox-http/src/rate_limited_stream.rs
@@ -23,7 +23,7 @@ pub enum RateLimiterTag {
pub type RateLimiterTags = Vec<RateLimiterTag>;
pub type RateLimiterCallback =
- dyn Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send;
+ dyn Fn(&[RateLimiterTag]) -> (Option<SharedRateLimit>,
Option<SharedRateLimit>) + Send;
/// A rate limited stream using [RateLimiter]
pub struct RateLimitedStream<S> {
@@ -75,12 +75,13 @@ impl<S> RateLimitedStream<S> {
/// Note: This function is called within an async context, so it
/// should be fast and must not block.
pub fn with_limiter_update_cb<
- F: Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) +
Send + 'static,
+ F: Fn(&[RateLimiterTag]) -> (Option<SharedRateLimit>,
Option<SharedRateLimit>) + Send + 'static,
>(
stream: S,
update_limiter_cb: F,
) -> Self {
- let (read_limiter, write_limiter) = update_limiter_cb();
+ let tag_handle = Some(Arc::new(Mutex::new(Vec::new())));
+ let (read_limiter, write_limiter) = update_limiter_cb(&[]);
Self {
read_limiter,
read_delay: None,
@@ -88,7 +89,7 @@ impl<S> RateLimitedStream<S> {
write_delay: None,
update_limiter_cb: Some(Box::new(update_limiter_cb)),
last_limiter_update: Instant::now(),
- tag_handle: None,
+ tag_handle,
last_tags: None,
stream,
}
@@ -109,7 +110,8 @@ impl<S> RateLimitedStream<S> {
if force_update ||
self.last_limiter_update.elapsed().as_secs() >= 5 {
self.last_limiter_update = Instant::now();
- let (read_limiter, write_limiter) = update_limiter_cb();
+ let tags = self.last_tags.as_ref().map(|tags|
tags.as_slice()).unwrap_or(&[]);
+ let (read_limiter, write_limiter) =
update_limiter_cb(tags);
self.read_limiter = read_limiter;
self.write_limiter = write_limiter;
}
@@ -127,11 +129,6 @@ impl<S> RateLimitedStream<S> {
pub fn tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
self.tag_handle.as_ref().map(Arc::clone)
}
-
- pub fn set_tag_handle(&mut self, handle: Arc<Mutex<RateLimiterTags>>) {
- self.tag_handle = Some(handle);
- self.last_tags = None;
- }
}
fn register_traffic(limiter: &dyn ShareableRateLimit, count: usize) ->
Option<Pin<Box<Sleep>>> {
More information about the pbs-devel
mailing list