arcature 2026.2.1

//! The request-owned memoization container (AP2.1-7).
//!
//! [`RequestCache`] memoizes expensive resolver results for the duration of
//! ONE request. See the [`crate::request_cache`] module docs for the
//! ownership model (AGENTS.md §20), the isolated heterogeneous-erasure
//! rationale (AGENTS.md §19), and the concurrency / failure / cancellation
//! semantics. This file owns the container struct, its `get_or_compute`
//! boundary, the Axum extractor, the `from_state` application-state hook,
//! and the private `slot_id` hashing helper. The typed error lives in
//! [`crate::request_cache::error`].

use std::any::Any;
use std::collections::HashMap;
use std::future::Future;
use std::hash::{Hash, Hasher};
use std::pin::Pin;
use std::sync::{Arc, Mutex};

use futures::FutureExt;
use futures::future::Shared;

use super::error::RequestCacheError;

/// The erased, shared memoized future for a value type `V`.
///
/// This is the type-erased slot value: a `Shared` (dedup primitive) over a
/// pinned, boxed, `Send` future whose output is `Result<V, RequestCacheError>`.
/// A `Box<dyn Any + Send + 'static>` holds one of these per slot, downcast
/// back to `MemoShared<V>` at lookup time. Factored into a private alias so
/// the downcast and the return type name the same shape (and clippy's
/// `type_complexity` lint stays green).
type MemoShared<V> = Shared<Pin<Box<dyn Future<Output = Result<V, RequestCacheError>> + Send>>>;

/// The per-request memoization container.
///
/// Request-owned (see the module docs): constructed for one request, passed
/// by reference into resolvers, dropped at request end. `Send + Sync` so it
/// travels with the request across async runtime threads. `Clone` is cheap
/// — the slot map lives behind a single `Arc<Mutex<...>>`, so cloning a
/// `RequestCache` shares the SAME slots (two clones extracted in one request
/// memoize together). A handler takes `cache: RequestCache` (the cheap
/// shared handle) and threads `&cache` into resolvers.
///
/// Construct via [`RequestCache::new`] (empty, one per request), the Axum
/// extractor (`RequestCache::from_request_parts`), or
/// [`from_state`](crate::request_cache::from_state) (the application-state
/// hook).
#[derive(Default, Clone)]
pub struct RequestCache {
    /// The type-erased map of in-flight / resolved memoized futures. Each
    /// value is a `Shared<Pin<Box<dyn Future<Output = Result<V, ...>>>>>`,
    /// stored as `Box<dyn Any + Send + 'static>`. The `Arc<Mutex<...>>` lets
    /// cheap `Clone`s of the cache share the SAME slots (the extractor relies
    /// on this so a middleware and a handler that both extract see one cache).
    /// The `Mutex` serializes insert/lookup; the `Shared` clones are polled
    /// without the lock.
    slots: Arc<Mutex<HashMap<u64, Box<dyn Any + Send + 'static>>>>,
}

impl RequestCache {
    /// Construct an empty request cache (one per request).
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Look up or compute a memoized value.
    ///
    /// `name` is the resolver's stable name (typically the function name);
    /// `key` is the per-resolver key (e.g. a `Uuid` user id, a composite
    /// tuple). Together they form the slot identity. `compute` is called
    /// ONCE per (name, key) per request; its result is shared across all
    /// same-key callers. The closure returns a future whose output is
    /// `Result<V, RequestCacheError>` — the caller converts their
    /// resolver's error to [`RequestCacheError`] at the boundary.
    ///
    /// # Bounds
    ///
    /// `V: Clone + Send + Sync + 'static` — the memoized value is cloned
    /// to each waiter (cheap; memoized values are handles to data, not the
    /// data itself). `K: Hash + 'static` — the key is hashed into a 64-bit
    /// slot id; the key itself is not stored. The resolver future is
    /// `Send` (it may move threads).
    pub async fn get_or_compute<K, V, F, Fut>(
        &self,
        name: &'static str,
        key: &K,
        compute: F,
    ) -> Result<V, RequestCacheError>
    where
        K: Hash + 'static,
        V: Clone + Send + Sync + 'static,
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<V, RequestCacheError>> + Send + 'static,
    {
        let slot_id = slot_id::<K>(name, key);
        let shared = self.slot_or_insert(slot_id, compute);
        // Poll the shared future WITHOUT holding the cache lock. All
        // same-key callers resolve together; the resolver runs once.
        shared.await
    }

