Confirm a recovered cache entry's version before serving it (ADR-0049)

crates/pacer-daemon/src/metrics.rs

@@ -250,6 +250,17 @@ pub struct Metrics {
     /// tomorrow. This counter is the only series that says the conditional itself was
     /// honoured, which is what the Mountpoint arm's unexplained 0 % hit rate needed.
     pub conditional_get_served: IntCounter,
+    /// Cached object headers checked against the backend before their first use by this
+    /// process (ADR-0049), by `outcome`: `current` (the backend still has this version)
+    /// or `stale` (it does not, and the header was replaced).
+    ///
+    /// The one series that shows a restart's recovered cache being re-trusted rather
+    /// than served blind: after a restart it climbs by one per object read, then stops.
+    pub revalidations: IntCounterVec,
+    /// Cached chunks dropped because their version witness did not match the object
+    /// version the read resolved (ADR-0049) — a chunk a restart recovered from before an
+    /// overwrite, or one filled before ADR-0049 with no witness at all.
+    pub stale_chunks: IntCounter,
     /// Whole-object cache fills that completed.
     pub fills_completed: IntCounter,
     /// Cache fills abandoned (client disconnect or short read).
@@ -1510,14 +1521,16 @@ struct CachePathCounters {
     misses: IntCounter,
     bypass: IntCounter,
     conditional_get_served: IntCounter,
+    revalidations: IntCounterVec,
+    stale_chunks: IntCounter,
     fills_completed: IntCounter,
     fills_aborted: IntCounter,
     bytes_from_cache: IntCounter,
     bytes_filled: IntCounter,
     writes_refused: IntCounter,
 }
 
-/// Register the nine counters of [`CachePathCounters`].
+/// Register the eleven series of [`CachePathCounters`].
 ///
 /// # Errors
 ///
@@ -1535,6 +1548,16 @@ fn register_cache_path_counters(registry: &Registry) -> anyhow::Result<CachePath
             "pacer_conditional_get_served_total",
             "If-Match GETs whose ETag matched, so the cache path was taken (ADR-0039)",
         )?,
