From 796f3f830aa7dfa4036c6eb47993054a83e6ae7d Mon Sep 17 00:00:00 2001 From: Alejandro Andraca Date: Mon, 5 Oct 2026 11:59:30 +0200 Subject: [PATCH] Confirm a recovered cache entry's version before serving it (ADR-0049) The cache directory is a hostPath, so a restarted daemon rebuilds both disk tiers from the previous process's files: the chunk store from its slot headers, foyer from its blocks. Neither knows whether what it recovered is still current. An object overwritten while the node was down came back as its old bytes, and so did one overwritten through PACER just before a restart: ChunkStore::remove only updates the in-memory index, and foyer keeps no tombstone for a removed entry. A cached header is now used only once this process has confirmed its object against the backend, so the first read of each object after a restart costs one HeadObject. Every node-mode fill records the version it filled, and a chunk is served only under the ETag the read resolved: locally, from a peer, and by a holder's one-sided write into client memory, whose response now carries the holder's witness. A mismatched local copy is forgotten so its re-fill can land; a mismatched peer copy is skipped and its holder sent an Invalidate. ETags are compared without quotes, because the scatter path tags chunks with CompleteMultipartUpload's quoted one. tests/daemon/restart.rs restarts a daemon over its own cache directory on both tiers: an overwrite while down, an invalidation before the restart, and an unchanged object, which must still be served from the recovered tier after one HeadObject and no re-fill. Fixes #64 Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/pacer-daemon/src/metrics.rs | 27 +- crates/pacer-daemon/src/peer.rs | 12 +- crates/pacer-daemon/src/proxy/fill.rs | 141 +++++++--- crates/pacer-daemon/src/proxy/mod.rs | 7 + crates/pacer-daemon/src/proxy/place.rs | 40 ++- crates/pacer-daemon/src/proxy/read.rs | 114 ++++++--- crates/pacer-daemon/src/proxy/revalidate.rs | 97 +++++++ crates/pacer-daemon/tests/common/mod.rs | 4 +- crates/pacer-daemon/tests/common/single.rs | 64 ++++- crates/pacer-daemon/tests/daemon.rs | 2 + crates/pacer-daemon/tests/daemon/restart.rs | 241 ++++++++++++++++++ crates/pacer-transport/src/efa/target.rs | 7 + ...evalidate-recovered-cache-after-restart.md | 78 ++++++ docs/adr/README.md | 1 + 14 files changed, 757 insertions(+), 78 deletions(-) create mode 100644 crates/pacer-daemon/src/proxy/revalidate.rs create mode 100644 crates/pacer-daemon/tests/daemon/restart.rs create mode 100644 docs/adr/0049-revalidate-recovered-cache-after-restart.md 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 |