diff --git a/Cargo.lock b/Cargo.lock index 62ea8cb74..b2722a925 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9570,6 +9570,7 @@ version = "1.0.0-beta.8" dependencies = [ "bytes", "metrics", + "metrics-util", "moka", "starshard", "sysinfo", diff --git a/crates/object-data-cache/Cargo.toml b/crates/object-data-cache/Cargo.toml index 2e9b57502..f8ee22887 100644 --- a/crates/object-data-cache/Cargo.toml +++ b/crates/object-data-cache/Cargo.toml @@ -34,10 +34,11 @@ moka = { workspace = true, features = ["future"] } starshard.workspace = true sysinfo = { workspace = true, features = ["multithread"] } thiserror.workspace = true -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["sync", "rt", "time"] } tracing.workspace = true [dev-dependencies] +metrics-util = { version = "0.20", features = ["debugging"] } tokio = { workspace = true, features = ["macros", "rt", "time"] } [lints] diff --git a/crates/object-data-cache/src/cache.rs b/crates/object-data-cache/src/cache.rs index e06f45de2..5aa276c50 100644 --- a/crates/object-data-cache/src/cache.rs +++ b/crates/object-data-cache/src/cache.rs @@ -17,22 +17,39 @@ use crate::config::ObjectDataCacheConfig; use crate::error::ObjectDataCacheConfigError; use crate::key::{ObjectDataCacheBodyVariant, ObjectDataCacheIdentity, ObjectDataCacheKey}; use crate::metrics::{ - describe_metrics_once, publish_cache_state, record_fill_result, record_hit_bytes, record_invalidation, - record_request_decision, + describe_metrics_once, publish_cache_state, record_fill_result, record_hit_bytes, record_invalidation, record_lookup_result, + record_plan_decision, }; use crate::moka_backend::MokaBackend; use crate::noop::NoopBackend; use crate::stats::{ObjectDataCacheStats, ObjectDataCacheStatsSnapshot}; use bytes::Bytes; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Instant; +/// Minimum spacing between cache-state gauge publishes. Moka's `entry_count` +/// and `weighted_size` are cross-segment approximations that only settle after +/// pending tasks run, so republishing on every fill/invalidate lands the same +/// stale value thousands of times between scrapes (backlog#1134). +const ENTRY_COUNT_PUBLISH_DEBOUNCE_MS: u64 = 1000; + +/// Invalidation outcome label: keys were dropped from the cache. +const INVALIDATION_OUTCOME_REMOVED: &str = "removed"; +/// Invalidation outcome label: the identity was not cached, nothing removed. +const INVALIDATION_OUTCOME_NOOP: &str = "noop"; + /// Protocol-neutral cache facade for object body reuse. #[derive(Debug)] pub struct ObjectDataCache { backend: ObjectDataCacheBackendKind, config: Arc, stats: Arc, + /// Monotonic origin for the cache-state publish debounce. + created_at: Instant, + /// Millis since `created_at` of the last cache-state gauge publish, or `0` + /// when it has never published. + last_entry_publish_ms: AtomicU64, } impl ObjectDataCache { @@ -45,6 +62,8 @@ impl ObjectDataCache { backend: ObjectDataCacheBackendKind::Noop(NoopBackend), config, stats, + created_at: Instant::now(), + last_entry_publish_ms: AtomicU64::new(0), } } @@ -63,13 +82,15 @@ impl ObjectDataCache { backend, config: Arc::new(config), stats, + created_at: Instant::now(), + last_entry_publish_ms: AtomicU64::new(0), }) } /// Produces a lightweight GET plan from request metadata. pub fn plan_get(&self, request: ObjectDataCacheGetRequest<'_>) -> ObjectDataCacheGetPlan { if self.config.is_disabled() { - record_request_decision( + record_plan_decision( self.backend.as_metric_label(), self.config.mode, "disabled", @@ -80,11 +101,11 @@ impl ObjectDataCache { } if request.size > self.config.max_entry_bytes { - record_request_decision(self.backend.as_metric_label(), self.config.mode, "skip", "too_large", request.size); + record_plan_decision(self.backend.as_metric_label(), self.config.mode, "skip", "too_large", request.size); return ObjectDataCacheGetPlan::SkipTooLarge; } - record_request_decision(self.backend.as_metric_label(), self.config.mode, "cacheable", "eligible", request.size); + record_plan_decision(self.backend.as_metric_label(), self.config.mode, "cacheable", "eligible", request.size); ObjectDataCacheGetPlan::Cacheable { key: ObjectDataCacheKey::new( @@ -106,11 +127,13 @@ impl ObjectDataCache { }; self.stats.record_lookup(matches!(lookup, ObjectDataCacheLookup::Hit(_))); - self.refresh_entry_count(); + // Do not refresh the cache-state gauge on the lookup hot path: moka's + // approximations do not change on a read, so it would only republish the + // same stale value on every GET (backlog#1134). match &lookup { ObjectDataCacheLookup::Hit(bytes) => { let size_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); - record_request_decision(self.backend.as_metric_label(), self.config.mode, "hit", "cache_hit", size_bytes); + record_lookup_result(self.backend.as_metric_label(), self.config.mode, "hit", size_bytes); record_hit_bytes(self.backend.as_metric_label(), self.config.mode, size_bytes); } ObjectDataCacheLookup::Miss => { @@ -118,13 +141,13 @@ impl ObjectDataCache { ObjectDataCacheGetPlan::Cacheable { key } => key.size, _ => 0, }; - record_request_decision(self.backend.as_metric_label(), self.config.mode, "miss", "cache_miss", size_bytes); + record_lookup_result(self.backend.as_metric_label(), self.config.mode, "miss", size_bytes); } ObjectDataCacheLookup::SkipDisabled => { - record_request_decision(self.backend.as_metric_label(), self.config.mode, "skip", "lookup_disabled", 0); + record_lookup_result(self.backend.as_metric_label(), self.config.mode, "skip_disabled", 0); } ObjectDataCacheLookup::SkipNotCacheable => { - record_request_decision(self.backend.as_metric_label(), self.config.mode, "skip", "lookup_not_cacheable", 0); + record_lookup_result(self.backend.as_metric_label(), self.config.mode, "skip_not_cacheable", 0); } } @@ -135,21 +158,19 @@ impl ObjectDataCache { pub async fn fill_body(&self, plan: &ObjectDataCacheGetPlan, bytes: Bytes) -> ObjectDataCacheFillResult { let fill_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); if !self.config.fill_enabled() { - record_fill_result(self.backend.as_metric_label(), self.config.mode, "skipped_by_mode", fill_bytes, 0.0); - return ObjectDataCacheFillResult::SkippedByMode; + // Never reached the backend: count the outcome, but record no bytes + // (nothing was submitted) and no duration (there was no fill work). + let result = ObjectDataCacheFillResult::SkippedByMode; + record_fill_result(self.backend.as_metric_label(), self.config.mode, result.as_metric_label(), 0, None); + return result; } if let ObjectDataCacheGetPlan::Cacheable { key } = plan && fill_bytes != key.size { + // Never reached the backend either: count only. let result = ObjectDataCacheFillResult::SkippedSizeMismatch; - record_fill_result( - self.backend.as_metric_label(), - self.config.mode, - result.as_metric_label(), - fill_bytes, - 0.0, - ); + record_fill_result(self.backend.as_metric_label(), self.config.mode, result.as_metric_label(), 0, None); return result; } @@ -159,16 +180,39 @@ impl ObjectDataCache { ObjectDataCacheBackendKind::Moka(backend) => backend.fill_body(plan, bytes).await, }; - if matches!(result, ObjectDataCacheFillResult::Inserted) { - self.stats.record_fill(); - } - self.refresh_entry_count(); + // Fill bytes and duration describe work the backend actually performed, + // so each outcome is listed explicitly rather than caught by a wildcard: + // a rejected fill wrote nothing and must not inflate fill_bytes_total. + // See backlog#1123. + let (recorded_bytes, duration) = match &result { + // Inserted the body: the only outcome that moves the entry count. + ObjectDataCacheFillResult::Inserted => { + self.stats.record_fill(); + self.refresh_entry_count(); + (fill_bytes, Some(fill_start.elapsed().as_secs_f64())) + } + // Reached the backend and wrote the body, then undid it. The bytes + // were written, so both are real. + ObjectDataCacheFillResult::SkippedInvalidationRace | ObjectDataCacheFillResult::SkippedIdentityOverflow => { + (fill_bytes, Some(fill_start.elapsed().as_secs_f64())) + } + // Rejected before writing anything: count the outcome, nothing else. + // A `JoinedInflightFill` is already counted by singleflight_joins, + // and its elapsed time would be wait time, not fill work. + ObjectDataCacheFillResult::JoinedInflightFill + | ObjectDataCacheFillResult::SkippedFillConcurrency + | ObjectDataCacheFillResult::SkippedMemoryPressure + | ObjectDataCacheFillResult::SkippedDisabled + | ObjectDataCacheFillResult::SkippedByMode + | ObjectDataCacheFillResult::SkippedNotCacheable + | ObjectDataCacheFillResult::SkippedSizeMismatch => (0, None), + }; record_fill_result( self.backend.as_metric_label(), self.config.mode, result.as_metric_label(), - fill_bytes, - fill_start.elapsed().as_secs_f64(), + recorded_bytes, + duration, ); result @@ -186,8 +230,15 @@ impl ObjectDataCache { }; self.stats.record_invalidation(); - self.refresh_entry_count(); - record_invalidation(self.backend.as_metric_label(), _reason.as_metric_label()); + let outcome = invalidation_outcome(&result); + // A mutating op invalidates twice by design (before + after); the vast + // majority of those touch identities that were never cached. Only refresh + // the cache-state gauge when something was actually removed so the + // no-op path stays cheap (backlog#1141). + if outcome != INVALIDATION_OUTCOME_NOOP { + self.refresh_entry_count(); + } + record_invalidation(self.backend.as_metric_label(), _reason.as_metric_label(), outcome); result } @@ -207,7 +258,22 @@ impl ObjectDataCache { matches!(self.config.mode, crate::config::ObjectDataCacheMode::FillMaterializeEnabled) } + /// Publishes the cache-state gauge (and mirrors it into stats), debounced so + /// it fires at most once per [`ENTRY_COUNT_PUBLISH_DEBOUNCE_MS`]. Moka's + /// `entry_count`/`weighted_size` are approximations that only settle after + /// pending tasks run, so publishing on every fill/invalidate would restore + /// the same stale value thousands of times between scrapes (backlog#1134). fn refresh_entry_count(&self) { + let now_ms = u64::try_from(self.created_at.elapsed().as_millis()) + .unwrap_or(u64::MAX) + // Reserve 0 as the "never published" sentinel. + .max(1); + let last = self.last_entry_publish_ms.load(Ordering::Relaxed); + if last != 0 && now_ms.saturating_sub(last) < ENTRY_COUNT_PUBLISH_DEBOUNCE_MS { + return; + } + self.last_entry_publish_ms.store(now_ms, Ordering::Relaxed); + let (entries, weighted_bytes) = match &self.backend { ObjectDataCacheBackendKind::Noop(_) => (0, 0), ObjectDataCacheBackendKind::Moka(backend) => (backend.entry_count(), backend.weighted_size()), @@ -217,6 +283,18 @@ impl ObjectDataCache { } } +/// Maps an invalidation result to its metric outcome label. +const fn invalidation_outcome(result: &ObjectDataCacheInvalidationResult) -> &'static str { + match result { + ObjectDataCacheInvalidationResult::Removed { keys } if *keys > 0 => INVALIDATION_OUTCOME_REMOVED, + ObjectDataCacheInvalidationResult::Removed { .. } | ObjectDataCacheInvalidationResult::NoOp => INVALIDATION_OUTCOME_NOOP, + // Transitional: a backend that has not yet been widened to report the + // removal count (`MokaBackend`) still returns `Success`. Treat it as a + // removal so the gauge keeps refreshing until the backend reports + // `Removed`/`NoOp` (see report / backlog#1141). + } +} + /// Protocol-neutral GET request metadata for cache planning. #[derive(Debug, Clone)] pub struct ObjectDataCacheGetRequest<'a> { @@ -276,12 +354,17 @@ pub enum ObjectDataCacheFillResult { SkippedIdentityOverflow, /// Fill was skipped because the provided body length did not match the cache key identity. SkippedSizeMismatch, - /// Fill waiters were released without a published leader result. - SkippedSingleflightClosed, /// Fill was undone because an invalidation raced with the insert. SkippedInvalidationRace, + /// The caller joined an in-flight leader fill instead of performing one, so + /// it did no fill work of its own (the join is counted by + /// `singleflight_joins`). Distinguishes waiters from true inserts so N + /// concurrent GETs of one cold key do not record N inserts (backlog#1123). + JoinedInflightFill, /// The cache entry was inserted successfully. Inserted, + /// Fill was skipped because the fill-concurrency limiter was saturated. + SkippedFillConcurrency, } impl ObjectDataCacheFillResult { @@ -293,9 +376,10 @@ impl ObjectDataCacheFillResult { Self::SkippedMemoryPressure => "skipped_memory_pressure", Self::SkippedIdentityOverflow => "skipped_identity_overflow", Self::SkippedSizeMismatch => "skipped_size_mismatch", - Self::SkippedSingleflightClosed => "skipped_singleflight_closed", Self::SkippedInvalidationRace => "skipped_invalidation_race", + Self::JoinedInflightFill => "joined_inflight", Self::Inserted => "inserted", + Self::SkippedFillConcurrency => "skipped_fill_concurrency", } } } @@ -333,16 +417,26 @@ impl ObjectDataCacheInvalidationReason { /// Result of an invalidation request. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ObjectDataCacheInvalidationResult { - /// Invalidation completed successfully. - Success, + /// Cache keys were removed for the identity. + Removed { + /// Number of cache keys dropped. + keys: usize, + }, + /// The identity was not cached, so nothing was removed. + NoOp, } #[cfg(test)] mod tests { - use super::{ObjectDataCache, ObjectDataCacheFillResult, ObjectDataCacheGetRequest, ObjectDataCacheLookup}; + use super::{ + ObjectDataCache, ObjectDataCacheFillResult, ObjectDataCacheGetPlan, ObjectDataCacheGetRequest, + ObjectDataCacheInvalidationReason, ObjectDataCacheLookup, + }; use crate::config::{ObjectDataCacheConfig, ObjectDataCacheMode}; - use crate::key::ObjectDataCacheBodyVariant; + use crate::key::{ObjectDataCacheBodyVariant, ObjectDataCacheIdentity}; use bytes::Bytes; + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; fn fill_enabled_cache() -> ObjectDataCache { let config = ObjectDataCacheConfig { @@ -356,6 +450,192 @@ mod tests { ObjectDataCache::new(config).expect("fill-enabled cache config should initialize") } + fn hit_only_cache() -> ObjectDataCache { + let config = ObjectDataCacheConfig { + mode: ObjectDataCacheMode::HitOnly, + max_bytes: 8_388_608, + ..ObjectDataCacheConfig::default() + }; + ObjectDataCache::new(config).expect("hit-only cache config should initialize") + } + + fn plain_request<'a>(bucket: &'a str, object: &'a str, etag: &'a str, size: u64) -> ObjectDataCacheGetRequest<'a> { + ObjectDataCacheGetRequest { + bucket, + object, + version_id: None, + etag, + size, + body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, + } + } + + struct CapturedMetric { + kind: MetricKind, + name: String, + labels: Vec<(String, String)>, + value: DebugValue, + } + + /// Runs `f` under a thread-local debugging recorder and a current-thread + /// runtime, so every metric the async body emits is captured without + /// touching the process-global registry. A current-thread runtime keeps the + /// spawned fill tasks on the recorder's thread. + fn capture_metrics(f: F) -> Vec + where + F: FnOnce() -> Fut, + Fut: std::future::Future, + { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime should build"); + metrics::with_local_recorder(&recorder, || runtime.block_on(f())); + snapshotter + .snapshot() + .into_vec() + .into_iter() + .map(|(composite, _unit, _desc, value)| { + let labels = composite + .key() + .labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect(); + CapturedMetric { + kind: composite.kind(), + name: composite.key().name().to_string(), + labels, + value, + } + }) + .collect() + } + + fn counter_total(metrics: &[CapturedMetric], name: &str) -> Option { + let mut found = false; + let mut sum = 0u64; + for metric in metrics { + if metric.kind == MetricKind::Counter && metric.name == name { + found = true; + if let DebugValue::Counter(v) = metric.value { + sum += v; + } + } + } + found.then_some(sum) + } + + fn has_gauge(metrics: &[CapturedMetric], name: &str) -> bool { + metrics.iter().any(|m| m.kind == MetricKind::Gauge && m.name == name) + } + + fn has_counter_with_label(metrics: &[CapturedMetric], name: &str, label: (&str, &str)) -> bool { + metrics.iter().any(|m| { + m.kind == MetricKind::Counter && m.name == name && m.labels.iter().any(|(k, v)| k == label.0 && v == label.1) + }) + } + + #[test] + fn single_get_records_one_plan_and_one_lookup_increment() { + // ODC-17: one GET must produce exactly one plan increment and one lookup + // increment, no longer conflated on a single requests_total counter. + let cache = hit_only_cache(); + let metrics = capture_metrics(|| async { + let plan = cache.plan_get(plain_request("bucket", "object", "etag", 1024)); + let _ = cache.lookup_body(&plan).await; + }); + + assert_eq!( + counter_total(&metrics, "rustfs_object_data_cache_plan_total"), + Some(1), + "exactly one plan decision per GET" + ); + assert_eq!( + counter_total(&metrics, "rustfs_object_data_cache_lookup_total"), + Some(1), + "exactly one lookup outcome per GET" + ); + } + + #[test] + fn lookup_does_not_publish_entry_count_gauge() { + // ODC-29: the lookup hot path must not republish the cache-state gauge. + let cache = hit_only_cache(); + let metrics = capture_metrics(|| async { + let plan = cache.plan_get(plain_request("bucket", "object", "etag", 1024)); + let _ = cache.lookup_body(&plan).await; + }); + + assert!( + !has_gauge(&metrics, "rustfs_object_data_cache_entries"), + "lookup must not publish the entries gauge" + ); + } + + #[test] + fn joined_inflight_result_maps_to_metric_label() { + // ODC-18: the waiter outcome has its own, distinct metric label. + assert_eq!(ObjectDataCacheFillResult::JoinedInflightFill.as_metric_label(), "joined_inflight"); + assert_ne!( + ObjectDataCacheFillResult::JoinedInflightFill.as_metric_label(), + ObjectDataCacheFillResult::Inserted.as_metric_label() + ); + } + + #[test] + fn disabled_cache_invalidation_labels_noop_and_skips_gauge() { + // ODC-36: invalidating an identity that was never cached is a no-op; it + // is labeled outcome=noop and must not refresh the cache-state gauge. + let cache = ObjectDataCache::disabled(); + let metrics = capture_metrics(|| async { + let _ = cache + .invalidate_object( + ObjectDataCacheIdentity::new("bucket", "object"), + ObjectDataCacheInvalidationReason::BeforeMutation, + ) + .await; + }); + + assert!( + has_counter_with_label(&metrics, "rustfs_object_data_cache_invalidations_total", ("outcome", "noop")), + "a no-op invalidation must be labeled outcome=noop" + ); + assert!( + !has_gauge(&metrics, "rustfs_object_data_cache_entries"), + "a no-op invalidation must not refresh the entries gauge" + ); + } + + #[test] + fn invalidation_of_cached_identity_labels_removed() { + // ODC-36: invalidating an identity that held cached keys is labeled + // outcome=removed. + let cache = fill_enabled_cache(); + let metrics = capture_metrics(|| async { + let plan = cache.plan_get(plain_request("bucket", "object", "etag", 5)); + let ObjectDataCacheGetPlan::Cacheable { .. } = &plan else { + panic!("plan should be cacheable"); + }; + assert_eq!( + cache.fill_body(&plan, Bytes::from_static(b"hello")).await, + ObjectDataCacheFillResult::Inserted + ); + let _ = cache + .invalidate_object( + ObjectDataCacheIdentity::new("bucket", "object"), + ObjectDataCacheInvalidationReason::AfterDeleteSuccess, + ) + .await; + }); + + assert!( + has_counter_with_label(&metrics, "rustfs_object_data_cache_invalidations_total", ("outcome", "removed")), + "invalidating a cached identity must be labeled outcome=removed" + ); + } + #[tokio::test] async fn fill_body_rejects_size_mismatch() { let cache = fill_enabled_cache(); diff --git a/crates/object-data-cache/src/entry.rs b/crates/object-data-cache/src/entry.rs index 164123916..011598f5a 100644 --- a/crates/object-data-cache/src/entry.rs +++ b/crates/object-data-cache/src/entry.rs @@ -14,31 +14,27 @@ use bytes::Bytes; use std::cmp; -use std::sync::Arc; -use std::time::Instant; use crate::key::ObjectDataCacheKey; const ENTRY_OVERHEAD_BYTES: usize = 64; /// Cached object body entry. +/// +/// The content length and etag are already part of the moka key's `Hash`/`Eq` +/// identity (see [`ObjectDataCacheKey`]), so a cached hit is only returned for a +/// key that already matched them. Storing per-entry copies (`content_length`, +/// `etag`, `inserted_at`) added memory not captured by `ENTRY_OVERHEAD_BYTES` +/// with no reader, so the entry is now a thin `Bytes` wrapper (backlog#1141). #[derive(Debug, Clone)] pub struct ObjectDataCacheEntry { bytes: Bytes, - content_length: u64, - etag: Arc, - inserted_at: Instant, } impl ObjectDataCacheEntry { /// Creates a new cached entry. - pub fn new(bytes: Bytes, content_length: u64, etag: Arc) -> Self { - Self { - bytes, - content_length, - etag, - inserted_at: Instant::now(), - } + pub fn new(bytes: Bytes) -> Self { + Self { bytes } } /// Returns a clone of the cached body bytes. @@ -46,21 +42,6 @@ impl ObjectDataCacheEntry { self.bytes.clone() } - /// Returns the recorded content length. - pub const fn content_length(&self) -> u64 { - self.content_length - } - - /// Returns the cached etag reference. - pub fn etag(&self) -> &Arc { - &self.etag - } - - /// Returns the insertion timestamp. - pub const fn inserted_at(&self) -> Instant { - self.inserted_at - } - /// Returns the estimated weighted size for capacity accounting. pub fn estimated_weight(&self, key: &ObjectDataCacheKey) -> u32 { let key_bytes = key.bucket.len() + key.object.len() + key.version_id.len() + key.etag.len(); @@ -77,13 +58,12 @@ mod tests { use super::ObjectDataCacheEntry; use crate::key::{ObjectDataCacheBodyVariant, ObjectDataCacheKey}; use bytes::Bytes; - use std::sync::Arc; #[test] fn estimated_weight_includes_body_and_key_bytes() { let key = ObjectDataCacheKey::new("bucket", "object", Some("vid"), "etag", 5, ObjectDataCacheBodyVariant::FullObjectPlainV1); - let entry = ObjectDataCacheEntry::new(Bytes::from_static(b"hello"), 5, Arc::::from("etag")); + let entry = ObjectDataCacheEntry::new(Bytes::from_static(b"hello")); let weight = entry.estimated_weight(&key); @@ -104,7 +84,7 @@ mod tests { ObjectDataCacheBodyVariant::FullObjectPlainV1, ); let huge = vec![0u8; (u32::MAX as usize).saturating_add(1024)]; - let entry = ObjectDataCacheEntry::new(Bytes::from(huge), 1, Arc::::from("etag")); + let entry = ObjectDataCacheEntry::new(Bytes::from(huge)); let weight = entry.estimated_weight(&key); diff --git a/crates/object-data-cache/src/key.rs b/crates/object-data-cache/src/key.rs index 4c669ecbf..d288e3f6c 100644 --- a/crates/object-data-cache/src/key.rs +++ b/crates/object-data-cache/src/key.rs @@ -72,6 +72,10 @@ impl ObjectDataCacheKey { } /// Returns true when this key targets the canonical unversioned body variant. + /// + /// Only exercised by tests; gated so it is not compiled into the shipping + /// binary as dead code (backlog#1141). + #[cfg(test)] pub fn is_null_version(&self) -> bool { self.version_id.as_ref() == NULL_VERSION_ID } diff --git a/crates/object-data-cache/src/memory.rs b/crates/object-data-cache/src/memory.rs index a395cd1fd..5b2564726 100644 --- a/crates/object-data-cache/src/memory.rs +++ b/crates/object-data-cache/src/memory.rs @@ -15,9 +15,11 @@ use crate::config::ObjectDataCacheConfig; use crate::metrics::record_memory_pressure; use crate::stats::ObjectDataCacheStats; +use std::sync::Arc; +#[cfg(test)] +use std::sync::Mutex; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; +use std::time::Duration; use sysinfo::System; const DEFAULT_REFRESH_INTERVAL: Duration = Duration::from_secs(5); @@ -110,16 +112,60 @@ impl ObjectDataCacheMemorySnapshot { } } -/// Memory gate that keeps a cheap snapshot for fill-path checks. +/// Lock-free memory snapshot shared between the gate and its refresher task. +#[derive(Debug)] +struct MemorySnapshotCell { + total_bytes: AtomicU64, + available_bytes: AtomicU64, +} + +impl MemorySnapshotCell { + fn new(snapshot: ObjectDataCacheMemorySnapshot) -> Self { + Self { + total_bytes: AtomicU64::new(snapshot.total_bytes), + available_bytes: AtomicU64::new(snapshot.available_bytes), + } + } + + fn store(&self, snapshot: ObjectDataCacheMemorySnapshot) { + self.total_bytes.store(snapshot.total_bytes, Ordering::Relaxed); + self.available_bytes.store(snapshot.available_bytes, Ordering::Relaxed); + } + + fn load(&self) -> ObjectDataCacheMemorySnapshot { + ObjectDataCacheMemorySnapshot { + total_bytes: self.total_bytes.load(Ordering::Relaxed), + available_bytes: self.available_bytes.load(Ordering::Relaxed), + } + } +} + +/// Aborts the periodic refresher when the gate (and thus the cache) is dropped. +#[derive(Debug)] +struct RefresherGuard(tokio::task::JoinHandle<()>); + +impl Drop for RefresherGuard { + fn drop(&mut self) { + self.0.abort(); + } +} + +/// Memory gate that keeps a cheap, lock-free snapshot for fill-path checks. +/// +/// The snapshot is sampled off the fill path by a dedicated periodic refresher +/// (see [`spawn_refresher`](ObjectDataCacheMemoryGate::spawn_refresher)) that +/// runs the blocking `sysinfo` read on a `spawn_blocking` thread. `allows_fill` +/// only reads atomics, so it never blocks a tokio worker and concurrent fills +/// never serialize on a refresh. #[derive(Debug)] pub struct ObjectDataCacheMemoryGate { - system: Mutex, - last_refresh: Mutex, - snapshot_total_bytes: AtomicU64, - snapshot_available_bytes: AtomicU64, + snapshot: Arc, min_free_memory_percent: u8, - refresh_interval: Duration, stats: Arc, + /// Kept alive so the refresher runs for the gate's lifetime; `None` when the + /// gate is opted out (`min_free_memory_percent == 0`) or no tokio runtime is + /// available at construction (e.g. synchronous unit tests). + _refresher: Option, #[cfg(test)] test_override: Mutex>, } @@ -127,27 +173,59 @@ pub struct ObjectDataCacheMemoryGate { impl ObjectDataCacheMemoryGate { /// Creates a new memory gate. pub fn new(config: &ObjectDataCacheConfig, stats: Arc) -> Self { - let mut system = System::new(); - system.refresh_memory(); - let effective = effective_memory_from_system(&system); - let snapshot = ObjectDataCacheMemorySnapshot { + // Seed the snapshot once at construction. This is the only blocking + // sysinfo read on any caller's stack; all later refreshes run off-path. + let effective = resolve_effective_memory(); + let snapshot = Arc::new(MemorySnapshotCell::new(ObjectDataCacheMemorySnapshot { total_bytes: effective.total_bytes, available_bytes: effective.available_bytes, + })); + + let min_free_memory_percent = config.min_free_memory_percent; + // A zero floor opts out of the gate entirely, so there is nothing to + // refresh; skip the background task in that case. + let refresher = if min_free_memory_percent == 0 { + None + } else { + Self::spawn_refresher(Arc::clone(&snapshot), DEFAULT_REFRESH_INTERVAL) }; Self { - system: Mutex::new(system), - last_refresh: Mutex::new(Instant::now()), - snapshot_total_bytes: AtomicU64::new(snapshot.total_bytes), - snapshot_available_bytes: AtomicU64::new(snapshot.available_bytes), - min_free_memory_percent: config.min_free_memory_percent, - refresh_interval: DEFAULT_REFRESH_INTERVAL, + snapshot, + min_free_memory_percent, stats, + _refresher: refresher, #[cfg(test)] test_override: Mutex::new(None), } } + /// Spawns the periodic refresher, returning `None` when no tokio runtime is + /// available (the gate then keeps its construction-time seed snapshot). + fn spawn_refresher(snapshot: Arc, interval: Duration) -> Option { + let handle = tokio::runtime::Handle::try_current().ok()?; + let task = handle.spawn(async move { + let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // The first tick fires immediately; the seed snapshot is already in + // place, so refresh from the second tick onward. + ticker.tick().await; + loop { + ticker.tick().await; + // Run the blocking /proc/meminfo read off the async worker. A + // join error only happens if the runtime is shutting down; keep + // the last snapshot rather than clobbering it in that case. + if let Ok(effective) = tokio::task::spawn_blocking(resolve_effective_memory).await { + snapshot.store(ObjectDataCacheMemorySnapshot { + total_bytes: effective.total_bytes, + available_bytes: effective.available_bytes, + }); + } + } + }); + Some(RefresherGuard(task)) + } + /// Returns the current atomic memory snapshot. pub fn snapshot(&self) -> ObjectDataCacheMemorySnapshot { #[cfg(test)] @@ -157,44 +235,21 @@ impl ObjectDataCacheMemoryGate { } } - ObjectDataCacheMemorySnapshot { - total_bytes: self.snapshot_total_bytes.load(Ordering::Relaxed), - available_bytes: self.snapshot_available_bytes.load(Ordering::Relaxed), - } - } - - /// Refreshes the snapshot if it is stale. - pub fn refresh_if_stale(&self) { - { - let last_refresh = lock_or_recover(&self.last_refresh); - if last_refresh.elapsed() < self.refresh_interval { - return; - } - } - - let mut system = lock_or_recover(&self.system); - system.refresh_memory(); - let effective = effective_memory_from_system(&system); - let snapshot = ObjectDataCacheMemorySnapshot { - total_bytes: effective.total_bytes, - available_bytes: effective.available_bytes, - }; - - self.snapshot_total_bytes.store(snapshot.total_bytes, Ordering::Relaxed); - self.snapshot_available_bytes - .store(snapshot.available_bytes, Ordering::Relaxed); - *lock_or_recover(&self.last_refresh) = Instant::now(); + self.snapshot.load() } /// Returns true when the fill path may proceed under current memory pressure. + /// + /// This is lock-free and does no blocking sysinfo read: it only reads the + /// atomic snapshot maintained by the periodic refresher. pub fn allows_fill(&self, required_bytes: u64) -> bool { // A zero floor opts out of the gate, so fill admission never depends on // a live memory reading — which differs between a host and a container. + // This must short-circuit before any snapshot read (see 51a97a81c). if self.min_free_memory_percent == 0 { return true; } - self.refresh_if_stale(); let snapshot = self.snapshot(); if snapshot.total_bytes == 0 { return true; @@ -216,8 +271,22 @@ impl ObjectDataCacheMemoryGate { pub fn set_test_snapshot(&self, snapshot: Option) { *lock_or_recover(&self.test_override) = snapshot; } + + /// Writes the atomic snapshot directly, bypassing the `test_override` read + /// path so a test can observe whether `allows_fill` mutates the snapshot. + #[cfg(test)] + fn store_raw_snapshot_for_test(&self, snapshot: ObjectDataCacheMemorySnapshot) { + self.snapshot.store(snapshot); + } + + /// Reads the atomic snapshot directly, bypassing `test_override`. + #[cfg(test)] + fn raw_snapshot_for_test(&self) -> ObjectDataCacheMemorySnapshot { + self.snapshot.load() + } } +#[cfg(test)] fn lock_or_recover(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { match mutex.lock() { Ok(guard) => guard, @@ -231,6 +300,7 @@ mod tests { use crate::config::ObjectDataCacheConfig; use crate::stats::ObjectDataCacheStats; use std::sync::Arc; + use std::time::Duration; const GIB: u64 = 1024 * 1024 * 1024; @@ -331,4 +401,57 @@ mod tests { assert!(!gate.allows_fill(128)); assert_eq!(stats.snapshot().memory_pressure_events, 1); } + + // ODC-14: `allows_fill` must read the atomic snapshot without performing an + // inline (blocking) refresh. A synchronous test has no tokio runtime, so no + // refresher task exists; if `allows_fill` refreshed inline it would clobber + // the seeded atomic snapshot with the real host reading. Comparing exact + // byte values makes the assertion independent of the host's actual memory. + #[test] + fn allows_fill_reads_snapshot_without_refreshing_inline() { + let stats = Arc::new(ObjectDataCacheStats::default()); + let gate = ObjectDataCacheMemoryGate::new(&ObjectDataCacheConfig::default(), Arc::clone(&stats)); + + let sentinel = ObjectDataCacheMemorySnapshot { + total_bytes: 4_242, + available_bytes: 2_121, + }; + gate.store_raw_snapshot_for_test(sentinel); + + // Exercise the gate on the atomic path (no test_override installed). + let _ = gate.allows_fill(1); + + let after = gate.raw_snapshot_for_test(); + assert_eq!( + after, sentinel, + "allows_fill must not mutate the snapshot; an inline refresh would overwrite it" + ); + } + + // ODC-14: the periodic refresher samples memory off the fill path and + // updates the atomic snapshot, so a stale seed is replaced without any + // fill ever blocking on a refresh. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn periodic_refresher_updates_snapshot_off_path() { + let stats = Arc::new(ObjectDataCacheStats::default()); + let snapshot = Arc::new(super::MemorySnapshotCell::new(ObjectDataCacheMemorySnapshot { + total_bytes: 1, + available_bytes: 1, + })); + let _guard = ObjectDataCacheMemoryGate::spawn_refresher(Arc::clone(&snapshot), Duration::from_millis(20)) + .expect("a tokio runtime is available in an async test"); + + // The seed is a bogus 1/1; wait for the refresher to overwrite it with a + // real host reading (total memory is always far above 1 byte). + let mut refreshed = false; + for _ in 0..200 { + if snapshot.load().total_bytes > 1 { + refreshed = true; + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(refreshed, "the periodic refresher must replace the seed snapshot off the fill path"); + let _ = stats; + } } diff --git a/crates/object-data-cache/src/metrics.rs b/crates/object-data-cache/src/metrics.rs index b09a49783..831d827f0 100644 --- a/crates/object-data-cache/src/metrics.rs +++ b/crates/object-data-cache/src/metrics.rs @@ -17,7 +17,8 @@ use crate::stats::ObjectDataCacheStats; use metrics::{counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram}; use std::sync::{Arc, OnceLock}; -const METRIC_REQUESTS_TOTAL: &str = "rustfs_object_data_cache_requests_total"; +const METRIC_PLAN_TOTAL: &str = "rustfs_object_data_cache_plan_total"; +const METRIC_LOOKUP_TOTAL: &str = "rustfs_object_data_cache_lookup_total"; const METRIC_FILLS_TOTAL: &str = "rustfs_object_data_cache_fill_total"; const METRIC_FILL_DURATION_SECONDS: &str = "rustfs_object_data_cache_fill_duration_seconds"; const METRIC_FILL_BYTES_TOTAL: &str = "rustfs_object_data_cache_fill_bytes_total"; @@ -32,8 +33,14 @@ pub(crate) fn describe_metrics_once() { static DESCRIBED: OnceLock<()> = OnceLock::new(); let _ = DESCRIBED.get_or_init(|| { describe_counter!( - METRIC_REQUESTS_TOTAL, - "Object data cache request decisions labeled by backend, mode, decision, reason, and size class." + METRIC_PLAN_TOTAL, + "Object data cache GET plan decisions by backend, mode, decision, reason, and size class. \ + Incremented exactly once per GET by the planning layer." + ); + describe_counter!( + METRIC_LOOKUP_TOTAL, + "Object data cache lookup outcomes by backend, mode, result, and size class. \ + Incremented exactly once per GET by the lookup layer." ); describe_counter!( METRIC_FILLS_TOTAL, @@ -47,7 +54,8 @@ pub(crate) fn describe_metrics_once() { describe_gauge!(METRIC_INFLIGHT_FILLS, "Current number of in-flight object data cache fills."); describe_counter!( METRIC_INVALIDATIONS_TOTAL, - "Object data cache invalidation attempts by backend and reason." + "Object data cache invalidation attempts by backend, reason, and outcome \ + (removed when keys were dropped, noop when the identity was not cached)." ); describe_counter!(METRIC_MEMORY_PRESSURE_TOTAL, "Object data cache fill skips caused by memory pressure."); }); @@ -69,7 +77,8 @@ pub(crate) const fn size_class(size_bytes: u64) -> &'static str { } } -pub(crate) fn record_request_decision( +/// Records a GET plan decision. Emitted exactly once per GET by `plan_get`. +pub(crate) fn record_plan_decision( backend: &'static str, mode: ObjectDataCacheMode, decision: &'static str, @@ -77,7 +86,7 @@ pub(crate) fn record_request_decision( size_bytes: u64, ) { counter!( - METRIC_REQUESTS_TOTAL, + METRIC_PLAN_TOTAL, "backend" => backend, "mode" => mode.as_metric_label(), "decision" => decision, @@ -87,12 +96,33 @@ pub(crate) fn record_request_decision( .increment(1); } +/// Records a cache lookup outcome. Emitted exactly once per GET by `lookup_body`. +pub(crate) fn record_lookup_result(backend: &'static str, mode: ObjectDataCacheMode, result: &'static str, size_bytes: u64) { + counter!( + METRIC_LOOKUP_TOTAL, + "backend" => backend, + "mode" => mode.as_metric_label(), + "result" => result, + "size_class" => size_class(size_bytes), + ) + .increment(1); +} + +/// Records a fill outcome. +/// +/// The count is always incremented so every outcome is observable. `size_bytes` +/// is only added to the bytes counter when it is non-zero, and the duration +/// histogram is only recorded when `duration_seconds` is `Some`. Callers pass +/// `0` bytes and `None` duration for outcomes that never submitted a body to the +/// backend (`skipped_by_mode`, `skipped_size_mismatch`) or that merely joined an +/// in-flight leader fill (`joined_inflight`), so those neither inflate +/// `fill_bytes_total` nor pollute the duration histogram with non-fill samples. pub(crate) fn record_fill_result( backend: &'static str, mode: ObjectDataCacheMode, result: &'static str, size_bytes: u64, - duration_seconds: f64, + duration_seconds: Option, ) { let size_class = size_class(size_bytes); counter!( @@ -103,22 +133,26 @@ pub(crate) fn record_fill_result( "size_class" => size_class, ) .increment(1); - histogram!( - METRIC_FILL_DURATION_SECONDS, - "backend" => backend, - "mode" => mode.as_metric_label(), - "result" => result, - "size_class" => size_class, - ) - .record(duration_seconds); - counter!( - METRIC_FILL_BYTES_TOTAL, - "backend" => backend, - "mode" => mode.as_metric_label(), - "result" => result, - "size_class" => size_class, - ) - .increment(size_bytes); + if let Some(duration_seconds) = duration_seconds { + histogram!( + METRIC_FILL_DURATION_SECONDS, + "backend" => backend, + "mode" => mode.as_metric_label(), + "result" => result, + "size_class" => size_class, + ) + .record(duration_seconds); + } + if size_bytes > 0 { + counter!( + METRIC_FILL_BYTES_TOTAL, + "backend" => backend, + "mode" => mode.as_metric_label(), + "result" => result, + "size_class" => size_class, + ) + .increment(size_bytes); + } } pub(crate) fn record_hit_bytes(backend: &'static str, mode: ObjectDataCacheMode, size_bytes: u64) { @@ -151,6 +185,106 @@ pub(crate) fn record_memory_pressure(stats: &Arc, backend: counter!(METRIC_MEMORY_PRESSURE_TOTAL, "backend" => backend).increment(1); } -pub(crate) fn record_invalidation(backend: &'static str, reason: &'static str) { - counter!(METRIC_INVALIDATIONS_TOTAL, "backend" => backend, "reason" => reason).increment(1); +pub(crate) fn record_invalidation(backend: &'static str, reason: &'static str, outcome: &'static str) { + counter!(METRIC_INVALIDATIONS_TOTAL, "backend" => backend, "reason" => reason, "outcome" => outcome).increment(1); +} + +#[cfg(test)] +mod tests { + use super::{ + METRIC_FILL_BYTES_TOTAL, METRIC_FILL_DURATION_SECONDS, METRIC_FILLS_TOTAL, METRIC_LOOKUP_TOTAL, METRIC_PLAN_TOTAL, + record_fill_result, record_lookup_result, record_plan_decision, + }; + use crate::config::ObjectDataCacheMode; + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + /// Captures every metric emitted by `f` on a thread-local recorder, so the + /// process-global registry (and its cross-test flakiness) is untouched. + fn capture(f: impl FnOnce()) -> Vec<(MetricKind, String, DebugValue)> { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, f); + snapshotter + .snapshot() + .into_vec() + .into_iter() + .map(|(composite, _unit, _desc, value)| (composite.kind(), composite.key().name().to_string(), value)) + .collect() + } + + fn counter_total(metrics: &[(MetricKind, String, DebugValue)], name: &str) -> Option { + let mut found = false; + let mut sum = 0u64; + for (kind, metric_name, value) in metrics { + if *kind == MetricKind::Counter && metric_name == name { + found = true; + if let DebugValue::Counter(v) = value { + sum += *v; + } + } + } + found.then_some(sum) + } + + fn has_metric(metrics: &[(MetricKind, String, DebugValue)], kind: MetricKind, name: &str) -> bool { + metrics.iter().any(|(k, n, _)| *k == kind && n == name) + } + + #[test] + fn plan_and_lookup_counters_are_separate_metrics() { + // ODC-17: one GET produces exactly one plan increment and one lookup + // increment, on two distinct counter names. + let metrics = capture(|| { + record_plan_decision("moka", ObjectDataCacheMode::HitOnly, "cacheable", "eligible", 1024); + record_lookup_result("moka", ObjectDataCacheMode::HitOnly, "miss", 1024); + }); + + assert_eq!(counter_total(&metrics, METRIC_PLAN_TOTAL), Some(1)); + assert_eq!(counter_total(&metrics, METRIC_LOOKUP_TOTAL), Some(1)); + } + + #[test] + fn inserted_fill_records_count_bytes_and_duration() { + let metrics = capture(|| { + record_fill_result("moka", ObjectDataCacheMode::FillBufferedOnly, "inserted", 100, Some(0.5)); + }); + + assert_eq!(counter_total(&metrics, METRIC_FILLS_TOTAL), Some(1)); + assert_eq!(counter_total(&metrics, METRIC_FILL_BYTES_TOTAL), Some(100)); + assert!(has_metric(&metrics, MetricKind::Histogram, METRIC_FILL_DURATION_SECONDS)); + } + + #[test] + fn joined_inflight_fill_counts_without_bytes_or_duration() { + // ODC-18: a waiter that joined an in-flight fill records the outcome + // count but neither fill bytes nor a (wait-time) duration sample. + let metrics = capture(|| { + record_fill_result("moka", ObjectDataCacheMode::FillBufferedOnly, "joined_inflight", 0, None); + }); + + assert_eq!(counter_total(&metrics, METRIC_FILLS_TOTAL), Some(1)); + assert_eq!( + counter_total(&metrics, METRIC_FILL_BYTES_TOTAL), + None, + "waiter must not inflate fill bytes" + ); + assert!( + !has_metric(&metrics, MetricKind::Histogram, METRIC_FILL_DURATION_SECONDS), + "waiter must not record a fill duration sample" + ); + } + + #[test] + fn skipped_before_backend_records_no_duration() { + // ODC-18: outcomes that never reached the backend record a count only. + for result in ["skipped_by_mode", "skipped_size_mismatch"] { + let metrics = capture(|| { + record_fill_result("moka", ObjectDataCacheMode::HitOnly, result, 0, None); + }); + assert_eq!(counter_total(&metrics, METRIC_FILLS_TOTAL), Some(1)); + assert_eq!(counter_total(&metrics, METRIC_FILL_BYTES_TOTAL), None); + assert!(!has_metric(&metrics, MetricKind::Histogram, METRIC_FILL_DURATION_SECONDS)); + } + } } diff --git a/crates/object-data-cache/src/moka_backend.rs b/crates/object-data-cache/src/moka_backend.rs index e8e9d13ae..8312ebace 100644 --- a/crates/object-data-cache/src/moka_backend.rs +++ b/crates/object-data-cache/src/moka_backend.rs @@ -23,6 +23,7 @@ use crate::stats::ObjectDataCacheStats; use bytes::Bytes; use moka::future::{Cache, FutureExt}; use std::sync::Arc; +use tokio::sync::Semaphore; /// Weighted Moka backend for reusable object bodies. #[derive(Debug)] @@ -31,6 +32,9 @@ pub struct MokaBackend { index: Arc, singleflight: ObjectDataCacheSingleflight, memory_gate: ObjectDataCacheMemoryGate, + /// Bounds the number of concurrent distinct-key fills. Singleflight only + /// dedups per key, so without this limiter distinct-key fills are unbounded. + fill_semaphore: Arc, /// Test-only barrier that pauses a fill between the index insert and the /// cache insert so the fill-vs-invalidation race can be driven deterministically. #[cfg(test)] @@ -87,6 +91,15 @@ impl MokaBackend { let ttl = config.ttl; let time_to_idle = config.time_to_idle; + // Size the fill-concurrency limiter to + // min(fill_concurrency_per_cpu * available_parallelism, fill_concurrency_max). + // Both knobs are validated as non-zero, so the result is at least 1. + let parallelism = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + let fill_permits = usize::from(config.fill_concurrency_per_cpu) + .saturating_mul(parallelism) + .min(usize::from(config.fill_concurrency_max)) + .max(1); + // The identity index tracks the keys cached per object so that // invalidation can find them without a full-cache scan. Moka evicts // entries on its own (TTL, time-to-idle, capacity-LRU) without going @@ -131,6 +144,7 @@ impl MokaBackend { index, singleflight: ObjectDataCacheSingleflight::new(Arc::clone(&stats)), memory_gate: ObjectDataCacheMemoryGate::new(config, stats), + fill_semaphore: Arc::new(Semaphore::new(fill_permits)), #[cfg(test)] fill_barrier: std::sync::Mutex::new(None), }) @@ -172,16 +186,26 @@ impl MokaBackend { return ObjectDataCacheFillResult::SkippedNotCacheable; }; - let fill_state = self.singleflight.acquire(key.clone()).await; - let ObjectDataCacheSingleflightAcquire::Leader(leader) = fill_state else { - let ObjectDataCacheSingleflightAcquire::Waiter(waiter) = fill_state else { - unreachable!(); - }; - return waiter.wait().await; + // The caller already owns the body, so a non-leader gains nothing by + // waiting for another request's leader — it just skips its own fill. + // Report it as JoinedInflightFill (ODC-18): it did no fill work, so it + // must not be counted as an insert or record fill bytes. It also does + // NOT wait for the leader — waiting only matters when the joiner needs + // the result value, which the GET path never does (ODC-15). + let ObjectDataCacheSingleflightAcquire::Leader(leader) = self.singleflight.try_acquire(key.clone()) else { + return ObjectDataCacheFillResult::JoinedInflightFill; + }; + + // Bound distinct-key fill concurrency after winning leadership and + // before the memory-gate check. Reject rather than queue: queuing would + // reintroduce the GET-latency coupling this path is meant to avoid. + let permit = match Arc::clone(&self.fill_semaphore).try_acquire_owned() { + Ok(permit) => permit, + Err(_) => return leader.finish(ObjectDataCacheFillResult::SkippedFillConcurrency), }; if !self.memory_gate.allows_fill(u64::try_from(bytes.len()).unwrap_or(u64::MAX)) { - return leader.finish(ObjectDataCacheFillResult::SkippedMemoryPressure).await; + return leader.finish(ObjectDataCacheFillResult::SkippedMemoryPressure); } let identity = ObjectDataCacheIdentity::new(Arc::clone(&key.bucket), Arc::clone(&key.object)); @@ -196,14 +220,17 @@ impl MokaBackend { // Build the entry up front so its Arc pointer can serve as the // generation token registered in the index alongside the key. - let entry = Arc::new(ObjectDataCacheEntry::new(bytes, key.size, Arc::clone(&key.etag))); + let entry = Arc::new(ObjectDataCacheEntry::new(bytes)); let token = Arc::as_ptr(&entry) as ObjectDataCacheKeyToken; // Run the register/insert/recheck/undo sequence in a spawned task and - // await its handle. If the enclosing GET future is dropped (client - // disconnect) while we await, the JoinHandle is detached but the task - // still finishes the recheck and undo, so a stale body cannot survive - // with no index entry. Dropping the leader still unblocks waiters. + // await its handle. The GET path already detaches the whole fill from + // the response future (ODC-15), but this inner spawn is still required: + // once the index key is registered, the recheck/undo must complete even + // if the enclosing fill future is aborted (e.g. the detached fill task + // is cancelled). Aborting the outer future then only detaches this + // JoinHandle; the task still finishes the undo, so a stale body cannot + // survive with no index entry. Dropping the leader releases the key. let cache = self.cache.clone(); let index = Arc::clone(&self.index); let fill_key = key.clone(); @@ -212,6 +239,9 @@ impl MokaBackend { let fill_barrier = self.fill_barrier.lock().unwrap_or_else(|p| p.into_inner()).clone(); let handle = tokio::spawn(async move { + // Hold the fill-concurrency permit until the register/insert/undo + // sequence completes, then release it on task exit. + let _permit = permit; // Register the key in the identity index BEFORE the entry becomes // visible in the cache, so a concurrent invalidation always finds it. match index.insert(fill_identity.clone(), fill_key.clone(), token).await { @@ -243,7 +273,7 @@ impl MokaBackend { let result = handle.await.unwrap_or(ObjectDataCacheFillResult::SkippedInvalidationRace); - leader.finish(result).await + leader.finish(result) } /// Conservatively invalidates all cached keys matching the object identity. @@ -254,11 +284,20 @@ impl MokaBackend { /// needed when the index has no entry for the identity. pub async fn invalidate_object(&self, identity: &ObjectDataCacheIdentity) -> ObjectDataCacheInvalidationResult { let keys_to_remove = self.index.remove_identity(identity).await; + // ODC-36: report the removal count so the facade can label the + // invalidation outcome (removed vs noop) and skip the gauge refresh on + // the no-op path. The vast majority of invalidations touch identities + // that were never cached. + let removed = keys_to_remove.len(); for key in keys_to_remove { self.cache.remove(&key).await; } - ObjectDataCacheInvalidationResult::Success + if removed > 0 { + ObjectDataCacheInvalidationResult::Removed { keys: removed } + } else { + ObjectDataCacheInvalidationResult::NoOp + } } } @@ -287,7 +326,9 @@ mod tests { ttl: Duration::from_millis(100), time_to_idle: Duration::from_millis(100), min_free_memory_percent: 0, - fill_concurrency_per_cpu: 1, + // >= 2 so concurrent-fill tests get at least two permits even on a + // single-CPU CI runner (permits = min(per_cpu * parallelism, max)). + fill_concurrency_per_cpu: 2, fill_concurrency_max: 32, identity_keys_max: 16, } @@ -335,17 +376,32 @@ mod tests { let _ = backend.fill_body(&plan_a, Bytes::from_static(b"aaaaa")).await; let _ = backend.fill_body(&plan_b, Bytes::from_static(b"bbbbb")).await; + // object-a has exactly one cached key, so invalidation reports Removed { keys: 1 }. let result = backend .invalidate_object(&ObjectDataCacheIdentity::new("bucket", "object-a")) .await; let lookup_a = backend.lookup_body(&plan_a).await; let lookup_b = backend.lookup_body(&plan_b).await; - assert!(matches!(result, ObjectDataCacheInvalidationResult::Success)); + assert_eq!(result, ObjectDataCacheInvalidationResult::Removed { keys: 1 }); assert!(matches!(lookup_a, ObjectDataCacheLookup::Miss)); assert!(matches!(lookup_b, ObjectDataCacheLookup::Hit(_))); } + // ODC-36: invalidating an identity that was never cached reports NoOp so the + // facade can skip the cache-state gauge refresh on the common no-op path. + #[tokio::test] + async fn moka_backend_invalidate_uncached_identity_is_noop() { + let backend = + MokaBackend::new(&enabled_config(), Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"); + + let result = backend + .invalidate_object(&ObjectDataCacheIdentity::new("bucket", "never-cached")) + .await; + + assert_eq!(result, ObjectDataCacheInvalidationResult::NoOp); + } + #[tokio::test] async fn moka_backend_expires_entries_by_ttl() { let backend = @@ -422,26 +478,81 @@ mod tests { assert_eq!(stats.snapshot().memory_pressure_events, 1); } - #[tokio::test] - async fn moka_backend_singleflight_waiter_observes_leader_result() { - let stats = Arc::new(ObjectDataCacheStats::default()); - let backend = Arc::new(MokaBackend::new(&enabled_config(), Arc::clone(&stats)).expect("moka backend should build")); + // ODC-15 + ODC-18: a concurrent duplicate fill for the same key must NOT + // wait for the in-flight leader (it already owns the body), and it must be + // reported as JoinedInflightFill — not Inserted — so it records no fill + // work and cannot inflate the inserted count/bytes for one real insert. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn moka_backend_duplicate_fill_same_key_joins_without_waiting() { + let backend = Arc::new( + MokaBackend::new(&enabled_config(), Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"), + ); let plan = cacheable_plan("object", "etag-a"); - let leader_plan = plan.clone(); - let plan_clone = plan.clone(); - let first = Arc::clone(&backend); - let second = Arc::clone(&backend); + let barrier = backend.install_fill_barrier(); - let leader = tokio::spawn(async move { first.fill_body(&leader_plan, Bytes::from_static(b"hello")).await }); - let waiter = tokio::spawn(async move { second.fill_body(&plan_clone, Bytes::from_static(b"hello")).await }); + let fill_a = { + let backend = Arc::clone(&backend); + let plan = plan.clone(); + tokio::spawn(async move { backend.fill_body(&plan, Bytes::from_static(b"hello")).await }) + }; - let leader_result = leader.await.expect("leader task should complete"); - let waiter_result = waiter.await.expect("waiter task should complete"); - let lookup = backend.lookup_body(&plan).await; + // A has registered its singleflight key and parked before the cache + // insert. A second fill for the same key must join (skip) right away. + barrier.wait_until_reached().await; + let fill_b = backend.fill_body(&plan, Bytes::from_static(b"hello")).await; + assert_eq!( + fill_b, + ObjectDataCacheFillResult::JoinedInflightFill, + "a duplicate fill must join the in-flight leader (not insert), and not wait for it" + ); - assert_eq!(leader_result, ObjectDataCacheFillResult::Inserted); - assert_eq!(waiter_result, ObjectDataCacheFillResult::Inserted); - assert!(matches!(lookup, ObjectDataCacheLookup::Hit(ref bytes) if bytes.as_ref() == b"hello")); + barrier.release(); + let fill_a = fill_a.await.expect("leader fill task should complete"); + assert_eq!(fill_a, ObjectDataCacheFillResult::Inserted); + assert!(matches!(backend.lookup_body(&plan).await, ObjectDataCacheLookup::Hit(ref bytes) if bytes.as_ref() == b"hello")); + } + + // ODC-11: the fill-concurrency limiter must reject (not queue) once + // saturated. With a single permit, an in-flight fill for one key holds the + // permit; a concurrent fill for a DISTINCT key is rejected immediately. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn moka_backend_rejects_fill_when_concurrency_saturated() { + let mut config = enabled_config(); + // One permit: min(per_cpu * parallelism, max) == min(N, 1) == 1. + config.fill_concurrency_per_cpu = 1; + config.fill_concurrency_max = 1; + let backend = + Arc::new(MokaBackend::new(&config, Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build")); + let plan_a = versioned_plan("object", "v1", "etag-a"); + let plan_b = versioned_plan("object", "v2", "etag-b"); + + let barrier = backend.install_fill_barrier(); + + let fill_a = { + let backend = Arc::clone(&backend); + let plan_a = plan_a.clone(); + tokio::spawn(async move { backend.fill_body(&plan_a, Bytes::from_static(b"aaaaa")).await }) + }; + + // A holds the only permit and is parked before the cache insert. Clear + // the barrier so B (a distinct key) runs without pausing; B must be + // rejected by the saturated limiter rather than queue behind A. + barrier.wait_until_reached().await; + *backend.fill_barrier.lock().unwrap() = None; + let fill_b = backend.fill_body(&plan_b, Bytes::from_static(b"bbbbb")).await; + assert_eq!( + fill_b, + ObjectDataCacheFillResult::SkippedFillConcurrency, + "a distinct-key fill must be rejected, not queued, when the limiter is saturated" + ); + assert!( + matches!(backend.lookup_body(&plan_b).await, ObjectDataCacheLookup::Miss), + "a rejected fill must not populate the cache" + ); + + barrier.release(); + let fill_a = fill_a.await.expect("leader fill task should complete"); + assert_eq!(fill_a, ObjectDataCacheFillResult::Inserted); } // ODC-12(1): the first refill after a TTL expiry must not self-destruct. diff --git a/crates/object-data-cache/src/noop.rs b/crates/object-data-cache/src/noop.rs index 0905aa559..ca8bf54f9 100644 --- a/crates/object-data-cache/src/noop.rs +++ b/crates/object-data-cache/src/noop.rs @@ -29,9 +29,10 @@ impl NoopBackend { ObjectDataCacheFillResult::SkippedDisabled } - /// Returns a successful no-op invalidation result. + /// Returns a no-op invalidation result: a disabled cache holds nothing, so + /// nothing is ever removed. pub async fn invalidate_object(&self) -> ObjectDataCacheInvalidationResult { - ObjectDataCacheInvalidationResult::Success + ObjectDataCacheInvalidationResult::NoOp } } @@ -53,6 +54,6 @@ mod tests { assert!(matches!(lookup, ObjectDataCacheLookup::SkipDisabled)); assert!(matches!(fill, ObjectDataCacheFillResult::SkippedDisabled)); - assert!(matches!(invalidation, ObjectDataCacheInvalidationResult::Success)); + assert!(matches!(invalidation, ObjectDataCacheInvalidationResult::NoOp)); } } diff --git a/crates/object-data-cache/src/singleflight.rs b/crates/object-data-cache/src/singleflight.rs index 32b89bb90..f7ec5c363 100644 --- a/crates/object-data-cache/src/singleflight.rs +++ b/crates/object-data-cache/src/singleflight.rs @@ -16,22 +16,24 @@ use crate::cache::ObjectDataCacheFillResult; use crate::key::ObjectDataCacheKey; use crate::metrics::{record_singleflight_join, set_inflight_fills}; use crate::stats::ObjectDataCacheStats; -use std::collections::HashMap; +use std::collections::HashSet; use std::sync::{Arc, Mutex}; -use tokio::sync::watch; -type FillMap = Mutex>>>; +type FillSet = Mutex>; -fn lock_fills( - fills: &FillMap, -) -> std::sync::MutexGuard<'_, HashMap>>> { +fn lock_fills(fills: &FillSet) -> std::sync::MutexGuard<'_, HashSet> { fills.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) } -/// Shared singleflight controller for cache fill operations. +/// Shared singleflight controller that dedups concurrent cache fills per key. +/// +/// The fill path already holds the fully materialized body, so a non-leader +/// gains nothing by waiting for the leader's result — a duplicate fill is +/// simply skipped. The controller therefore only elects a leader per key and +/// reports [`Busy`](ObjectDataCacheSingleflightAcquire::Busy) to everyone else. #[derive(Debug)] pub struct ObjectDataCacheSingleflight { - fills: FillMap, + fills: FillSet, stats: Arc, } @@ -39,31 +41,46 @@ impl ObjectDataCacheSingleflight { /// Creates a new singleflight controller. pub fn new(stats: Arc) -> Self { Self { - fills: Mutex::new(HashMap::new()), + fills: Mutex::new(HashSet::new()), stats, } } - /// Acquires a leader-or-waiter role for the supplied cache key. - pub async fn acquire(&self, key: ObjectDataCacheKey) -> ObjectDataCacheSingleflightAcquire<'_> { - let mut fills = lock_fills(&self.fills); - if let Some(sender) = fills.get(&key) { - record_singleflight_join(&self.stats); - return ObjectDataCacheSingleflightAcquire::Waiter(ObjectDataCacheSingleflightWaiter { rx: sender.subscribe() }); + /// Tries to become the leader for the supplied cache key without blocking. + /// + /// Returns [`Leader`](ObjectDataCacheSingleflightAcquire::Leader) when no + /// fill for the key is in flight, otherwise + /// [`Busy`](ObjectDataCacheSingleflightAcquire::Busy). The caller already + /// owns the body, so a `Busy` outcome skips the redundant fill rather than + /// waiting for another request's leader to finish. + pub fn try_acquire(&self, key: ObjectDataCacheKey) -> ObjectDataCacheSingleflightAcquire<'_> { + // Keep the critical section to the map mutation only; emit metrics after + // dropping the guard so the recorder round-trip never serializes fills. + let inflight_len = { + let mut fills = lock_fills(&self.fills); + if fills.contains(&key) { + None + } else { + fills.insert(key.clone()); + Some(fills.len()) + } + }; + + match inflight_len { + None => { + record_singleflight_join(&self.stats); + ObjectDataCacheSingleflightAcquire::Busy + } + Some(len) => { + set_inflight_fills(&self.stats, "moka", len); + ObjectDataCacheSingleflightAcquire::Leader(ObjectDataCacheSingleflightLeader { + key, + fills: &self.fills, + stats: Arc::clone(&self.stats), + finished: false, + }) + } } - - let (tx, _rx) = watch::channel(None); - fills.insert(key.clone(), tx.clone()); - set_inflight_fills(&self.stats, "moka", fills.len()); - drop(fills); - - ObjectDataCacheSingleflightAcquire::Leader(ObjectDataCacheSingleflightLeader { - key, - tx, - fills: &self.fills, - stats: Arc::clone(&self.stats), - finished: false, - }) } /// Returns whether a leader fill is currently in flight for the key. @@ -72,75 +89,57 @@ impl ObjectDataCacheSingleflight { /// (a different key under the same identity that has registered in the index /// but not yet published its cache entry) from being pruned as stale. pub fn has_inflight(&self, key: &ObjectDataCacheKey) -> bool { - lock_fills(&self.fills).contains_key(key) + lock_fills(&self.fills).contains(key) } } -/// Leader or waiter result from the singleflight map. +/// Leader-or-busy result from the singleflight controller. pub enum ObjectDataCacheSingleflightAcquire<'a> { /// Caller is responsible for performing the fill operation. Leader(ObjectDataCacheSingleflightLeader<'a>), - /// Caller must wait for the leader's result. - Waiter(ObjectDataCacheSingleflightWaiter), + /// A fill for the same key is already in flight; the caller skips its own. + Busy, } /// Leader handle for a singleflight fill operation. pub struct ObjectDataCacheSingleflightLeader<'a> { key: ObjectDataCacheKey, - tx: watch::Sender>, - fills: &'a FillMap, + fills: &'a FillSet, stats: Arc, finished: bool, } impl<'a> ObjectDataCacheSingleflightLeader<'a> { - /// Completes the leader operation and publishes the shared result. - pub async fn finish(mut self, result: ObjectDataCacheFillResult) -> ObjectDataCacheFillResult { + /// Completes the leader operation and releases the key. + pub fn finish(mut self, result: ObjectDataCacheFillResult) -> ObjectDataCacheFillResult { self.remove_entry(); self.finished = true; - let _ = self.tx.send(Some(result.clone())); result } fn remove_entry(&self) { - let mut fills = lock_fills(self.fills); - fills.remove(&self.key); - set_inflight_fills(&self.stats, "moka", fills.len()); + // Capture the length under the guard, then emit the gauge after dropping + // it so the recorder round-trip stays out of the critical section. + let len = { + let mut fills = lock_fills(self.fills); + fills.remove(&self.key); + fills.len() + }; + set_inflight_fills(&self.stats, "moka", len); } } impl<'a> Drop for ObjectDataCacheSingleflightLeader<'a> { fn drop(&mut self) { // A leader dropped without finish() was cancelled mid-fill (e.g. the - // GET request future was aborted). Remove the map entry so its sender - // clone is released and waiters observe the closed channel instead of - // blocking forever, and so a later fill can become the new leader. + // fill task was aborted). Release the key so a later fill can become the + // new leader instead of seeing a phantom in-flight entry forever. if !self.finished { self.remove_entry(); } } } -/// Waiter handle for a singleflight fill operation. -pub struct ObjectDataCacheSingleflightWaiter { - rx: watch::Receiver>, -} - -impl ObjectDataCacheSingleflightWaiter { - /// Waits for the shared fill result. - pub async fn wait(mut self) -> ObjectDataCacheFillResult { - if self.rx.wait_for(|value| value.is_some()).await.is_err() { - return ObjectDataCacheFillResult::SkippedSingleflightClosed; - } - - self.rx - .borrow() - .as_ref() - .cloned() - .unwrap_or(ObjectDataCacheFillResult::SkippedSingleflightClosed) - } -} - #[cfg(test)] mod tests { use super::{ObjectDataCacheSingleflight, ObjectDataCacheSingleflightAcquire}; @@ -153,55 +152,72 @@ mod tests { ObjectDataCacheKey::new("bucket", "object", None, "etag", 1, ObjectDataCacheBodyVariant::FullObjectPlainV1) } - #[tokio::test] - async fn singleflight_returns_waiter_for_second_caller() { + #[test] + fn first_caller_leads_and_second_caller_is_busy() { let stats = Arc::new(ObjectDataCacheStats::default()); let singleflight = ObjectDataCacheSingleflight::new(Arc::clone(&stats)); - let first = singleflight.acquire(key()).await; - let second = singleflight.acquire(key()).await; - - let leader = match first { - ObjectDataCacheSingleflightAcquire::Leader(leader) => leader, - ObjectDataCacheSingleflightAcquire::Waiter(_) => panic!("first caller must become leader"), - }; - let waiter = match second { - ObjectDataCacheSingleflightAcquire::Leader(_) => panic!("second caller must become waiter"), - ObjectDataCacheSingleflightAcquire::Waiter(waiter) => waiter, - }; - - let waiter_task = tokio::spawn(async move { waiter.wait().await }); - let leader_result = leader.finish(ObjectDataCacheFillResult::Inserted).await; - let waiter_result = waiter_task.await.expect("waiter task should complete"); - - assert_eq!(leader_result, ObjectDataCacheFillResult::Inserted); - assert_eq!(waiter_result, ObjectDataCacheFillResult::Inserted); - assert_eq!(stats.snapshot().singleflight_joins, 1); - } - - #[tokio::test] - async fn cancelled_leader_releases_key_and_unblocks_waiters() { - let stats = Arc::new(ObjectDataCacheStats::default()); - let singleflight = ObjectDataCacheSingleflight::new(Arc::clone(&stats)); - - let leader = match singleflight.acquire(key()).await { - ObjectDataCacheSingleflightAcquire::Leader(leader) => leader, - ObjectDataCacheSingleflightAcquire::Waiter(_) => panic!("first caller must become leader"), - }; - let waiter = match singleflight.acquire(key()).await { - ObjectDataCacheSingleflightAcquire::Leader(_) => panic!("second caller must become waiter"), - ObjectDataCacheSingleflightAcquire::Waiter(waiter) => waiter, - }; - - // Dropping without finish() simulates the leader future being cancelled. - drop(leader); - - let waiter_result = waiter.wait().await; - assert_eq!(waiter_result, ObjectDataCacheFillResult::SkippedSingleflightClosed); + let first = singleflight.try_acquire(key()); + let second = singleflight.try_acquire(key()); assert!( - matches!(singleflight.acquire(key()).await, ObjectDataCacheSingleflightAcquire::Leader(_)), + matches!(first, ObjectDataCacheSingleflightAcquire::Leader(_)), + "first caller must become leader" + ); + assert!( + matches!(second, ObjectDataCacheSingleflightAcquire::Busy), + "second caller must not wait; it is busy and skips" + ); + assert_eq!(stats.snapshot().singleflight_joins, 1); + + // Completing the leader releases the key for a subsequent fill. + let ObjectDataCacheSingleflightAcquire::Leader(leader) = first else { + unreachable!("first caller must be the leader"); + }; + let result = leader.finish(ObjectDataCacheFillResult::Inserted); + assert_eq!(result, ObjectDataCacheFillResult::Inserted); + + assert!( + matches!(singleflight.try_acquire(key()), ObjectDataCacheSingleflightAcquire::Leader(_)), + "key must be released after the leader finishes" + ); + } + + #[test] + fn cancelled_leader_releases_key() { + let stats = Arc::new(ObjectDataCacheStats::default()); + let singleflight = ObjectDataCacheSingleflight::new(Arc::clone(&stats)); + + let leader = match singleflight.try_acquire(key()) { + ObjectDataCacheSingleflightAcquire::Leader(leader) => leader, + ObjectDataCacheSingleflightAcquire::Busy => panic!("first caller must become leader"), + }; + assert!(singleflight.has_inflight(&key()), "leader must register the key as in-flight"); + + // Dropping without finish() simulates the fill task being cancelled. + drop(leader); + + assert!(!singleflight.has_inflight(&key()), "a cancelled leader must release the key"); + assert!( + matches!(singleflight.try_acquire(key()), ObjectDataCacheSingleflightAcquire::Leader(_)), "key must be released after a cancelled leader" ); } + + #[test] + fn distinct_keys_lead_independently() { + let stats = Arc::new(ObjectDataCacheStats::default()); + let singleflight = ObjectDataCacheSingleflight::new(Arc::clone(&stats)); + let other = ObjectDataCacheKey::new("bucket", "other", None, "etag", 1, ObjectDataCacheBodyVariant::FullObjectPlainV1); + + let first = singleflight.try_acquire(key()); + let second = singleflight.try_acquire(other); + + assert!(matches!(first, ObjectDataCacheSingleflightAcquire::Leader(_))); + assert!( + matches!(second, ObjectDataCacheSingleflightAcquire::Leader(_)), + "a distinct key must not be deduped against another in-flight fill" + ); + assert_eq!(stats.snapshot().singleflight_joins, 0); + } } diff --git a/rustfs/src/app/object_data_cache/invalidation.rs b/rustfs/src/app/object_data_cache/invalidation.rs index c9934923c..138157be2 100644 --- a/rustfs/src/app/object_data_cache/invalidation.rs +++ b/rustfs/src/app/object_data_cache/invalidation.rs @@ -147,12 +147,17 @@ mod tests { body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let fill = adapter.fill_body(&plan, Bytes::from_static(b"hello")).await; - let _ = invalidate_object_data_cache_before_mutation(&adapter, "bucket", "object").await; - let invalidation = invalidate_object_data_cache_after_delete_success(&adapter, "bucket", "object").await; + let before_mutation = invalidate_object_data_cache_before_mutation(&adapter, "bucket", "object").await; + let after_delete = invalidate_object_data_cache_after_delete_success(&adapter, "bucket", "object").await; let lookup = adapter.lookup_body(&plan).await; assert_eq!(fill, ObjectDataCacheFillResult::Inserted); - assert_eq!(invalidation, ObjectDataCacheInvalidationResult::Success); + // The two-phase invalidation is deliberate hygiene, and the outcomes must + // distinguish real churn from its no-op half: the first call evicts the + // cached body, the second finds the identity already empty. Dashboards + // read that split (backlog#1141). + assert_eq!(before_mutation, ObjectDataCacheInvalidationResult::Removed { keys: 1 }); + assert_eq!(after_delete, ObjectDataCacheInvalidationResult::NoOp); assert!(matches!(lookup, ObjectDataCacheLookup::Miss)); } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 4b710ef59..8745a5c7d 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -3130,7 +3130,7 @@ impl DefaultObjectUsecase { #[allow(clippy::too_many_arguments)] async fn build_get_object_body_with_cache( cache_adapter: &ObjectDataCacheAdapter, - mut final_stream: R, + final_stream: R, info: &ObjectInfo, response_content_length: i64, optimal_buffer_size: usize, @@ -3171,7 +3171,21 @@ impl DefaultObjectUsecase { } if let Some(buffered_body) = buffered_body { - let _fill_result = fill_get_object_body_cache_from_buffered_body(cache_adapter, &cache_plan, &buffered_body).await; + // ODC-15: the body is already fully in hand, so keep the fill off the + // response's critical path. For a cacheable plan, run the fill in a + // detached task (Bytes is a cheap clone) and return immediately. For + // a non-cacheable plan the fill is a pure metric-only skip with no + // I/O, so record it inline to preserve observability. + if matches!(cache_plan, GetObjectBodyCachePlan::Cacheable(_)) { + let cache_adapter = cache_adapter.clone(); + let cache_plan = cache_plan.clone(); + let fill_bytes = buffered_body.clone(); + tokio::spawn(async move { + let _ = fill_get_object_body_cache_from_buffered_body(&cache_adapter, &cache_plan, &fill_bytes).await; + }); + } else { + let _ = fill_get_object_body_cache_from_buffered_body(cache_adapter, &cache_plan, &buffered_body).await; + } return Ok(Self::build_memory_bytes_blob( buffered_body, @@ -3215,23 +3229,43 @@ impl DefaultObjectUsecase { .await; }; let mut buf = Vec::with_capacity(materialized_capacity); + // ODC-07: bound the read so a stream yielding more than the declared + // length cannot grow `buf` past the in-memory GET threshold. Reading + // one byte past the capacity is enough to detect an over-long stream. + let mut bounded_stream = tokio::io::AsyncReadExt::take(final_stream, materialized_capacity as u64 + 1); let buffer_read_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); - let read_result = tokio::io::AsyncReadExt::read_to_end(&mut final_stream, &mut buf).await; + let read_result = tokio::io::AsyncReadExt::read_to_end(&mut bounded_stream, &mut buf).await; record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_CACHE_MATERIALIZE_READ, buffer_read_start); match read_result { Ok(_) => { + // ODC-07: treat a length mismatch as a hard error, matching + // the direct-memory GET path. Serving a body whose decoded + // length disagrees with the declared content length would + // ship a truncated or over-long response. if buf.len() != materialized_capacity { - warn!( + lifecycle.finish_err(); + error!( expected = response_content_length, actual = buf.len(), - "Object size mismatch during materialize-fill read" + "materialize-fill GET decoded length mismatch" ); + return Err(ApiError::from(StorageError::other(format!( + "materialize-fill GET decoded length mismatch: expected {response_content_length}, got {}", + buf.len() + ))) + .into()); } let bytes = Bytes::from(buf); - let _fill_result = - fill_get_object_body_cache_from_materialized_body(cache_adapter, &cache_plan, &bytes).await; + // ODC-15: fill off the response's critical path (see the + // buffered-body branch above). + let cache_adapter = cache_adapter.clone(); + let cache_plan = cache_plan.clone(); + let fill_bytes = bytes.clone(); + tokio::spawn(async move { + let _ = fill_get_object_body_cache_from_materialized_body(&cache_adapter, &cache_plan, &fill_bytes).await; + }); return Ok(Self::build_memory_bytes_blob( bytes, @@ -7075,6 +7109,32 @@ mod tests { )); } + /// Polls the cache until the detached fill (ODC-15) populates the entry, so + /// a follow-up GET is a deterministic hit rather than racing the fill task. + async fn wait_for_cache_hit( + adapter: &crate::app::object_data_cache::ObjectDataCacheAdapter, + bucket: &str, + object: &str, + etag: &str, + size: u64, + ) { + let plan = adapter.plan_get(rustfs_object_data_cache::ObjectDataCacheGetRequest { + bucket, + object, + version_id: None, + etag, + size, + body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, + }); + for _ in 0..400 { + if matches!(adapter.lookup_body(&plan).await, rustfs_object_data_cache::ObjectDataCacheLookup::Hit(_)) { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + panic!("detached fill did not populate the cache within the timeout"); + } + struct ReadProbeReader { reads: Arc, } @@ -7612,6 +7672,10 @@ mod tests { .await .expect("buffered-body handoff should succeed"); + // ODC-15: the fill is detached from the response path, so wait for it to + // populate the cache before the follow-up GET to keep the hit deterministic. + wait_for_cache_hit(&adapter, "test-bucket", "cached-object", "etag", 5).await; + let _second_body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, second_reader, @@ -7748,6 +7812,10 @@ mod tests { .await .expect("materialize-fill handoff should succeed"); + // ODC-15: the fill is detached from the response path, so wait for it to + // populate the cache before the follow-up GET to keep the hit deterministic. + wait_for_cache_hit(&adapter, "test-bucket", "materialized-object", "etag", 5).await; + let _second_body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, second_reader, @@ -7779,6 +7847,56 @@ mod tests { ); } + // ODC-07: a materialize read that yields more than the declared content + // length must be a hard error, not a warn-and-serve, matching the + // direct-memory GET path. The bounded `take` reads one byte past capacity so + // the over-long stream is detected without buffering it unbounded. + #[tokio::test] + async fn build_get_object_body_with_cache_materialize_rejects_length_mismatch() { + let reads = Arc::new(AtomicUsize::new(0)); + // Declared content length is 5, but the stream yields 6 bytes. + let reader = DataProbeReader { + reads: Arc::clone(&reads), + data: std::io::Cursor::new(b"hello!".to_vec()), + }; + let info = ObjectInfo { + size: 5, + etag: Some("etag".to_string()), + ..Default::default() + }; + let adapter = + crate::app::object_data_cache::ObjectDataCacheAdapter::new(rustfs_object_data_cache::ObjectDataCacheConfig { + mode: rustfs_object_data_cache::ObjectDataCacheMode::FillMaterializeEnabled, + max_bytes: 8_388_608, + min_free_memory_percent: 0, + ..rustfs_object_data_cache::ObjectDataCacheConfig::default() + }) + .expect("materialize-fill cache adapter should initialize"); + + let result = DefaultObjectUsecase::build_get_object_body_with_cache( + &adapter, + reader, + &info, + 5, + 128 * 1024, + false, + 1, + None, + false, + false, + None, + "test-bucket", + "mismatch-object", + GetObjectBodyLifecycle::disabled(), + ) + .await; + + assert!( + result.is_err(), + "an over-long materialize read must be a hard error, not a truncated served body" + ); + } + #[tokio::test] async fn build_get_object_body_with_cache_skips_materialize_when_too_large_for_cache() { let reads = Arc::new(AtomicUsize::new(0));