new generic cache and refresher

2023-10-05 17:11:03 +01:00
parent 04bf153701
commit 00cdc5010a
3 changed files with 279 additions and 0 deletions
@@ -0,0 +1,141 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::ops::Deref;
+use std::sync::Arc;
+use std::time::Duration;
+use thiserror::Error;
+use time::OffsetDateTime;
+use tokio::sync::{RwLock, RwLockReadGuard};
+
+#[derive(Debug, Error)]
+#[error("the cache item has not been initialised")]
+pub struct UninitialisedCache;
+
+pub struct SharedCache<T>(Arc<RwLock<CachedItem<T>>>);
+
+impl<T> Clone for SharedCache<T> {
+    fn clone(&self) -> Self {
+        SharedCache(Arc::clone(&self.0))
+    }
+}
+
+impl<T> Default for SharedCache<T> {
+    fn default() -> Self {
+        SharedCache(Arc::new(RwLock::new(CachedItem { inner: None })))
+    }
+}
+
+impl<T> SharedCache<T> {
+    pub(crate) fn new() -> Self {
+        SharedCache::default()
+    }
+
+    pub(crate) async fn update(&self, value: T) {
+        let mut guard = self.0.write().await;
+        if let Some(ref mut existing) = guard.inner {
+            existing.update(value)
+        } else {
+            guard.inner = Some(Cache::new(value))
+        }
+    }
+
+    pub(crate) async fn get(&self) -> Result<RwLockReadGuard<'_, Cache<T>>, UninitialisedCache> {
+        let guard = self.0.read().await;
+        RwLockReadGuard::try_map(guard, |a| a.inner.as_ref()).map_err(|_| UninitialisedCache)
+    }
+
+    // ignores expiration data
+    #[allow(dead_code)]
+    pub(crate) async fn unchecked_get_inner(
+        &self,
+    ) -> Result<RwLockReadGuard<'_, T>, UninitialisedCache> {
+        Ok(RwLockReadGuard::map(self.get().await?, |a| &a.value))
+    }
+}
+
+impl<T> From<Cache<T>> for SharedCache<T> {
+    fn from(value: Cache<T>) -> Self {
+        SharedCache(Arc::new(RwLock::new(value.into())))
+    }
+}
+
+#[derive(Default)]
+pub(crate) struct CachedItem<T> {
+    inner: Option<Cache<T>>,
+}
+
+impl<T> CachedItem<T> {
+    #[allow(dead_code)]
+    fn get(&self) -> Result<&Cache<T>, UninitialisedCache> {
+        self.inner.as_ref().ok_or(UninitialisedCache)
+    }
+}
+
+impl<T> From<Cache<T>> for CachedItem<T> {
+    fn from(value: Cache<T>) -> Self {
+        CachedItem { inner: Some(value) }
+    }
+}
+
+pub struct Cache<T> {
+    value: T,
+    as_at: i64,
+}
+
+impl<T> Cache<T> {
+    fn new(value: T) -> Self {
+        Cache {
+            value,
+            as_at: current_unix_timestamp(),
+        }
+    }
+
+    fn update(&mut self, value: T) {
+        self.value = value;
+        self.as_at = current_unix_timestamp()
+    }
+
+    #[allow(dead_code)]
+    pub fn has_expired(&self, ttl: Duration, now: Option<OffsetDateTime>) -> bool {
+        let now = now.unwrap_or(OffsetDateTime::now_utc()).unix_timestamp();
+        let diff = now - self.as_at;
+
+        diff > (ttl.as_secs() as i64)
+    }
+
+    #[allow(dead_code)]
+    pub fn timestamp(&self) -> i64 {
+        self.as_at
+    }
+
+    #[allow(dead_code)]
+    pub fn into_inner(self) -> T {
+        self.value
+    }
+}
+
+impl<T> Deref for Cache<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.value
+    }
+}
+
+impl<T> Default for Cache<T>
+where
+    T: Default,
+{
+    fn default() -> Self {
+        Cache {
+            value: T::default(),
+            as_at: 0,
+        }
+    }
+}
+
+fn current_unix_timestamp() -> i64 {
+    let now = OffsetDateTime::now_utc();
+    now.unix_timestamp()
+}
@@ -1,7 +1,13 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
 use serde::Serialize;
 use std::ops::Deref;
 use time::OffsetDateTime;

