Re: [pbs-devel] [PATCH proxmox v2 2/3] http: track user tag updates on rate-limited streams

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

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@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>>> {


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