Implement TTL Sieve to replace permanently stuck Sieve

crates/lib/caching/memory_caching/src/arc_ttl_sieve.rs

@@ -0,0 +1,159 @@
+use std::hash::Hash;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use anyhow::{anyhow, bail};
+use sieve_cache::SieveCache;
+
+use errors::AnyhowResult;
+
+#[derive(Clone)]
+struct SieveItemWithTtl<V: Clone + ?Sized> {
+  value: V,
+  stored_at: Instant,
+}
+
+impl <V: Clone + ?Sized> SieveItemWithTtl<V> {
+  fn new(value: V) -> Self {
+    Self {
+      value,
+      stored_at: Instant::now(),
+    }
+  }
+}
+
+#[derive(Clone)]
+pub struct ArcTtlSieve<K: Eq + Hash + Clone, V: Clone + ?Sized> {
+  cache: Arc<Mutex<SieveCache<K, SieveItemWithTtl<V>>>>,
+  /// Purge items from the sieve if they're requested after this duration.
+  /// Do not bump the expiry on successful lookups against the sieve.
+  ttl_duration: Duration,
+}
+
+impl <K: Eq + Hash + Clone, V: Clone + ?Sized> ArcTtlSieve<K, V> {
+
+  pub fn with_capacity_and_ttl_duration(capacity: usize, ttl_duration: Duration) -> AnyhowResult<Self> {
+    let cache = SieveCache::new(capacity)
+        .map_err(|err| anyhow!("could not create cache: {:?}", err))?;
+    Ok(Self {
+      cache: Arc::new(Mutex::new(cache)),
+      ttl_duration,
+    })
+  }
+
+  pub fn store(&self, key: K, value: V) -> AnyhowResult<()> {
+    match self.cache.lock() {
+      Err(e) => bail!("could not unlock mutex to write: {:?}", e),
+      Ok(mut cache) => {
+        let item = SieveItemWithTtl::new(value);
+        cache.insert(key, item);
+      },
+    };
+    Ok(())
+  }
+
+  pub fn store_copy(&self, key: &K, value: &V) -> AnyhowResult<()> {
+    match self.cache.lock() {
+      Err(e) => bail!("could not unlock mutex to write: {:?}", e),
+      Ok(mut cache) => {
+        let item = SieveItemWithTtl::new(value.clone());
+        cache.insert(key.clone(), item);
+      },
+    };
+    Ok(())
+  }
+
+  pub fn get_copy(&self, key: &K) -> AnyhowResult<Option<V>> {
+    match self.cache.lock() {
+      Err(e) => bail!("could not unlock mutex to read: {:?}", e),
+      Ok(mut cache) => {
+        let maybe_item = cache.get(key);
+        if let Some(item) = maybe_item {
+          if item.stored_at.elapsed() < self.ttl_duration {
+            return Ok(Some(item.value.clone()));
+          } else {
+            // NB: The item has elapsed its TTL, so we'll drop it.
+            cache.remove(key);
+          }
+        }
+        return Ok(None);
+      },
+    }
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use std::thread;
+  use std::time::Duration;
+  use crate::arc_ttl_sieve::ArcTtlSieve;
+
+  #[test]
+  fn test_get_copy() {
+    let sieve : ArcTtlSieve<String, String> =
+        ArcTtlSieve::with_capacity_and_ttl_duration(1, Duration::from_secs(100))
+            .unwrap();
+
+    let key = "key".to_string();
+    let value = "value".to_string();
+
+    sieve.store_copy(&key, &value).unwrap();
+    let maybe_value = sieve.get_copy(&key).unwrap();
+    assert_eq!(maybe_value, Some("value".to_string()));
+  }
+
+  #[test]
+  fn test_simple_expire() {
+    let sieve : ArcTtlSieve<String, String> =
+        ArcTtlSieve::with_capacity_and_ttl_duration(1, Duration::ZERO)
+            .unwrap();
+
+    let key = "key".to_string();
+    let value = "value".to_string();
+
+    sieve.store_copy(&key, &value).unwrap();
+    let maybe_value = sieve.get_copy(&key).unwrap();
+
+    assert_eq!(maybe_value, None);
+  }
+
+  #[test]
+  fn test_simple_get_then_expire() {
+    let sieve : ArcTtlSieve<String, String> =
+        ArcTtlSieve::with_capacity_and_ttl_duration(1, Duration::from_secs(2))
+            .unwrap();
+
+    let key = "key".to_string();
+    let value = "value".to_string();
+
+    sieve.store_copy(&key, &value).unwrap();
+
+    let mut count = 0;
+
+    for _ in 0 .. 10_000_000 {
+      let maybe_value = sieve.get_copy(&key).unwrap();
+
+      if maybe_value.is_none() {
+        break;
+      } else {
+        assert_eq!(maybe_value, Some("value".to_string()));
+        count += 1;
+      }
+    }
+
+    // NB: We should be able to retrieve the item roughly over a million times in two seconds,
+    // but in case there's an interrupt we'll only check that we got it a few times.
+    assert!(count > 3);
+
+    let mut maybe_value = sieve.get_copy(&key).unwrap();
+
+    if maybe_value.is_some() {
+      // The TTL should have expired by now, but just in case...
+      thread::sleep(Duration::from_secs(2));
+    }
+
+    maybe_value = sieve.get_copy(&key).unwrap();
+
+    assert_eq!(maybe_value, None);
+  }
+}

crates/lib/caching/memory_caching/src/lib.rs

@@ -15,6 +15,7 @@
 #![allow(non_snake_case)]
 
 pub mod arc_sieve;
+pub mod arc_ttl_sieve;
 pub mod multi_item_ttl_cache;
 pub mod single_item_ttl_cache;
 pub mod ttl_key_counter;

crates/service/web/storyteller_web/src/main.rs

@@ -61,6 +61,7 @@ use config::shared_constants::DEFAULT_RUST_LOG;
 use email_sender::smtp_email_sender::SmtpEmailSender;
 use errors::AnyhowResult;
 use memory_caching::arc_sieve::ArcSieve;
+use memory_caching::arc_ttl_sieve::ArcTtlSieve;
 use memory_caching::single_item_ttl_cache::SingleItemTtlCache;
 use mysql_queries::mediators::badge_granter::BadgeGranter;
 use mysql_queries::mediators::firehose_publisher::FirehosePublisher;
@@ -449,8 +450,9 @@ async fn main() -> AnyhowResult<()> {
           easyenv::get_env_duration_seconds_or_default(
             "QUEUE_STATS_CACHE_TTL_SECONDS",
             Duration::from_secs(60))),
-        featured_media_files_sieve: ArcSieve::with_capacity(
-          easyenv::get_env_num("FEATURED_MEDIA_FILES_CACHE_SIZE", 25)?
+        featured_media_files_sieve: ArcTtlSieve::with_capacity_and_ttl_duration(
+          easyenv::get_env_num("FEATURED_MEDIA_FILES_CACHE_SIZE", 25)?,
+          easyenv::get_env_duration_seconds_or_default("FEATURED_MEDIA_FILES_TTL_SECONDS", Duration::from_secs(60)),
         )?,
       }
     },

