rust_mc_status/core/

cache.rs

//! Response cache with TTL, LRU eviction, and in-flight deduplication.
//!
//! ## Overview
//!
//! [`ResponseCache`] stores the full [`ServerStatus`] result for each
//! `(protocol, address)` cache key with a configurable TTL.  On a cache hit
//! the stored status is returned instantly with `latency = 0.0` and
//! `cached = true`.
//!
//! The cache is **disabled by default** (`TTL = Duration::ZERO`).  Enable it
//! via [`McClientBuilder::response_cache`](crate::McClientBuilder::response_cache).
//!
//! ## In-flight deduplication (thundering herd protection)
//!
//! When many tasks concurrently miss the cache for the same key, only **one**
//! real ping is performed.  The rest subscribe via a `tokio::sync::broadcast`
//! channel and receive the result as soon as it arrives.
//!
//! ```text
//! task-1  ──► cache miss ──► starts real ping ──────────────► insert + broadcast
//! task-2  ──► cache miss ──► sees in-flight   ──► subscribe ──────────────────────► receive
//! task-3  ──► cache miss ──► sees in-flight   ──► subscribe ──────────────────────► receive
//! ```
//!
//! The caller must call [`insert`](ResponseCache::insert) or
//! [`insert_error`](ResponseCache::insert_error) after the ping completes —
//! this broadcasts the result to all subscribers and removes the in-flight
//! entry.  If the sender is dropped without broadcasting, subscribers receive
//! `Err(RecvError::Closed)` and fall through to a direct ping.

use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use lru::LruCache;
use tokio::sync::{RwLock, broadcast};

use crate::error::McError;
use crate::models::ServerStatus;

/// Default LRU capacity when [`McClientBuilder::response_cache_ttl`](crate::McClientBuilder::response_cache_ttl)
/// is used without an explicit size.
pub(crate) const DEFAULT_RESPONSE_CACHE_SIZE: usize = 256;

/// Maximum number of concurrent subscribers per in-flight key.
///
/// If more than 64 tasks simultaneously miss for the same key, excess
/// subscribers will receive a `RecvError::Closed` and fall through to a
/// direct ping.  This is a very conservative limit — real workloads
/// rarely see more than a handful of concurrent misses.
const INFLIGHT_CHANNEL_CAPACITY: usize = 64;

// ─── Internal entry ───────────────────────────────────────────────────────────

#[derive(Clone)]
struct Entry {
    status:    ServerStatus,
    /// Wall-clock time when this entry was stored.  Used to compute TTL expiry.
    stored_at: SystemTime,
}

// ─── ResponseCache ────────────────────────────────────────────────────────────

/// Response cache: LRU + TTL + in-flight deduplication.
///
/// Cheaply `Clone`-able — all clones share the same underlying `Arc` state.
///
/// A TTL of [`Duration::ZERO`] disables the cache entirely — [`get`](Self::get)
/// always returns `None` and [`insert`](Self::insert) is a no-op.
#[derive(Clone)]
pub struct ResponseCache {
    /// Completed results indexed by cache key.
    entries: Arc<RwLock<LruCache<String, Entry>>>,
    /// In-flight keys mapped to their broadcast sender.
    ///
    /// A key is present here while a real ping is in progress.
    /// Removed when the ping completes (via [`insert`](Self::insert) or
    /// [`insert_error`](Self::insert_error)).
    inflight: Arc<RwLock<std::collections::HashMap<
        String,
        broadcast::Sender<Result<ServerStatus, String>>,
    >>>,
    /// Cache TTL.  `Duration::ZERO` = disabled.
    ttl: Duration,
}

impl ResponseCache {
    /// Create a disabled cache (`TTL = Duration::ZERO`).
    ///
    /// All operations are no-ops.  No memory is allocated for entries.
    pub fn disabled() -> Self {
        Self::new(Duration::ZERO, 1)
    }