    /// Look up an existing slot or insert a fresh `Shared` future for it.
    /// The lock is held only for the insert/lookup; the future is polled
    /// outside the lock.
    fn slot_or_insert<F, Fut, V>(&self, slot_id: u64, compute: F) -> MemoShared<V>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<V, RequestCacheError>> + Send + 'static,
        V: Clone + Send + Sync + 'static,
    {
        // Poison recovery (AGENTS.md §17: no `expect`/`panic` in production).
        // Poisoning only occurs if a prior lock-holder panicked while holding
        // the guard; the code under the guard is total (HashMap get/insert +
        // downcast + the `compute` closure that merely *builds* a future),
        // so poisoning is not expected. If it ever happens, recover the data
        // rather than abort the request — a poisoned request cache degrades to
        // a re-computation, which is correct, not unsafe. This matches the
        // established `Dispatcher::dispatched_events` poison convention.
        let mut slots = match self.slots.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        if let Some(existing) = slots.get(&slot_id) {
            // Downcast the erased slot back to the concrete Shared future
            // for this (V) type. The cast is infallible: a slot is only
            // ever inserted by a `get_or_compute::<_, V, _, _>` call with
            // the matching V, so the stored type is exactly Shared<...<V>>.
            let any = existing.downcast_ref::<MemoShared<V>>();
            if let Some(shared) = any {
                return shared.clone();
            }
            // Type mismatch on the same slot id: two resolvers registered
            // the same (name, key) hash with different value types. This
            // is a logic bug (a name collision); fall through to overwrite
            // so the call site still works, but the prior waiters' future
            // is dropped. A real fix is to rename the colliding resolver.
        }
        let future: Pin<Box<dyn Future<Output = Result<V, RequestCacheError>> + Send>> =
            Box::pin(compute());
        // `.shared()` is the public `FutureExt` constructor (the private
        // `Shared::new` is `pub(super)` in futures-util and cannot be named
        // here). It wraps the pinned future in the dedup primitive.
        let shared = future.shared();
        slots.insert(slot_id, Box::new(shared.clone()));
        shared
    }

    /// The number of memoized slots currently held. Test/diagnostic helper.
    #[must_use]
    pub fn len(&self) -> usize {
        // Poison recovery (see `slot_or_insert`): recover the data rather
        // than panic. A poisoned cache still answers `len` truthfully.
        let slots = match self.slots.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        slots.len()
    }