+pub(crate) mod cache;
+pub(crate) mod refresher;
+
 #[derive(Serialize, Clone)]
 pub struct Cache<T> {
    pub value: T,
@@ -0,0 +1,132 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::support::caching::cache::SharedCache;
+use nym_task::TaskClient;
+use std::time::Duration;
+use tokio::time::interval;
+
+pub struct CacheRefresher<T, E> {
+    name: String,
+    refreshing_interval: Duration,
+
+    // TODO: the Send + Sync bounds are only required for the `start` method. could we maybe make it less restrictive?
+    provider: Box<dyn CacheItemProvider<Error = E, Item = T> + Send + Sync>,
+    shared_cache: SharedCache<T>,
+    // triggers: Vec<Box<dyn RefreshTriggerTrait>>,
+}
+
+#[async_trait]
+pub trait CacheItemProvider {
+    type Item;
+    type Error: std::error::Error;
+
+    async fn try_refresh(&self) -> Result<Self::Item, Self::Error>;
+}
+
+// pub struct TriggerFailure;
+//
+// #[async_trait]
+// pub trait RefreshTriggerTrait {
+//     async fn triggerred(&mut self) -> Result<(), TriggerFailure>;
+// }
+//
+// // TODO: how to get rid of `T: Send + Sync`? it really doesn't need to be Send + Sync
+// // since it's wrapped in Shared<T> internally anyway
+// #[async_trait]
+// impl<T> RefreshTriggerTrait for watch::Receiver<T>
+// where
+//     T: Send + Sync,
+// {
+//     async fn triggerred(&mut self) -> Result<(), TriggerFailure> {
+//         self.changed().await.map_err(|err| {
+//             error!("failed to process refresh trigger: {err}");
+//             TriggerFailure
+//         })
+//     }
+// }
+
+impl<T, E> CacheRefresher<T, E>
+where
+    E: std::error::Error,
+{
+    pub(crate) fn new(
+        item_provider: Box<dyn CacheItemProvider<Error = E, Item = T> + Send + Sync>,
+        refreshing_interval: Duration,
+    ) -> Self {
+        CacheRefresher {
+            name: "GenericCacheRefresher".to_string(),
+            refreshing_interval,
+            provider: item_provider,
+            shared_cache: SharedCache::new(),
+        }
+    }
+
+    pub(crate) fn new_with_initial_value(
+        item_provider: Box<dyn CacheItemProvider<Error = E, Item = T> + Send + Sync>,
+        refreshing_interval: Duration,
+        shared_cache: SharedCache<T>,
+    ) -> Self {
+        CacheRefresher {
+            name: "GenericCacheRefresher".to_string(),
+            refreshing_interval,
+            provider: item_provider,
+            shared_cache,
+        }
+    }
+
+    #[must_use]
+    pub(crate) fn named(mut self, name: impl Into<String>) -> Self {
+        self.name = name.into();
+        self
+    }
+
+    #[allow(dead_code)]
+    pub(crate) fn get_shared_cache(&self) -> SharedCache<T> {
+        self.shared_cache.clone()
+    }
+
+    async fn do_refresh_cache(&self) {
+        match self.provider.try_refresh().await {
+            Ok(updated_items) => {
+                self.shared_cache.update(updated_items).await;
+            }
+            Err(err) => {
+                error!("{}: failed to refresh the cache: {err}", self.name)
+            }
+        }
+    }
+
+    pub async fn refresh(&self, task_client: &mut TaskClient) {
+        log::info!("{}: refreshing cache state", self.name);
+
+        tokio::select! {
+            biased;
+            _ = task_client.recv() => {
+                log::trace!("{}: Received shutdown while refreshing cache", self.name)
+            }
+            _ = self.do_refresh_cache() => (),
+        }
+    }
+
+    pub async fn run(&self, mut task_client: TaskClient) {
+        let mut refresh_interval = interval(self.refreshing_interval);
+        while !task_client.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = task_client.recv() => {
+                    log::trace!("{}: Received shutdown", self.name)
+                }
+                _ = refresh_interval.tick() => self.refresh(&mut task_client).await,
+            }
+        }
+    }
+
+    pub fn start(self, task_client: TaskClient)
+    where
+        T: Send + Sync + 'static,
+        E: Send + Sync + 'static,
+    {
+        tokio::spawn(async move { self.run(task_client).await });
+    }
+}