holger-server-lib 0.6.9

//! Proxy/group backend: chains a primary [`RepositoryBackendTrait`] with an
//! ordered upstream fallback list — the Nexus-style "proxy repo" over pull-through
//! (e.g. a writable-znippy `/cache` fronting crates.io). Itself a
//! `RepositoryBackendTrait`, so it slots transparently into FastRoutes wiring
//! wherever a single backend is expected.
//!
//! Flow: `fetch` tries primary then each upstream in declaration order, returning
//! the first hit; an upstream hit is written THROUGH into the primary when it is
//! writable, so the next fetch is a local hit. `handle_http2_request` mirrors this
//! but keys off status — primary wins unless it 404s, then the first non-404
//! upstream response is served (else the primary's 404). Everything else
//! (`put`, `list`, `archive_*`, `name`/`format`) delegates to the primary alone;
//! upstreams are read-only proxies.
//!
//! Gotcha: the write-through cache fill is strictly best-effort — a `put` failure
//! is logged and swallowed, never surfaced, so a cache-fill error can never fail a
//! read the caller already has bytes for.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use traits::{ArtifactEntry, ArtifactFormat, ArtifactId, ArchiveInfo, RepositoryBackendTrait};

/// Upper bound on live negative-cache entries per proxy repo. A miss-memoizing
/// cache must never grow without limit (a scanner probing many absent
/// coordinates would otherwise balloon it), so inserts prune expired entries
/// first and then refuse once the map is full — the cache degrades to
/// re-querying rather than leaking memory.
const NEG_CAP: usize = 4096;

/// Wraps a primary repository backend with an ordered upstream fallback chain —
/// the Nexus-style "proxy / group" repo (chain repos in series).
///
/// * `fetch` — tries primary, then each upstream in declaration order; returns the
///   first hit found. **Caching**: when the primary is writable, an upstream hit is
///   written through into the primary (best-effort) so the next fetch is a local
///   hit — a pull-through cache (e.g. a writable-znippy `/cache` over crates.io).
/// * `handle_http2_request` — returns the primary's response unless the status is
///   404, in which case it walks the upstream chain and returns the first non-404
///   response (or the primary's 404 if all upstreams also 404).
/// * `put` — always delegates to the primary; upstreams are read-only proxies.
pub struct ProxyBackend {
    primary: Arc<dyn RepositoryBackendTrait>,
    upstreams: Vec<Arc<dyn RepositoryBackendTrait>>,
    /// Negative-cache TTL. `Duration::ZERO` (the default) ⇒ the negative cache is
    /// **off** and every miss re-walks the upstream chain (behaviour identical to
    /// before this field existed).
    neg_ttl: Duration,
    /// Remembered misses: `key → expiry Instant`. A coordinate/HTTP path that
    /// missed the primary AND every upstream is recorded here so a repeat within
    /// the TTL short-circuits the (slow / network) upstream walk. Guarded by a
    /// `Mutex` so the `&self` read path can update it.
    neg_cache: Mutex<HashMap<String, Instant>>,
}

impl ProxyBackend {
    /// A proxy with the negative cache **disabled** (`neg_ttl == 0`) — the
    /// original constructor, byte-for-byte the prior pull-through behaviour.
    pub fn new(
        primary: Arc<dyn RepositoryBackendTrait>,
        upstreams: Vec<Arc<dyn RepositoryBackendTrait>>,
    ) -> Self {
        Self {
            primary,
            upstreams,
            neg_ttl: Duration::ZERO,
            neg_cache: Mutex::new(HashMap::new()),
        }
    }

    /// Enable negative caching with a `secs`-second TTL (a `secs` of `0` leaves it
    /// disabled). Additive builder — existing callers that use [`new`](Self::new)
    /// are unaffected.
    pub fn with_negative_cache_secs(mut self, secs: u64) -> Self {
        self.neg_ttl = Duration::from_secs(secs);
        self
    }

    /// Whether the negative cache is active.
    #[inline]
    fn neg_enabled(&self) -> bool {
        !self.neg_ttl.is_zero()
    }

    /// Cache key for a coordinate miss (`fetch`). NUL-separated so distinct
    /// coordinates can never alias; `F` prefix keeps it disjoint from HTTP keys.
    fn fetch_key(id: &ArtifactId) -> String {
        format!("F\u{0}{}\u{0}{}\u{0}{}", id.namespace.as_deref().unwrap_or(""), id.name, id.version)
    }

