fix(gifs): bound shard cache memory

fluxer_gifs/Cargo.toml

@@ -14,7 +14,7 @@ fluxer-svc = { path = "../fluxer_svc", default-features = false }
 hmac = "0.13.0"
 moka = { version = "0.12.15", features = ["future", "sync"] }
 reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
-serde = { version = "1.0.228", features = ["derive"] }
+serde = { version = "1.0.228", features = ["derive", "rc"] }
 serde_json = "1.0.150"
 sha2 = "0.11.0"
 tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }

fluxer_gifs/src/klipy.rs

@@ -9,7 +9,7 @@ use serde_json::Value;
 use std::collections::{BTreeMap, HashSet};
 use std::sync::LazyLock;
 use std::time::Duration;
-use tokio::time::sleep;
+use tokio::time::{sleep, timeout};
 
 const KLIPY_BASE_URL: &str = "https://api.klipy.com/v2";
 const KLIPY_DIRECT_BASE_URL: &str = "https://api.klipy.com/api/v1";
@@ -18,6 +18,8 @@ const CLIENT_KEY: &str = "fluxer";
 const MAX_RETRIES: usize = 3;
 const BACKOFF_BASE_DELAY: Duration = Duration::from_secs(1);
 const KLIPY_RESPONSE_LIMIT_BYTES: usize = 512 * 1024;
+const MAX_FEATURED_CATEGORIES: usize = 50;
+const FEATURED_CATEGORIES_FETCH_TIMEOUT: Duration = Duration::from_secs(30);
 const FLUXER_USER_AGENT: &str = "Fluxerbot/1.0 (+https://fluxer.app)";
 const KLIPY_PROVIDER_NAME: &str = "klipy";
 const KLIPY_FEATURED_CATEGORY_REFRESH_COUNTRY: &str = "US";