    /// Create an enabled cache with the given `ttl` and LRU `capacity`.
    ///
    /// Entries older than `ttl` are treated as expired on next access.
    /// When the cache is full the least-recently-used entry is evicted.
    pub fn new(ttl: Duration, capacity: usize) -> Self {
        let cap = NonZeroUsize::new(capacity.max(1))
            .expect("response cache capacity must be > 0");
        Self {
            entries:  Arc::new(RwLock::new(LruCache::new(cap))),
            inflight: Arc::new(RwLock::new(std::collections::HashMap::new())),
            ttl,
        }
    }

    /// `true` when the cache is active (TTL > 0).
    pub fn is_enabled(&self) -> bool { !self.ttl.is_zero() }

    /// Current TTL setting.
    pub fn ttl(&self) -> Duration { self.ttl }

    /// Try to return a non-expired cached [`ServerStatus`] for `key`.
    ///
    /// Returns `None` when:
    /// - the cache is disabled,
    /// - no entry exists for `key`, or
    /// - the entry has expired (age ≥ TTL).
    ///
    /// On a hit, the returned status has `cached = true` and `latency = 0.0`.
    pub async fn get(&self, key: &str) -> Option<ServerStatus> {
        if !self.is_enabled() { return None; }
        let mut cache = self.entries.write().await;
        let entry     = cache.get(key)?;
        if entry.stored_at.elapsed().unwrap_or(Duration::MAX) >= self.ttl {
            return None;
        }
        let mut status = entry.status.clone();
        status.cached  = true;
        status.latency = 0.0;
        Some(status)
    }

    /// Store a successful [`ServerStatus`] under `key` and broadcast it to
    /// any tasks waiting on the in-flight channel for this key.
    ///
    /// If the cache is disabled this is a no-op (but the broadcast still fires
    /// if an in-flight entry exists, releasing any waiting subscribers).
    pub async fn insert(&self, key: String, status: ServerStatus) {
        if !self.is_enabled() { return; }

        // Store in LRU first so subscribers can immediately read from cache
        {
            let entry = Entry { status: status.clone(), stored_at: SystemTime::now() };
            self.entries.write().await.put(key.clone(), entry);
        }

        // Notify in-flight subscribers and free the slot
        let sender = self.inflight.write().await.remove(&key);
        if let Some(tx) = sender {
            let _ = tx.send(Ok(status)); // ignore Err (no subscribers) — that's fine
        }
    }

    /// Notify in-flight subscribers of a failure for `key`, then free the slot.
    ///
    /// The error message is broadcast as a `String` so subscribers can surface
    /// a meaningful error without cloning the original `McError`.
    pub async fn insert_error(&self, key: &str, err: &McError) {
        let sender = self.inflight.write().await.remove(key);
        if let Some(tx) = sender {
            let _ = tx.send(Err(err.to_string()));
        }
    }

    /// Attempt to join an existing in-flight request for `key`, or register
    /// this task as the designated pinger.
    ///
    /// Returns:
    /// - `Some(receiver)` — another task is already pinging; wait on this
    ///   channel for the result.
    /// - `None` — this task is first; it **must** call [`insert`](Self::insert)
    ///   or [`insert_error`](Self::insert_error) when the ping completes.
    ///
    /// Always returns `None` when the cache is disabled.
    pub async fn join_or_register(
        &self,
        key: &str,
    ) -> Option<broadcast::Receiver<Result<ServerStatus, String>>> {
        if !self.is_enabled() { return None; }
        let mut inflight = self.inflight.write().await;
        if let Some(tx) = inflight.get(key) {
            // Subscribe to the existing in-flight ping
            return Some(tx.subscribe());
        }
        // Register as the pinger
        let (tx, _) = broadcast::channel(INFLIGHT_CHANNEL_CAPACITY);
        inflight.insert(key.to_string(), tx);
        None
    }

    /// Clear all cached entries and all in-flight registrations.
    pub async fn clear(&self) {
        self.entries.write().await.clear();
        self.inflight.write().await.clear();
    }

    /// Number of entries currently in the LRU (includes expired entries that
    /// have not yet been accessed and evicted).
    pub async fn len(&self) -> usize {
        self.entries.read().await.len()
    }
}