+        revalidations: int_counter_vec(
+            registry,
+            "pacer_cache_revalidations_total",
+            "Cached object headers checked against the backend before this process first used them, by outcome current|stale (ADR-0049)",
+            &["outcome"],
+        )?,
+        stale_chunks: c(
+            "pacer_cache_stale_chunks_total",
+            "Cached chunks dropped because their version witness did not match the object version read (ADR-0049)",
+        )?,
         fills_completed: c("pacer_fills_completed_total", "Whole-object cache fills")?,
         fills_aborted: c(
             "pacer_fills_aborted_total",
@@ -2358,6 +2381,8 @@ impl Metrics {
             cache_misses: cache_path.misses,
             cache_bypass: cache_path.bypass,
             conditional_get_served: cache_path.conditional_get_served,
+            revalidations: cache_path.revalidations,
+            stale_chunks: cache_path.stale_chunks,
             fills_completed: cache_path.fills_completed,
             fills_aborted: cache_path.fills_aborted,
             bytes_from_cache: cache_path.bytes_from_cache,

crates/pacer-daemon/src/peer.rs

@@ -901,6 +901,10 @@ fn content_range_total(content_range: Option<&str>) -> Option<u64> {
 /// for). Without a fill, a closed channel ends the pump.
 async fn pump_read_through(mut st: PumpState) {
     let mut sent: u64 = 0;
+    // The fill's version witness (ADR-0049): the ETag of the very response these bytes
+    // are, which no header lookup could improve on. Taken before `meta` moves into the
+    // first frame.
+    let e_tag = st.meta.e_tag.clone();
     let mut first = Some(st.meta);
     let mut failed = false;
     loop {
@@ -949,10 +953,12 @@ async fn pump_read_through(mut st: PumpState) {
         // `admit` bool above), so this holds the claim this fill's insert needs
         // fenced against a write's `Invalidate` racing it (gh22, ADR-0044) —
         // see `insert_fenced`.
+        let cached = match e_tag {
+            Some(e_tag) => CachedChunk::versioned(body, e_tag),
+            None => CachedChunk::new(body),
+        };
         let inserted = match st.guard.as_ref() {
-            Some(guard) => {
-                insert_fenced(&st.tier, &st.cache_key, CachedChunk::new(body), guard).await
-            }
+            Some(guard) => insert_fenced(&st.tier, &st.cache_key, cached, guard).await,
             None => false,
         };
         if inserted {

crates/pacer-daemon/src/proxy/fill.rs

@@ -44,15 +44,15 @@ use std::sync::{Arc, Mutex};
 use bytes::{Bytes, BytesMut};
 use futures::StreamExt;
 use pacer_backend::retry::{BackendReadError, BackendReadErrorKind, ChunkRead, RetryPolicy};
-use pacer_cache::chunk::{CachedChunk, ChunkConfig};
+use pacer_cache::chunk::{CachedChunk, ChunkConfig, ObjectHeader};
 use pacer_cache::tier::ChunkTier;
 use pacer_cache::ReadDecision;
 use pacer_ring::directory::Tier;
 use pacer_transport::TransportError;
 use prometheus::{IntCounter, IntGauge};
 use s3s::{s3_error, S3Result};
 use tokio::sync::broadcast;
-use tracing::{trace, warn};
+use tracing::{debug, trace, warn};
 
 use crate::cachefill::ChunkFill;
 use crate::metrics::Metrics;
@@ -119,6 +119,11 @@ pub(super) struct FillCtx {
     pub(super) bucket: String,
     pub(super) key: String,
     pub(super) object_len: u64,
+    /// The object version this read serves: the ETag of the header it resolved — one
+    /// this process confirmed against the backend (ADR-0049) or, in `requester` mode,
+    /// the one this GET's probe just reported. Every chunk this read fills is tagged
+    /// with it, and a cached chunk is served only under it ([`Self::witness_matches`]).
+    pub(super) e_tag: Option<String>,
     /// `true` when a completed fetch of a missed chunk should be inserted
     /// (`CacheAndFill` + size-admitted); `false` bypasses the fill (`no-store`
     /// or a below-`min`/above-`max` object).
@@ -166,12 +171,16 @@ impl PacerProxy {
     ///
     /// `warm` comes back `false`; the warm path sets it with struct-update syntax
     /// rather than as an eighth argument here.
+    ///
+    /// The object's length and version both come from `header`, rather than as two
+    /// arguments, so a read cannot be built against one version's length and another's
+    /// ETag.
     pub(super) fn fill_ctx(
         &self,
         object_key: String,
         bucket: String,
         key: String,
-        object_len: u64,
+        header: &ObjectHeader,
         decision: ReadDecision,
         requester: Option<RequesterRead>,
     ) -> FillCtx {
@@ -188,7 +197,8 @@ impl PacerProxy {
             object_key,
             bucket,
             key,
-            object_len,
+            object_len: header.object_len,
+            e_tag: header.e_tag.clone(),
             admit: decision == ReadDecision::CacheAndFill,
             no_fill: decision == ReadDecision::CacheNoFill,
             #[cfg(feature = "efa")]
@@ -727,6 +737,19 @@ pub(crate) async fn insert_fenced(
 /// Poison **before** forget: a fill whose insert lands between the two is
 /// undone by `insert_fenced`'s after-check; one that finished and released its
 /// claim before this call is covered by the forget.
+/// Whether two ETags name the same object version (ADR-0049): both present, and equal
+/// once the surrounding quotes are dropped.
+///
+/// Absent on either side is never "the same" — there is nothing to compare. And quotes
+/// are dropped because each source spells the ETag its own way: `head_backend` stores it
+/// unquoted, `CompleteMultipartUpload` returns it quoted, and both end up as witnesses.
+pub(super) fn same_version(a: Option<&str>, b: Option<&str>) -> bool {
+    match (a, b) {
+        (Some(a), Some(b)) => a.trim_matches('"') == b.trim_matches('"'),
+        _ => false,
+    }
+}
+
 pub(crate) async fn forget_fenced(filling: &FillRegistry, tier: &ChunkTier, key: &str) {
     filling.poison(key);
     tier.forget(key).await;
@@ -755,14 +778,10 @@ impl FillCtx {
     /// fall back to the backend and are never client-visible.
     pub(super) async fn resolve_chunk(&self, idx: u64) -> S3Result<Bytes> {
         let chunk_key = self.chunk.chunk_key(&self.object_key, idx);
-        if let Ok(Some(c)) = self.tier.get_chunk(&chunk_key).await {
-            // A stale entry in requester mode is a miss: resolved below, and a home's
-            // fill overwrites it under the current witness.
-            if self.witness_matches(c.e_tag.as_deref()) {
-                self.metrics.cache_hits.inc();
-                self.metrics.bytes_from_cache.inc_by(c.body.len() as u64);
-                return Ok(c.body);
-            }
+        if let Some(c) = self.cached_current(&chunk_key).await {
+            self.metrics.cache_hits.inc();
+            self.metrics.bytes_from_cache.inc_by(c.body.len() as u64);
+            return Ok(c.body);
         }
         // A chunk this node co-homes (or single-node) is read through the
         // backend and filled here (ADR-0016 layer 2: all R homes fill).
@@ -988,9 +1007,11 @@ impl FillCtx {
                 .fetch_blob(source, chunk_key, None, self.no_fill)
                 .await
             {
-                // A holder's copy of an older version (requester mode's witness
-                // check): skip it as if it were not cached, never serve it.
-                Ok(blob) if !self.witness_matches(blob.e_tag.as_deref()) => {}
+                // A holder's copy of another version: skip it as if it were not cached,
+                // never serve it — and tell the holder, which cannot know (ADR-0049).
+                Ok(blob) if !self.witness_matches(blob.e_tag.as_deref()) => {
+                    self.spawn_forget_at(source, chunk_key);
+                }
                 Ok(blob) => match collect_blob(blob).await {
                     Ok(bytes) => {
                         self.metrics.peer_fetches.inc();
@@ -1218,29 +1239,80 @@ impl FillCtx {
         self.fill.cached_bytes(data, &self.metrics)
     }
 
-    /// The cache entry for `data`. In `requester` mode it carries the probe's ETag
-    /// as its version witness, which [`Self::witness_matches`] checks on every
-    /// later read (ADR-0041, planning/30 § 3.4); node mode is unchanged.
+    /// The cache entry for `data`, carrying this read's version ([`Self::e_tag`]) as its
+    /// witness, which [`Self::witness_matches`] checks on every later read — in both
+    /// auth modes (ADR-0041 for `requester`, ADR-0049 for `node`).
     fn to_cached(&self, data: &Bytes) -> CachedChunk {
         let body = self.cached_bytes(data);
-        match self.requester.as_ref().and_then(|r| r.e_tag.clone()) {
+        match self.e_tag.clone() {
             Some(e_tag) => CachedChunk::versioned(body, e_tag),
             None => CachedChunk::new(body),
         }
     }
 
-    /// Whether a cached chunk whose witness is `e_tag` may be served to this read.
+    /// Whether a cached chunk whose witness is `e_tag` may be served to this read: only
+    /// when it is the version this read resolved ([`Self::e_tag`]).
     ///
-    /// Always, in node mode (ADR-0015's contract, unchanged). In `requester` mode
-    /// only when it matches the ETag this GET's probe just reported: a chunk filled
-    /// under an older version — overwritten through this proxy or around it — is a
-    /// miss and is re-read, never served. A probe with no ETag serves nothing from
-    /// the cache, because there is nothing to compare against.
-    fn witness_matches(&self, e_tag: Option<&str>) -> bool {
-        match &self.requester {
-            None => true,
-            Some(r) => r.e_tag.is_some() && r.e_tag.as_deref() == e_tag,
+    /// A chunk filled under an older version — overwritten through this proxy, around it,
+    /// or while this node was down and then recovered from disk by a restart (#64) — is a
+    /// miss and is re-read, never served. So is a chunk with no witness at all, which is
+    /// what every `node`-mode fill wrote before ADR-0049. A read whose own version has no
+    /// ETag serves nothing from the cache, because there is nothing to compare against.
+    ///
+    /// Compared by [`same_version`], which ignores the quotes the scatter path's witnesses
+    /// carry and the header's ETag does not.
+    pub(super) fn witness_matches(&self, e_tag: Option<&str>) -> bool {
+        same_version(self.e_tag.as_deref(), e_tag)
+    }
+
+    /// Ask `holder` to drop its copy of `chunk_key`, which just answered this read under
+    /// another version's witness (ADR-0049). Fire-and-forget, like an ADR-0017 announce.
+    ///
+    /// Without this a stale copy at a home outlives the read that found it: every
+    /// requester skips it and reads the backend, while the home — which still "holds" the
+    /// key — never reads it through again, so the chunk is uncached cluster-wide until the
+    /// home's own clients happen to read it or LRU reaches it. A restart that recovered
+    /// a pre-overwrite copy is the case that produces one.
+    ///
+    /// Safe if this read is the one that is behind: the holder forgets a current copy and
+    /// refills it on the next read through, which costs one backend read and serves
+    /// nothing wrong.
+    pub(super) fn spawn_forget_at(&self, holder: &pacer_ring::NodeId, chunk_key: &str) {
+        let Some(cluster) = &self.cluster else {
+            return;
+        };
+        let transport = Arc::clone(&cluster.transport);
+        let (holder, chunk_key) = (holder.clone(), chunk_key.to_owned());
+        tokio::spawn(async move {
+            if let Err(e) = transport.invalidate(&holder, &chunk_key).await {
+                debug!(key = %chunk_key, holder = %holder.name(), error = %e,
+                    "could not tell a holder its copy is another version; it serves a miss until evicted");
+            }
+        });
+    }
+
+    /// This node's cached copy of `chunk_key`, if it holds one of the version this read
+    /// serves — the one local-hit check, shared by the body path ([`Self::resolve_chunk`])
+    /// and the delivery path (`deliver_window`) so neither can serve what the other refuses.
+    ///
+    /// A copy under any other witness is **forgotten** here, not just skipped: the slot
+    /// store's `put` is a no-op for a key it already holds (chunks are immutable within
+    /// one version), so a re-fill under the new witness would otherwise never replace it,
+    /// and every later read would miss on it until LRU got there. A plain forget rather
+    /// than a fenced one: there is no fill of *this* version to fence — a fill racing it
+    /// is at worst forgotten too, and costs one more miss.
+    ///
+    /// A tier read error is a miss as well — the caller's fallback is the same.
+    pub(super) async fn cached_current(&self, chunk_key: &str) -> Option<CachedChunk> {
+        let cached = self.tier.get_chunk(chunk_key).await.ok().flatten()?;
+        if self.witness_matches(cached.e_tag.as_deref()) {
+            return Some(cached);
         }
+        self.tier.forget(chunk_key).await;
+        self.metrics.stale_chunks.inc();
+        trace!(key = %chunk_key, witness = ?cached.e_tag, current = ?self.e_tag,
+            "cached chunk is not the version this read serves; forgotten");
+        None
     }
 
     /// Insert a freshly-fetched chunk into the cache, guarded so only one fill
@@ -1333,6 +1405,17 @@ pub(super) async fn collect_blob(blob: pacer_transport::BlobStream) -> anyhow::R
 mod tests {
     use super::*;
 
+    /// ADR-0049's comparison: quotes are spelling, absence is never a match.
+    #[test]
+    fn same_version_ignores_quotes_and_never_matches_an_absent_etag() {
+        assert!(same_version(Some("abc-3"), Some("\"abc-3\"")));
+        assert!(same_version(Some("\"abc-3\""), Some("abc-3")));
+        assert!(!same_version(Some("abc-3"), Some("abc-4")));
+        assert!(!same_version(None, Some("abc-3")));
+        assert!(!same_version(Some("abc-3"), None));
+        assert!(!same_version(None, None));
+    }
+
     /// Build a [`pacer_transport::BlobStream`] whose body is `parts` in order,
     /// with `len` = the total byte count (as the transport sets it).
     fn blob_stream(parts: Vec<Bytes>) -> pacer_transport::BlobStream {

crates/pacer-daemon/src/proxy/mod.rs

@@ -22,6 +22,8 @@
 //! * `place` — one window's bytes into that memory, by copy or one-sided WRITE
 //!   (ADR-0026 point 4, ADR-0018, ADR-0030).
 //! * `warm` — the warm-only GET: every covering chunk resolved, no body (ADR-0048).
+//! * `revalidate` — which cached headers this process has confirmed against the
+//!   backend, so a restart re-checks what it recovered (ADR-0049).
 //!
 //! Everything a *passthrough* op does is here in full, because there is nothing to
 //! it: count the op, resolve the bucket alias (ADR-0002), forward to `inner`.
@@ -45,6 +47,7 @@ mod deliver;
 mod fill;
 mod place;
 mod read;
+mod revalidate;
 mod target;
 mod warm;
 mod write;
@@ -95,6 +98,9 @@ pub struct PacerProxy {
     /// Chunk keys with a fill in flight, and what each one will hand a second
     /// arrival (ADR-0040). Shared with the peer server (one fill per key node-wide).
     filling: FillRegistry,
+    /// Objects whose cached header this process has confirmed against the backend
+    /// (ADR-0049). Empty at startup, so every header a restart recovered is checked once.
+    revalidated: revalidate::Revalidated,
     /// Cluster tier (Phase 2): ring + peer transport. None = single-node.
     cluster: Option<Cluster>,
     /// Backend shape (ADR-0023). Gates the Express-only request normalization
@@ -164,6 +170,7 @@ impl PacerProxy {
             // (ADR-0002): an empty map resolves every bucket to itself.
             bucket_map: HashMap::new(),
             filling: FillRegistry::new(),
+            revalidated: revalidate::Revalidated::default(),
             cluster: None,
             // Defaults to Express (ADR-0002/0023) so callers that don't set a
             // backend shape keep the historical behavior; main.rs overrides it