@@ -295,6 +297,19 @@ impl KlipyClient {
         &self,
         api_key: &str,
         locale: &str,
+    ) -> anyhow::Result<Vec<GifCategoryTag>> {
+        timeout(
+            FEATURED_CATEGORIES_FETCH_TIMEOUT,
+            self.fetch_featured_categories(api_key, locale),
+        )
+        .await
+        .context("KLIPY featured categories request timed out")?
+    }
+
+    async fn fetch_featured_categories(
+        &self,
+        api_key: &str,
+        locale: &str,
     ) -> anyhow::Result<Vec<GifCategoryTag>> {
         let normalized_locale = normalize_locale(locale);
         let response: TagsResponse = self
@@ -317,6 +332,7 @@ impl KlipyClient {
             .map(|tag| tag.searchterm.trim().to_owned())
             .filter(|term| !term.is_empty())
             .filter(|term| seen.insert(term.clone()))
+            .take(MAX_FEATURED_CATEGORIES)
             .collect::<Vec<_>>();
 
         let mut categories = Vec::with_capacity(search_terms.len());
@@ -438,7 +454,7 @@ impl KlipyClient {
     where
         T: serde::de::DeserializeOwned,
     {
-        let response = self
+        let mut response = self
             .http_client
             .get(url)
             .send()
@@ -460,13 +476,21 @@ impl KlipyClient {
         {
             anyhow::bail!("KLIPY response declared more than {KLIPY_RESPONSE_LIMIT_BYTES} bytes");
         }
-        let bytes = response
-            .bytes()
+        let initial_capacity = response
+            .content_length()
+            .and_then(|length| usize::try_from(length).ok())
+            .unwrap_or_default();
+        let mut bytes = Vec::with_capacity(initial_capacity);
+        while let Some(chunk) = response
+            .chunk()
             .await
             .map_err(reqwest::Error::without_url)
-            .with_context(|| format!("KLIPY {label} response body failed"))?;
-        if bytes.len() > KLIPY_RESPONSE_LIMIT_BYTES {
-            anyhow::bail!("KLIPY {label} response exceeded {KLIPY_RESPONSE_LIMIT_BYTES} bytes");
+            .with_context(|| format!("KLIPY {label} response body failed"))?
+        {
+            if bytes.len().saturating_add(chunk.len()) > KLIPY_RESPONSE_LIMIT_BYTES {
+                anyhow::bail!("KLIPY {label} response exceeded {KLIPY_RESPONSE_LIMIT_BYTES} bytes");
+            }
+            bytes.extend_from_slice(&chunk);
         }
         serde_json::from_slice(&bytes)
             .with_context(|| format!("failed to parse KLIPY {label} response"))

fluxer_gifs/src/main.rs

@@ -32,7 +32,7 @@ async fn main() -> anyhow::Result<()> {
             fluxer_svc::router::run_router(&config, router, transport).await
         }
         Mode::Shard => {
-            let shard = GifsShard::new(&config)?;
+            let shard = GifsShard::new()?;
             fluxer_svc::shard::run_shard(&config, shard, transport).await
         }
     }

fluxer_gifs/src/shard_impl.rs

@@ -2,15 +2,21 @@
 
 use crate::klipy::{KlipyClient, build_share_url, extract_slug_from_url, resolve_cache_key};
 use crate::media_proxy::MediaProxyUrlBuilder;
-use crate::types::{GifCategoryTag, GifItem, GifRequest, GifServiceResponse};
-use fluxer_svc::config::ServiceConfig;
+use crate::types::{GifCategoryTag, GifItem, GifMediaFormat, GifRequest, GifServiceResponse};
+use fluxer_svc::config::optional_env;
 use fluxer_svc::shard::ShardService;
-use moka::future::Cache;
+use moka::notification::RemovalCause;
+use moka::ops::compute::{CompResult, Op};
+use moka::{Expiry, future::Cache};
 use std::collections::HashSet;
+use std::fmt::Write;
 use std::future::Future;
-use std::sync::Arc;
+use std::mem::size_of;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
-use tokio::sync::Mutex;
+use tokio::sync::{Semaphore, TryAcquireError};
+use tokio::time::MissedTickBehavior;
 
 const SEARCH_SOFT_TTL: Duration = Duration::from_secs(30);
 const SEARCH_HARD_TTL: Duration = Duration::from_secs(5 * 60);
@@ -22,6 +28,24 @@ const CATEGORIES_SOFT_TTL: Duration = Duration::from_secs(24 * 60 * 60);
 const CATEGORIES_HARD_TTL: Duration = Duration::from_secs(48 * 60 * 60);
 const RESOLVE_SOFT_TTL: Duration = Duration::from_secs(30 * 60);
 const RESOLVE_HARD_TTL: Duration = Duration::from_secs(2 * 60 * 60);
+const DEFAULT_SHARD_CACHE_MAX_BYTES: u64 = 512 * 1024 * 1024;
+const MIN_SHARD_CACHE_MAX_BYTES: u64 = 16 * 1024 * 1024;
+const CACHE_BUDGET_PARTS: u64 = 16;
+const GIF_LIST_CACHE_PARTS: u64 = 10;
+const CATEGORY_CACHE_PARTS: u64 = 1;
+const SUGGESTION_CACHE_PARTS: u64 = 1;
+const RESOLVED_CACHE_PARTS: u64 = 4;
+const MAX_CONCURRENT_REFRESHES: usize = 8;
+const CACHE_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(30);
+const CACHE_ENTRY_OVERHEAD_BYTES: u64 = 256;
+const BTREE_ENTRY_OVERHEAD_BYTES: u64 = 64;
+const ARC_HEADER_BYTES: u64 = (2 * size_of::<usize>()) as u64;
+const METRICS_ORDERING: Ordering = Ordering::Relaxed;
+
+const _: () = assert!(
+    GIF_LIST_CACHE_PARTS + CATEGORY_CACHE_PARTS + SUGGESTION_CACHE_PARTS + RESOLVED_CACHE_PARTS
+        == CACHE_BUDGET_PARTS
+);
 
 #[derive(Clone)]
 pub struct GifsShard {
@@ -30,20 +54,51 @@ pub struct GifsShard {
 
 struct GifsShardInner {
     klipy: KlipyClient,
-    gif_lists: Cache<String, Cached<Vec<GifItem>>>,
-    categories: Cache<String, Cached<Vec<GifCategoryTag>>>,
-    suggestions: Cache<String, Cached<Vec<String>>>,
-    resolved: Cache<String, Cached<Option<GifItem>>>,
-    refreshing: Mutex<HashSet<String>>,
+    gif_lists: ManagedCache<Vec<GifItem>>,
+    categories: ManagedCache<Vec<GifCategoryTag>>,
+    suggestions: ManagedCache<Vec<String>>,
+    resolved: ManagedCache<Option<GifItem>>,
+    refreshing: Mutex<HashSet<RefreshKey>>,
+    refresh_permits: Arc<Semaphore>,
+}
+
+struct ManagedCache<T> {
+    cache: Cache<String, Cached<T>>,
+    telemetry: Arc<CacheTelemetry>,
+    max_weight_bytes: u64,
+}
+
+impl<T> Clone for ManagedCache<T> {
+    fn clone(&self) -> Self {
+        Self {
+            cache: self.cache.clone(),
+            telemetry: Arc::clone(&self.telemetry),
+            max_weight_bytes: self.max_weight_bytes,
+        }
+    }
 }
 
-#[derive(Debug, Clone)]
 struct Cached<T> {
-    data: T,
+    data: Arc<T>,
     stored_at: Instant,
+    generation: u64,
+    policy: CachePolicy,
+    retained_bytes: u32,
 }
 
-#[derive(Debug, Clone, Copy)]
+impl<T> Clone for Cached<T> {
+    fn clone(&self) -> Self {
+        Self {
+            data: Arc::clone(&self.data),
+            stored_at: self.stored_at,
+            generation: self.generation,
+            policy: self.policy,
+            retained_bytes: self.retained_bytes,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct CachePolicy {
     soft_ttl: Duration,
     hard_ttl: Duration,
@@ -51,15 +106,252 @@ struct CachePolicy {
 
 impl CachePolicy {
     const fn new(soft_ttl: Duration, hard_ttl: Duration) -> Self {
+        assert!(soft_ttl.as_nanos() <= hard_ttl.as_nanos());
         Self { soft_ttl, hard_ttl }
     }
 }
 
-impl<T> Cached<T> {
-    fn new(data: T) -> Self {
+struct CachedExpiry;
+
+impl<T> Expiry<String, Cached<T>> for CachedExpiry {
+    fn expire_after_create(
+        &self,
+        _key: &String,
+        value: &Cached<T>,
+        _created_at: Instant,
+    ) -> Option<Duration> {
+        Some(value.policy.hard_ttl)
+    }
+
+    fn expire_after_update(
+        &self,
+        _key: &String,
+        value: &Cached<T>,
+        _updated_at: Instant,
+        _duration_until_expiry: Option<Duration>,
+    ) -> Option<Duration> {
+        Some(value.policy.hard_ttl)
+    }
+}
+
+#[derive(Clone, Hash, PartialEq, Eq)]
+struct RefreshKey {
+    cache_name: &'static str,
+    key: String,
+}
+
+struct CacheTelemetry {
+    name: &'static str,
+    hits_total: AtomicU64,
+    misses_total: AtomicU64,
+    size_evictions_total: AtomicU64,
+    expired_evictions_total: AtomicU64,
+    refresh_success_total: AtomicU64,
+    refresh_error_total: AtomicU64,
+    refresh_superseded_total: AtomicU64,
+    refresh_duplicate_total: AtomicU64,
+    refresh_saturated_total: AtomicU64,
+    refresh_in_flight: AtomicU64,
+    next_generation: AtomicU64,
+}
+
+impl CacheTelemetry {
+    fn new(name: &'static str) -> Self {
         Self {
-            data,
+            name,
+            hits_total: AtomicU64::new(0),
+            misses_total: AtomicU64::new(0),
+            size_evictions_total: AtomicU64::new(0),
+            expired_evictions_total: AtomicU64::new(0),
+            refresh_success_total: AtomicU64::new(0),
+            refresh_error_total: AtomicU64::new(0),
+            refresh_superseded_total: AtomicU64::new(0),
+            refresh_duplicate_total: AtomicU64::new(0),
+            refresh_saturated_total: AtomicU64::new(0),
+            refresh_in_flight: AtomicU64::new(0),
+            next_generation: AtomicU64::new(1),
+        }
+    }
+
+    fn next_generation(&self) -> u64 {
+        self.next_generation
+            .fetch_update(METRICS_ORDERING, METRICS_ORDERING, |generation| {
+                generation.checked_add(1)
+            })
+            .expect("GIF cache generation exhausted")
+    }
+
+    fn record_removal(&self, cause: RemovalCause) {
+        match cause {
+            RemovalCause::Expired => {
+                self.expired_evictions_total.fetch_add(1, METRICS_ORDERING);
+            }
+            RemovalCause::Size => {
+                self.size_evictions_total.fetch_add(1, METRICS_ORDERING);
+            }
+            RemovalCause::Explicit | RemovalCause::Replaced => {}
+        }
+    }
+}
+
+struct RefreshRegistration {
+    inner: Arc<GifsShardInner>,
+    telemetry: Arc<CacheTelemetry>,
+    key: RefreshKey,
+}
+
+impl Drop for RefreshRegistration {
+    fn drop(&mut self) {
+        let removed = self
+            .inner
+            .refreshing
+            .lock()
+            .expect("GIF refresh registry mutex poisoned")
+            .remove(&self.key);
+        assert!(removed, "GIF refresh registry lost an active key");
+        self.telemetry
+            .refresh_in_flight
+            .fetch_sub(1, METRICS_ORDERING);
+    }
+}
+
+#[derive(Clone, Copy)]
+struct CacheBudgets {
+    total: u64,
+    gif_lists: u64,
+    categories: u64,
+    suggestions: u64,
+    resolved: u64,
+}
+
+impl CacheBudgets {
+    fn from_env() -> anyhow::Result<Self> {
+        let total = match optional_env("FLUXER_GIFS_SHARD_CACHE_MAX_BYTES") {
+            Some(value) => value.parse::<u64>().map_err(|error| {
+                anyhow::anyhow!("invalid FLUXER_GIFS_SHARD_CACHE_MAX_BYTES: {error}")
+            })?,
+            None => DEFAULT_SHARD_CACHE_MAX_BYTES,
+        };
+        if total < MIN_SHARD_CACHE_MAX_BYTES {
+            anyhow::bail!(
+                "FLUXER_GIFS_SHARD_CACHE_MAX_BYTES must be at least {MIN_SHARD_CACHE_MAX_BYTES}"
+            );
+        }
+
+        let part = total / CACHE_BUDGET_PARTS;
+        let categories = part.saturating_mul(CATEGORY_CACHE_PARTS);
+        let suggestions = part.saturating_mul(SUGGESTION_CACHE_PARTS);
+        let resolved = part.saturating_mul(RESOLVED_CACHE_PARTS);
+        let gif_lists = total
+            .checked_sub(categories)
+            .and_then(|remaining| remaining.checked_sub(suggestions))
+            .and_then(|remaining| remaining.checked_sub(resolved))
+            .expect("GIF cache budget partition exceeds total");
+        assert!(gif_lists >= part.saturating_mul(GIF_LIST_CACHE_PARTS));
+
+        Ok(Self {
+            total,
+            gif_lists,
+            categories,
+            suggestions,
+            resolved,
+        })
+    }
+}
+
+trait HeapSize {
+    fn heap_bytes(&self) -> u64;
+}
+
+impl HeapSize for String {
+    fn heap_bytes(&self) -> u64 {
+        usize_bytes(self.capacity())
+    }
+}
+
+impl<T> HeapSize for Vec<T>
+where
+    T: HeapSize,
+{
+    fn heap_bytes(&self) -> u64 {
+        self.iter().fold(
+            usize_bytes(self.capacity()).saturating_mul(usize_bytes(size_of::<T>())),
+            |bytes, value| bytes.saturating_add(value.heap_bytes()),
+        )
+    }
+}
+
+impl<T> HeapSize for Option<T>
+where
+    T: HeapSize,
+{
+    fn heap_bytes(&self) -> u64 {
+        self.as_ref().map(HeapSize::heap_bytes).unwrap_or_default()
+    }
+}
+
+impl HeapSize for GifMediaFormat {
+    fn heap_bytes(&self) -> u64 {
+        self.src
+            .heap_bytes()
+            .saturating_add(self.proxy_src.heap_bytes())
+    }
+}
+
+impl HeapSize for GifItem {
+    fn heap_bytes(&self) -> u64 {
+        let string_bytes = [
+            &self.id,
+            &self.slug,
+            &self.provider,
+            &self.title,
+            &self.url,
+            &self.src,
+            &self.proxy_src,
+        ]
+        .into_iter()
+        .fold(0_u64, |bytes, value| {
+            bytes.saturating_add(value.heap_bytes())
+        });
+        let media_bytes = self.media.iter().fold(0_u64, |bytes, (key, value)| {
+            bytes
+                .saturating_add(usize_bytes(size_of::<String>()))
+                .saturating_add(usize_bytes(size_of::<GifMediaFormat>()))
+                .saturating_add(BTREE_ENTRY_OVERHEAD_BYTES)
+                .saturating_add(key.heap_bytes())
+                .saturating_add(value.heap_bytes())
+        });
+
+        string_bytes
+            .saturating_add(self.placeholder.heap_bytes())
+            .saturating_add(media_bytes)
+    }
+}
+
+impl HeapSize for GifCategoryTag {
+    fn heap_bytes(&self) -> u64 {
+        self.name
+            .heap_bytes()
+            .saturating_add(self.src.heap_bytes())
+            .saturating_add(self.proxy_src.heap_bytes())
+            .saturating_add(self.gif.heap_bytes())
+    }
+}
+
+impl<T> Cached<T>
+where
+    T: HeapSize,
+{
+    fn new(data: T, policy: CachePolicy, generation: u64) -> Self {
+        let retained_bytes = ARC_HEADER_BYTES
+            .saturating_add(usize_bytes(size_of::<T>()))
+            .saturating_add(data.heap_bytes());
+        Self {
+            data: Arc::new(data),
             stored_at: Instant::now(),
+            generation,
+            policy,
+            retained_bytes: u32::try_from(retained_bytes).unwrap_or(u32::MAX).max(1),
         }
     }
 
@@ -69,52 +361,69 @@ impl<T> Cached<T> {
 }
 
 impl GifsShard {
-    pub fn new(config: &ServiceConfig) -> anyhow::Result<Self> {
+    pub fn new() -> anyhow::Result<Self> {
         let media_proxy = MediaProxyUrlBuilder::from_env()?;
         let klipy = KlipyClient::new(media_proxy)?;
-        let max_capacity = config.cache_max_entries;
-        let max_cache_ttl = CATEGORIES_HARD_TTL;
-        Ok(Self {
-            inner: Arc::new(GifsShardInner {
-                klipy,
-                gif_lists: build_cache(max_capacity, max_cache_ttl),
-                categories: build_cache(max_capacity, max_cache_ttl),
-                suggestions: build_cache(max_capacity, max_cache_ttl),
-                resolved: build_cache(max_capacity, max_cache_ttl),
-                refreshing: Mutex::new(HashSet::new()),
-            }),
-        })
+        let budgets = CacheBudgets::from_env()?;
+        tracing::info!(
+            total_cache_max_bytes = budgets.total,
+            gif_list_cache_max_bytes = budgets.gif_lists,
+            category_cache_max_bytes = budgets.categories,
+            suggestion_cache_max_bytes = budgets.suggestions,
+            resolved_cache_max_bytes = budgets.resolved,
+            "configured GIF shard cache budgets"
+        );
+
+        let inner = Arc::new(GifsShardInner {
+            klipy,
+            gif_lists: build_cache("gif_lists", budgets.gif_lists),
+            categories: build_cache("categories", budgets.categories),
+            suggestions: build_cache("suggestions", budgets.suggestions),
+            resolved: build_cache("resolved", budgets.resolved),
+            refreshing: Mutex::new(HashSet::new()),
+            refresh_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_REFRESHES)),
+        });
+        spawn_cache_maintenance(&inner);
+        Ok(Self { inner })
     }
 
     async fn get_cached<T, Fetch, Fut>(
         &self,
-        cache: Cache<String, Cached<T>>,
+        cache: ManagedCache<T>,
         key: String,
         policy: CachePolicy,
         fetch: Fetch,
-    ) -> anyhow::Result<T>
+    ) -> anyhow::Result<Arc<T>>
     where
-        T: Clone + Send + Sync + 'static,
+        T: HeapSize + Send + Sync + 'static,
         Fetch: Fn() -> Fut + Clone + Send + Sync + 'static,
         Fut: Future<Output = anyhow::Result<T>> + Send + 'static,
     {
-        if let Some(cached) = cache.get(&key).await {
+        if let Some(cached) = cache.cache.get(&key).await {
+            assert_eq!(cached.policy, policy, "GIF cache policy mismatch for {key}");
             let age = cached.age();
-            if age <= policy.soft_ttl {
-                return Ok(cached.data);
-            }
             if age <= policy.hard_ttl {
-                self.trigger_background_refresh(cache, key, policy, fetch);
+                cache.telemetry.hits_total.fetch_add(1, METRICS_ORDERING);
+                if age > policy.soft_ttl {
+                    self.trigger_background_refresh(cache, key, policy, cached.generation, fetch);
+                }
                 return Ok(cached.data);
             }
-            cache.invalidate(&key).await;
+            cache.cache.invalidate(&key).await;
         }
 
+        cache.telemetry.misses_total.fetch_add(1, METRICS_ORDERING);
         let fetch_for_load = fetch.clone();
+        let telemetry_for_load = Arc::clone(&cache.telemetry);
         let cached = cache
+            .cache
             .try_get_with(key, async move {
                 let data = fetch_for_load().await?;
-                Ok::<Cached<T>, anyhow::Error>(Cached::new(data))
+                Ok::<Cached<T>, anyhow::Error>(Cached::new(
+                    data,
+                    policy,
+                    telemetry_for_load.next_generation(),
+                ))
             })
             .await
             .map_err(|error| anyhow::anyhow!("{}", error.as_ref()))?;
@@ -123,36 +432,101 @@ impl GifsShard {
 
     fn trigger_background_refresh<T, Fetch, Fut>(
         &self,
-        cache: Cache<String, Cached<T>>,
+        cache: ManagedCache<T>,
         key: String,
-        _policy: CachePolicy,
+        policy: CachePolicy,
+        observed_generation: u64,
         fetch: Fetch,
     ) where
-        T: Clone + Send + Sync + 'static,
+        T: HeapSize + Send + Sync + 'static,
         Fetch: Fn() -> Fut + Clone + Send + Sync + 'static,
         Fut: Future<Output = anyhow::Result<T>> + Send + 'static,
     {
-        let this = self.clone();
-        tokio::spawn(async move {
-            {
-                let mut refreshing = this.inner.refreshing.lock().await;
-                if !refreshing.insert(key.clone()) {
-                    return;
-                }
+        let permit = match self.inner.refresh_permits.clone().try_acquire_owned() {
+            Ok(permit) => permit,
+            Err(TryAcquireError::NoPermits) => {
+                cache
+                    .telemetry
+                    .refresh_saturated_total
+                    .fetch_add(1, METRICS_ORDERING);
+                return;
             }
+            Err(TryAcquireError::Closed) => panic!("GIF refresh semaphore unexpectedly closed"),
+        };
+        let refresh_key = RefreshKey {
+            cache_name: cache.telemetry.name,
+            key: key.clone(),
+        };
+        {
+            let mut refreshing = self
+                .inner
+                .refreshing
+                .lock()
+                .expect("GIF refresh registry mutex poisoned");
+            if !refreshing.insert(refresh_key.clone()) {
+                cache
+                    .telemetry
+                    .refresh_duplicate_total
+                    .fetch_add(1, METRICS_ORDERING);
+                return;
+            }
+        }
+        cache
+            .telemetry
+            .refresh_in_flight
+            .fetch_add(1, METRICS_ORDERING);
+        let registration = RefreshRegistration {
+            inner: Arc::clone(&self.inner),
+            telemetry: Arc::clone(&cache.telemetry),
+            key: refresh_key,
+        };
 
-            let result = fetch().await;
-            match result {
+        tokio::spawn(async move {
+            let _permit = permit;
+            let _registration = registration;
+            match fetch().await {
                 Ok(data) => {
-                    cache.insert(key.clone(), Cached::new(data)).await;
+                    let replacement = Cached::new(data, policy, cache.telemetry.next_generation());
+                    let result = cache
+                        .cache
+                        .entry(key.clone())
+                        .and_compute_with(move |current| {
+                            let operation = match current {
+                                Some(entry) if entry.value().generation == observed_generation => {
+                                    Op::Put(replacement)
+                                }
+                                Some(_) => Op::Nop,
+                                None => Op::Put(replacement),
+                            };
+                            std::future::ready(operation)
+                        })
+                        .await;
+                    match result {
+                        CompResult::Inserted(_) | CompResult::ReplacedWith(_) => {
+                            cache
+                                .telemetry
+                                .refresh_success_total
+                                .fetch_add(1, METRICS_ORDERING);
+                        }
+                        CompResult::Unchanged(_) => {
+                            cache
+                                .telemetry
+                                .refresh_superseded_total
+                                .fetch_add(1, METRICS_ORDERING);
+                        }
+                        CompResult::StillNone(_) | CompResult::Removed(_) => {
+                            panic!("GIF cache refresh produced an impossible compute result");
+                        }
+                    }
                 }
                 Err(error) => {
+                    cache
+                        .telemetry
+                        .refresh_error_total
+                        .fetch_add(1, METRICS_ORDERING);
                     tracing::debug!(error = %error, cache_key = %key, "background GIF cache refresh failed");
                 }
             }
-
-            let mut refreshing = this.inner.refreshing.lock().await;
-            refreshing.remove(&key);
         });
     }
 
@@ -334,6 +708,76 @@ impl GifsShard {
             .await?;
         Ok(GifServiceResponse::Resolved { gif })
     }
+
+    fn render_cache_metrics<T>(&self, output: &mut String, cache: &ManagedCache<T>) {
+        let telemetry = cache.telemetry.as_ref();
+        let name = telemetry.name;
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_budget_bytes{{cache=\"{name}\"}} {}",
+            cache.max_weight_bytes
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_entries{{cache=\"{name}\"}} {}",
+            cache.cache.entry_count()
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_weighted_size_bytes{{cache=\"{name}\"}} {}",
+            cache.cache.weighted_size()
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_hits_total{{cache=\"{name}\"}} {}",
+            telemetry.hits_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_misses_total{{cache=\"{name}\"}} {}",
+            telemetry.misses_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_evictions_total{{cache=\"{name}\",cause=\"size\"}} {}",
+            telemetry.size_evictions_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_evictions_total{{cache=\"{name}\",cause=\"expired\"}} {}",
+            telemetry.expired_evictions_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"success\"}} {}",
+            telemetry.refresh_success_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"error\"}} {}",
+            telemetry.refresh_error_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"superseded\"}} {}",
+            telemetry.refresh_superseded_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"duplicate\"}} {}",
+            telemetry.refresh_duplicate_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"saturated\"}} {}",
+            telemetry.refresh_saturated_total.load(METRICS_ORDERING)
+        );
+        let _ = writeln!(
+            output,
+            "fluxer_gifs_shard_cache_refreshes_in_flight{{cache=\"{name}\"}} {}",
+            telemetry.refresh_in_flight.load(METRICS_ORDERING)
+        );
+    }
 }
 
 impl ShardService for GifsShard {
@@ -344,6 +788,36 @@ impl ShardService for GifsShard {
         "gifs"
     }
 
+    fn render_prometheus_metrics(&self, output: &mut String) {
+        let _ = writeln!(output, "# TYPE fluxer_gifs_shard_cache_budget_bytes gauge");
+        let _ = writeln!(output, "# TYPE fluxer_gifs_shard_cache_entries gauge");
+        let _ = writeln!(
+            output,
+            "# TYPE fluxer_gifs_shard_cache_weighted_size_bytes gauge"
+        );
+        let _ = writeln!(output, "# TYPE fluxer_gifs_shard_cache_hits_total counter");
+        let _ = writeln!(
+            output,
+            "# TYPE fluxer_gifs_shard_cache_misses_total counter"
+        );
+        let _ = writeln!(
+            output,
+            "# TYPE fluxer_gifs_shard_cache_evictions_total counter"
+        );
+        let _ = writeln!(
+            output,
+            "# TYPE fluxer_gifs_shard_cache_refreshes_total counter"
+        );
+        let _ = writeln!(
+            output,
+            "# TYPE fluxer_gifs_shard_cache_refreshes_in_flight gauge"
+        );
+        self.render_cache_metrics(output, &self.inner.gif_lists);
+        self.render_cache_metrics(output, &self.inner.categories);
+        self.render_cache_metrics(output, &self.inner.suggestions);
+        self.render_cache_metrics(output, &self.inner.resolved);
+    }
+
     async fn handle(&self, request: GifRequest) -> anyhow::Result<GifServiceResponse> {
         let response = match request {
             GifRequest::IsAvailable { api_key } => self.handle_available(api_key).await,
@@ -399,12 +873,57 @@ impl ShardService for GifsShard {
     }
 }
 
-fn build_cache<T>(max_capacity: u64, time_to_live: Duration) -> Cache<String, Cached<T>>
+fn usize_bytes(value: usize) -> u64 {
+    u64::try_from(value).expect("usize must fit in u64")
+}
+
+fn cache_entry_weight<T>(key: &String, value: &Cached<T>) -> u32 {
+    let bytes = CACHE_ENTRY_OVERHEAD_BYTES
+        .saturating_add(usize_bytes(size_of::<Cached<T>>()))
+        .saturating_add(usize_bytes(size_of::<String>()))
+        .saturating_add(usize_bytes(key.capacity()))
+        .saturating_add(u64::from(value.retained_bytes));
+    u32::try_from(bytes).unwrap_or(u32::MAX).max(1)
+}
+
+fn build_cache<T>(name: &'static str, max_weight_bytes: u64) -> ManagedCache<T>
 where
-    T: Clone + Send + Sync + 'static,
+    T: HeapSize + Send + Sync + 'static,
 {
-    Cache::builder()
-        .max_capacity(max_capacity)
-        .time_to_live(time_to_live)
-        .build()
+    assert!(max_weight_bytes > 0);
+    let telemetry = Arc::new(CacheTelemetry::new(name));
+    let listener_telemetry = Arc::clone(&telemetry);
+    let cache = Cache::<String, Cached<T>>::builder()
+        .max_capacity(max_weight_bytes)
+        .weigher(cache_entry_weight::<T>)
+        .expire_after(CachedExpiry)
+        .eviction_listener(move |_key, _value, cause| {
+            listener_telemetry.record_removal(cause);
+        })
+        .build();
+    ManagedCache {
+        cache,
+        telemetry,
+        max_weight_bytes,
+    }
+}
+
+fn spawn_cache_maintenance(inner: &Arc<GifsShardInner>) {
+    let weak_inner = Arc::downgrade(inner);
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(CACHE_MAINTENANCE_INTERVAL);
+        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+        loop {
+            interval.tick().await;
+            let Some(inner) = weak_inner.upgrade() else {
+                return;
+            };
+            tokio::join!(
+                inner.gif_lists.cache.run_pending_tasks(),
+                inner.categories.cache.run_pending_tasks(),
+                inner.suggestions.cache.run_pending_tasks(),
+                inner.resolved.cache.run_pending_tasks(),
+            );
+        }
+    });
 }

fluxer_gifs/src/types.rs

@@ -2,6 +2,7 @@
 
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "op", rename_all_fields = "snake_case")]
@@ -56,16 +57,16 @@ pub enum GifServiceResponse {
     Available {
         available: bool,
     },
-    SearchResults(Vec<GifItem>),
+    SearchResults(Arc<Vec<GifItem>>),
     Featured {
-        gifs: Vec<GifItem>,
-        categories: Vec<GifCategoryTag>,
+        gifs: Arc<Vec<GifItem>>,
+        categories: Arc<Vec<GifCategoryTag>>,
     },
-    TrendingResults(Vec<GifItem>),
-    Suggestions(Vec<String>),
+    TrendingResults(Arc<Vec<GifItem>>),
+    Suggestions(Arc<Vec<String>>),
     Registered,
     Resolved {
-        gif: Option<GifItem>,
+        gif: Arc<Option<GifItem>>,
     },
     ShareUrl {
         url: String,

fluxer_svc/src/metrics.rs

@@ -2,9 +2,11 @@
 
 use libc::{CLOCK_MONOTONIC, clock_gettime, timespec};
 use std::fmt::Write;
+use std::sync::Arc;
 use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
 
 const ORDERING: Ordering = Ordering::Relaxed;
+pub(crate) type AdditionalMetricsRenderer = Arc<dyn Fn(&mut String) + Send + Sync>;
 
 const HISTOGRAM_BUCKETS_MS: &[u64] = &[
     1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000,
@@ -71,6 +73,7 @@ pub struct ServiceMetrics {
     pub shard_forwards_total: AtomicU64,
     pub request_duration: Histogram,
     start_ms: AtomicI64,
+    additional_renderer: Option<AdditionalMetricsRenderer>,
 }
 
 impl Default for ServiceMetrics {
@@ -83,11 +86,19 @@ impl Default for ServiceMetrics {
             shard_forwards_total: AtomicU64::new(0),
             request_duration: Histogram::new(),
             start_ms: AtomicI64::new(0),
+            additional_renderer: None,
         }
     }
 }
 
 impl ServiceMetrics {
+    pub(crate) fn with_additional_renderer(renderer: AdditionalMetricsRenderer) -> Self {
+        Self {
+            additional_renderer: Some(renderer),
+            ..Self::default()
+        }
+    }
+
     pub fn init(&self) {
         self.start_ms.store(now_ms(), ORDERING);
     }
@@ -167,6 +178,10 @@ impl ServiceMetrics {
             uptime_ms as f64 / 1000.0,
         );
 
+        if let Some(renderer) = self.additional_renderer.as_ref() {
+            renderer(&mut out);
+        }
+
         out
     }
 }

fluxer_svc/src/shard.rs

@@ -16,6 +16,7 @@ pub trait ShardService: Send + Sync + 'static {
     type Response: serde::Serialize + serde::de::DeserializeOwned + Send + 'static;
 
     fn service_name(&self) -> &str;
+    fn render_prometheus_metrics(&self, _output: &mut String) {}
     fn handle(
         &self,
         request: Self::Request,
@@ -36,7 +37,12 @@ where
     let shard_subject = format!("svc.{name}.shard.{shard_id}");
     let health_addr = config.listen_addr;
 
-    let metrics = Arc::new(ServiceMetrics::default());
+    let metrics_service = service.clone();
+    let additional_renderer: Arc<dyn Fn(&mut String) + Send + Sync> =
+        Arc::new(move |output| metrics_service.render_prometheus_metrics(output));
+    let metrics = Arc::new(ServiceMetrics::with_additional_renderer(
+        additional_renderer,
+    ));
     metrics.init();
 
     let is_serving = Arc::new(AtomicBool::new(false));