diff --git a/crates/pacer-daemon/src/metrics.rs b/crates/pacer-daemon/src/metrics.rs index a1e0400..cdea53a 100644 --- a/crates/pacer-daemon/src/metrics.rs +++ b/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,6 +1521,8 @@ 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, @@ -1517,7 +1530,7 @@ struct CachePathCounters { 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) -> Option { /// 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 { diff --git a/crates/pacer-daemon/src/proxy/fill.rs b/crates/pacer-daemon/src/proxy/fill.rs index 0437feb..6ab6b3f 100644 --- a/crates/pacer-daemon/src/proxy/fill.rs +++ b/crates/pacer-daemon/src/proxy/fill.rs @@ -44,7 +44,7 @@ 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; @@ -52,7 +52,7 @@ 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, /// `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, ) -> 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 { 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 { + 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) -> pacer_transport::BlobStream { diff --git a/crates/pacer-daemon/src/proxy/mod.rs b/crates/pacer-daemon/src/proxy/mod.rs index 7a9d824..1ef2c1f 100644 --- a/crates/pacer-daemon/src/proxy/mod.rs +++ b/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, /// 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 diff --git a/crates/pacer-daemon/src/proxy/place.rs b/crates/pacer-daemon/src/proxy/place.rs index 8f1ef42..307d9ad 100644 --- a/crates/pacer-daemon/src/proxy/place.rs +++ b/crates/pacer-daemon/src/proxy/place.rs @@ -252,7 +252,9 @@ impl FillCtx { // when it was not. `stage_seconds{stage="cache_read"}` plus // `stage_seconds{stage="copy"}` is the whole of `chunk_seconds` on a local hit. let read_started = std::time::Instant::now(); - let entry = self.tier.get_chunk(&chunk_key).await; + // The same local-hit check the body path makes (ADR-0049), so a chunk a restart + // recovered from before an overwrite is refused here too, not delivered. + let entry = self.cached_current(&chunk_key).await; self.metrics .delivery .stage_seconds @@ -260,7 +262,7 @@ impl FillCtx { .observe(read_started.elapsed().as_secs_f64()); // Local hit: a `memcpy`, and NO registration — the fast path never touches // `ibv_reg_mr`, which is the entire point of `ClientMemory`'s laziness. - if let Ok(Some(chunk)) = entry { + if let Some(chunk) = entry { self.metrics.cache_hits.inc(); self.metrics.bytes_from_cache.inc_by(window.len as u64); return self @@ -599,6 +601,12 @@ impl FillCtx { .fetch_blob(source, chunk_key, None, self.no_fill) .await { + // A holder's copy of another version — one it recovered from before an + // overwrite, say (ADR-0049): skip it as if it were not cached, as + // `fetch_from_peer` does. + 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(); @@ -740,7 +748,20 @@ impl FillCtx { ) .await; match delivered { - Ok(ChunkDelivery::Landed { bytes, crc32 }) => { + // The holder wrote a copy of another version (ADR-0049): not a delivery. The + // caller moves on, and whichever source answers next overwrites the window + // before any response leaves this node. + Ok(ChunkDelivery::Landed { ref e_tag, .. }) + if !self.witness_matches(e_tag.as_deref()) => + { + None + } + Ok(ChunkDelivery::Streamed(ref blob)) + if !self.witness_matches(blob.e_tag.as_deref()) => + { + None + } + Ok(ChunkDelivery::Landed { bytes, crc32, .. }) => { self.metrics.peer_fetches.inc(); self.metrics.bytes_from_peers.inc_by(bytes); self.metrics @@ -823,8 +844,19 @@ impl FillCtx { match delivered { // `crc32` is `None` by construction on this arm — the daemon registered this // window, so `run_delivery` digests it by reading it back, which proves what is + // A copy of another version (ADR-0049) is not a delivery: see the token arm. + Ok(pacer_transport::efa::ChunkDelivery::Landed { ref e_tag, .. }) + if !self.witness_matches(e_tag.as_deref()) => + { + None + } + Ok(pacer_transport::efa::ChunkDelivery::Streamed(ref blob)) + if !self.witness_matches(blob.e_tag.as_deref()) => + { + None + } // IN the client's memory rather than what a holder says it sent. - Ok(pacer_transport::efa::ChunkDelivery::Landed { bytes, crc32: _ }) => { + Ok(pacer_transport::efa::ChunkDelivery::Landed { bytes, .. }) => { self.metrics.peer_fetches.inc(); self.metrics.bytes_from_peers.inc_by(bytes); self.metrics diff --git a/crates/pacer-daemon/src/proxy/read.rs b/crates/pacer-daemon/src/proxy/read.rs index 4b1a12b..8a1230e 100644 --- a/crates/pacer-daemon/src/proxy/read.rs +++ b/crates/pacer-daemon/src/proxy/read.rs @@ -40,7 +40,7 @@ use tokio_stream::wrappers::ReceiverStream; use tracing::warn; use super::cluster::is_home; -use super::fill::{FillCtx, RequesterRead}; +use super::fill::{same_version, FillCtx, RequesterRead}; use super::PacerProxy; use crate::warm::SkipReason; @@ -48,38 +48,42 @@ use crate::warm::SkipReason; /// `get_object` because it is not a read. const OP_WARM: &str = "warm_object"; -/// Resolve the object header (length + response metadata) needed to compute -/// the covering chunk set. A cache hit returns the stored header; a miss -/// issues a backend `HeadObject` (a metadata op — no per-GB retrieval -/// charge, ADR-0015). Returns `(header, was_cached)` so the caller inserts -/// a freshly-discovered header once admission is decided. +/// `outcome` label values of `pacer_cache_revalidations_total` (ADR-0049). Consts for +/// the reason the fill path's labels are: a dashboard depends on the exact string. /// -/// A free function for the same reason [`super::cluster::chunk_sources`] is: the pre-flight -/// query needs an -/// object's length to compute the same covering chunk set the read will, and a second HEAD -/// path could disagree about which lengths are cached. Its cache read also *warms* the header -/// the GET that follows will want, so the pre-flight is not purely a cost. +/// The backend still has the version the cached header describes. +const REVALIDATION_CURRENT: &str = "current"; +/// It does not: the header was replaced by the backend's current one. +const REVALIDATION_STALE: &str = "stale"; + +/// This node's cached header for `object_key`, if the foyer tier holds one. +/// +/// A read failure is logged and treated as absent, because the caller's fallback — a +/// `HeadObject` — is what a miss costs anyway. +async fn cached_header(tier: &ChunkTier, object_key: &str) -> Option { + match tier.cache().get(object_key).await { + Ok(Some(entry)) => entry.value().as_header().cloned(), + Ok(None) => None, + Err(e) => { + warn!(key = %object_key, error = %e, "cache read failed; heading backend"); + None + } + } +} + +/// The backend's current header for `bucket`/`key`, by `HeadObject` (a metadata op — no +/// per-GB retrieval charge, ADR-0015). /// /// # Errors /// /// Maps a backend `NoSuchKey` to the same S3 error the passthrough would /// return; any other backend failure surfaces as `InternalError`. -pub(super) async fn header_for( - tier: &ChunkTier, +async fn head_backend( backend: &aws_sdk_s3::Client, object_key: &str, bucket: &str, key: &str, -) -> S3Result<(ObjectHeader, bool)> { - match tier.cache().get(object_key).await { - Ok(Some(entry)) => { - if let Some(h) = entry.value().as_header() { - return Ok((h.clone(), true)); - } - } - Ok(None) => {} - Err(e) => warn!(key = %object_key, error = %e, "cache read failed; heading backend"), - } +) -> S3Result { let head = backend .head_object() .bucket(bucket) @@ -102,7 +106,7 @@ pub(super) async fn header_for( .content_length() .and_then(|l| u64::try_from(l).ok()) .ok_or_else(|| s3_error!(InternalError, "backend HeadObject without content length"))?; - Ok((object_header_from_head(&head, object_len), false)) + Ok(object_header_from_head(&head, object_len)) } /// Map a backend `HeadObjectOutput` onto an [`ObjectHeader`], `object_len` @@ -112,7 +116,7 @@ pub(super) async fn header_for( /// otherwise report a length for a *different* version than the ETag this /// header is about to carry. /// -/// The one mapping every caller shares: [`header_for`]'s cache-miss HEAD and +/// The one mapping every caller shares: [`head_backend`]'s HEAD and /// the scatter path's inline post-Complete `HeadObject` both need it, and issue #25 is /// exactly the bug that opened up when a second call site built an /// `ObjectHeader` by hand instead of reusing this one. @@ -219,7 +223,7 @@ pub(crate) fn representation_output_fields( /// not guess for. Letting s3s decide it means the one place quoting and weakness are /// parsed is the one place upstream tests them. /// -/// Ours comes from [`header_for`], which stores it unquoted, so the comparison is a plain +/// Ours comes from [`head_backend`], which stores it unquoted, so the comparison is a plain /// string equality between two already-normalised strong values. fn if_match_allows_cache(if_match: Option<&dto::ETagCondition>, header_etag: Option<&str>) -> bool { let Some(condition) = if_match else { @@ -295,19 +299,59 @@ impl PacerProxy { } /// Resolve the object header (length + response metadata) needed to compute - /// the covering chunk set — see [`header_for`], which this delegates to so the - /// pre-flight query resolves an object's length exactly as the read does. + /// the covering chunk set. Returns `(header, was_cached)` so the caller inserts + /// a freshly-discovered header once admission is decided. + /// + /// A cached header is used as-is only once this process has confirmed its object + /// against the backend (ADR-0049). Until then it may be one the disk tier recovered + /// from a previous process, describing a version that has since been overwritten, so + /// the first read of each object costs one `HeadObject`: a matching ETag confirms the + /// cached header, anything else replaces it with the backend's. A miss costs the same + /// `HeadObject` it always did, and confirms the object too. + /// + /// Two ETags are only "the same version" when both exist. A backend that reports none + /// gives nothing to compare, so its objects are re-headed rather than trusted — and + /// their chunks, whose witness is that same absent ETag, are never served from cache + /// ([`FillCtx::witness_matches`]). /// /// # Errors /// - /// See [`header_for`]. + /// See [`head_backend`]. A confirmation that cannot reach the backend fails the GET + /// exactly as a miss would: a recovered header is never served unconfirmed. async fn header_for( &self, object_key: &str, bucket: &str, key: &str, ) -> S3Result<(ObjectHeader, bool)> { - header_for(&self.tier, &self.backend, object_key, bucket, key).await + let cached = cached_header(&self.tier, object_key).await; + if let Some(header) = &cached { + if self.revalidated.is_confirmed(object_key) { + return Ok((header.clone(), true)); + } + } + let current = head_backend(&self.backend, object_key, bucket, key).await?; + self.revalidated.confirm(object_key); + let Some(header) = cached else { + return Ok((current, false)); + }; + if same_version(header.e_tag.as_deref(), current.e_tag.as_deref()) { + self.count_revalidation(REVALIDATION_CURRENT); + return Ok((header, true)); + } + // Removed rather than left for `remember_header` to overwrite: a `no-store` read + // inserts nothing, and the stale header would then pass as confirmed. + self.tier.cache().remove(object_key); + self.count_revalidation(REVALIDATION_STALE); + Ok((current, false)) + } + + /// Count one cached header checked against the backend (ADR-0049). + fn count_revalidation(&self, outcome: &str) { + self.metrics + .revalidations + .with_label_values(&[outcome]) + .inc(); } /// Record what the header resolution found: cache a freshly-discovered header @@ -427,7 +471,7 @@ impl PacerProxy { /// # Errors /// /// `InvalidRange` for a range the object cannot satisfy, and whatever the - /// header resolution ([`header_for`]) or the first unresolvable chunk + /// header resolution ([`Self::header_for`]) or the first unresolvable chunk /// ([`FillCtx::resolve_chunk`]) fails with. pub(super) async fn serve_get( &self, @@ -524,7 +568,7 @@ impl PacerProxy { // 4. Serve the covering chunks in order through the bounded look-ahead // pipeline, filling misses per chunk. - let ctx = self.fill_ctx(object_key, bucket, key, header.object_len, decision, None); + let ctx = self.fill_ctx(object_key, bucket, key, &header, decision, None); if warm { return self.warm_range(ctx, &header, start, end).await; } @@ -638,7 +682,7 @@ impl PacerProxy { object_key, bucket, key, - header.object_len, + &header, decision, Some(RequesterRead { held, @@ -966,7 +1010,7 @@ mod tests { use super::if_match_allows_cache; use s3s::dto::{ETag, ETagCondition}; - /// Our own ETag, as [`super::header_for`] stores it: unquoted. + /// Our own ETag, as [`super::head_backend`] stores it: unquoted. const OURS: &str = "d41d8cd98f00b204e9800998ecf8427e"; /// The condition as s3s would have parsed it off the wire, so these cases exercise the @@ -985,7 +1029,7 @@ mod tests { #[test] fn the_etag_the_client_named_matches_ours() { // Sent quoted, as every S3 client sends it and RFC 9110 requires; s3s unquotes, - // and `header_for` stored ours unquoted, so the two meet already normalised. + // and `head_backend` stored ours unquoted, so the two meet already normalised. assert!(if_match_allows_cache( Some(&condition(&format!("\"{OURS}\""))), Some(OURS) diff --git a/crates/pacer-daemon/src/proxy/revalidate.rs b/crates/pacer-daemon/src/proxy/revalidate.rs new file mode 100644 index 0000000..f6fd30a --- /dev/null +++ b/crates/pacer-daemon/src/proxy/revalidate.rs @@ -0,0 +1,97 @@ +//! Which cached object headers this process has confirmed against the backend +//! (ADR-0049). +//! +//! A header can reach the cache two ways: a `HeadObject` this process issued, or the +//! disk tier recovering one a previous process wrote. Nothing in the header says which, +//! and the second kind may describe a version the backend no longer has — an overwrite +//! made while this node was down was never sent to it, and an invalidation it did +//! receive only updated an in-memory index the restart threw away (#64). So a cached +//! header is used only once its object is in this set; anything else costs one +//! `HeadObject` first. +//! +//! The set starts empty in every process, which is the whole mechanism: it is how a +//! restart forgets what it trusted without anything having to be written to disk. + +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; + +/// Objects the set holds before it starts over. +/// +/// Clearing rather than evicting one at a time: the only cost of forgetting an object +/// is one more `HeadObject` on its next read, so an exact LRU would buy nothing that a +/// reset does not, at the price of bookkeeping on every GET. 65 536 object keys of a +/// few hundred bytes each is a few tens of MiB at worst — far below the RAM tier — and +/// well above the object count of the checkpoint and model sets this cache is for, so +/// in practice the reset never happens. +const CONFIRMED_CAPACITY: usize = 1 << 16; + +/// The object keys whose cached header this process has confirmed is current. Cheap to +/// clone: every clone shares one set. +#[derive(Clone, Default)] +pub(crate) struct Revalidated { + confirmed: Arc>>, +} + +impl Revalidated { + /// Whether `object_key`'s cached header may be used without asking the backend. + pub(crate) fn is_confirmed(&self, object_key: &str) -> bool { + self.lock().contains(object_key) + } + + /// Record that the backend just reported `object_key`'s current version — by a + /// `HeadObject` this process issued, whichever way it came out. + pub(crate) fn confirm(&self, object_key: &str) { + let mut confirmed = self.lock(); + if confirmed.len() >= CONFIRMED_CAPACITY { + confirmed.clear(); + } + confirmed.insert(object_key.to_owned()); + } + + /// Lock the set. One hash operation is all any holder does with it, so a poisoned + /// lock means a panic inside `HashSet` itself and is not recoverable. + fn lock(&self) -> std::sync::MutexGuard<'_, HashSet> { + self.confirmed + .lock() + .expect("the revalidation set lock is never held across a panic-capable call") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_fresh_set_confirms_nothing() { + assert!(!Revalidated::default().is_confirmed("bucket/key")); + } + + #[test] + fn a_confirmed_key_stays_confirmed_and_others_do_not() { + let set = Revalidated::default(); + set.confirm("bucket/a"); + assert!(set.is_confirmed("bucket/a")); + assert!(!set.is_confirmed("bucket/b")); + } + + #[test] + fn clones_share_one_set() { + let set = Revalidated::default(); + set.clone().confirm("bucket/a"); + assert!(set.is_confirmed("bucket/a")); + } + + /// Full means start over, keeping the key that triggered it — never a set that + /// grows without bound against a workload that touches each key once. + #[test] + fn a_full_set_starts_over_with_the_new_key() { + let set = Revalidated::default(); + for n in 0..CONFIRMED_CAPACITY { + set.confirm(&format!("bucket/{n}")); + } + set.confirm("bucket/next"); + assert!(set.is_confirmed("bucket/next")); + assert!(!set.is_confirmed("bucket/0")); + assert_eq!(set.lock().len(), 1); + } +} diff --git a/crates/pacer-daemon/tests/common/mod.rs b/crates/pacer-daemon/tests/common/mod.rs index c1b15b3..dd8f8fb 100644 --- a/crates/pacer-daemon/tests/common/mod.rs +++ b/crates/pacer-daemon/tests/common/mod.rs @@ -56,8 +56,8 @@ mod single; // make the callers spell it out. pub use fleet::{node_parts, serve_node, serve_peer, NodeParts, NodeSpec}; pub use single::{ - daemon_core, daemon_core_over, wait_until_connection_refused, BackendPair, Daemon, DaemonCore, - DaemonSpec, + daemon_core, daemon_core_in, daemon_core_over, wait_until_connection_refused, BackendPair, + Daemon, DaemonCore, DaemonSpec, TierKind, }; use std::future::Future; diff --git a/crates/pacer-daemon/tests/common/single.rs b/crates/pacer-daemon/tests/common/single.rs index 06750e9..505c80b 100644 --- a/crates/pacer-daemon/tests/common/single.rs +++ b/crates/pacer-daemon/tests/common/single.rs @@ -13,6 +13,8 @@ use aws_sdk_s3::primitives::ByteStream; use bytes::Bytes; use pacer_backend::BackendType; use pacer_cache::chunk::ChunkConfig; +use pacer_cache::slot::SLOT_HEADER_BYTES; +use pacer_cache::store::{ChunkStore, ReadShape, StoreConfig}; use pacer_cache::tier::ChunkTier; use pacer_daemon::auth::PlaceholderAuth; use pacer_daemon::listen::ListenLimits; @@ -155,15 +157,69 @@ pub async fn daemon_core(spec: DaemonSpec) -> DaemonCore { /// If the cache cannot be brought up. pub async fn daemon_core_over(spec: DaemonSpec, backend: BackendPair) -> DaemonCore { let cache_dir = tempfile::tempdir().expect("a temp dir for the cache"); + let mut core = daemon_core_in(spec, backend, cache_dir.path(), TierKind::Foyer).await; + core.dirs.push(cache_dir); + core +} + +/// Which backend holds chunk bodies — `config.diskTier`'s two values. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TierKind { + /// foyer's hybrid cache holds chunks and headers alike. + Foyer, + /// ADR-0033's slot store holds chunks; foyer keeps the headers. + Store, +} + +/// Subdirectory of the cache dir for the slot store's extents — `main`'s layout, so a +/// reopen over the same directory finds what the previous daemon wrote where `main` +/// would look. +const CHUNK_STORE_SUBDIR: &str = "chunk-store"; + +/// Slots the store gets: room for every chunk an arm writes, with the eviction this +/// suite is not about kept out of the way. +const STORE_SLOTS: u64 = 64; + +/// [`daemon_core_over`] over a cache directory the caller owns, on either tier. +/// +/// The directory is borrowed, not moved in, because the arms this exists for close one +/// daemon and open the next over the **same** directory — what a pod restart on a +/// `hostPath` cache does. `dirs` comes back empty; the caller keeps the directory alive. +/// +/// # Panics +/// +/// If the cache or the store cannot be opened. +pub async fn daemon_core_in( + spec: DaemonSpec, + backend: BackendPair, + cache_dir: &std::path::Path, + kind: TierKind, +) -> DaemonCore { let metrics = Metrics::new().expect("a fresh registry"); let cache = if spec.foyer_metrics { - build_cache_with_metrics(cache_dir.path(), spec.cache, &metrics).await + build_cache_with_metrics(cache_dir, spec.cache, &metrics).await } else { - build_cache(cache_dir.path(), spec.cache).await + build_cache(cache_dir, spec.cache).await }; // No slab: these daemons have no RDMA plane, so cached chunks belong on the heap // (ADR-0028's default). - let tier = ChunkTier::foyer(cache, Default::default()); + let tier = match kind { + TierKind::Foyer => ChunkTier::foyer(cache, Default::default()), + TierKind::Store => { + let chunk_size = usize::try_from(spec.chunk_size).expect("a test chunk fits usize"); + let store = ChunkStore::open(StoreConfig { + dir: cache_dir.join(CHUNK_STORE_SUBDIR), + chunk_size, + capacity_bytes: STORE_SLOTS * (SLOT_HEADER_BYTES as u64 + spec.chunk_size), + verify_body: true, + read_shape: ReadShape::default(), + read_concurrency: 0, + }) + .await + .expect("opening the chunk store"); + ChunkTier::with_store(cache, store, Default::default()) + } + }; let proxy = PacerProxy::new( backend.daemon, tier.clone(), @@ -182,7 +238,7 @@ pub async fn daemon_core_over(spec: DaemonSpec, backend: BackendPair) -> DaemonC tier, backend: backend.truth, chunk_size: spec.chunk_size, - dirs: vec![cache_dir], + dirs: Vec::new(), } } diff --git a/crates/pacer-daemon/tests/daemon.rs b/crates/pacer-daemon/tests/daemon.rs index 25d04ab..397f862 100644 --- a/crates/pacer-daemon/tests/daemon.rs +++ b/crates/pacer-daemon/tests/daemon.rs @@ -85,6 +85,8 @@ mod fill_coalesce; mod get_bypass; #[path = "daemon/listener.rs"] mod listener; +#[path = "daemon/restart.rs"] +mod restart; #[path = "daemon/scatter/mod.rs"] mod scatter; #[path = "daemon/warm.rs"] diff --git a/crates/pacer-daemon/tests/daemon/restart.rs b/crates/pacer-daemon/tests/daemon/restart.rs new file mode 100644 index 0000000..c428f83 --- /dev/null +++ b/crates/pacer-daemon/tests/daemon/restart.rs @@ -0,0 +1,241 @@ +//! A daemon restarted over the cache directory it left behind (#64). +//! +//! The cache directory is a `hostPath` (`cache.hostPath` in the chart), so it outlives the +//! pod: a rolling update, an OOM kill and a crash all bring the next daemon up over the +//! previous one's files, and both disk tiers rebuild from them — the slot store from its +//! slot headers (ADR-0033), foyer from its blocks (recovery on by default, every in-memory +//! entry flushed to disk on close). Each arm here closes one daemon the way `main`'s drain +//! does, overwrites an object one way or another, opens a second daemon over the same +//! directory and the same backend, and requires the **current** bytes back. +//! +//! The overwrite always keeps the object's length. A different length would change the +//! header's `object_len` and so the covering chunks' bounds, but not their keys +//! (`{object_key}#{chunk_size}:{index}`) — same-length is the case where nothing but a +//! version check can tell the old chunks from the new ones. + +use aws_sdk_s3::primitives::ByteStream; +use bytes::Bytes; +use rstest::rstest; + +use crate::common::{ + create_test_bucket, daemon_core_in, fs_backend_service, metric_value, seeded_body, + wait_for_fills, BackendPair, CacheSpec, Daemon, DaemonSpec, TierKind, BUCKET, +}; + +/// Chunks in the object under test: more than one, so a stale read is a stale +/// *chunk set* rather than one entry, and the object clears the default admission floor +/// (4 MiB at the default 1 MiB chunk) with room to spare. +const OBJECT_CHUNKS: u64 = 6; + +/// Seed of the body every arm reads first. +const OLD_SEED: u8 = 1; +/// Seed of the overwrite — any other value; [`seeded_body`] makes the two differ at +/// every offset. +const NEW_SEED: u8 = 2; + +/// A memory tier smaller than the object, so foyer has demoted its chunks to disk +/// before the overwrite: a chunk foyer still held in memory would be removed outright +/// and never flushed, and the arm would pass without exercising recovery at all. +const SMALL_MEM_TIER: usize = 2 << 20; + +/// The key every arm reads. +const KEY: &str = "restart/overwritten.bin"; + +/// One backend and one cache directory, outliving any number of daemons over them. +struct Node { + backend: s3s::service::S3Service, + creds: aws_sdk_s3::config::Credentials, + kind: TierKind, + spec: DaemonSpec, + cache_dir: tempfile::TempDir, + _backend_dir: tempfile::TempDir, +} + +impl Node { + async fn new(kind: TierKind, cache: CacheSpec) -> Self { + let backend_dir = tempfile::tempdir().expect("a temp dir for the backend"); + let (backend, creds) = fs_backend_service(backend_dir.path()); + create_test_bucket(&BackendPair::over(&backend, &creds).truth).await; + Self { + backend, + creds, + kind, + spec: DaemonSpec { + cache, + ..DaemonSpec::default() + }, + cache_dir: tempfile::tempdir().expect("a temp dir for the cache"), + _backend_dir: backend_dir, + } + } + + /// Start a daemon over this node's directory — a fresh one the first time, a + /// restart every time after. + async fn start(&self) -> Daemon { + daemon_core_in( + self.spec, + BackendPair::over(&self.backend, &self.creds), + self.cache_dir.path(), + self.kind, + ) + .await + .in_process() + } + + fn object_len(&self) -> usize { + usize::try_from(self.spec.chunk_size * OBJECT_CHUNKS).expect("a test object fits") + } +} + +/// Stop `daemon` the way `main`'s drain does: close the tier, which flushes foyer's +/// memory tier to disk. +async fn stop(daemon: Daemon) { + daemon + .tier + .cache() + .close() + .await + .expect("closing the cache tier"); +} + +/// Write `body` under [`KEY`] with `client` — the daemon's for a write through PACER, +/// the backend's for one around it. +async fn put(client: &aws_sdk_s3::Client, body: &Bytes) { + client + .put_object() + .bucket(BUCKET) + .key(KEY) + .body(ByteStream::from(body.clone())) + .send() + .await + .expect("overwriting the object"); +} + +/// Read [`KEY`] through `daemon` and fail unless the bytes are `want`, naming which +/// version came back instead — "the old body" is the bug, anything else is a different one. +async fn assert_serves(daemon: &Daemon, want: &Bytes, old: &Bytes, when: &str) { + let got = daemon.get(KEY).await; + assert!( + got == *want, + "{when}: the daemon served {} bytes that are {}", + got.len(), + if got == *old { + "the OLD body — the pre-overwrite version came back" + } else { + "neither version" + } + ); +} + +/// Read the object once so every chunk is filled, and wait until the fills land. +async fn warm(daemon: &Daemon, old: &Bytes) { + assert_serves(daemon, old, old, "the first read").await; + wait_for_fills(&daemon.metrics, OBJECT_CHUNKS).await; +} + +/// The issue's case: the object is overwritten while this node's daemon is down, so no +/// invalidation can reach it. The restarted daemon must not serve what it cached before. +#[rstest] +#[case::foyer(TierKind::Foyer)] +#[case::store(TierKind::Store)] +#[tokio::test(flavor = "multi_thread")] +async fn an_overwrite_while_down_is_served_after_restart(#[case] kind: TierKind) { + let node = Node::new(kind, CacheSpec::default()).await; + let old = seeded_body(OLD_SEED, node.object_len()); + let new = seeded_body(NEW_SEED, node.object_len()); + + let first = node.start().await; + put(&first.backend, &old).await; + warm(&first, &old).await; + stop(first).await; + + // Around PACER, with no daemon running: nothing is told. + let truth = BackendPair::over(&node.backend, &node.creds).truth; + put(&truth, &new).await; + + let second = node.start().await; + assert_serves(&second, &new, &old, "after restart").await; +} + +/// A write through PACER invalidates this node before it stops — and the restart must +/// not undo that invalidation by recovering the removed entries from disk. +/// +/// `read_before_stop` is the confound this arm exists to keep visible. A read between the +/// write and the stop refills every chunk, and the store hands a fill the slot freed most +/// recently — the one its stale copy just left — so the refill overwrites the old slot +/// header on disk and there is nothing left to recover. Without that read the invalidated +/// slots are still on disk at the restart, which is the case a write followed by a +/// rolling update produces. +#[rstest] +#[tokio::test(flavor = "multi_thread")] +async fn an_invalidation_before_restart_survives_it( + #[values( + (TierKind::Foyer, CacheSpec { mem_capacity: SMALL_MEM_TIER, ..CacheSpec::default() }), + (TierKind::Store, CacheSpec::default()) + )] + tier: (TierKind, CacheSpec), + #[values(true, false)] read_before_stop: bool, +) { + let node = Node::new(tier.0, tier.1).await; + let old = seeded_body(OLD_SEED, node.object_len()); + let new = seeded_body(NEW_SEED, node.object_len()); + + let first = node.start().await; + put(&first.backend, &old).await; + warm(&first, &old).await; + put(&first.client, &new).await; + if read_before_stop { + assert_serves(&first, &new, &old, "after the write, before restart").await; + } + stop(first).await; + + let second = node.start().await; + assert_serves(&second, &new, &old, "after restart").await; +} + +/// What the check above must not cost: a restart over an **unchanged** object serves +/// every chunk from the tier it recovered, after exactly one `HeadObject` to confirm the +/// version (ADR-0049). Without this arm, the two above would also pass on a daemon that +/// simply never trusted its disk tier again — correct, and a full re-read of every +/// checkpoint after each rolling update. +#[rstest] +#[case::foyer(TierKind::Foyer)] +#[case::store(TierKind::Store)] +#[tokio::test(flavor = "multi_thread")] +async fn an_unchanged_object_is_served_from_the_recovered_tier(#[case] kind: TierKind) { + let node = Node::new(kind, CacheSpec::default()).await; + let old = seeded_body(OLD_SEED, node.object_len()); + + let first = node.start().await; + put(&first.backend, &old).await; + warm(&first, &old).await; + stop(first).await; + + let second = node.start().await; + assert_serves(&second, &old, &old, "after restart").await; + let text = second.scrape(); + assert_eq!( + metric_value( + &text, + "pacer_cache_revalidations_total", + &[("outcome", "current")] + ), + 1.0, + "the recovered header is confirmed once, by one HeadObject" + ); + assert_eq!( + second.metrics.fills_completed.get(), + 0, + "every chunk came from the recovered tier: nothing was re-read and re-filled" + ); + assert_eq!( + second.metrics.stale_chunks.get(), + 0, + "no recovered chunk was refused" + ); + assert_eq!( + second.metrics.cache_hits.get(), + OBJECT_CHUNKS, + "one hit per chunk" + ); +} diff --git a/crates/pacer-transport/src/efa/target.rs b/crates/pacer-transport/src/efa/target.rs index 92746e8..19f05c4 100644 --- a/crates/pacer-transport/src/efa/target.rs +++ b/crates/pacer-transport/src/efa/target.rs @@ -140,6 +140,11 @@ pub enum ChunkDelivery { /// back, so the digest has to come from the side that held the bytes /// (ADR-0030 point 7). crc32: Option, + /// The holder's version witness for the chunk it wrote — the ETag its cached copy + /// was filled under. The requester compares it with the version its read resolved + /// (ADR-0049) and treats a mismatch as no delivery, so the window is written again + /// from a current source before any response leaves this node. + e_tag: Option, }, /// The holder streamed the body instead (it had no cached AH for this /// requester, the body did not fit the offered window, or its own WRITE @@ -300,6 +305,7 @@ impl EfaRdmaTransport { // The daemon registered this window, so `run_delivery` digests it by reading it // back — a stronger check than anything the holder could report. crc32: None, + e_tag: meta.e_tag, }) } @@ -404,6 +410,7 @@ impl EfaRdmaTransport { Ok(ChunkDelivery::Landed { bytes: meta.total_len, crc32: meta.written_crc32, + e_tag: meta.e_tag, }) } } diff --git a/docs/adr/0049-revalidate-recovered-cache-after-restart.md b/docs/adr/0049-revalidate-recovered-cache-after-restart.md new file mode 100644 index 0000000..9450568 --- /dev/null +++ b/docs/adr/0049-revalidate-recovered-cache-after-restart.md @@ -0,0 +1,78 @@ +# ADR-0049: A restarted daemon confirms each object's version before serving it from cache + +Date: 2026-10-05 · Status: **Accepted.** Fixes +[#64](https://github.com/aws-samples/sample-pacer/issues/64). Extends the version witness +[0041](0041-requester-identity-auth-mode.md) introduced for `auth.mode: requester` to +`auth.mode: node`. + +## Context + +The cache directory is a `hostPath` on the node (`cache.hostPath`), so it outlives the daemon's +pod. A rolling update, an OOM kill or a crash brings the next daemon up over the previous one's +files, and both disk tiers rebuild from them: the chunk store ([0033](0033-chunk-store-owns-the-disk-tier.md)) by +scanning its slot headers, and foyer by its own recovery, which is on by default and which a +graceful shutdown feeds by flushing every in-memory entry to disk. + +Nothing recovered that way says whether it is still current. Two cases serve the wrong bytes, +and a test that restarts a daemon over its own directory reproduces both on both tiers +(`crates/pacer-daemon/tests/daemon/restart.rs`): + +1. **An overwrite while the node was down.** A node that is not in the ring is sent no + invalidation ([0007](0007-write-through-read-after-write.md)), so when it comes back it serves the + chunks and the header it had before the write. +2. **An invalidation the node did receive.** `ChunkStore::remove` updates only the in-memory + index; the slot on disk keeps its header until a later write reuses it. foyer keeps no + tombstone for a removed entry unless its tombstone log is enabled, and it is not. A restart + therefore recovers entries an invalidation had removed. With foyer the result can be a mix: + chunks still in memory at the invalidation are gone, chunks already demoted to disk come + back. + +In node mode a cached chunk carries no version witness at all ([0015](0015-chunk-granular-caching.md) +left it unchecked), so there is nothing to compare a recovered chunk against. + +## Decision + +**A cached object header is used only once this process has confirmed it against the backend, +and a cached chunk is served only under the ETag of the header the read resolved — in both auth +modes.** + +1. **Headers are confirmed once per object per process.** The proxy keeps the set of object keys + whose current version it has seen from the backend. It starts empty, so after a restart the + first read of each object sends one `HeadObject` even when a header is cached. A matching + ETag confirms the cached header; anything else replaces it with the backend's and removes the + cached one. A cache miss sends the same `HeadObject` it always did and confirms the object + too. Steady state is unchanged: one `HeadObject` per object per process, not per read. +2. **The set is bounded and starts over when full** (65 536 objects). Forgetting an object costs + one more `HeadObject` on its next read, so an exact LRU would buy nothing a reset does not. +3. **Every node-mode fill records the version it filled.** The proxy's own fills are tagged with + the ETag of the header the read resolved; a peer's read-through is tagged with the ETag of the + ranged GET response it filled from. The write paths already recorded one. +4. **A chunk is served only when its witness equals the read's ETag**, locally and from a peer, + on the body path and on the delivery path ([0026](0026-client-supplied-target-memory.md)), + including a holder's one-sided write into client memory, whose response now carries the + holder's witness. A mismatched local copy is **forgotten**, not just skipped, because the chunk + store's `put` ignores a key it already holds and the re-fill would otherwise never replace it. + A mismatched peer copy is skipped and the holder is sent an `Invalidate` for that key, so a + stale copy at a home does not stay uncached cluster-wide. +5. **A confirmation that cannot reach the backend fails the read**, as a miss already does. A + recovered header is never served unconfirmed. +6. **No ETag, no cache.** An object whose backend reports no ETag has nothing to compare, so its + header is re-checked on every read and its chunks are never served from cache. S3 always + returns an ETag. + +## Consequences + +- **A restart keeps its warm cache.** Over an unchanged object, the restarted daemon sends one + `HeadObject` and serves every chunk from the recovered tier; nothing is re-read. +- **Invalidation stays an in-memory operation.** No write is added to the invalidation path; the + witness check makes a resurrected entry a miss instead of preventing the resurrection. +- **Chunks filled before this change have no witness**, so the first read of each after the + upgrade re-reads it from the backend. A rolling upgrade also re-reads chunks that peers still + on the previous version serve, since those carry no witness either. +- **Two new series:** `pacer_cache_revalidations_total{outcome="current"|"stale"}` counts cached + headers checked against the backend, and `pacer_cache_stale_chunks_total` counts chunks dropped + for the wrong witness. After a restart the first climbs by one per object read, then stops. +- **Not covered:** an object overwritten around PACER while the node stays up is still served from + cache until it is evicted, as before. Bounding that is + [#72](https://github.com/aws-samples/sample-pacer/issues/72), which can reuse this mechanism by + letting a confirmation expire. diff --git a/docs/adr/README.md b/docs/adr/README.md index 4553686..d62f0cd 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -72,3 +72,4 @@ one variable, interleaved repetitions, raw client report and the script alongsid | [0046](0046-fence-reads-that-claim-only-to-insert.md) | **Accepted. Claims a defect removed, not a rate.** Amends [0044](0044-poison-in-flight-fills-on-invalidate.md), whose fence covered only the coalescing leader: `fetch_from_backend` (coalescing off, `Busy`, a follower's fallback), the layer-1 admit and the peer read-through all read first and claim only to insert, so an `Invalidate` during their read had no claim to poison and the pre-write bytes were cached. Each now takes a `ReadTicket` before its read; `poison` bumps the key's epoch, and the insert refuses a stale ticket as it does a poisoned claim. A read begun after the invalidation still fills. Fixes [#55](https://github.com/aws-samples/sample-pacer/issues/55) | 2026-09-29 | | [0047](0047-cachering-operator-renders-the-chart.md) | **Accepted.** A namespaced `CacheRing` (`pacer.io/v1alpha1`) declares one ring, and `pacer-operator` reconciles it by running the shipped chart through `helm template` (the chart reads no cluster state, so that render is exactly what an install of the same values would produce) rather than re-implementing its derivations and refusals. `spec.values` are the chart values; objects are server-side applied and owned by the ring, re-applied on every pass so drift is undone, and pruned against `status.inventory` only when they still name the ring as owner; a failed render applies nothing and leaves the last good objects running. Status reports rollout counts and **members** by the daemons' own membership rule, plus `Rendered`/`Applied`/`Ready`. Trust boundary: creating a ring is equivalent to running its node-level pods, Pod Security admission still applies, and the operator refuses objects outside the ring's namespace, outside a kind allowlist, or binding a ClusterRole. Karpenter node provisioning is refused and deferred to an administrator-owned class | 2026-09-29 | | [0048](0048-warm-only-get.md) | A GET carrying `x-pacer-warm: 1` runs the read path through admission and resolves every covering chunk, then answers header-only (`x-pacer-warmed: `) instead of streaming a body — reusing ADR-0026's header-only shape and, in `auth.mode: requester`, its own authorization probe. Synchronous by design (ADR-0011's reasoning: a fill nobody waits on has no back-pressure), so a large object is the caller's job to slice, which `pacer-daemon warm` does. Skips [0016](0016-multi-copy-replication.md) layer-1 admission — a warm must not leave a copy on the node that ran it. A daemon that predates this degrades to streaming the body, warming the object all the same. In `requester` mode the command forwards inside TLS to the daemon's listener, never through a `CONNECT` tunnel. Fills one of each chunk's R homes; filling all of them is #38 | 2026-09-29 | +| [0049](0049-revalidate-recovered-cache-after-restart.md) | A cached object header is used only once this process has confirmed it against the backend: the first read of each object after a restart sends one `HeadObject`, and a mismatched header is replaced. Every `node`-mode fill now records the version it filled, and a chunk is served — locally, from a peer, or by a holder's write into client memory — only under the ETag the read resolved; a mismatched local copy is forgotten, a mismatched peer copy is skipped and its holder told. Fixes a restart serving bytes from before an overwrite made while the node was down, and entries an invalidation had removed coming back from disk. A restart over unchanged objects still serves from the recovered tier | 2026-10-05 |