server/lib/src/proxy.rs
//! 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.
//!
//! Gotcha: an upstream that ERRORS (connection refused, DNS failure, TLS error, a
//! non-404 status) is logged, counted as `result="upstream_error"`, and SKIPPED —
//! the walk continues to the next member, because "first hit wins" is the whole
//! contract and one dead upstream must not blind the server to the reachable ones.
//! Only when no upstream hit AND at least one errored is the failure surfaced, and
//! that case deliberately does NOT populate the negative cache: the coordinate is
//! unknown, not known-absent, and memoizing a transient network fault would keep
//! answering 404 for the whole TTL after the upstream came back.
//!
//! # What the cache is allowed to believe
//!
//! A pull-through cache re-publishes what it caches: a znippy-backed `/cache`
//! generates its sparse index from what it stores, under holger's OWN sha256. So
//! whatever holger caches, holger vouches for — and cargo's end-to-end digest
//! check then verifies holger's hash of whatever arrived rather than the
//! registry's hash of the crate. Two gates follow from that, and both need one
//! thing the plain backend trait does not carry: what the ORIGIN advertises for a
//! coordinate. That is [`Upstream::advertised_sha256`], attached per upstream with
//! [`ProxyUpstream::attested`].
//!
//! * **Cache-fill gate** ([`ProxyBackend::cache_verdict`]) — upstream bytes are
//! written through only when the upstream's advertised digest MATCHES them. A
//! mismatch fails the whole read (substituted bytes are not served either). An
//! attesting upstream that advertises NOTHING for the coordinate is served but
//! **not cached**: holger will not mint its own attestation over bytes nobody
//! vouched for. An upstream with no attestation seam at all is cached as
//! before — holger cannot ask, and silently turning every existing
//! pull-through cache into a pass-through would be a different (and unasked-for)
//! change; that gap is closed by attaching a seam, not by refusing to cache.
//! * **Serve-time gate** ([`ProxyBackend::primary_verdict`]) — a primary hit in a
//! repo with an attesting upstream is held against the SAME advertised digest
//! before it is served, because the primary is both the cache and the local
//! publish target: a writer publishing `serde 1.0.0` into a pull-through cache
//! would otherwise permanently shadow the registry's. A local copy that differs
//! from what the origin publishes is not served; the walk falls through to the
//! upstream and serves the real one. A coordinate the origin does not publish at
//! all (a genuinely local crate) is served untouched. Verdicts are memoized in a
//! bounded, TTL'd map so this costs ONE upstream metadata probe per coordinate,
//! not one per request.
use crate::lock::LockOrRecover as _;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
/// An origin's reply as the proxy keeps it: status, headers, body.
type CachedReply = (u16, Vec<(String, String)>, Vec<u8>);
use traits::{ArtifactFormat, ArtifactId, RepositoryBackendTrait, Upstream};
/// 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;
/// Upper bound on live attestation verdicts (see [`AttestVerdict`]). Same
/// reasoning as [`NEG_CAP`]: a memo that a scanner can grow must be bounded, and
/// a full map degrades to re-probing rather than leaking.
const ATTEST_CAP: usize = 4096;
/// How long an attestation verdict is trusted before the upstream is asked
/// again. Published artifacts are immutable, so this is long; it exists so a
/// coordinate that appears upstream *later* is eventually noticed, and so a
/// wrong answer cannot outlive the process by design.
const ATTEST_TTL: Duration = Duration::from_secs(3600);
/// One member of the upstream chain: the backend to read through, and — when the
/// upstream can state what the origin published — the attestation seam the two
/// digest gates ask through.
pub struct ProxyUpstream {
backend: Arc<dyn RepositoryBackendTrait>,
attestation: Option<Arc<dyn Upstream>>,
}
impl ProxyUpstream {
/// An upstream with no attestation seam: holger cannot ask it what the origin
/// published, and the gates have nothing to compare against.
pub fn plain(backend: Arc<dyn RepositoryBackendTrait>) -> Self {
Self {
backend,
attestation: None,
}
}
/// An upstream that states what the origin published (every
/// [`Upstream`]): its bytes are checked before they are cached, and a
/// local copy of the same coordinate is checked before it is served.
pub fn attested<T: Upstream + 'static>(upstream: Arc<T>) -> Self {
Self {
backend: upstream.clone(),
attestation: Some(upstream),
}
}
/// The same thing from two already-erased handles — for the wiring engine,
/// which holds a repository's backend as `Arc<dyn RepositoryBackendTrait>`
/// and would otherwise have to recover the concrete upstream type to reach
/// [`attested`](Self::attested).
pub fn with_attestation(
backend: Arc<dyn RepositoryBackendTrait>,
attestation: Arc<dyn Upstream>,
) -> Self {
Self {
backend,
attestation: Some(attestation),
}
}
/// The wrapped backend (the read itself always goes through this).
pub fn backend(&self) -> &Arc<dyn RepositoryBackendTrait> {
&self.backend
}
}
/// A memoized answer to "does the origin publish this coordinate, and is the
/// local copy what it published?".
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AttestVerdict {
/// The origin publishes this coordinate and the local bytes match it.
Matches,
/// The origin publishes no digest for this coordinate — it is local-only as
/// far as holger can tell, and nothing upstream contradicts serving it.
NotPublished,
}
/// What to do with a primary (local) hit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PrimaryVerdict {
/// Serve it — it is the origin's artifact, or the origin has no opinion.
Serve,
/// The origin publishes a DIFFERENT artifact under this coordinate: the local
/// copy is a substitution (a local publish shadowing the registry, or a cache
/// poisoned before this gate existed). Fall through to the upstream.
Shadowed,
}
/// What to do with upstream bytes on the way into the cache.
#[derive(Debug, Clone, PartialEq, Eq)]
enum CacheVerdict {
/// Digest matched, or the upstream has no attestation seam at all.
Cache,
/// Serve, but do not cache — holger would otherwise re-attest bytes nobody
/// vouched for.
ServeOnly(String),
/// The upstream's own advertised digest says these are not the artifact:
/// refuse the read outright.
Refuse { expected: String, actual: String },
}
/// 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<ProxyUpstream>,
/// 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>>,
/// Remembered attestation verdicts: `coordinate key → (verdict, expiry)`, so
/// the serve-time gate costs ONE upstream metadata probe per coordinate per
/// [`ATTEST_TTL`] instead of one per request. Bounded by [`ATTEST_CAP`].
attest_cache: Mutex<HashMap<String, (AttestVerdict, Instant)>>,
}
impl ProxyBackend {
/// A proxy with the negative cache **disabled** (`neg_ttl == 0`) and upstreams
/// that carry no attestation seam — 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::with_upstreams(
primary,
upstreams.into_iter().map(ProxyUpstream::plain).collect(),
)
}
/// A proxy whose upstreams may carry an [`Upstream::advertised_sha256`] seam
/// ([`ProxyUpstream::attested`]) — the constructor the two digest gates need.
pub fn with_upstreams(
primary: Arc<dyn RepositoryBackendTrait>,
upstreams: Vec<ProxyUpstream>,
) -> Self {
Self {
primary,
upstreams,
neg_ttl: Duration::ZERO,
neg_cache: Mutex::new(HashMap::new()),
attest_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 = self.neg_cache.lock_or_recover();
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,
};
{
let mut cache = self.neg_cache.lock_or_recover();
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);
}
}
}
/// How many of this proxy's upstreams can be asked what the origin published.
///
/// Public because the two digest gates are only as live as the WIRING that
/// attaches an attestation seam, and that wiring runs in `wire_holger`, a
/// layer above this one. Both gates degrade silently to the old behaviour
/// when this is zero — which is correct for an upstream that genuinely cannot
/// be asked, and a regression indistinguishable from correct when the wiring
/// breaks. This is how a test tells those two apart.
pub fn attested_upstream_count(&self) -> usize {
self.upstreams
.iter()
.filter(|u| u.attestation.is_some())
.count()
}
// === The two digest gates (see the module note) ===
/// Every upstream that can state what the origin published.
fn attestors(&self) -> impl Iterator<Item = (&str, &Arc<dyn Upstream>)> {
self.upstreams
.iter()
.filter_map(|u| u.attestation.as_ref().map(|a| (u.backend.name(), a)))
}
/// A live (non-expired) memoized verdict for `key`; expired entries are
/// evicted in passing.
fn attest_hit(&self, key: &str) -> Option<AttestVerdict> {
let mut cache = self.attest_cache.lock_or_recover();
match cache.get(key) {
Some(&(v, expiry)) if expiry > Instant::now() => Some(v),
Some(_) => {
cache.remove(key);
None
}
None => None,
}
}
/// Memoize `verdict` for `key`, bounded exactly like the negative cache.
fn attest_remember(&self, key: &str, verdict: AttestVerdict) {
let now = Instant::now();
let Some(expiry) = now.checked_add(ATTEST_TTL) else {
return;
};
{
let mut cache = self.attest_cache.lock_or_recover();
if cache.len() >= ATTEST_CAP {
cache.retain(|_, (_, exp)| *exp > now);
}
if cache.len() < ATTEST_CAP {
cache.insert(key.to_string(), (verdict, expiry));
}
}
}
/// Forget what was believed about `key` — called on a local `put`, so a
/// publish can never ride a verdict formed before it.
fn attest_forget(&self, key: &str) {
{
let mut cache = self.attest_cache.lock_or_recover();
cache.remove(key);
}
}
/// **Serve-time gate.** Is this local (primary) copy the artifact the origin
/// publishes under that coordinate?
///
/// Only asks when an upstream can answer, and asks at most once per
/// coordinate per [`ATTEST_TTL`]. An upstream that cannot be reached is not
/// evidence of anything, so the local copy is served (availability: a dead
/// crates.io must not empty the cache) — the mismatch verdict requires a
/// POSITIVE statement from the origin.
fn primary_verdict(&self, id: &ArtifactId, data: &[u8]) -> PrimaryVerdict {
if self.attestors().next().is_none() {
return PrimaryVerdict::Serve; // nothing can state otherwise
}
let key = Self::fetch_key(id);
if let Some(v) = self.attest_hit(&key) {
return match v {
AttestVerdict::Matches | AttestVerdict::NotPublished => PrimaryVerdict::Serve,
};
}
let actual = nornir_hash::sha256_hex(data);
for (name, attestor) in self.attestors() {
match attestor.advertised_sha256(id) {
Ok(Some(expected)) if expected == actual => {
self.attest_remember(&key, AttestVerdict::Matches);
return PrimaryVerdict::Serve;
}
Ok(Some(expected)) => {
// The origin publishes a different artifact under this
// coordinate. NOT memoized: the local store may be corrected
// (re-cached from upstream) and must not stay condemned.
log::error!(
"proxy: {}/{} in {} is NOT what upstream {name} publishes \
(origin sha256 {expected}, local {actual}) — refusing to serve the local \
copy and falling through to the upstream",
id.name,
id.version,
self.primary.name()
);
crate::grpc::functional_status(
"holger-proxy/attestation",
"local_copy_shadows_upstream_refused",
true,
&format!("{}/{} in {}", id.name, id.version, self.primary.name()),
);
return PrimaryVerdict::Shadowed;
}
Ok(None) => continue, // this origin makes no claim; ask the next
Err(e) => {
log::warn!(
"proxy: could not ask upstream {name} what it publishes for {}/{}: {e}",
id.name,
id.version
);
return PrimaryVerdict::Serve; // unreachable ≠ contradicted
}
}
}
// Every attestor answered "no claim": a local-only artifact.
self.attest_remember(&key, AttestVerdict::NotPublished);
PrimaryVerdict::Serve
}
/// **Cache-fill gate.** May these upstream bytes be written through into the
/// primary — i.e. may holger start vouching for them?
fn cache_verdict(
&self,
upstream: &ProxyUpstream,
id: &ArtifactId,
data: &[u8],
) -> CacheVerdict {
let Some(attestor) = upstream.attestation.as_ref() else {
// No seam to ask through. Cached as before — see the module note.
return CacheVerdict::Cache;
};
match attestor.advertised_sha256(id) {
Ok(Some(expected)) => {
let actual = nornir_hash::sha256_hex(data);
if expected == actual {
CacheVerdict::Cache
} else {
CacheVerdict::Refuse { expected, actual }
}
}
Ok(None) => CacheVerdict::ServeOnly(format!(
"upstream {} advertises no digest for {}/{}",
upstream.backend.name(),
id.name,
id.version
)),
Err(e) => CacheVerdict::ServeOnly(format!(
"upstream {} could not be asked what it publishes for {}/{}: {e}",
upstream.backend.name(),
id.name,
id.version
)),
}
}
}
/// A [`traits::Decorator`] over the PRIMARY: every method reaches it by
/// construction (path mapping, listing, archive, delete, streaming uploads —
/// upstreams share the repository's format and the primary owns the local
/// layout), except the reads and writes the pull-through changes below.
impl ProxyBackend {
/// **A cargo sparse-index answer, with the origin's line for every version the
/// origin publishes** (Holger G0). The writable primary of a pull-through cache
/// builds its own index for what it stores, and a crate it CACHED has no deps
/// sidecar, so its line says `"deps":[]`, `"features":{}` and no `features2`. That
/// line must not shadow the origin's: cargo would resolve the crate with none of its
/// dependencies or features. So for a Rust index path both answers are read, the
/// origin's lines win per version, and versions only the primary holds (a local
/// publish) keep the primary's line. `None` = not an index answer to merge: the
/// primary's response stands as it is.
fn cargo_index_with_the_origins_lines(
&self,
method: &str,
suburl: &str,
body: &[u8],
primary: &CachedReply,
) -> Option<CachedReply> {
if method != "GET"
|| primary.0 != 200
|| self.upstreams.is_empty()
|| self.primary.format() != ArtifactFormat::Rust
|| !suburl.contains("/index/")
|| suburl.ends_with("config.json")
|| self.primary.coordinate(suburl).is_some()
{
return None;
}
let versions = |b: &[u8]| -> Option<Vec<(String, String)>> {
let text = std::str::from_utf8(b).ok()?;
let mut out = Vec::new();
for line in text.lines().filter(|l| !l.trim().is_empty()) {
let v: serde_json::Value = serde_json::from_str(line).ok()?;
out.push((v.get("vers")?.as_str()?.to_string(), line.to_string()));
}
(!out.is_empty()).then_some(out)
};
let local = versions(&primary.2)?;
let origin = self.upstreams.iter().find_map(|u| {
match u.backend.handle_http2_request(method, suburl, body) {
Ok((200, _, b)) => versions(&b),
_ => None,
}
})?;
let mut lines: Vec<String> = origin.iter().map(|(_, l)| l.clone()).collect();
for (v, l) in &local {
if !origin.iter().any(|(o, _)| o == v) {
lines.push(l.clone());
}
}
Some((primary.0, primary.1.clone(), lines.join("\n").into_bytes()))
}
}
impl traits::Decorator 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()
}
/// The primary's, minus streaming reads: a read in pieces straight off the
/// primary would skip the pull-through walk, so a proxied `GET` takes the
/// whole-body door and its pieces are sliced from [`fetch`](Self::fetch).
fn caps(&self) -> traits::Caps {
traits::Caps {
streaming_reads: false,
..self.primary.caps()
}
}
fn around<T>(
&self,
_: traits::Access,
call: impl FnOnce(&dyn RepositoryBackendTrait) -> anyhow::Result<T>,
) -> anyhow::Result<T> {
call(self.primary.as_ref())
}
fn fetch_chunks(
&self,
_id: &ArtifactId,
_chunk_bytes: usize,
) -> anyhow::Result<Option<traits::ChunkIter>> {
traits::unsupported(self.primary.name(), traits::Cap::StreamingReads)
}
/// A streamed local publish drops the coordinate's verdict, as `put` does.
fn put_from_reader(
&self,
id: &ArtifactId,
body: &mut dyn std::io::Read,
size_hint: Option<u64>,
) -> anyhow::Result<()> {
self.attest_forget(&Self::fetch_key(id));
self.primary.put_from_reader(id, body, size_hint)
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
// Set when the primary DID hold the coordinate but the copy contradicts
// the origin: the upstream serves the read, and nothing is written
// through over the local copy (repairing an immutable archive in place is
// not this gate's job — it refuses to serve the substitution, loudly).
let mut shadowed = false;
if let Some(data) = self.primary.fetch(id)? {
// Serve-time gate: the primary is BOTH the cache and the local publish
// target, so a local hit is not automatically the origin's artifact.
// A copy that contradicts what the origin publishes is not served;
// the walk below serves the real one instead.
match self.primary_verdict(id, &data) {
PrimaryVerdict::Serve => {
crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::PrimaryHit);
return Ok(Some(data));
}
PrimaryVerdict::Shadowed => shadowed = true,
}
}
// 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);
}
}
// THE QUEUED PULL-THROUGH (`crate::pull_through`). Only when a door
// installed the queue for this walk, and only into a cache that can
// keep the bytes (a read-only primary would be asked again forever):
// the walk below becomes ONE durable `Job::Fetch` and this request is
// answered `FetchQueued` (503 + Retry-After at the door). Nothing
// installed ⇒ `None` ⇒ the inline walk, exactly as before.
if !shadowed && crate::pull_through::armed() && self.primary.is_writable() {
match crate::pull_through::intercept(self.primary.name(), id) {
None => {}
Some(Err(e)) => return Err(e),
Some(Ok(())) => {
if let Some(data) = self.primary.fetch(id)? {
crate::metrics::global()
.record_proxy(crate::metrics::ProxyOutcome::PrimaryHit);
return Ok(Some(data));
}
}
}
}
// Walk the chain. An upstream that ERRORS (connection refused, DNS
// failure, TLS error, a non-404 status) is logged and SKIPPED, not
// propagated: the contract is "try each upstream in order, return the
// first hit", and one unreachable member must not blind the server to the
// reachable ones. The `?` here used to abort the whole walk, so a repo
// with upstreams ["crates-io", "local-mirror"] returned 500 for every
// artifact — including the ones local-mirror held — the moment crates.io
// became unreachable.
let mut last_err: Option<anyhow::Error> = None;
for upstream in &self.upstreams {
let hit = match upstream.backend.fetch(id) {
Ok(hit) => hit,
Err(e) => {
log::warn!(
"proxy: upstream {} errored on {}/{}, skipping: {e}",
upstream.backend.name(),
id.name,
id.version
);
crate::upstreams::record(
upstream.backend.name(),
crate::upstreams::Outcome::Error(&e.to_string()),
crate::clock::now_unix_ms(),
);
crate::metrics::global()
.record_proxy(crate::metrics::ProxyOutcome::UpstreamError);
last_err = Some(e);
continue;
}
};
let Some(data) = hit else {
crate::upstreams::record(
upstream.backend.name(),
crate::upstreams::Outcome::Miss,
crate::clock::now_unix_ms(),
);
continue;
};
{
log::debug!(
"proxy: cache miss on {}, served from upstream {}",
self.primary.name(),
upstream.backend.name()
);
// Cache-fill gate: holger re-publishes what it caches under its
// OWN digest, so it may only cache bytes the origin vouches for.
match self.cache_verdict(upstream, id, &data) {
CacheVerdict::Refuse { expected, actual } => {
// Not merely uncacheable — not the artifact. Do not serve
// it either; a substitution passed to the client is the
// same compromise one request earlier.
crate::upstreams::record(
upstream.backend.name(),
crate::upstreams::Outcome::Error(
"served bytes that do not match the digest it advertises",
),
crate::clock::now_unix_ms(),
);
crate::grpc::functional_status(
"holger-proxy/attestation",
"upstream_bytes_refused_on_digest_mismatch",
true,
&format!(
"{}/{} from {}",
id.name,
id.version,
upstream.backend.name()
),
);
return Err(anyhow::anyhow!(
"upstream {} served {}/{} that does not match the digest it advertises \
(advertised sha256 {expected}, bytes hash to {actual}) — refusing",
upstream.backend.name(),
id.name,
id.version
));
}
CacheVerdict::ServeOnly(why) => {
log::warn!(
"proxy: serving {}/{} from upstream {} WITHOUT caching it — {why}",
id.name,
id.version,
upstream.backend.name()
);
crate::grpc::functional_status(
"holger-proxy/attestation",
"unattested_upstream_bytes_not_cached",
true,
&format!("{}/{}", id.name, id.version),
);
}
CacheVerdict::Cache => {
// 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.
// Skipped when the primary already holds a contradicting
// copy (see `shadowed`).
if self.primary.is_writable() && !shadowed {
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::upstreams::record(
upstream.backend.name(),
crate::upstreams::Outcome::Hit,
crate::clock::now_unix_ms(),
);
crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::UpstreamHit);
// …and to the consumer this walk runs for (`Cost.upstream_fetch`).
crate::pull_through::note_upstream_fetch(data.len() as u64);
return Ok(Some(data));
}
}
// No upstream produced a hit. If some of them errored, the coordinate is
// NOT known-absent — it is unknown — so surface the failure and do not
// memoize: negative-caching a transient network fault would keep serving
// 404 for the whole TTL after the upstream recovered.
if let Some(e) = last_err {
return Err(e.context("every upstream either errored or missed"));
}
// Missed primary AND every upstream, all of them cleanly — 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)
}
/// Local publish. Delegated to the primary as before — a pull-through repo
/// that is also a team's publish target is a shipped shape (`/sparring` in the
/// dev pair) and refusing it here would break it. What DOES change: the
/// attestation verdict for the coordinate is dropped, so the next read
/// re-asks the origin instead of riding a verdict formed before this write.
/// A publish that shadows an upstream coordinate is caught there.
fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
self.attest_forget(&Self::fetch_key(id));
self.primary.put(id, data)
}
/// The proxy answers through its own [`handle_http2_request`](Self::handle_http2_request)
/// (the upstream walk on a miss); the blanket forward to the primary alone would skip it.
fn handle_http2_request_bytes(
&self,
method: &str,
suburl: &str,
body: &[u8],
) -> anyhow::Result<traits::HttpBytesReply> {
traits::Decorator::handle_http2_request(self, method, suburl, body)
.map(|(s, h, b)| (s, h, traits::Bytes::from(b)))
}
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);
if let Some(merged) =
self.cargo_index_with_the_origins_lines(method, suburl, body, &resp)
{
return Ok(merged);
}
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);
}
}
// The queued pull-through, for a GET the primary can name as an
// artifact (an index path it cannot map stays inline — it is small and
// it is what the client reads first). See `fetch` above.
if method == "GET" && crate::pull_through::armed() && self.primary.is_writable() {
if let Some(id) = self.primary.coordinate(suburl) {
match crate::pull_through::intercept(self.primary.name(), &id) {
None => {}
Some(Err(e)) => return Err(e),
Some(Ok(())) => {
let again = self.primary.handle_http2_request(method, suburl, body)?;
if again.0 != 404 {
crate::metrics::global()
.record_proxy(crate::metrics::ProxyOutcome::PrimaryHit);
return Ok(again);
}
}
}
}
}
// THE CACHE FILL, for a GET the primary names as an artifact (a crate
// download): go through `fetch`, which walks the upstreams, holds the
// bytes against what the origin publishes and writes them through into
// the primary. The primary then answers in its own words, so the next
// download is local. An index path the primary cannot name stays on the
// HTTP walk below.
if method == "GET" && self.primary.is_writable() {
if let Some(id) = self.primary.coordinate(suburl) {
match traits::Decorator::fetch(self, &id) {
Ok(Some(data)) => {
let again = self.primary.handle_http2_request(method, suburl, body)?;
if again.0 != 404 {
return Ok(again);
}
// Served but not cached (no digest to hold it against).
let octet = vec![(
"Content-Type".to_string(),
"application/octet-stream".to_string(),
)];
return Ok((200, octet, data.to_vec()));
}
// Every upstream was asked and none holds it.
Ok(None) => return Ok(resp),
Err(e) => log::warn!(
"proxy: cache fill of {suburl} failed, walking the upstreams by path: {e}"
),
}
}
}
// Same skip-don't-abort rule as `fetch` above: a dead upstream must not
// take the reachable ones down with it.
let mut last_err: Option<anyhow::Error> = None;
for upstream in &self.upstreams {
let up_resp = match upstream.backend.handle_http2_request(method, suburl, body) {
Ok(r) => r,
Err(e) => {
log::warn!(
"proxy: upstream {} errored on HTTP {} {}, skipping: {e}",
upstream.backend.name(),
method,
suburl
);
crate::upstreams::record(
upstream.backend.name(),
crate::upstreams::Outcome::Error(&e.to_string()),
crate::clock::now_unix_ms(),
);
crate::metrics::global()
.record_proxy(crate::metrics::ProxyOutcome::UpstreamError);
last_err = Some(e);
continue;
}
};
let outcome = match up_resp.0 {
200..=299 => crate::upstreams::Outcome::Hit,
404 => crate::upstreams::Outcome::Miss,
// Served as before (first non-404 wins); on the board it did not serve.
_ => crate::upstreams::Outcome::Error("answered with an error status"),
};
crate::upstreams::record(
upstream.backend.name(),
outcome,
crate::clock::now_unix_ms(),
);
if up_resp.0 != 404 {
log::debug!(
"proxy: HTTP {} {} — primary 404, served from upstream {}",
method,
suburl,
upstream.backend.name()
);
crate::metrics::global().record_proxy(crate::metrics::ProxyOutcome::UpstreamHit);
crate::pull_through::note_upstream_fetch(up_resp.2.len() as u64);
return Ok(up_resp);
}
}
// An errored upstream means "unknown", not "absent" — surface it and skip
// the negative cache (see `fetch`).
if let Some(e) = last_err {
return Err(e.context("every upstream either errored or 404'd"));
}
// Primary 404 + every upstream cleanly 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)
}
}
#[cfg(test)]
mod tests {
use super::*;
use traits::ArtifactFormat;
/// Every trait method reaches the primary by construction, except the
/// streaming read the pull-through takes over: around a primary with every
/// cap the proxy answers all of them but that one, honestly.
#[test]
fn a_proxy_forwards_every_method_to_its_primary() {
use traits::backend::conformance::{check, Everything};
let all = ProxyBackend::new(Arc::new(Everything::new("all")), vec![]);
assert!(!traits::RepositoryBackendTrait::caps(&all).streaming_reads);
assert!(
traits::RepositoryBackendTrait::caps(&all).delete
&& traits::RepositoryBackendTrait::caps(&all).streaming_uploads
);
check(&all).unwrap();
check(&ProxyBackend::new(
Arc::new(Everything::core_only("core")),
vec![],
))
.unwrap();
}
#[cfg(feature = "testmatrix")]
use traits::test_support::fstatus;
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 caps(&self) -> traits::Caps {
traits::Caps::NONE
}
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<traits::Bytes>> {
let key = format!("{}/{}", id.name, id.version);
Ok(self.data.get(&key).cloned().map(traits::Bytes::from))
}
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()))
}
}
/// An upstream that is DOWN: every call errors, the way
/// `SkidbladnirUpstream::fetch` does on connection-refused / DNS failure /
/// TLS error / a non-404 status.
struct DeadBackend {
name: String,
}
impl DeadBackend {
fn new(name: &str) -> Arc<Self> {
Arc::new(Self { name: name.into() })
}
}
impl RepositoryBackendTrait for DeadBackend {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
anyhow::bail!("connection refused")
}
fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("connection refused")
}
fn handle_http2_request(
&self,
_method: &str,
_suburl: &str,
_body: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
anyhow::bail!("connection refused")
}
}
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",
);
}
/// A DEAD upstream earlier in the chain must be skipped, not fatal.
///
/// RED-when-broken: `upstream.fetch(id)?` inside the walk propagated the
/// first Err, so a repo configured `ron_upstreams: ["crates-io",
/// "local-mirror"]` served 500 for EVERY artifact — including the ones
/// local-mirror held — as soon as crates.io became unreachable.
#[test]
fn fetch_skips_erroring_upstream_and_serves_from_the_next_one() {
let primary = StubBackend::always_404("primary");
let dead = DeadBackend::new("crates-io");
let live = StubBackend::with_artifact("local-mirror", "serde", "1.0", b"serde-bytes");
let proxy = ProxyBackend::new(primary, vec![dead, live]);
let result = proxy
.fetch(&artifact("serde", "1.0"))
.expect("a dead upstream must not fail the read");
assert_eq!(result.as_deref(), Some(b"serde-bytes".as_ref()));
}
/// When NO upstream hits and at least one errored, the coordinate is unknown
/// rather than known-absent: surface the error and do NOT negative-cache it
/// (memoizing a transient network fault would keep 404ing for the whole TTL
/// after the upstream recovered).
#[test]
fn fetch_errors_when_every_upstream_errored_and_none_hit() {
let primary = StubBackend::always_404("primary");
let proxy = ProxyBackend::new(primary, vec![DeadBackend::new("crates-io")]);
let err = proxy
.fetch(&artifact("serde", "1.0"))
.expect_err("an all-errored walk is unknown, not absent");
assert!(
err.to_string().contains("every upstream"),
"error should say the whole chain failed, got: {err}"
);
}
/// The HTTP door follows the same rule as `fetch`.
#[test]
fn http_skips_erroring_upstream_and_serves_from_the_next_one() {
let primary = StubBackend::always_404("primary");
let dead = DeadBackend::new("crates-io");
let live = StubBackend::always_200("local-mirror", b"upstream-body");
let proxy = ProxyBackend::new(primary, vec![dead, live]);
let (status, _, body) = proxy
.handle_http2_request("GET", "/repo/index/se/rd/serde", b"")
.expect("a dead upstream must not fail the request");
assert_eq!(status, 200);
assert_eq!(body, b"upstream-body");
}
/// **A cached crate keeps the origin's index line** (Holger G0, found by dwarves'
/// Gate B). The writable primary of a pull-through cache generates its own sparse
/// index for what it stores, and a crate it CACHED (not published) has no deps
/// sidecar, so its line read `"deps":[]` and `"features":{}` and shadowed the
/// origin's: the second crate built through the cache failed with "serde does not
/// have that feature `derive`". A version the origin publishes is answered with the
/// origin's line; a version only the primary has (a local publish) keeps its own.
#[test]
fn a_cached_crates_index_line_is_the_origins_and_a_local_version_stays() {
let local =
br#"{"name":"serde","vers":"1.0.0","deps":[],"cksum":"aa","features":{},"yanked":false}
{"name":"serde","vers":"9.9.9","deps":[],"cksum":"bb","features":{},"yanked":false}"#;
let origin = br#"{"name":"serde","vers":"1.0.0","deps":[{"name":"serde_derive","req":"^1","features":[],"optional":true,"default_features":true,"target":null,"kind":"normal"}],"cksum":"aa","features":{"derive":["serde_derive"]},"features2":{"std":["dep:x"]},"yanked":false,"v":2}
{"name":"serde","vers":"1.0.1","deps":[],"cksum":"cc","features":{},"yanked":false}"#;
let proxy = ProxyBackend::new(
StubBackend::always_200("cache", local),
vec![StubBackend::always_200("crates-io", origin)],
);
let (status, _, body) = proxy
.handle_http2_request("GET", "/cache/index/se/rd/serde", b"")
.unwrap();
assert_eq!(status, 200);
let lines: Vec<serde_json::Value> = String::from_utf8(body)
.unwrap()
.lines()
.map(|l| serde_json::from_str(l).unwrap())
.collect();
let by = |v: &str| {
lines
.iter()
.find(|l| l["vers"] == v)
.unwrap_or_else(|| panic!("{v} in {lines:?}"))
};
assert_eq!(
by("1.0.0")["features"]["derive"][0],
"serde_derive",
"the origin's line, features and all"
);
assert_eq!(by("1.0.0")["features2"]["std"][0], "dep:x");
assert_eq!(by("1.0.0")["deps"][0]["name"], "serde_derive");
assert_eq!(
by("9.9.9")["cksum"],
"bb",
"a local-only version keeps its line"
);
assert_eq!(
by("1.0.1")["cksum"],
"cc",
"the origin's other versions are offered too"
);
assert_eq!(lines.len(), 3);
// The config file is not an index: the primary's answer stands.
let (_, _, body) = proxy
.handle_http2_request("GET", "/cache/index/config.json", b"")
.unwrap();
assert_eq!(body, local.to_vec());
}
#[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 caps(&self) -> traits::Caps {
traits::Caps::NONE
}
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<traits::Bytes>> {
Ok(self
.store
.lock()
.unwrap()
.get(&format!("{}/{}", id.name, id.version))
.cloned()
.map(traits::Bytes::from))
}
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 caps(&self) -> traits::Caps {
traits::Caps::NONE
}
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<traits::Bytes>> {
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",
);
}
// === The digest gates (§1, §4) ===
/// An upstream that can state what the ORIGIN published — the shape
/// `RustRegistryUpstream` (crates.io's index `cksum`) and every Skidbladnir
/// harbour have. `digests` is what it ADVERTISES; `data` is what it SERVES,
/// and the two are set independently so a compromised-CDN scenario is just a
/// row where they differ.
struct AttestedStub {
name: String,
data: std::sync::Mutex<std::collections::HashMap<String, Vec<u8>>>,
digests: std::sync::Mutex<std::collections::HashMap<String, Option<String>>>,
probes: AtomicUsize,
auth: traits::UpstreamAuth,
}
impl AttestedStub {
fn new(name: &str) -> Arc<Self> {
Arc::new(Self {
name: name.into(),
data: Default::default(),
digests: Default::default(),
probes: AtomicUsize::new(0),
auth: traits::UpstreamAuth::None,
})
}
fn key(id: &ArtifactId) -> String {
format!("{}/{}", id.name, id.version)
}
/// Publish `bytes` honestly: served AND advertised under their own digest.
fn publishes(&self, name: &str, version: &str, bytes: &[u8]) {
let k = format!("{name}/{version}");
self.data.lock().unwrap().insert(k.clone(), bytes.to_vec());
self.digests
.lock()
.unwrap()
.insert(k, Some(nornir_hash::sha256_hex(bytes)));
}
/// Advertise one artifact but serve another — a poisoned mirror/CDN edge.
fn serves_something_else(&self, name: &str, version: &str, served: &[u8]) {
self.data
.lock()
.unwrap()
.insert(format!("{name}/{version}"), served.to_vec());
}
/// Serve bytes while advertising nothing for them.
fn serves_unattested(&self, name: &str, version: &str, bytes: &[u8]) {
let k = format!("{name}/{version}");
self.data.lock().unwrap().insert(k.clone(), bytes.to_vec());
self.digests.lock().unwrap().insert(k, None);
}
}
impl RepositoryBackendTrait for AttestedStub {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
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<traits::Bytes>> {
Ok(self
.data
.lock()
.unwrap()
.get(&Self::key(id))
.cloned()
.map(traits::Bytes::from))
}
fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
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()))
}
}
impl traits::Upstream for AttestedStub {
fn base_url(&self) -> &str {
"https://origin.test"
}
fn auth(&self) -> &traits::UpstreamAuth {
&self.auth
}
fn metadata(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::ArtifactMeta>> {
self.probes.fetch_add(1, Ordering::SeqCst);
Ok(self
.digests
.lock()
.unwrap()
.get(&Self::key(id))
.cloned()
.map(|sha256| traits::ArtifactMeta {
size: None,
sha256,
content_type: None,
}))
}
}
/// **Upstream bytes that contradict the upstream's own advertised digest are
/// neither cached nor served.**
///
/// RED-when-broken, and it WAS broken: `fetch` wrote the fetched bytes
/// straight into the writable primary and returned them. A poisoned CDN edge
/// serving a different `serde 1.0.0` was cached under holger's OWN sha256,
/// and the sparse index a znippy `/cache` generates then attested the
/// substitution to cargo. Remove the `CacheVerdict::Refuse` arm and this test
/// sees the substituted bytes come back.
#[test]
fn upstream_bytes_that_contradict_the_advertised_digest_are_refused() {
let real = b"the crate the registry published";
let primary = WritableStub::new("cache");
let upstream = AttestedStub::new("crates-io");
upstream.publishes("serde", "1.0.0", real);
upstream.serves_something_else("serde", "1.0.0", b"substituted bytes");
let proxy = ProxyBackend::with_upstreams(
primary.clone(),
vec![ProxyUpstream::attested(upstream.clone())],
);
let err = proxy
.fetch(&artifact("serde", "1.0.0"))
.expect_err("substituted upstream bytes must fail the read");
let msg = err.to_string();
assert!(
msg.contains("does not match the digest it advertises"),
"{msg}"
);
assert!(
primary
.fetch(&artifact("serde", "1.0.0"))
.unwrap()
.is_none(),
"and nothing was written through into the cache"
);
#[cfg(feature = "testmatrix")]
fstatus(
"holger-proxy/attestation",
"upstream_bytes_refused_on_digest_mismatch",
msg.contains("does not match the digest it advertises"),
"poisoned upstream bytes neither served nor cached",
);
}
/// Bytes that DO match what the upstream advertises are served and cached —
/// the gate rejects substitutions, not caching.
#[test]
fn upstream_bytes_that_match_the_advertised_digest_are_cached() {
let real = b"rand-crate-bytes";
let primary = WritableStub::new("cache");
let upstream = AttestedStub::new("crates-io");
upstream.publishes("rand", "0.8.5", real);
let proxy = ProxyBackend::with_upstreams(
primary.clone(),
vec![ProxyUpstream::attested(upstream.clone())],
);
assert_eq!(
proxy.fetch(&artifact("rand", "0.8.5")).unwrap().as_deref(),
Some(real.as_ref())
);
assert_eq!(
primary
.fetch(&artifact("rand", "0.8.5"))
.unwrap()
.as_deref(),
Some(real.as_ref()),
"verified bytes are written through as before"
);
}
/// **An upstream that advertises NOTHING for a coordinate is served but not
/// cached.** holger does not mint its own attestation over bytes nobody
/// vouched for — the deliberate call: availability of the read is kept, the
/// re-attestation is not.
#[test]
fn unattested_upstream_bytes_are_served_but_never_cached() {
let primary = WritableStub::new("cache");
let upstream = AttestedStub::new("crates-io");
upstream.serves_unattested("mystery", "1.0.0", b"bytes nobody vouches for");
let proxy = ProxyBackend::with_upstreams(
primary.clone(),
vec![ProxyUpstream::attested(upstream.clone())],
);
assert_eq!(
proxy
.fetch(&artifact("mystery", "1.0.0"))
.unwrap()
.as_deref(),
Some(b"bytes nobody vouches for".as_ref()),
"the read still succeeds"
);
assert!(
primary
.fetch(&artifact("mystery", "1.0.0"))
.unwrap()
.is_none(),
"but the cache does not start vouching for them"
);
#[cfg(feature = "testmatrix")]
fstatus(
"holger-proxy/attestation",
"unattested_upstream_bytes_not_cached",
primary
.fetch(&artifact("mystery", "1.0.0"))
.unwrap()
.is_none(),
"served, not cached",
);
}
/// An upstream with NO attestation seam is cached exactly as before — the
/// gate cannot invent a claim, and refusing to cache every such upstream
/// would silently turn every existing pull-through repo into a pass-through.
#[test]
fn an_upstream_with_no_attestation_seam_is_cached_as_before() {
let primary = WritableStub::new("cache");
let upstream = StubBackend::with_artifact("plain-mirror", "rand", "0.8.5", b"rand-bytes");
let proxy = ProxyBackend::new(primary.clone(), vec![upstream]);
assert!(proxy.fetch(&artifact("rand", "0.8.5")).unwrap().is_some());
assert_eq!(
primary
.fetch(&artifact("rand", "0.8.5"))
.unwrap()
.as_deref(),
Some(b"rand-bytes".as_ref()),
"unchanged behaviour for an upstream that cannot be asked"
);
}
/// **A local publish cannot shadow an upstream coordinate.** (§4)
///
/// RED-when-broken, and it WAS broken: `fetch` returned any primary hit
/// first, and the primary is both the cache AND the local publish target — so
/// a Writer publishing `serde 1.0.0` into a pull-through cache permanently
/// served their bytes in place of the registry's, to every client of that
/// repo. The serve-time gate holds a local hit against what the origin
/// publishes: the substitution is refused and the registry's artifact is
/// served instead. Delete the `PrimaryVerdict::Shadowed` arm and the
/// attacker's bytes come back.
#[test]
fn a_local_publish_cannot_shadow_an_upstream_coordinate() {
let real = b"the serde crates.io published";
let primary = WritableStub::new("cache");
let upstream = AttestedStub::new("crates-io");
upstream.publishes("serde", "1.0.0", real);
let proxy = ProxyBackend::with_upstreams(
primary.clone(),
vec![ProxyUpstream::attested(upstream.clone())],
);
// A writer publishes their own `serde 1.0.0` into the cache repo.
proxy
.put(&artifact("serde", "1.0.0"), b"pwned")
.expect("the publish itself still works");
assert_eq!(
primary
.fetch(&artifact("serde", "1.0.0"))
.unwrap()
.as_deref(),
Some(b"pwned".as_ref()),
"it really is in the local store"
);
// …and is never what a reader gets.
let served = proxy.fetch(&artifact("serde", "1.0.0")).unwrap();
assert_eq!(
served.as_deref(),
Some(real.as_ref()),
"the registry's artifact is served, not the local substitution"
);
#[cfg(feature = "testmatrix")]
fstatus(
"holger-proxy/attestation",
"local_copy_shadows_upstream_refused",
served.as_deref() == Some(real.as_ref()),
"a local publish of an upstream coordinate is not served",
);
}
/// The serve-time gate must not break the legitimate cases: a coordinate the
/// origin does not publish (a genuinely local crate) is served from the
/// primary untouched, and a correctly cached copy is served without the
/// upstream being re-probed on every request.
#[test]
fn local_only_artifacts_are_served_and_the_origin_is_probed_once_per_coordinate() {
let real = b"cached-from-upstream";
let primary = WritableStub::new("cache");
let upstream = AttestedStub::new("crates-io");
upstream.publishes("serde", "1.0.0", real);
let proxy = ProxyBackend::with_upstreams(
primary.clone(),
vec![ProxyUpstream::attested(upstream.clone())],
);
// A private crate the origin knows nothing about: served, every time.
primary
.put(&artifact("inhouse", "0.1.0"), b"our own crate")
.unwrap();
for _ in 0..5 {
assert_eq!(
proxy
.fetch(&artifact("inhouse", "0.1.0"))
.unwrap()
.as_deref(),
Some(b"our own crate".as_ref()),
"a local-only artifact is not held hostage by the gate"
);
}
let after_local = upstream.probes.load(Ordering::SeqCst);
assert!(
after_local <= 2,
"the origin is asked once, not once per read (got {after_local})"
);
// A correctly cached upstream artifact: same story.
primary.put(&artifact("serde", "1.0.0"), real).unwrap();
let before = upstream.probes.load(Ordering::SeqCst);
for _ in 0..5 {
assert_eq!(
proxy.fetch(&artifact("serde", "1.0.0")).unwrap().as_deref(),
Some(real.as_ref())
);
}
let probes = upstream.probes.load(Ordering::SeqCst) - before;
assert!(probes <= 2, "the verdict is memoized (probes: {probes})");
}
/// An upstream that cannot be REACHED is not evidence that the local copy is
/// wrong: the cache keeps serving. (A dead crates.io must not empty a
/// mirror.) Only a positive, contradicting statement from the origin refuses.
#[test]
fn an_unreachable_origin_does_not_condemn_the_local_copy() {
struct Unreachable {
name: String,
auth: traits::UpstreamAuth,
}
impl RepositoryBackendTrait for Unreachable {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
anyhow::bail!("connection refused")
}
fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
anyhow::bail!("connection refused")
}
}
impl traits::Upstream for Unreachable {
fn base_url(&self) -> &str {
"https://origin.test"
}
fn auth(&self) -> &traits::UpstreamAuth {
&self.auth
}
fn metadata(&self, _: &ArtifactId) -> anyhow::Result<Option<traits::ArtifactMeta>> {
anyhow::bail!("connection refused")
}
}
let primary = WritableStub::new("cache");
primary
.put(&artifact("serde", "1.0.0"), b"cached earlier")
.unwrap();
let proxy = ProxyBackend::with_upstreams(
primary,
vec![ProxyUpstream::attested(Arc::new(Unreachable {
name: "crates-io".into(),
auth: traits::UpstreamAuth::None,
}))],
);
assert_eq!(
proxy.fetch(&artifact("serde", "1.0.0")).unwrap().as_deref(),
Some(b"cached earlier".as_ref()),
"an unreachable origin is not a contradiction"
);
}
/// 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"
);
}
}