[PATCH datacenter-manager 3/4] api: resources: subscriptions: switch over to pdm_cache

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

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
 server/src/api/resources.rs | 82 ++++++++++++++-----------------------
 1 file changed, 31 insertions(+), 51 deletions(-)

diff --git a/server/src/api/resources.rs b/server/src/api/resources.rs
index 50315b11..c22e33c5 100644
--- a/server/src/api/resources.rs
+++ b/server/src/api/resources.rs
@@ -30,9 +30,10 @@ use proxmox_schema::{api, parse_boolean};
 use proxmox_sortable_macro::sortable;
 use proxmox_subscription::SubscriptionStatus;
 use pve_api_types::{ClusterResource, ClusterResourceNetworkType, ClusterResourceType};
+use serde::{Deserialize, Serialize};
 
 use crate::metric_collection::top_entities;
-use crate::{connection, views};
+use crate::{connection, pdm_cache, views};
 
 pub const ROUTER: Router = Router::new()
     .get(&list_subdirs_api_method!(SUBDIRS))
@@ -798,15 +799,11 @@ async fn get_top_entities(
     Ok(res)
 }
 
-#[derive(Clone)]
+#[derive(Clone, Serialize, Deserialize)]
 struct CachedSubscriptionState {
     node_info: HashMap<String, Option<NodeSubscriptionInfo>>,
-    timestamp: i64,
 }
 
-static SUBSCRIPTION_CACHE: LazyLock<RwLock<HashMap<String, CachedSubscriptionState>>> =
-    LazyLock::new(|| RwLock::new(HashMap::new()));
-
 /// Get the subscription state for a given remote.
 ///
 /// If recent enough cached data is available, it is returned
@@ -815,66 +812,49 @@ pub 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) {
+    if let Some(cached_subscription) =
+        get_cached_subscription_info(remote.id.clone(), max_age).await?
+    {
         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);
+        update_cached_subscription_info(remote.id.clone(), node_info.clone()).await?;
         Ok(node_info)
     }
 }
 
-fn get_cached_subscription_info(remote: &str, max_age: u64) -> Option<CachedSubscriptionState> {
-    let cache = SUBSCRIPTION_CACHE
-        .read()
-        .expect("subscription mutex poisoned");
+const SUBSCRIPTION_STATE_CACHE_KEY: &str = "subscription-state";
 
-    if max_age == 0 {
-        return None;
-    }
-    if let Some(cached_subscription) = cache.get(remote) {
-        let now = proxmox_time::epoch_i64();
-        let diff = now - cached_subscription.timestamp;
+async fn get_cached_subscription_info(
+    remote: String,
+    max_age: u64,
+) -> Result<Option<CachedSubscriptionState>, Error> {
+    tokio::task::spawn_blocking(move || {
+        let cache = pdm_cache::instance().read_remote(&remote)?;
 
-        if diff >= max_age as i64 || diff < 0 {
-            // value is too old or from the future
-            None
-        } else {
-            Some(cached_subscription.clone())
-        }
-    } else {
-        None
-    }
+        Ok(cache.get_with_max_age(SUBSCRIPTION_STATE_CACHE_KEY, max_age as i64)?)
+    })
+    .await?
 }
 
 /// 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");
+async fn update_cached_subscription_info(
+    remote: String,
+    node_info: HashMap<String, Option<NodeSubscriptionInfo>>,
+) -> Result<(), Error> {
+    tokio::task::spawn_blocking(move || {
+        let cache = pdm_cache::instance().write_remote(&remote)?;
 
-    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,
-        },
-    );
+        Ok(cache.set(
+            SUBSCRIPTION_STATE_CACHE_KEY,
+            &CachedSubscriptionState {
+                node_info: node_info,
+            },
+        )?)
+    })
+    .await?
 }
 
 /// Maps a list of node subscription infos into a single [`RemoteSubscriptionState`]
-- 
2.47.3