Re: [pdm-devel] [PATCH datacenter-manager 2/3] server: introduce ApiCache abstraction

["Lukas Wagner" <l.wagner@proxmox.com>]

A couple of smaller comments inline.

In general: We should have unit tests for new or refactored code like this.
Should be pretty simple for ApiCache, the only 'problem' I see are the calls to
`proxmox_time::epoch_i64`, maybe ApiCache could have a test-only
`set_time` function to inject a timestamp for test cases. Maybe there
are better solutions for this though.

On Mon Oct 13, 2025 at 10:33 AM CEST, Dominik Csapak wrote:
> Abstract our resources/subscription cache mechanism away into a struct
> that takes care of the max-age logic and fetching/inserting.
>
> This can be useful to have in general, and we don't need to copy it
> every time.
>
> We still have to use a LazyLock at the usage site and give the correct
> Type, but we can deduplicate most of the access/update logic.
>
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
> note: there may be better ways to abstract that away, i just went
> with the most obvious way to not duplicate the methods cache/max-age
> handling we had.
>
>  server/src/api/resources.rs                  | 139 +++----------------
>  server/src/api_cache.rs                      |  87 ++++++++++++
>  server/src/lib.rs                            |   1 +
>  server/src/metric_collection/top_entities.rs |   2 +-
>  4 files changed, 106 insertions(+), 123 deletions(-)
>  create mode 100644 server/src/api_cache.rs
>
> diff --git a/server/src/api/resources.rs b/server/src/api/resources.rs
> index 39a400c3..5b7ac524 100644
> --- a/server/src/api/resources.rs
> +++ b/server/src/api/resources.rs
> @@ -1,6 +1,6 @@
>  use std::collections::HashMap;
>  use std::str::FromStr;
> -use std::sync::{LazyLock, RwLock};
> +use std::sync::LazyLock;
>  
>  use anyhow::{bail, format_err, Error};
>  use futures::future::join_all;
> @@ -28,6 +28,7 @@ use proxmox_sortable_macro::sortable;
>  use proxmox_subscription::SubscriptionStatus;
>  use pve_api_types::{ClusterResource, ClusterResourceType};
>  
> +use crate::api_cache::ApiCache;
>  use crate::connection;
>  use crate::metric_collection::top_entities;
>  
> @@ -521,14 +522,8 @@ async fn get_top_entities(
>      Ok(res)
>  }
>  
> -#[derive(Clone)]
> -struct CachedSubscriptionState {
> -    node_info: HashMap<String, Option<NodeSubscriptionInfo>>,
> -    timestamp: i64,
> -}
> -
> -static SUBSCRIPTION_CACHE: LazyLock<RwLock<HashMap<String, CachedSubscriptionState>>> =
> -    LazyLock::new(|| RwLock::new(HashMap::new()));
> +static SUBSCRIPTION_CACHE: LazyLock<ApiCache<HashMap<String, Option<NodeSubscriptionInfo>>>> =
> +    LazyLock::new(ApiCache::new);
>  
>  /// Get the subscription state for a given remote.
>  ///
> @@ -538,63 +533,11 @@ async fn get_subscription_info_for_remote(
>      remote: &Remote,
>      max_age: u64,
>  ) -> Result<HashMap<String, Option<NodeSubscriptionInfo>>, Error> {
> -    if let Some(cached_subscription) = get_cached_subscription_info(&remote.id, max_age) {
> -        Ok(cached_subscription.node_info)
> -    } else {
> -        let node_info = fetch_remote_subscription_info(remote).await?;
> -        let now = proxmox_time::epoch_i64();
> -        update_cached_subscription_info(&remote.id, &node_info, now);
> -        Ok(node_info)
> -    }
> -}
> -
> -fn get_cached_subscription_info(remote: &str, max_age: u64) -> Option<CachedSubscriptionState> {
> -    let cache = SUBSCRIPTION_CACHE
> -        .read()
> -        .expect("subscription mutex poisoned");
> -
> -    if let Some(cached_subscription) = cache.get(remote) {
> -        let now = proxmox_time::epoch_i64();
> -        let diff = now - cached_subscription.timestamp;
> -
> -        if diff > max_age as i64 || diff < 0 {
> -            // value is too old or from the future
> -            None
> -        } else {
> -            Some(cached_subscription.clone())
> -        }
> -    } else {
> -        None
> -    }
> -}
> -
> -/// Update cached subscription data.
> -///
> -/// If the cache already contains more recent data we don't insert the passed resources.
> -fn update_cached_subscription_info(
> -    remote: &str,
> -    node_info: &HashMap<String, Option<NodeSubscriptionInfo>>,
> -    now: i64,
> -) {
> -    // there is no good way to recover from this, so panicking should be fine
> -    let mut cache = SUBSCRIPTION_CACHE
> -        .write()
> -        .expect("subscription mutex poisoned");
> -
> -    if let Some(cached_resource) = cache.get(remote) {
> -        // skip updating if the data is new enough
> -        if cached_resource.timestamp >= now {
> -            return;
> -        }
> -    }
> -
> -    cache.insert(
> -        remote.into(),
> -        CachedSubscriptionState {
> -            node_info: node_info.clone(),
> -            timestamp: now,
> -        },
> -    );
> +    SUBSCRIPTION_CACHE
> +        .get_updated_value(&remote.id, max_age as i64, |_| async move {
> +            fetch_remote_subscription_info(remote).await
> +        })
> +        .await
>  }
>  
>  /// Maps a list of node subscription infos into a single [`RemoteSubscriptionState`]
> @@ -687,14 +630,7 @@ async fn fetch_remote_subscription_info(
>      Ok(list)
>  }
>  
> -#[derive(Clone)]
> -pub struct CachedResources {
> -    pub resources: Vec<Resource>,
> -    pub timestamp: i64,
> -}
> -
> -static CACHE: LazyLock<RwLock<HashMap<String, CachedResources>>> =
> -    LazyLock::new(|| RwLock::new(HashMap::new()));
> +static CACHE: LazyLock<ApiCache<Vec<Resource>>> = LazyLock::new(ApiCache::new);
>  
>  /// Get resources for a given remote.
>  ///
> @@ -702,57 +638,16 @@ static CACHE: LazyLock<RwLock<HashMap<String, CachedResources>>> =
>  /// instead of calling out to the remote.
>  async fn get_resources_for_remote(remote: &Remote, max_age: u64) -> Result<Vec<Resource>, Error> {
>      let remote_name = remote.id.to_owned();
> -    if let Some(cached_resource) = get_cached_resources(&remote_name, max_age) {
> -        Ok(cached_resource.resources)
> -    } else {
> -        let resources = fetch_remote_resource(remote).await?;
> -        let now = proxmox_time::epoch_i64();
> -        update_cached_resources(&remote_name, &resources, now);
> -        Ok(resources)
> -    }
> +    CACHE
> +        .get_updated_value(&remote_name, max_age as i64, |_| async move {
> +            fetch_remote_resource(remote).await
> +        })
> +        .await
>  }
>  
>  /// Read cached resource data from the cache
> -pub fn get_cached_resources(remote: &str, max_age: u64) -> Option<CachedResources> {
> -    // there is no good way to recover from this, so panicking should be fine
> -    let cache = CACHE.read().expect("mutex poisoned");
> -
> -    if let Some(cached_resource) = cache.get(remote) {
> -        let now = proxmox_time::epoch_i64();
> -        let diff = now - cached_resource.timestamp;
> -
> -        if diff > max_age as i64 || diff < 0 {
> -            // value is too old or from the future
> -            None
> -        } else {
> -            Some(cached_resource.clone())
> -        }
> -    } else {
> -        None
> -    }
> -}
> -
> -/// Update cached resource data.
> -///
> -/// If the cache already contains more recent data we don't insert the passed resources.
> -fn update_cached_resources(remote: &str, resources: &[Resource], now: i64) {
> -    // there is no good way to recover from this, so panicking should be fine
> -    let mut cache = CACHE.write().expect("mutex poisoned");
> -
> -    if let Some(cached_resource) = cache.get(remote) {
> -        // skip updating if existing data is newer
> -        if cached_resource.timestamp >= now {
> -            return;
> -        }
> -    }
> -
> -    cache.insert(
> -        remote.into(),
> -        CachedResources {
> -            timestamp: now,
> -            resources: resources.into(),
> -        },
> -    );
> +pub fn get_cached_resources(remote: &str, max_age: u64) -> Option<Vec<Resource>> {
> +    CACHE.get_cached_value(remote, max_age as i64)
>  }
>  
>  /// Fetch remote resources and map to pdm-native data types.
> diff --git a/server/src/api_cache.rs b/server/src/api_cache.rs
> new file mode 100644
> index 00000000..1339fe64
> --- /dev/null
> +++ b/server/src/api_cache.rs
> @@ -0,0 +1,87 @@
> +use std::{collections::HashMap, future::Future, sync::RwLock};
> +
> +/// Holds the given data per remote with a timestamp indicating the last update
> +pub struct ApiCache<CacheType> {
> +    data: RwLock<HashMap<String, (i64, CacheType)>>,
> +}
> +
> +impl<CacheType> ApiCache<CacheType> {
> +    /// Creates a new instance of `ApiCache`
> +    pub fn new() -> Self {
> +        Self {
> +            data: RwLock::new(HashMap::new()),
> +        }
> +    }
> +
> +    fn update_cache(&self, remote: &str, data: CacheType, now: i64) {
> +        // there is no good way to recover from this, so panicking should be fine
> +        let mut cache = self.data.write().expect("cache mutex poisoned");
> +
> +        if let Some((timestamp, _)) = cache.get(remote) {
> +            // skip updating if existing data is newer
> +            if *timestamp >= now {
> +                return;
> +            }
> +        }
> +
> +        cache.insert(remote.into(), (now, data));
> +    }
> +}
> +
> +impl<CacheType> Default for ApiCache<CacheType> {
> +    fn default() -> Self {
> +        Self::new()
> +    }
> +}
> +
> +impl<CacheType> ApiCache<CacheType>
> +where
> +    CacheType: Clone,
> +{
> +    /// Returns the cached value for the given remote if it's not older than `max_age`
> +    pub fn get_cached_value(&self, remote: &str, max_age: i64) -> Option<CacheType> {

tiny nit:

I think since the whole thing itself is nothing more than a cache, the
'cached' part is already implied, so maybe just call it 'get_value'?

> +        let cache = self.data.read().expect("cache mutex poisoned");
> +        if let Some((timestamp, cached_data)) = cache.get(remote) {
> +            let now = proxmox_time::epoch_i64();
> +            let diff = now - timestamp;
> +            if diff > max_age || diff < 0 {
> +                // value is too old or from the future
> +                None
> +            } else {
> +                Some(cached_data.clone())
> +            }
> +        } else {
> +            None
> +        }
> +    }
> +}
> +

v ^ You can merge these two impl blocks.

> +impl<CacheType> ApiCache<CacheType>
> +where
> +    CacheType: Clone,
> +{
> +    /// Returns data for the given remote that is not older than `max_age` either
> +    /// from the cache or updated via the given `update_func`
> +    pub async fn get_updated_value<S, F, R>(

tiny nit: How about `get_value_or_update`? Since you only update if it is too old
or does not exist?

> +        &self,
> +        remote: &str,
> +        max_age: i64,
> +        update_func: F,
> +    ) -> Result<CacheType, anyhow::Error>
> +    where
> +        S: Into<CacheType>,
> +        F: Fn(&str) -> R,
> +        R: Future<Output = Result<S, anyhow::Error>>,
> +    {
> +        if let Some(cached_data) = self.get_cached_value(remote, max_age) {
> +            Ok(cached_data)
> +        } else {
> +            let data = update_func(remote).await?;
> +            let now = proxmox_time::epoch_i64();
> +
> +            let data: CacheType = data.into();
> +            self.update_cache(remote, data.clone(), now);
> +            Ok(data)
> +        }
> +    }
> +}
> diff --git a/server/src/lib.rs b/server/src/lib.rs
> index 964807eb..1f8508c9 100644
> --- a/server/src/lib.rs
> +++ b/server/src/lib.rs
> @@ -2,6 +2,7 @@
>  
>  pub mod acl;
>  pub mod api;
> +pub mod api_cache;
>  pub mod auth;
>  pub mod context;
>  pub mod env;
> diff --git a/server/src/metric_collection/top_entities.rs b/server/src/metric_collection/top_entities.rs
> index ea121eef..715d0eeb 100644
> --- a/server/src/metric_collection/top_entities.rs
> +++ b/server/src/metric_collection/top_entities.rs
> @@ -50,7 +50,7 @@ pub fn calculate_top(
>          if let Some(data) =
>              crate::api::resources::get_cached_resources(remote_name, i64::MAX as u64)
>          {
> -            for res in data.resources {
> +            for res in data {
>                  let id = res.id().to_string();
>                  let name = format!("pve/{remote_name}/{id}");
>                  match &res {



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