crates/service/web/storyteller_web/src/state/server_state.rs

@@ -7,7 +7,7 @@ use actix_helpers::middleware::banned_ip_filter::ip_ban_list::ip_ban_list::IpBan
 use billing_component::stripe::stripe_config::StripeConfig;
 use cloud_storage::bucket_client::BucketClient;
 use email_sender::smtp_email_sender::SmtpEmailSender;
-use memory_caching::arc_sieve::ArcSieve;
+use memory_caching::arc_ttl_sieve::ArcTtlSieve;
 use memory_caching::single_item_ttl_cache::SingleItemTtlCache;
 use mysql_queries::mediators::badge_granter::BadgeGranter;
 use mysql_queries::mediators::firehose_publisher::FirehosePublisher;
@@ -197,7 +197,7 @@ pub struct EphemeralInMemoryCaches {
   pub leaderboard: SingleItemTtlCache<LeaderboardInfo>,
 
   /// Cache of featured media files
-  pub featured_media_files_sieve: ArcSieve<ListFeaturedMediaFilesQueryParams, FeaturedMediaFileListPage>,
+  pub featured_media_files_sieve: ArcTtlSieve<ListFeaturedMediaFilesQueryParams, FeaturedMediaFileListPage>,
 }
 
 #[derive(Clone)]