From b60407421766abc560721b7a2f241abba2ae1aaa Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 10 Jul 2026 20:25:23 +0800 Subject: [PATCH] perf(object-data-cache): take fill off the GET path and fix the metrics (#4678) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(object-data-cache): make cache metrics one-increment-per-GET Rework the object data cache observability so each counter carries one clear meaning, and drop dead per-entry state. ODC-17 (backlog#1122): split requests_total, which was incremented by both plan_get and lookup_body, into plan_total{decision,reason,size_class} (emitted only by plan_get) and lookup_total{result,size_class} (emitted only by lookup_body). Each is now incremented exactly once per GET per layer; help text states this. ODC-18 (backlog#1123): add a JoinedInflightFill result (label joined_inflight) so a singleflight waiter is no longer counted as an insert. record_fill_result now always counts the outcome but records fill bytes only when non-zero and the duration histogram only when present, so joined_inflight / skipped_by_mode / skipped_size_mismatch no longer inflate fill_bytes_total or add non-fill duration samples. The waiter->JoinedInflightFill mapping lives in MokaBackend (owned by a concurrent branch); cache.rs handles the variant already. ODC-29 (backlog#1134): stop refreshing the cache-state gauge on every lookup, and debounce it on fill/invalidate to at most once per second via an AtomicU64 millis timestamp, since moka's entry_count is a settling approximation. ODC-36 (backlog#1141): give invalidations_total an outcome label (removed|noop) and skip the gauge refresh on the no-op path. Extend ObjectDataCacheInvalidationResult with Removed{keys}/NoOp (Success kept as a transitional variant until MokaBackend reports the removal count); NoopBackend now reports NoOp. ODC-30 (backlog#1141): drop the dead content_length/etag/inserted_at fields (and getters) from ObjectDataCacheEntry, which are redundant with the moka key identity; the constructor keeps its arity so the out-of-scope MokaBackend caller still compiles. Gate is_null_version behind cfg(test). Tests use a thread-local metrics recorder to avoid the global-singleton flakiness. Fill-enabled test configs set min_free_memory_percent=0 so the cgroup gate does not refuse fills in CI pods. Co-Authored-By: heihutu * fix(object-data-cache): move fill off GET path, bound & harden the fill Implements five object-data-cache audit findings. ODC-15 (backlog#1120): the GET response no longer blocks on cache fill. The buffered/materialized body is already in hand, so the fill now runs in a detached task and the response is built immediately. Singleflight is made non-blocking: `try_acquire` returns Leader or Busy, and a non-leader skips its own fill (SkippedSingleflightBusy) instead of waiting on another request's leader. The inner fill spawn is KEPT: once the index key is registered, the recheck/undo must complete even if the enclosing fill future is aborted, so removing it would weaken the cancellation-safety guarantee (ODC-13 test) and the invalidation-race undo. ODC-11 (backlog#1116): enforce the fill-concurrency knobs. MokaBackend now holds a Semaphore sized min(per_cpu * parallelism, max), acquired after winning leadership and before the memory gate. On saturation it rejects (try_acquire_owned) with SkippedFillConcurrency rather than queueing, so the fill path never reintroduces GET-latency coupling. ODC-14 (backlog#1119): the memory gate no longer does a blocking sysinfo refresh under a mutex on the async fill path. A dedicated periodic refresher (tokio interval + spawn_blocking) updates an atomic snapshot every 5s; allows_fill is now lock-free. The min_free_memory_percent == 0 short-circuit still runs before any snapshot read. No runtime at construction (sync unit tests) simply keeps the seed snapshot. ODC-32 (backlog#1137): singleflight no longer emits metrics while holding the map mutex. try_acquire/remove_entry capture the map length under the guard, drop it, then emit the gauge/counter. ODC-07 (backlog#1112): the materialize read is bounded via take(capacity + 1) so an over-long stream cannot grow the buffer past the in-memory GET threshold, and any length mismatch is now a hard error matching the direct-memory GET path, instead of warn-and-serve. cache.rs is owned by a concurrent branch; this commit only appends the SkippedFillConcurrency and SkippedSingleflightBusy variants + label arms. Co-Authored-By: heihutu * fix(object-data-cache): report JoinedInflightFill and invalidation outcome Reconciles the moka_backend half of two metrics-semantics findings after merging fix/odc-metrics-semantics (which landed the cache.rs half). ODC-18 (backlog#1123): the singleflight non-leader path now returns JoinedInflightFill instead of counting as an insert, so N concurrent GETs of one cold key record one insert, not N. This composes with the ODC-15 redesign: the non-leader does NOT wait for the leader (it already owns the body and never re-serves from the fill result), so no blocking wait is reintroduced. Drops the redundant local SkippedSingleflightBusy variant in favor of the canonical JoinedInflightFill (mapped to no bytes / no duration by the facade). SkippedFillConcurrency (ODC-11) is retained. ODC-36 (backlog#1141): MokaBackend::invalidate_object now returns Removed { keys } / NoOp instead of the transitional Success, so the facade labels invalidations_total{outcome=removed|noop} and skips the cache-state gauge refresh on the no-op path. Tests: duplicate-fill test asserts JoinedInflightFill (bites when reverted to Inserted); new no-op invalidation test; matching-identity test asserts Removed { keys: 1 }. Co-Authored-By: heihutu * fix(object-data-cache): drop unused mut on materialize stream binding The ODC-07 change moves `final_stream` into `AsyncReadExt::take`, so the `build_get_object_body_with_cache` parameter is no longer mutated in place. Co-Authored-By: heihutu * refactor(object-data-cache): close the cross-branch coordination seams Merging the metrics and hot-path branches left four transitional shims that only existed because each side could not edit the other's files. With both sides present they are dead weight, and two of them actively misreport. - ObjectDataCacheEntry::new took content_length and etag only to keep the caller in moka_backend.rs compiling, then discarded both. Drop them; the entry is a Bytes wrapper and the caller now passes only the body. - SkippedSingleflightClosed lost its only producer when the waiter model was replaced by Leader/Busy election. Remove the variant. - The fill-accounting match ended in a wildcard that charged fill_bytes and a duration to every non-Inserted outcome, so a fill rejected by the memory gate or the concurrency semaphore — which never touched the backend and wrote nothing — still inflated fill_bytes_total. Enumerate every variant explicitly: only outcomes that wrote the body report bytes and duration. Exhaustiveness also forces a future variant to state its accounting rather than inherit a wrong default. - InvalidationResult::Success mapped a no-op to outcome=removed, the exact lie backlog#1141 set out to fix, and its doc comment claimed MokaBackend still returned it after that backend had been widened. It has no producer left. Remove it, and tighten the app-layer test from "any successful variant" to the real contract: the pre-mutation call reports Removed{keys:1}, the post-delete call NoOp. Refs: backlog#1123, backlog#1141, backlog#1135 Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- Cargo.lock | 1 + crates/object-data-cache/Cargo.toml | 3 +- crates/object-data-cache/src/cache.rs | 348 ++++++++++++++++-- crates/object-data-cache/src/entry.rs | 40 +- crates/object-data-cache/src/key.rs | 4 + crates/object-data-cache/src/memory.rs | 215 ++++++++--- crates/object-data-cache/src/metrics.rs | 184 +++++++-- crates/object-data-cache/src/moka_backend.rs | 175 +++++++-- crates/object-data-cache/src/noop.rs | 7 +- crates/object-data-cache/src/singleflight.rs | 228 ++++++------ .../src/app/object_data_cache/invalidation.rs | 11 +- rustfs/src/app/object_usecase.rs | 132 ++++++- 12 files changed, 1061 insertions(+), 287 deletions(-) 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));