diff --git a/crates/lib/caching/memory_caching/src/arc_ttl_sieve.rs b/crates/lib/caching/memory_caching/src/arc_ttl_sieve.rs new file mode 100644 index 0000000000..597ffec93e --- /dev/null +++ b/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 { + value: V, + stored_at: Instant, +} + +impl SieveItemWithTtl { + fn new(value: V) -> Self { + Self { + value, + stored_at: Instant::now(), + } + } +} + +#[derive(Clone)] +pub struct ArcTtlSieve { + cache: Arc>>>, + /// 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 ArcTtlSieve { + + pub fn with_capacity_and_ttl_duration(capacity: usize, ttl_duration: Duration) -> AnyhowResult { + 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> { + 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 = + 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 = + 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 = + 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); + } +} diff --git a/crates/lib/caching/memory_caching/src/lib.rs b/crates/lib/caching/memory_caching/src/lib.rs index 7bad86985d..7dc511d182 100644 --- a/crates/lib/caching/memory_caching/src/lib.rs +++ b/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; diff --git a/crates/service/web/storyteller_web/src/main.rs b/crates/service/web/storyteller_web/src/main.rs index cb17d0e9fb..6ff796ab8d 100644 --- a/crates/service/web/storyteller_web/src/main.rs +++ b/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)), )?, } }, diff --git a/crates/service/web/storyteller_web/src/state/server_state.rs b/crates/service/web/storyteller_web/src/state/server_state.rs index 54b189a32d..77d3a5cf93 100644 --- a/crates/service/web/storyteller_web/src/state/server_state.rs +++ b/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, /// Cache of featured media files - pub featured_media_files_sieve: ArcSieve, + pub featured_media_files_sieve: ArcTtlSieve, } #[derive(Clone)]