    /// Cache key for an HTTP miss (`method` + `suburl`). `H` prefix keeps it
    /// disjoint from coordinate keys.
    fn http_key(method: &str, suburl: &str) -> String {
        format!("H\u{0}{}\u{0}{}", method, suburl)
    }

    /// `true` iff `key` has a live (non-expired) negative entry. An expired entry
    /// is evicted in passing (lazy expiry) and reported as a miss.
    fn neg_hit(&self, key: &str) -> bool {
        if !self.neg_enabled() {
            return false;
        }
        let now = Instant::now();
        let mut cache = match self.neg_cache.lock() {
            Ok(c) => c,
            Err(_) => return false, // poisoned lock ⇒ behave as a cache miss
        };
        match cache.get(key) {
            Some(&expiry) if expiry > now => true,
            Some(_) => {
                cache.remove(key); // expired: evict, treat as miss
                false
            }
            None => false,
        }
    }

    /// Remember `key` as a miss for `neg_ttl`. Prunes expired entries first and
    /// refuses once [`NEG_CAP`] live entries are held (bounded memory). No-op when
    /// the cache is disabled.
    fn neg_remember(&self, key: &str) {
        if !self.neg_enabled() {
            return;
        }
        let now = Instant::now();
        let expiry = match now.checked_add(self.neg_ttl) {
            Some(e) => e,
            None => return,
        };
        if let Ok(mut cache) = self.neg_cache.lock() {
            if cache.len() >= NEG_CAP {
                cache.retain(|_, &mut exp| exp > now); // drop expired to make room
            }
            if cache.len() < NEG_CAP {
                cache.insert(key.to_string(), expiry);
            }
        }
    }
}

impl RepositoryBackendTrait for ProxyBackend {
    fn name(&self) -> &str {
        self.primary.name()
    }

    fn format(&self) -> ArtifactFormat {
        self.primary.format()
    }

    fn is_writable(&self) -> bool {
        self.primary.is_writable()
    }

    fn has_archive(&self) -> bool {
        self.primary.has_archive()
    }

    fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
        if let Some(data) = self.primary.fetch(id)? {
            crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::PrimaryHit);
            return Ok(Some(data));
        }
        // Negative cache: a coordinate known-absent within the TTL short-circuits
        // the upstream walk (the primary is still checked first above, so a later
        // local publish is never shadowed). Consulted only after the primary miss.
        let key = self.neg_enabled().then(|| Self::fetch_key(id));
        if let Some(k) = &key {
            if self.neg_hit(k) {
                crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::NegcacheHit);
                crate::grpc::functional_status(
                    "holger-proxy/negcache",
                    "fetch_neg_hit_short_circuit",
                    true,
                    &id.name,
                );
                return Ok(None);
            }
        }
        for upstream in &self.upstreams {
            if let Some(data) = upstream.fetch(id)? {
                log::debug!(
                    "proxy: cache miss on {}, served from upstream {}",
                    self.primary.name(),
                    upstream.name()
                );
                // Write-through: populate the writable primary so the next fetch
                // is a local hit. Best-effort — a cache-fill failure must never
                // fail the read the caller already has bytes for.
                if self.primary.is_writable() {
                    if let Err(e) = self.primary.put(id, &data) {
                        log::warn!(
                            "proxy: cache-fill of {}/{} into {} failed: {e}",
                            id.name,
                            id.version,
                            self.primary.name()
                        );
                    }
                }
                crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::UpstreamHit);
                return Ok(Some(data));
            }
        }
        // Missed primary AND every upstream — remember it for the TTL.
        if let Some(k) = &key {
            self.neg_remember(k);
        }
        crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::Miss);
        Ok(None)
    }

    fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
        self.primary.put(id, data)
    }

    fn handle_http2_request(
        &self,
        method: &str,
        suburl: &str,
        body: &[u8],
    ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
        let resp = self.primary.handle_http2_request(method, suburl, body)?;
        if resp.0 != 404 {
            crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::PrimaryHit);
            return Ok(resp);
        }
        // Negative cache: a path known-404 within the TTL returns the primary's
        // 404 without re-walking the upstreams. Only read requests are memoized —
        // a write's 404 is not a stable "absent" fact.
        let cacheable = matches!(method, "GET" | "HEAD");
        let key = (self.neg_enabled() && cacheable).then(|| Self::http_key(method, suburl));
        if let Some(k) = &key {
            if self.neg_hit(k) {
                crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::NegcacheHit);
                crate::grpc::functional_status(
                    "holger-proxy/negcache",
                    "http_neg_hit_short_circuit",
                    true,
                    suburl,
                );
                return Ok(resp);
            }
        }
        for upstream in &self.upstreams {
            let up_resp = upstream.handle_http2_request(method, suburl, body)?;
            if up_resp.0 != 404 {
                log::debug!(
                    "proxy: HTTP {} {} — primary 404, served from upstream {}",
                    method,
                    suburl,
                    upstream.name()
                );
                crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::UpstreamHit);
                return Ok(up_resp);
            }
        }
        // Primary 404 + every upstream 404 — remember the miss for the TTL.
        if let Some(k) = &key {
            self.neg_remember(k);
        }
        crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::Miss);
        Ok(resp)
    }

    fn list(
        &self,
        name_filter: Option<&str>,
        limit: usize,
    ) -> anyhow::Result<Vec<ArtifactEntry>> {
        self.primary.list(name_filter, limit)
    }

    fn archive_files(&self, prefix: Option<&str>) -> anyhow::Result<Vec<String>> {
        self.primary.archive_files(prefix)
    }

    fn archive_info(&self) -> anyhow::Result<ArchiveInfo> {
        self.primary.archive_info()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use traits::ArtifactFormat;

    /// Emit one functional-status row for a real check. Gated behind
    /// `--features testmatrix` so release builds strip it (dep is optional).
    #[cfg(feature = "testmatrix")]
    fn fstatus(component: &str, check: &str, ok: bool, detail: &str) {
        nornir_testmatrix::functional_status(component, check, ok, detail);
    }

    struct StubBackend {
        name: String,
        data: std::collections::HashMap<String, Vec<u8>>,
        http_status: u16,
        http_body: Vec<u8>,
        writable: bool,
    }

    impl StubBackend {
        fn always_200(name: &str, body: &[u8]) -> Arc<Self> {
            Arc::new(Self {
                name: name.into(),
                data: Default::default(),
                http_status: 200,
                http_body: body.to_vec(),
                writable: false,
            })
        }
        fn always_404(name: &str) -> Arc<Self> {
            Arc::new(Self {
                name: name.into(),
                data: Default::default(),
                http_status: 404,
                http_body: b"Not found".to_vec(),
                writable: false,
            })
        }
        fn with_artifact(name: &str, crate_name: &str, version: &str, body: &[u8]) -> Arc<Self> {
            let mut data = std::collections::HashMap::new();
            data.insert(format!("{}/{}", crate_name, version), body.to_vec());
            Arc::new(Self {
                name: name.into(),
                data,
                http_status: 404,
                http_body: b"Not found".to_vec(),
                writable: false,
            })
        }
    }

    impl RepositoryBackendTrait for StubBackend {
        fn name(&self) -> &str { &self.name }
        fn format(&self) -> ArtifactFormat { ArtifactFormat::Rust }
        fn is_writable(&self) -> bool { self.writable }

        fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            let key = format!("{}/{}", id.name, id.version);
            Ok(self.data.get(&key).cloned())
        }

        fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> {
            anyhow::bail!("read-only stub")
        }

        fn handle_http2_request(
            &self,
            _method: &str,
            _suburl: &str,
            _body: &[u8],
        ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            Ok((self.http_status, Vec::new(), self.http_body.clone()))
        }
    }

    fn artifact(name: &str, version: &str) -> ArtifactId {
        ArtifactId { namespace: None, name: name.into(), version: version.into() }
    }

    #[test]
    fn fetch_returns_primary_hit_without_consulting_upstreams() {
        let primary = StubBackend::with_artifact("primary", "tokio", "1.0", b"tokio-bytes");
        let upstream = StubBackend::with_artifact("upstream", "tokio", "1.0", b"upstream-bytes");
        let proxy = ProxyBackend::new(primary, vec![upstream]);

        let result = proxy.fetch(&artifact("tokio", "1.0")).unwrap();
        assert_eq!(result.as_deref(), Some(b"tokio-bytes".as_ref()), "primary hit must win");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "fetch_primary_hit_wins",
            result.as_deref() == Some(b"tokio-bytes".as_ref()),
            &format!("primary hit served {:?} (upstream not consulted)", result.as_deref().map(<[u8]>::len)),
        );
    }

    #[test]
    fn fetch_falls_through_to_upstream_on_primary_miss() {
        let primary = StubBackend::always_404("primary");
        let upstream = StubBackend::with_artifact("upstream", "serde", "1.0", b"serde-bytes");
        let proxy = ProxyBackend::new(primary, vec![upstream]);

        let result = proxy.fetch(&artifact("serde", "1.0")).unwrap();
        assert_eq!(result.as_deref(), Some(b"serde-bytes".as_ref()), "upstream must serve on primary miss");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "fetch_falls_through_to_upstream",
            result.as_deref() == Some(b"serde-bytes".as_ref()),
            "primary 404 -> upstream served the bytes",
        );
    }

    #[test]
    fn fetch_returns_none_when_all_miss() {
        let primary = StubBackend::always_404("primary");
        let upstream = StubBackend::always_404("upstream");
        let proxy = ProxyBackend::new(primary, vec![upstream]);

        let result = proxy.fetch(&artifact("missing", "1.0")).unwrap();
        assert!(result.is_none());

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "fetch_none_when_all_miss",
            result.is_none(),
            "primary + upstream both 404 -> None",
        );
    }

    #[test]
    fn http_returns_primary_on_non_404() {
        let primary = StubBackend::always_200("primary", b"primary-body");
        let upstream = StubBackend::always_200("upstream", b"upstream-body");
        let proxy = ProxyBackend::new(primary, vec![upstream]);

        let (status, _, body) = proxy.handle_http2_request("GET", "/repo/index/config.json", b"").unwrap();
        assert_eq!(status, 200);
        assert_eq!(body, b"primary-body");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "http_returns_primary_on_non_404",
            status == 200 && body == b"primary-body",
            &format!("status={status} body=primary-body (upstream skipped)"),
        );
    }

    #[test]
    fn http_falls_through_to_upstream_on_primary_404() {
        let primary = StubBackend::always_404("primary");
        let upstream = StubBackend::always_200("upstream", b"upstream-body");
        let proxy = ProxyBackend::new(primary, vec![upstream]);

        let (status, _, body) = proxy.handle_http2_request("GET", "/repo/index/se/rd/serde", b"").unwrap();
        assert_eq!(status, 200);
        assert_eq!(body, b"upstream-body");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "http_falls_through_to_upstream_on_404",
            status == 200 && body == b"upstream-body",
            &format!("primary 404 -> upstream status={status} body=upstream-body"),
        );
    }

    #[test]
    fn http_returns_404_when_all_404() {
        let primary = StubBackend::always_404("primary");
        let upstream1 = StubBackend::always_404("upstream1");
        let upstream2 = StubBackend::always_404("upstream2");
        let proxy = ProxyBackend::new(primary, vec![upstream1, upstream2]);

        let (status, _, _) = proxy.handle_http2_request("GET", "/x", b"").unwrap();
        assert_eq!(status, 404);

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "http_404_when_all_404",
            status == 404,
            &format!("primary + 2 upstreams all 404 -> status {status}"),
        );
    }

    #[test]
    fn proxy_name_and_format_delegate_to_primary() {
        let primary = StubBackend::always_404("my-primary-repo");
        let proxy = ProxyBackend::new(primary, vec![]);
        assert_eq!(proxy.name(), "my-primary-repo");
        assert_eq!(proxy.format(), ArtifactFormat::Rust);

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "name_and_format_delegate_to_primary",
            proxy.name() == "my-primary-repo" && proxy.format() == ArtifactFormat::Rust,
            &format!("name={} format={:?}", proxy.name(), proxy.format()),
        );
    }

    /// A writable in-memory backend that actually stores `put`s, so we can prove
    /// the proxy writes an upstream hit THROUGH into the primary (cache fill).
    struct WritableStub {
        name: String,
        store: std::sync::Mutex<std::collections::HashMap<String, Vec<u8>>>,
    }
    impl WritableStub {
        fn new(name: &str) -> Arc<Self> {
            Arc::new(Self { name: name.into(), store: std::sync::Mutex::new(Default::default()) })
        }
    }
    impl RepositoryBackendTrait for WritableStub {
        fn name(&self) -> &str { &self.name }
        fn format(&self) -> ArtifactFormat { ArtifactFormat::Rust }
        fn is_writable(&self) -> bool { true }
        fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            Ok(self.store.lock().unwrap().get(&format!("{}/{}", id.name, id.version)).cloned())
        }
        fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
            self.store.lock().unwrap().insert(format!("{}/{}", id.name, id.version), data.to_vec());
            Ok(())
        }
        fn handle_http2_request(&self, _m: &str, _s: &str, _b: &[u8])
            -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            Ok((404, Vec::new(), b"Not found".to_vec()))
        }
    }

    /// CACHING: a miss on a writable primary served from upstream is written
    /// through, so the primary then holds it (a pull-through cache, e.g. /cache
    /// over crates.io). Inject upstream bytes, assert the primary is filled.
    #[test]
    fn fetch_caches_upstream_hit_into_writable_primary() {
        let primary = WritableStub::new("cache");
        let upstream = StubBackend::with_artifact("crates-io", "rand", "0.8.5", b"rand-crate-bytes");
        let proxy = ProxyBackend::new(primary.clone(), vec![upstream]);

        assert!(primary.fetch(&artifact("rand", "0.8.5")).unwrap().is_none(), "primary empty before");
        let got = proxy.fetch(&artifact("rand", "0.8.5")).unwrap();
        assert_eq!(got.as_deref(), Some(b"rand-crate-bytes".as_ref()), "served from upstream on miss");

        let cached = primary.fetch(&artifact("rand", "0.8.5")).unwrap();
        assert_eq!(
            cached.as_deref(),
            Some(b"rand-crate-bytes".as_ref()),
            "write-through must fill the writable primary on an upstream hit"
        );

        #[cfg(feature = "testmatrix")]
        fstatus(
            "proxy",
            "fetch_caches_upstream_hit_into_writable_primary",
            cached.as_deref() == Some(b"rand-crate-bytes".as_ref()),
            "upstream hit written through into the writable primary",
        );
    }

    // === Negative caching (§2) ===

    use std::sync::atomic::{AtomicUsize, Ordering};

    /// An always-miss upstream that COUNTS how many times it is consulted — the
    /// probe the negative-cache tests use to prove the upstream is (or is not)
    /// re-walked on a repeated miss.
    struct CountingMissUpstream {
        name: String,
        fetches: AtomicUsize,
        https: AtomicUsize,
    }
    impl CountingMissUpstream {
        fn new(name: &str) -> Arc<Self> {
            Arc::new(Self { name: name.into(), fetches: AtomicUsize::new(0), https: AtomicUsize::new(0) })
        }
    }
    impl RepositoryBackendTrait for CountingMissUpstream {
        fn name(&self) -> &str { &self.name }
        fn format(&self) -> ArtifactFormat { ArtifactFormat::Rust }
        fn is_writable(&self) -> bool { false }
        fn fetch(&self, _id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            self.fetches.fetch_add(1, Ordering::SeqCst);
            Ok(None) // always a miss
        }
        fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> { anyhow::bail!("read-only") }
        fn handle_http2_request(
            &self,
            _method: &str,
            _suburl: &str,
            _body: &[u8],
        ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            self.https.fetch_add(1, Ordering::SeqCst);
            Ok((404, Vec::new(), b"Not found".to_vec()))
        }
    }

    /// RED-when-broken: with the negative cache ON, a second fetch of a coordinate
    /// that missed everywhere must NOT re-consult the upstream (served from the
    /// negative cache). If the short-circuit is removed the upstream is hit twice
    /// and this fails.
    #[test]
    fn negative_cache_short_circuits_repeated_fetch_miss() {
        let primary = StubBackend::always_404("primary");
        let upstream = CountingMissUpstream::new("crates-io");
        let proxy = ProxyBackend::new(primary, vec![upstream.clone()])
            .with_negative_cache_secs(60);

        assert_eq!(proxy.fetch(&artifact("ghost", "9.9.9")).unwrap(), None, "first: miss");
        assert_eq!(upstream.fetches.load(Ordering::SeqCst), 1, "first miss walks the upstream");
        assert_eq!(proxy.fetch(&artifact("ghost", "9.9.9")).unwrap(), None, "second: still a miss");
        assert_eq!(
            upstream.fetches.load(Ordering::SeqCst),
            1,
            "second miss is served from the negative cache — upstream NOT re-walked"
        );
        // A DIFFERENT coordinate is not shadowed — it still walks the upstream.
        assert_eq!(proxy.fetch(&artifact("other", "1.0")).unwrap(), None);
        assert_eq!(upstream.fetches.load(Ordering::SeqCst), 2, "a distinct miss is not cached under the same key");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-proxy/negcache",
            "fetch_neg_hit_short_circuit",
            upstream.fetches.load(Ordering::SeqCst) == 2,
            "repeat miss served from cache; distinct miss walks upstream",
        );
    }

    /// The negative cache is off by default (`new`): every miss re-walks the
    /// upstream. Proves the caching is what changed the count in the test above,
    /// and that the default path is byte-for-byte the prior behaviour (L2).
    #[test]
    fn negative_cache_disabled_requeries_every_fetch_miss() {
        let primary = StubBackend::always_404("primary");
        let upstream = CountingMissUpstream::new("crates-io");
        let proxy = ProxyBackend::new(primary, vec![upstream.clone()]); // no TTL ⇒ disabled

        assert_eq!(proxy.fetch(&artifact("ghost", "9.9.9")).unwrap(), None);
        assert_eq!(proxy.fetch(&artifact("ghost", "9.9.9")).unwrap(), None);
        assert_eq!(
            upstream.fetches.load(Ordering::SeqCst),
            2,
            "with the negative cache OFF, every miss re-queries the upstream"
        );
    }

    /// A negative-cached miss must NOT shadow a later local publish: the primary
    /// is always checked first, so once the artifact appears in the primary the
    /// proxy serves it even though a miss was cached earlier.
    #[test]
    fn negative_cache_never_shadows_a_later_primary_hit() {
        let primary = WritableStub::new("cache");
        let upstream = CountingMissUpstream::new("crates-io");
        let proxy = ProxyBackend::new(primary.clone(), vec![upstream.clone()])
            .with_negative_cache_secs(60);

        // Miss everywhere → negative entry cached.
        assert_eq!(proxy.fetch(&artifact("rand", "0.8")).unwrap(), None);
        assert_eq!(upstream.fetches.load(Ordering::SeqCst), 1);
        // The artifact is later published straight into the primary.
        primary.put(&artifact("rand", "0.8"), b"rand-bytes").unwrap();
        // The proxy must serve it (primary checked before the negative cache).
        assert_eq!(
            proxy.fetch(&artifact("rand", "0.8")).unwrap().as_deref(),
            Some(b"rand-bytes".as_ref()),
            "a later primary publish is served, never shadowed by the cached miss"
        );
        assert_eq!(upstream.fetches.load(Ordering::SeqCst), 1, "primary hit never touches the upstream");
    }

    /// RED-when-broken (HTTP path): a second GET for a path that 404'd everywhere
    /// must not re-walk the upstream when the negative cache is on.
    #[test]
    fn negative_cache_short_circuits_repeated_http_miss() {
        let primary = StubBackend::always_404("primary");
        let upstream = CountingMissUpstream::new("crates-io");
        let proxy = ProxyBackend::new(primary, vec![upstream.clone()])
            .with_negative_cache_secs(60);

        let (s1, _, _) = proxy.handle_http2_request("GET", "/proxy/absent.crate", &[]).unwrap();
        assert_eq!(s1, 404);
        assert_eq!(upstream.https.load(Ordering::SeqCst), 1, "first 404 walks the upstream");
        let (s2, _, _) = proxy.handle_http2_request("GET", "/proxy/absent.crate", &[]).unwrap();
        assert_eq!(s2, 404, "still 404");
        assert_eq!(
            upstream.https.load(Ordering::SeqCst),
            1,
            "repeat GET served the primary 404 from the negative cache — upstream NOT re-walked"
        );
    }

    /// RED-when-broken (wiring): driving the REAL `ProxyBackend` read paths must
    /// bump the process-global `holger_proxy_requests_total{result="…"}` family on
    /// the matching outcome — a primary hit bumps `primary_hit`, a full miss bumps
    /// `miss`, an upstream serve bumps `upstream_hit`, a negcache short-circuit
    /// bumps `negcache_hit`. Each outcome is triggered in bulk (`N`) and asserted as
    /// a `>= N` DELTA against a pre-snapshot, so a dropped bump fails even though the
    /// registry is a shared process-global (concurrent unit tests can only add a
    /// handful, never `N`). The exact-count RED guard for the family itself lives in
    /// `metrics::tests::proxy_outcomes_count_and_render_by_result_label`.
    #[test]
    fn proxy_outcomes_bump_the_global_metrics_family() {
        use crate::metrics::{global, ProxyOutcome};
        const N: u64 = 64;

        let snap = |o| global().proxy_count(o);
        let (p0, u0, n0, m0) = (
            snap(ProxyOutcome::PrimaryHit),
            snap(ProxyOutcome::UpstreamHit),
            snap(ProxyOutcome::NegcacheHit),
            snap(ProxyOutcome::Miss),
        );

        // primary_hit: primary serves the artifact.
        let ph = ProxyBackend::new(
            StubBackend::with_artifact("primary", "tokio", "1.0", b"bytes"),
            vec![StubBackend::always_404("up")],
        );
        for _ in 0..N {
            assert!(ph.fetch(&artifact("tokio", "1.0")).unwrap().is_some());
        }

        // upstream_hit: primary misses, upstream serves.
        let uh = ProxyBackend::new(
            StubBackend::always_404("primary"),
            vec![StubBackend::with_artifact("up", "serde", "1.0", b"bytes")],
        );
        for _ in 0..N {
            assert!(uh.fetch(&artifact("serde", "1.0")).unwrap().is_some());
        }

        // miss: primary + every upstream miss (negative cache off ⇒ real miss each).
        let ms = ProxyBackend::new(
            StubBackend::always_404("primary"),
            vec![StubBackend::always_404("up")],
        );
        for _ in 0..N {
            assert!(ms.fetch(&artifact("ghost", "9.9")).unwrap().is_none());
        }

        // negcache_hit: first miss primes the negative cache, the next N-1 short-circuit.
        let nc = ProxyBackend::new(
            StubBackend::always_404("primary"),
            vec![StubBackend::always_404("up")],
        )
        .with_negative_cache_secs(600);
        for _ in 0..N {
            assert!(nc.fetch(&artifact("absent", "0.0")).unwrap().is_none());
        }

        let (p1, u1, n1, m1) = (
            snap(ProxyOutcome::PrimaryHit),
            snap(ProxyOutcome::UpstreamHit),
            snap(ProxyOutcome::NegcacheHit),
            snap(ProxyOutcome::Miss),
        );
        assert!(p1 - p0 >= N, "primary_hit must advance by >= {N} (got {})", p1 - p0);
        assert!(u1 - u0 >= N, "upstream_hit must advance by >= {N} (got {})", u1 - u0);
        // The negcache proxy contributes one real miss (priming) + N-1 negcache hits.
        assert!(m1 - m0 >= N, "miss must advance by >= {N} (got {})", m1 - m0);
        assert!(n1 - n0 >= N - 1, "negcache_hit must advance by >= {} (got {})", N - 1, n1 - n0);

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-proxy/metrics",
            "proxy_outcomes_bump_global_family",
            p1 - p0 >= N && u1 - u0 >= N && m1 - m0 >= N && n1 - n0 >= N - 1,
            "primary/upstream/miss/negcache outcomes each advanced holger_proxy_requests_total",
        );
    }

    /// Expiry is honoured: an entry whose TTL has elapsed is treated as a miss and
    /// the upstream is re-walked. Driven deterministically by inserting an
    /// already-past expiry (no sleep, no flake).
    #[test]
    fn negative_cache_expired_entry_is_a_miss() {
        let primary = StubBackend::always_404("primary");
        let upstream = CountingMissUpstream::new("crates-io");
        let proxy = ProxyBackend::new(primary, vec![upstream.clone()])
            .with_negative_cache_secs(60);

        let key = ProxyBackend::fetch_key(&artifact("ghost", "9.9.9"));
        // Plant an entry that expired one second ago.
        let past = Instant::now().checked_sub(Duration::from_secs(1)).unwrap();
        proxy.neg_cache.lock().unwrap().insert(key.clone(), past);
        assert!(!proxy.neg_hit(&key), "an expired entry reads as a miss");
        // And a fetch therefore re-walks the upstream (does not short-circuit).
        assert_eq!(proxy.fetch(&artifact("ghost", "9.9.9")).unwrap(), None);
        assert_eq!(upstream.fetches.load(Ordering::SeqCst), 1, "expired negative entry ⇒ upstream re-queried");
    }
}