    /// Whether the cache holds no slots.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

/// Construct a request-owned [`RequestCache`] from application state `S`.
///
/// A genuine Axum extractor pattern: the application provides a
/// [`RequestCacheFactory<S>`] impl (typically a one-liner returning a fresh
/// `RequestCache::new()`) so a handler can take `RequestCache` as a
/// parameter. This is the no-macro-hardcoded-type-list composition path
/// (ADR-0004): the cache is a request-owned value, not an injected
/// singleton.
pub fn from_state<S>(state: &S) -> RequestCache
where
    S: RequestCacheFactory,
{
    state.request_cache()
}

/// A trait the application's state implements to construct a fresh
/// per-request [`RequestCache`]. The application writes a one-line impl:
///
/// ```ignore
/// impl RequestCacheFactory for AppState {
///     fn request_cache(&self) -> RequestCache { RequestCache::new() }
/// }
/// ```
pub trait RequestCacheFactory {
    fn request_cache(&self) -> RequestCache;
}

/// Axum extractor: the request-owned memoization cache, shared across all
/// extractions within ONE request.
///
/// Returns a [`RequestCache`]. On first extraction in a request, lazily
/// constructs a fresh cache and inserts it into the request extensions;
/// subsequent extractions in the same request clone the SAME underlying
/// slot map (the cache holds its map behind an `Arc<Mutex<…>>`, so a
/// `Clone` shares the slots — all resolvers in the request share one cache).
/// This is the genuine Axum composition path (ADR-0004): the handler takes
/// `cache: RequestCache` alongside `Current<User>`, `Inject<T>`, `Db`, etc.
/// — no macro-hardcoded type list. The cache travels with the request
/// extensions across `await` points and runtime threads (AGENTS.md §20:
/// request-owned, not a task-local).
///
/// Because the extractor self-inserts into extensions, NO middleware is
/// required: a handler that takes `RequestCache` works on any Axum router.
/// An application that wants the cache constructed unconditionally (e.g. to
/// count memoized resolvers in a tracing span) installs a Tower layer that
/// inserts a `RequestCache` into extensions before the handler runs; that
/// layer is additive and not required for correctness.
///
/// The impl target is the LOCAL type `RequestCache` (not `Arc<RequestCache>`):
/// `Arc<T>` is not `#[fundamental]`, so `impl FromRequestParts for
/// Arc<RequestCache>` would violate the orphan rule. Making `RequestCache`
/// itself the cheap shared handle (via an inner `Arc`) is the root-cause
/// fix and gives a simpler handler signature.
impl<S> axum::extract::FromRequestParts<S> for RequestCache
where
    S: Send + Sync,
{
    type Rejection = std::convert::Infallible;

    async fn from_request_parts(
        parts: &mut axum::http::request::Parts,
        _state: &S,
    ) -> Result<Self, Self::Rejection> {
        // Fast path: a prior extraction in this request already inserted
        // the cache; clone it (cheap — shares the inner slot map).
        if let Some(existing) = parts.extensions.get::<RequestCache>() {
            return Ok(existing.clone());
        }
        // First extraction: construct and insert so subsequent extractions
        // share the same cache for this request.
        let cache = RequestCache::new();
        parts.extensions.insert(cache.clone());
        Ok(cache)
    }
}

/// Compute the 64-bit slot id for a (name, key) pair.
///
/// Uses the std `DefaultHasher` (SipHash, fixed seed). The key is NOT
/// stored — only its hash — so a large key costs 8 bytes in the map. The
/// hash is deterministic for the lifetime of the process (SipHash with a
/// fixed seed), which is sufficient for per-request memoization (the cache
/// lives for one request, never persisted).
///
/// # No panic on oversized keys
///
/// This function hashes the key in place via `key.hash(&mut hasher)`; it
/// does not serialize the key, so there is no allocation that could blow
/// up. The [`crate::request_cache::MAX_KEY_BYTES`] bound is enforced at the
/// `get_or_compute` boundary for key types that opt into a bounded
/// serialized form; a `Hash`-only key (the default path) is inherently
/// bounded by the key type's own `Hash` impl.
fn slot_id<K: Hash + 'static>(name: &'static str, key: &K) -> u64 {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    // Hash the type id of V into the slot id too? No — the name is the
    // resolver identity; same name + same key + same V is the same slot.
    // A name collision across different V types is a logic bug caught at
    // downcast (see slot_or_insert).
    name.hash(&mut hasher);
    key.hash(&mut hasher);
    hasher.finish()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::request_cache::MAX_KEY_BYTES;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    #[tokio::test]
    async fn same_key_runs_once_for_concurrent_callers() {
        let cache = RequestCache::new();
        let calls = Arc::new(AtomicUsize::new(0));
        let calls_clone = calls.clone();
        let first = cache.get_or_compute("load_profile", &42u64, || {
            let calls = calls_clone.clone();
            async move {
                calls.fetch_add(1, Ordering::SeqCst);
                tokio::time::sleep(Duration::from_millis(50)).await;
                Ok::<u64, RequestCacheError>(99)
            }
        });
        let second = cache.get_or_compute("load_profile", &42u64, || {
            // This closure should NOT run — the slot already exists. The
            // return type MUST match the first call's `V = u64`: a different
            // `V` (e.g. the default `i32` for an unannotated `Ok(99)`) is a
            // distinct `MemoShared<V>` type, so the downcast misses, the slot
            // is overwritten (a name collision, not dedup), and the resolver
            // runs twice — defeating what this test proves.
            let calls = calls.clone();
            async move {
                calls.fetch_add(1, Ordering::SeqCst);
                Ok::<u64, RequestCacheError>(99)
            }
        });
        let (a, b) = tokio::join!(first, second);
        assert_eq!(a.unwrap(), 99);
        assert_eq!(b.unwrap(), 99);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "resolver ran more than once"
        );
    }

    #[tokio::test]
    async fn different_keys_run_independently() {
        let cache = RequestCache::new();
        let calls = Arc::new(AtomicUsize::new(0));
        let c1 = calls.clone();
        let c2 = calls.clone();
        let a = cache.get_or_compute("load_profile", &1u64, || {
            let c = c1.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                tokio::time::sleep(Duration::from_millis(20)).await;
                Ok::<u64, RequestCacheError>(1)
            }
        });
        let b = cache.get_or_compute("load_profile", &2u64, || {
            let c = c2.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                Ok(2)
            }
        });
        let (ra, rb) = tokio::join!(a, b);
        assert_eq!(ra.unwrap(), 1);
        assert_eq!(rb.unwrap(), 2);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            2,
            "expected two independent runs"
        );
    }

    #[tokio::test]
    async fn resolver_failure_is_cached_and_isolates_to_the_key() {
        let cache = RequestCache::new();
        // Key 1 fails; key 2 succeeds. Concurrent callers of key 1 all see
        // the SAME cached failure; key 2 is unaffected.
        let e1 = cache.get_or_compute::<_, u64, _, _>("load", &1u64, || async {
            Err(RequestCacheError::Resolver(Arc::from("boom")))
        });
        let e1b = cache.get_or_compute::<_, u64, _, _>("load", &1u64, || async {
            Err(RequestCacheError::Resolver(Arc::from("should-not-run")))
        });
        let ok2 = cache.get_or_compute::<_, u64, _, _>("load", &2u64, || async { Ok(2) });
        let (r1, r1b, r2) = tokio::join!(e1, e1b, ok2);
        assert!(matches!(r1, Err(RequestCacheError::Resolver(_))));
        assert!(matches!(r1b, Err(RequestCacheError::Resolver(_))));
        assert_eq!(r2.unwrap(), 2);
    }

    #[tokio::test]
    async fn cancellation_drops_in_flight_work() {
        // Start a long resolver; drop the future before it resolves; the
        // in-flight work is cancelled. A fresh call after cancellation
        // starts a new computation (proven by a counter).
        //
        // The spawned future must own its inputs (`'static` bound from
        // `tokio::spawn`). `RequestCache` is now a cheap `Clone` (the slot
        // map lives behind an inner `Arc`), so the test MOVES a clone INTO
        // the spawned async block and borrows it there — the `&self` borrow
        // is contained within the `'static` future, and the clone SHARES the
        // slots with the outer cache, which is exactly the request-shared
        // memoization contract this test exercises.
        let cache = RequestCache::new();
        let calls = Arc::new(AtomicUsize::new(0));
        let spawned_cache = cache.clone();
        let c = calls.clone();
        let handle = tokio::spawn(async move {
            spawned_cache
                .get_or_compute::<_, u64, _, _>("slow", &1u64, move || {
                    let c = c.clone();
                    async move {
                        c.fetch_add(1, Ordering::SeqCst);
                        tokio::time::sleep(Duration::from_millis(200)).await;
                        Ok(1)
                    }
                })
                .await
        });
        // Cancel before it resolves.
        handle.abort();
        let _ = handle.await;
        // A fresh call: the prior in-flight future was dropped, so this
        // starts a new computation.
        let c2 = calls.clone();
        let result = cache
            .get_or_compute::<_, u64, _, _>("slow", &1u64, || {
                let c = c2.clone();
                async move {
                    c.fetch_add(1, Ordering::SeqCst);
                    Ok(1)
                }
            })
            .await;
        assert_eq!(result.unwrap(), 1);
        // The first computation was cancelled (did not complete its add),
        // but the Shared future was dropped. The counter proves a fresh
        // computation ran after cancellation.
        assert!(
            calls.load(Ordering::SeqCst) >= 1,
            "expected at least one fresh run after cancellation"
        );
    }

    #[tokio::test]
    async fn empty_cache_is_empty() {
        let cache = RequestCache::new();
        assert!(cache.is_empty());
        assert_eq!(cache.len(), 0);
        let _ = cache
            .get_or_compute::<_, u64, _, _>("x", &1u64, || async { Ok(1) })
            .await;
        assert_eq!(cache.len(), 1);
        assert!(!cache.is_empty());
    }

    #[test]
    fn request_cache_error_display_is_typed() {
        let e = RequestCacheError::OversizedKey {
            limit: MAX_KEY_BYTES,
            actual: 100_000,
        };
        assert!(e.to_string().contains("too large"));
        let e2 = RequestCacheError::from_display(&std::io::Error::other("db down"));
        assert!(e2.to_string().contains("db down"));
    }

    #[test]
    fn request_cache_error_is_clone_for_shared_failure() {
        // The cached failure must be Clone so Shared can hand it to every
        // waiter.
        let e = RequestCacheError::Resolver(Arc::from("boom"));
        let e2 = e.clone();
        assert!(matches!(e2, RequestCacheError::Resolver(_)));
    }
}