diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 9b7987387..d018c2cc2 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -351,8 +351,8 @@ pub mod notification { pub mod object { pub use crate::object_api::{ - BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectBodyCacheHook, GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, - RangedDecompressReader, StreamConsumer, register_get_object_body_cache_hook, + BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectBodyCacheHook, GetObjectReader, ObjectInfo, ObjectMutationHook, ObjectOptions, + PutObjReader, RangedDecompressReader, StreamConsumer, register_get_object_body_cache_hook, register_object_mutation_hook, }; } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 8bfb9af2c..fc51e33c8 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -2261,6 +2261,9 @@ pub async fn expire_transitioned_object( opts.transition.expire_restored = true; return match api.delete_object(&oi.bucket, &oi.name, opts).await { Ok(dobj) => { + // Drop any cached restored-copy body so it does not sit resident + // until TTL after the copy is expired (ODC-26). + crate::object_api::notify_object_mutation(&oi.bucket, &oi.name).await; //audit_log_lifecycle(*oi, ILMExpiry, tags, traceFn); Ok(dobj) } @@ -2293,6 +2296,10 @@ pub async fn expire_transitioned_object( schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await; + // The transitioned version is gone; evict any cached body for this object + // so it does not linger until TTL (ODC-26). + crate::object_api::notify_object_mutation(&oi.bucket, &oi.name).await; + //audit_log_lifecycle(oi, ILMExpiry, tags); emit_transitioned_expiration_event(oi, &dobj); @@ -2802,6 +2809,13 @@ pub async fn apply_expiry_on_non_transitioned_objects( } }; schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await; + + // The object (or all its versions, for delete_all) was expired; evict any + // cached body so dead bytes do not sit resident until TTL (ODC-26). The + // cache identity is the decoded object name used by GET, not the + // encode_dir_object form passed to delete_object. + crate::object_api::notify_object_mutation(&oi.bucket, &oi.name).await; + //debug!("dobj: {:?}", dobj); if dobj.name.is_empty() { dobj = oi.clone(); diff --git a/crates/ecstore/src/client/object_handlers_common.rs b/crates/ecstore/src/client/object_handlers_common.rs index c51d4f1bf..7ac4595a4 100644 --- a/crates/ecstore/src/client/object_handlers_common.rs +++ b/crates/ecstore/src/client/object_handlers_common.rs @@ -103,6 +103,11 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ if errors.get(i).and_then(|err| err.as_ref()).is_some() { continue; } + // Evict any cached body for the successfully deleted noncurrent + // version so it does not sit resident until TTL (ODC-26). + if let Some(target) = to_del.get(i) { + crate::object_api::notify_object_mutation(bucket, &target.object_name).await; + } let Some(replication_state) = replication_candidates.get(i).and_then(|c| c.clone()) else { continue; }; diff --git a/crates/ecstore/src/object_api/body_cache_hook.rs b/crates/ecstore/src/object_api/body_cache_hook.rs index d7cb265f6..a56ec614a 100644 --- a/crates/ecstore/src/object_api/body_cache_hook.rs +++ b/crates/ecstore/src/object_api/body_cache_hook.rs @@ -21,8 +21,9 @@ //! (after the reader is built) means a hit no longer saves any disk I/O. use crate::object_api::ObjectInfo; +use crate::object_api::hook_slot::HookSlot; use bytes::Bytes; -use std::sync::{Arc, RwLock}; +use std::sync::Arc; /// Serves full-object GET bodies from a cache keyed by object identity. /// @@ -35,14 +36,15 @@ pub trait GetObjectBodyCacheHook: Send + Sync + 'static { async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option; } -// `RwLock>>` rather than `ArcSwapOption`: arc-swap's -// `RefCnt` is implemented only for the sized `Arc` (it stores a thin -// `*mut T`), so it cannot hold an `Arc` without a -// sized newtype wrapper. The probe reads this slot once per full-object GET, -// but the read guard only clones an `Arc`, which is negligible next to the -// metadata quorum fan-out already completed before the probe. Registration is -// a startup / config-reload event, so writer contention is a non-issue. -static GET_OBJECT_BODY_CACHE_HOOK: RwLock>> = RwLock::new(None); +// A `HookSlot` (RwLock>>) rather than `ArcSwapOption`: +// arc-swap's `RefCnt` is implemented only for the sized `Arc` (it stores a +// thin `*mut T`), so it cannot hold an `Arc` +// without a sized newtype wrapper. The probe reads this slot once per +// full-object GET, but the read guard only clones an `Arc`, which is negligible +// next to the metadata quorum fan-out already completed before the probe. +// Registration is a startup / config-reload event, so writer contention is a +// non-issue. +static GET_OBJECT_BODY_CACHE_HOOK: HookSlot = HookSlot::new(); /// Register (or re-register) the process-wide GET body cache hook. /// @@ -52,31 +54,22 @@ static GET_OBJECT_BODY_CACHE_HOOK: RwLock /// to the original adapter while every usecase-layer fill and invalidation /// targeted the replacement, silently degrading the feature to a 0% hit rate /// and stranding entries in the unreachable cache until their TTL (backlog#1126). -/// -/// Replacing a *different* hook instance is logged at WARN: in production the -/// hook is installed exactly once per process, so a swap to a distinct instance -/// signals an unexpected re-init and orphans the previous adapter's cache. pub fn register_get_object_body_cache_hook(hook: Arc) { - let mut slot = GET_OBJECT_BODY_CACHE_HOOK.write().unwrap_or_else(|e| e.into_inner()); - if let Some(previous) = slot.as_ref() - && !Arc::ptr_eq(previous, &hook) - { - tracing::warn!( - "GET object body cache hook re-registered with a different instance; \ - the previous adapter's cache is now unreachable by ecstore's GET probe" - ); - } - *slot = Some(hook); + GET_OBJECT_BODY_CACHE_HOOK.register( + hook, + "GET object body cache hook re-registered with a different instance; \ + the previous adapter's cache is now unreachable by ecstore's GET probe", + ); } /// The registered hook, if any. pub(crate) fn get_object_body_cache_hook() -> Option> { - GET_OBJECT_BODY_CACHE_HOOK.read().unwrap_or_else(|e| e.into_inner()).clone() + GET_OBJECT_BODY_CACHE_HOOK.get() } /// Test-only: unregister the hook so tests can register and clear the slot /// deterministically without leaking a hook into unrelated tests. #[cfg(test)] pub(crate) fn clear_get_object_body_cache_hook() { - *GET_OBJECT_BODY_CACHE_HOOK.write().unwrap_or_else(|e| e.into_inner()) = None; + GET_OBJECT_BODY_CACHE_HOOK.clear(); } diff --git a/crates/ecstore/src/object_api/hook_slot.rs b/crates/ecstore/src/object_api/hook_slot.rs new file mode 100644 index 000000000..9ac2e99e8 --- /dev/null +++ b/crates/ecstore/src/object_api/hook_slot.rs @@ -0,0 +1,126 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! A process-wide, re-registrable slot for a single `Arc` hook. +//! +//! The GET body cache hook and the object mutation hook both need the same +//! shape: one global slot that starts empty, is (re-)registered from the app +//! layer at startup, is read on the hot/delete paths, and warns when a +//! re-registration swaps in a *different* instance (an unexpected re-init that +//! orphans the previous adapter's cache). This type holds that logic once so +//! the two callers cannot drift apart. + +use std::sync::{Arc, RwLock}; + +/// A global slot holding at most one `Arc` hook. +/// +/// Registration atomically swaps the stored hook. Poisoned locks recover in +/// place: a panicked writer cannot leave an `Option>` in an unsound +/// state, so there is nothing to salvage. +pub(crate) struct HookSlot { + inner: RwLock>>, +} + +impl HookSlot { + /// Creates an empty slot. `const` so it can initialize a `static`. + pub(crate) const fn new() -> Self { + Self { + inner: RwLock::new(None), + } + } + + /// Registers (or re-registers) the hook. + /// + /// Replacing a *different* instance logs `warn_on_replace` at WARN: in + /// production a hook is installed exactly once per process, so a swap to a + /// distinct instance signals an unexpected re-init that leaves the previous + /// adapter's cache unreachable. + pub(crate) fn register(&self, hook: Arc, warn_on_replace: &str) { + let mut slot = self.inner.write().unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(previous) = slot.as_ref() + && !Arc::ptr_eq(previous, &hook) + { + tracing::warn!("{warn_on_replace}"); + } + *slot = Some(hook); + } + + /// Returns the registered hook, if any. + pub(crate) fn get(&self) -> Option> { + self.inner.read().unwrap_or_else(|poisoned| poisoned.into_inner()).clone() + } + + /// Clears the slot. Test-only: production never unregisters a hook. + #[cfg(test)] + pub(crate) fn clear(&self) { + *self.inner.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + } +} + +#[cfg(test)] +mod tests { + use super::HookSlot; + use std::sync::Arc; + + trait Marker: Send + Sync { + fn id(&self) -> u32; + } + struct Impl(u32); + impl Marker for Impl { + fn id(&self) -> u32 { + self.0 + } + } + + #[test] + fn empty_slot_reads_none() { + let slot: HookSlot = HookSlot::new(); + assert!(slot.get().is_none()); + } + + #[test] + fn register_then_get_returns_the_hook() { + let slot: HookSlot = HookSlot::new(); + slot.register(Arc::new(Impl(7)), "unused"); + assert_eq!(slot.get().map(|h| h.id()), Some(7)); + } + + #[test] + fn re_registration_swaps_to_the_latest_instance() { + // The load-bearing #1126 guarantee: the newest registration wins, so a + // rebuilt AppContext leaves ecstore pointed at the current adapter + // rather than a stranded first-wins one. + let slot: HookSlot = HookSlot::new(); + slot.register(Arc::new(Impl(1)), "unused"); + slot.register(Arc::new(Impl(2)), "unused"); + assert_eq!(slot.get().map(|h| h.id()), Some(2)); + } + + #[test] + fn re_registering_the_same_arc_is_idempotent() { + let slot: HookSlot = HookSlot::new(); + let hook: Arc = Arc::new(Impl(9)); + slot.register(Arc::clone(&hook), "unused"); + slot.register(hook, "unused"); + assert_eq!(slot.get().map(|h| h.id()), Some(9)); + } + + #[test] + fn clear_empties_the_slot() { + let slot: HookSlot = HookSlot::new(); + slot.register(Arc::new(Impl(3)), "unused"); + slot.clear(); + assert!(slot.get().is_none()); + } +} diff --git a/crates/ecstore/src/object_api/mod.rs b/crates/ecstore/src/object_api/mod.rs index d08cc4285..f6efa90a6 100644 --- a/crates/ecstore/src/object_api/mod.rs +++ b/crates/ecstore/src/object_api/mod.rs @@ -53,6 +53,8 @@ pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M mod body_cache_hook; +mod hook_slot; +mod object_mutation_hook; mod readers; mod types; @@ -60,5 +62,7 @@ mod types; pub(crate) use body_cache_hook::clear_get_object_body_cache_hook; pub(crate) use body_cache_hook::get_object_body_cache_hook; pub use body_cache_hook::{GetObjectBodyCacheHook, register_get_object_body_cache_hook}; +pub(crate) use object_mutation_hook::notify_object_mutation; +pub use object_mutation_hook::{ObjectMutationHook, register_object_mutation_hook}; pub use readers::*; pub use types::*; diff --git a/crates/ecstore/src/object_api/object_mutation_hook.rs b/crates/ecstore/src/object_api/object_mutation_hook.rs new file mode 100644 index 000000000..d9b0f6424 --- /dev/null +++ b/crates/ecstore/src/object_api/object_mutation_hook.rs @@ -0,0 +1,119 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Write-side counterpart to [`super::body_cache_hook`]. +//! +//! ecstore terminates object bodies from paths that never pass through the +//! app-layer usecases: lifecycle/scanner expiry, non-current version cleanup, +//! and restored-copy expiry all delete objects directly on the store. The GET +//! body cache is keyed on `(bucket, object, versionId, etag, size, variant)` +//! and every lookup follows a fresh metadata quorum, so a stale entry can never +//! be *served* after its object is gone (the GET fails at metadata resolution +//! first). What lingers is dead body bytes resident until TTL, which evict live +//! hot entries — a hygiene and capacity problem, not a correctness one. +//! +//! This hook lets the app-layer cache adapter drop those bodies from its index +//! the moment ecstore removes the object (ODC-26, backlog#1131). + +use crate::object_api::hook_slot::HookSlot; +use std::sync::Arc; + +/// Invalidates cached object bodies after an ecstore-internal mutation removed +/// them. Keyed on `(bucket, object)`; the implementation invalidates every +/// cached version/etag under that identity. +#[async_trait::async_trait] +pub trait ObjectMutationHook: Send + Sync + 'static { + async fn after_object_mutation(&self, bucket: &str, object: &str); +} + +// A `HookSlot`, for the same reason as the GET hook slot: arc-swap cannot hold +// an unsized `Arc`. Registration is a startup event and +// each delete-path invocation only clones an `Arc` under the read guard, +// negligible next to the delete it accompanies. +static OBJECT_MUTATION_HOOK: HookSlot = HookSlot::new(); + +/// Register (or re-register) the process-wide object mutation hook. +/// +/// Re-registration atomically swaps to `hook`, mirroring the GET body hook so a +/// rebuilt `AppContext` leaves ecstore's delete paths pointed at the newest +/// adapter. +pub fn register_object_mutation_hook(hook: Arc) { + OBJECT_MUTATION_HOOK.register( + hook, + "object mutation cache hook re-registered with a different instance; \ + the previous adapter's cache is now unreachable by ecstore's delete paths", + ); +} + +/// The registered hook, if any. +fn object_mutation_hook() -> Option> { + OBJECT_MUTATION_HOOK.get() +} + +/// Invoke the registered hook for `(bucket, object)`, if one is installed. +/// +/// A single `None` branch when the cache feature is off, so the ecstore delete +/// paths pay nothing beyond one relaxed lock read when unconfigured. +pub(crate) async fn notify_object_mutation(bucket: &str, object: &str) { + if let Some(hook) = object_mutation_hook() { + hook.after_object_mutation(bucket, object).await; + } +} + +/// Test-only: unregister the hook so tests can register and clear the slot +/// deterministically without leaking a hook into unrelated tests. +#[cfg(test)] +pub(crate) fn clear_object_mutation_hook() { + OBJECT_MUTATION_HOOK.clear(); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + struct RecordingHook { + calls: Arc>>, + } + + #[async_trait::async_trait] + impl ObjectMutationHook for RecordingHook { + async fn after_object_mutation(&self, bucket: &str, object: &str) { + self.calls.lock().unwrap().push((bucket.to_string(), object.to_string())); + } + } + + #[tokio::test] + #[serial_test::serial(object_mutation_hook)] + async fn notify_invokes_registered_hook_with_identity() { + clear_object_mutation_hook(); + let calls = Arc::new(Mutex::new(Vec::new())); + register_object_mutation_hook(Arc::new(RecordingHook { + calls: Arc::clone(&calls), + })); + + notify_object_mutation("bucket", "photos/a.jpg").await; + + assert_eq!(&*calls.lock().unwrap(), &[("bucket".to_string(), "photos/a.jpg".to_string())]); + clear_object_mutation_hook(); + } + + #[tokio::test] + #[serial_test::serial(object_mutation_hook)] + async fn notify_without_registered_hook_is_noop() { + clear_object_mutation_hook(); + // Must not panic when no hook is installed (the cache feature is off). + notify_object_mutation("bucket", "object").await; + } +} diff --git a/crates/object-data-cache/src/cache.rs b/crates/object-data-cache/src/cache.rs index 9e4f1ce01..229138f32 100644 --- a/crates/object-data-cache/src/cache.rs +++ b/crates/object-data-cache/src/cache.rs @@ -258,25 +258,80 @@ impl ObjectDataCache { /// Invalidates all cache entries associated with the object identity. pub async fn invalidate_object( &self, - _identity: ObjectDataCacheIdentity, - _reason: ObjectDataCacheInvalidationReason, + identity: ObjectDataCacheIdentity, + reason: ObjectDataCacheInvalidationReason, ) -> ObjectDataCacheInvalidationResult { let result = match &self.backend { ObjectDataCacheBackendKind::Noop(backend) => backend.invalidate_object().await, - ObjectDataCacheBackendKind::Moka(backend) => backend.invalidate_object(&_identity).await, + ObjectDataCacheBackendKind::Moka(backend) => backend.invalidate_object(&identity).await, }; + self.finish_invalidation(result, reason) + } + /// Invalidates every cached body under `bucket`/`prefix`. + /// + /// Drops the cached bodies of every identity in `bucket` whose object key + /// starts with `prefix`. Backed by a full identity-index scan, so it must + /// stay on the rare force-delete and admin paths and never touch the GET or + /// fill hot path (ODC-27, backlog#1132). + pub async fn invalidate_prefix( + &self, + bucket: &str, + prefix: &str, + reason: ObjectDataCacheInvalidationReason, + ) -> ObjectDataCacheInvalidationResult { + let result = match &self.backend { + ObjectDataCacheBackendKind::Noop(backend) => backend.invalidate_prefix().await, + ObjectDataCacheBackendKind::Moka(backend) => backend.invalidate_prefix(bucket, prefix).await, + }; + self.finish_invalidation(result, reason) + } + + /// Invalidates every cached body in `bucket`. + /// + /// Backed by a full identity-index scan; keep it on the rare bucket-delete + /// and admin paths only (ODC-28, backlog#1133). + pub async fn invalidate_bucket( + &self, + bucket: &str, + reason: ObjectDataCacheInvalidationReason, + ) -> ObjectDataCacheInvalidationResult { + let result = match &self.backend { + ObjectDataCacheBackendKind::Noop(backend) => backend.invalidate_bucket().await, + ObjectDataCacheBackendKind::Moka(backend) => backend.invalidate_bucket(bucket).await, + }; + self.finish_invalidation(result, reason) + } + + /// Drops every cached body and resets the identity index. + /// + /// The only production remediation for a poisoned or stale entry short of a + /// node restart (ODC-C2, backlog#1143). Rare admin path only. + pub async fn clear(&self, reason: ObjectDataCacheInvalidationReason) -> ObjectDataCacheInvalidationResult { + let result = match &self.backend { + ObjectDataCacheBackendKind::Noop(backend) => backend.clear().await, + ObjectDataCacheBackendKind::Moka(backend) => backend.clear().await, + }; + self.finish_invalidation(result, reason) + } + + /// Shared post-processing for every invalidation primitive: bump the + /// invalidation counter, refresh the cache-state gauge only when something + /// was removed, and emit the outcome-labelled metric. A mutating op + /// invalidates twice by design (before + after) and the vast majority of + /// those touch identities that were never cached, so the no-op path skips + /// the gauge refresh (backlog#1141). + fn finish_invalidation( + &self, + result: ObjectDataCacheInvalidationResult, + reason: ObjectDataCacheInvalidationReason, + ) -> ObjectDataCacheInvalidationResult { self.stats.record_invalidation(); 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); - + record_invalidation(self.backend.as_metric_label(), reason.as_metric_label(), outcome); result } @@ -285,6 +340,11 @@ impl ObjectDataCache { self.stats.snapshot() } + /// Returns the configured runtime mode, for admin status reporting. + pub fn mode(&self) -> crate::config::ObjectDataCacheMode { + self.config.mode + } + /// Returns true when the cache facade is fully disabled. pub fn is_disabled(&self) -> bool { self.config.is_disabled() @@ -438,6 +498,15 @@ pub enum ObjectDataCacheInvalidationReason { AfterCopySuccess, /// Invalidation after a successful complete multipart upload. AfterCompleteMultipartSuccess, + /// Invalidation after an ecstore-internal lifecycle/scanner expiry deleted + /// the object body (ODC-26). + AfterLifecycleExpiry, + /// Invalidation after a forced prefix delete removed every object under a + /// prefix (ODC-27). + AfterPrefixDelete, + /// Invalidation after a bucket delete removed every object in the bucket + /// (ODC-28). + AfterBucketDelete, /// Manual invalidation requested by the caller. Manual, } @@ -450,6 +519,9 @@ impl ObjectDataCacheInvalidationReason { Self::AfterDeleteSuccess => "after_delete_success", Self::AfterCopySuccess => "after_copy_success", Self::AfterCompleteMultipartSuccess => "after_complete_multipart_success", + Self::AfterLifecycleExpiry => "after_lifecycle_expiry", + Self::AfterPrefixDelete => "after_prefix_delete", + Self::AfterBucketDelete => "after_bucket_delete", Self::Manual => "manual", } } @@ -471,7 +543,7 @@ pub enum ObjectDataCacheInvalidationResult { mod tests { use super::{ ObjectDataCache, ObjectDataCacheFillResult, ObjectDataCacheGetPlan, ObjectDataCacheGetRequest, - ObjectDataCacheInvalidationReason, ObjectDataCacheLookup, + ObjectDataCacheInvalidationReason, ObjectDataCacheInvalidationResult, ObjectDataCacheLookup, }; use crate::config::{ObjectDataCacheConfig, ObjectDataCacheMode}; use crate::key::{ObjectDataCacheBodyVariant, ObjectDataCacheIdentity}; @@ -678,6 +750,96 @@ mod tests { ); } + #[test] + fn new_invalidation_reasons_map_to_distinct_labels() { + assert_eq!( + ObjectDataCacheInvalidationReason::AfterLifecycleExpiry.as_metric_label(), + "after_lifecycle_expiry" + ); + assert_eq!( + ObjectDataCacheInvalidationReason::AfterPrefixDelete.as_metric_label(), + "after_prefix_delete" + ); + assert_eq!( + ObjectDataCacheInvalidationReason::AfterBucketDelete.as_metric_label(), + "after_bucket_delete" + ); + } + + #[test] + fn prefix_invalidation_of_cached_identity_labels_reason_and_removed() { + // ODC-27: a prefix flush that drops a cached body is labelled with its + // own reason and outcome=removed so dashboards attribute the churn. + let cache = fill_enabled_cache(); + let metrics = capture_metrics(|| async { + let plan = cache.plan_get(plain_request("bucket", "photos/a", "etag", 5)); + assert_eq!( + cache.fill_body(&plan, Bytes::from_static(b"hello")).await, + ObjectDataCacheFillResult::Inserted + ); + let result = cache + .invalidate_prefix("bucket", "photos/", ObjectDataCacheInvalidationReason::AfterPrefixDelete) + .await; + assert_eq!(result, ObjectDataCacheInvalidationResult::Removed { keys: 1 }); + }); + + assert!(has_counter_with_label( + &metrics, + "rustfs_object_data_cache_invalidations_total", + ("reason", "after_prefix_delete") + )); + assert!(has_counter_with_label( + &metrics, + "rustfs_object_data_cache_invalidations_total", + ("outcome", "removed") + )); + } + + #[test] + fn bucket_invalidation_of_uncached_bucket_labels_noop() { + // ODC-28: a bucket flush that matched nothing is a no-op and must not + // refresh the entries gauge. + let cache = fill_enabled_cache(); + let metrics = capture_metrics(|| async { + let result = cache + .invalidate_bucket("empty-bucket", ObjectDataCacheInvalidationReason::AfterBucketDelete) + .await; + assert_eq!(result, ObjectDataCacheInvalidationResult::NoOp); + }); + + assert!(has_counter_with_label( + &metrics, + "rustfs_object_data_cache_invalidations_total", + ("outcome", "noop") + )); + assert!( + !has_gauge(&metrics, "rustfs_object_data_cache_entries"), + "a no-op bucket flush must not refresh the entries gauge" + ); + } + + #[test] + fn clear_of_cached_cache_labels_manual_and_removed() { + // ODC-C2: an admin clear reuses the Manual reason and reports removed. + let cache = fill_enabled_cache(); + let metrics = capture_metrics(|| async { + let plan = cache.plan_get(plain_request("bucket", "object", "etag", 5)); + assert_eq!( + cache.fill_body(&plan, Bytes::from_static(b"hello")).await, + ObjectDataCacheFillResult::Inserted + ); + let result = cache.clear(ObjectDataCacheInvalidationReason::Manual).await; + assert_eq!(result, ObjectDataCacheInvalidationResult::Removed { keys: 1 }); + assert!(matches!(cache.lookup_body(&plan).await, ObjectDataCacheLookup::Miss)); + }); + + assert!(has_counter_with_label( + &metrics, + "rustfs_object_data_cache_invalidations_total", + ("reason", "manual") + )); + } + #[tokio::test] async fn fill_body_rejects_size_mismatch() { let cache = fill_enabled_cache(); diff --git a/crates/object-data-cache/src/config.rs b/crates/object-data-cache/src/config.rs index b6782c69d..629a59d7c 100644 --- a/crates/object-data-cache/src/config.rs +++ b/crates/object-data-cache/src/config.rs @@ -54,6 +54,11 @@ pub enum ObjectDataCacheMode { } impl ObjectDataCacheMode { + /// Stable lowercase identifier for this mode, for admin status reporting. + pub const fn as_str(self) -> &'static str { + self.as_metric_label() + } + pub(crate) const fn as_metric_label(self) -> &'static str { match self { Self::Disabled => "disabled", diff --git a/crates/object-data-cache/src/memory.rs b/crates/object-data-cache/src/memory.rs index 5b2564726..5215780f7 100644 --- a/crates/object-data-cache/src/memory.rs +++ b/crates/object-data-cache/src/memory.rs @@ -153,7 +153,7 @@ impl Drop for RefresherGuard { /// 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 +/// (the private `spawn_refresher` task) 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. diff --git a/crates/object-data-cache/src/moka_backend.rs b/crates/object-data-cache/src/moka_backend.rs index 8312ebace..e8b8d3524 100644 --- a/crates/object-data-cache/src/moka_backend.rs +++ b/crates/object-data-cache/src/moka_backend.rs @@ -293,6 +293,48 @@ impl MokaBackend { self.cache.remove(&key).await; } + Self::invalidation_result(removed) + } + + /// Invalidates every cached key of an identity in `bucket` whose object key + /// starts with `prefix`. Full index scan; rare force-delete/admin path only. + pub async fn invalidate_prefix(&self, bucket: &str, prefix: &str) -> ObjectDataCacheInvalidationResult { + let keys_to_remove = self + .index + .remove_matching(|identity| identity.bucket.as_ref() == bucket && identity.object.starts_with(prefix)) + .await; + let removed = keys_to_remove.len(); + for key in keys_to_remove { + self.cache.remove(&key).await; + } + Self::invalidation_result(removed) + } + + /// Invalidates every cached key of every identity in `bucket`. Full index + /// scan; rare bucket-delete/admin path only. + pub async fn invalidate_bucket(&self, bucket: &str) -> ObjectDataCacheInvalidationResult { + let keys_to_remove = self + .index + .remove_matching(|identity| identity.bucket.as_ref() == bucket) + .await; + let removed = keys_to_remove.len(); + for key in keys_to_remove { + self.cache.remove(&key).await; + } + Self::invalidation_result(removed) + } + + /// Drops every cached body and resets the identity index. The index scan + /// yields the tracked key count for the outcome label; `invalidate_all` + /// then clears moka in one call (also dropping any entry not tracked in the + /// index). Rare admin `clear()` path only. + pub async fn clear(&self) -> ObjectDataCacheInvalidationResult { + let removed = self.index.remove_matching(|_| true).await.len(); + self.cache.invalidate_all(); + Self::invalidation_result(removed) + } + + const fn invalidation_result(removed: usize) -> ObjectDataCacheInvalidationResult { if removed > 0 { ObjectDataCacheInvalidationResult::Removed { keys: removed } } else { @@ -402,6 +444,109 @@ mod tests { assert_eq!(result, ObjectDataCacheInvalidationResult::NoOp); } + fn bucketed_plan(bucket: &str, object: &str, etag: &str) -> ObjectDataCacheGetPlan { + ObjectDataCacheGetPlan::Cacheable { + key: ObjectDataCacheKey::new(bucket, object, None, etag, 5, ObjectDataCacheBodyVariant::FullObjectPlainV1), + } + } + + // ODC-27: prefix invalidation drops only the identities under the prefix and + // leaves every other cached body intact. + #[tokio::test] + async fn moka_backend_invalidate_prefix_removes_only_matching_prefix() { + let mut config = enabled_config(); + config.ttl = Duration::from_secs(30); + config.time_to_idle = Duration::from_secs(30); + let backend = MokaBackend::new(&config, Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"); + let plan_a = cacheable_plan("photos/a.jpg", "etag-a"); + let plan_b = cacheable_plan("photos/b.jpg", "etag-b"); + let plan_c = cacheable_plan("videos/c.mp4", "etag-c"); + + let _ = backend.fill_body(&plan_a, Bytes::from_static(b"aaaaa")).await; + let _ = backend.fill_body(&plan_b, Bytes::from_static(b"bbbbb")).await; + let _ = backend.fill_body(&plan_c, Bytes::from_static(b"ccccc")).await; + + let result = backend.invalidate_prefix("bucket", "photos/").await; + + assert_eq!(result, ObjectDataCacheInvalidationResult::Removed { keys: 2 }); + assert!(matches!(backend.lookup_body(&plan_a).await, ObjectDataCacheLookup::Miss)); + assert!(matches!(backend.lookup_body(&plan_b).await, ObjectDataCacheLookup::Miss)); + assert!( + matches!(backend.lookup_body(&plan_c).await, ObjectDataCacheLookup::Hit(_)), + "an object outside the prefix must survive" + ); + } + + // ODC-27: a prefix that matches nothing cached is a no-op. + #[tokio::test] + async fn moka_backend_invalidate_prefix_uncached_is_noop() { + let backend = + MokaBackend::new(&enabled_config(), Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"); + let _ = backend + .fill_body(&cacheable_plan("photos/a.jpg", "e"), Bytes::from_static(b"aaaaa")) + .await; + + let result = backend.invalidate_prefix("bucket", "videos/").await; + + assert_eq!(result, ObjectDataCacheInvalidationResult::NoOp); + } + + // ODC-28: bucket invalidation drops every identity in the bucket and leaves + // other buckets untouched. + #[tokio::test] + async fn moka_backend_invalidate_bucket_removes_only_that_bucket() { + let mut config = enabled_config(); + config.ttl = Duration::from_secs(30); + config.time_to_idle = Duration::from_secs(30); + let backend = MokaBackend::new(&config, Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"); + let plan_a = bucketed_plan("bucket-a", "o1", "etag-a"); + let plan_b = bucketed_plan("bucket-a", "o2", "etag-b"); + let plan_other = bucketed_plan("bucket-b", "o3", "etag-c"); + + let _ = backend.fill_body(&plan_a, Bytes::from_static(b"aaaaa")).await; + let _ = backend.fill_body(&plan_b, Bytes::from_static(b"bbbbb")).await; + let _ = backend.fill_body(&plan_other, Bytes::from_static(b"ccccc")).await; + + let result = backend.invalidate_bucket("bucket-a").await; + + assert_eq!(result, ObjectDataCacheInvalidationResult::Removed { keys: 2 }); + assert!(matches!(backend.lookup_body(&plan_a).await, ObjectDataCacheLookup::Miss)); + assert!(matches!(backend.lookup_body(&plan_b).await, ObjectDataCacheLookup::Miss)); + assert!( + matches!(backend.lookup_body(&plan_other).await, ObjectDataCacheLookup::Hit(_)), + "an object in another bucket must survive a bucket flush" + ); + } + + // ODC-C2: clear drops every cached body and empties the identity index. + #[tokio::test] + async fn moka_backend_clear_removes_everything() { + let mut config = enabled_config(); + config.ttl = Duration::from_secs(30); + config.time_to_idle = Duration::from_secs(30); + let backend = MokaBackend::new(&config, Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"); + let plan_a = bucketed_plan("bucket-a", "o1", "etag-a"); + let plan_b = bucketed_plan("bucket-b", "o2", "etag-b"); + + let _ = backend.fill_body(&plan_a, Bytes::from_static(b"aaaaa")).await; + let _ = backend.fill_body(&plan_b, Bytes::from_static(b"bbbbb")).await; + + let result = backend.clear().await; + + assert_eq!(result, ObjectDataCacheInvalidationResult::Removed { keys: 2 }); + assert!(matches!(backend.lookup_body(&plan_a).await, ObjectDataCacheLookup::Miss)); + assert!(matches!(backend.lookup_body(&plan_b).await, ObjectDataCacheLookup::Miss)); + assert_eq!(backend.index.identity_count().await, 0, "clear must empty the identity index"); + } + + #[tokio::test] + async fn moka_backend_clear_empty_cache_is_noop() { + let backend = + MokaBackend::new(&enabled_config(), Arc::new(ObjectDataCacheStats::default())).expect("moka backend should build"); + + assert_eq!(backend.clear().await, ObjectDataCacheInvalidationResult::NoOp); + } + #[tokio::test] async fn moka_backend_expires_entries_by_ttl() { let backend = diff --git a/crates/object-data-cache/src/noop.rs b/crates/object-data-cache/src/noop.rs index ca8bf54f9..bbad8ef5c 100644 --- a/crates/object-data-cache/src/noop.rs +++ b/crates/object-data-cache/src/noop.rs @@ -34,6 +34,21 @@ impl NoopBackend { pub async fn invalidate_object(&self) -> ObjectDataCacheInvalidationResult { ObjectDataCacheInvalidationResult::NoOp } + + /// Prefix invalidation on a disabled cache removes nothing. + pub async fn invalidate_prefix(&self) -> ObjectDataCacheInvalidationResult { + ObjectDataCacheInvalidationResult::NoOp + } + + /// Bucket invalidation on a disabled cache removes nothing. + pub async fn invalidate_bucket(&self) -> ObjectDataCacheInvalidationResult { + ObjectDataCacheInvalidationResult::NoOp + } + + /// Clearing a disabled cache removes nothing. + pub async fn clear(&self) -> ObjectDataCacheInvalidationResult { + ObjectDataCacheInvalidationResult::NoOp + } } #[cfg(test)] diff --git a/crates/object-data-cache/src/starshard_index.rs b/crates/object-data-cache/src/starshard_index.rs index 21c7f0a71..d867b222e 100644 --- a/crates/object-data-cache/src/starshard_index.rs +++ b/crates/object-data-cache/src/starshard_index.rs @@ -98,6 +98,42 @@ impl StarshardIdentityIndex { .map_or_else(Vec::new, |set| set.cloned()) } + /// Removes every tracked identity whose key matches `predicate`, returning + /// all keys that were dropped so the caller can evict them from the cache. + /// + /// This performs a **full shard scan** (`keys()` snapshots every identity, + /// then each match is removed under its shard write lock). It is therefore + /// restricted to the rare prefix-delete, bucket-delete, and admin + /// `clear()`/flush paths — it must never run on the GET or fill hot path. + /// A `true`-returning predicate clears the whole index (used by `clear()`). + /// + /// The snapshot/remove split leaves a small window: an identity inserted + /// after the snapshot but before removal is not visited, and a key added to + /// a matched identity between snapshot and its `remove` is still dropped + /// from the index (via `remove`) but returned for cache eviction, so no + /// tracked body is stranded. This is acceptable hygiene slack on these rare + /// paths (backlog#1132/#1133/#1143). + pub async fn remove_matching(&self, predicate: F) -> Vec + where + F: Fn(&ObjectDataCacheIdentity) -> bool, + { + let identities: Vec = self + .by_object + .keys() + .await + .into_iter() + .filter(|identity| predicate(identity)) + .collect(); + + let mut removed = Vec::new(); + for identity in identities { + if let Some(key_set) = self.by_object.remove(&identity).await { + removed.extend(key_set.cloned()); + } + } + removed + } + /// Removes a single evicted key tracked under an identity, but only when the /// evicted entry's generation token still matches the tracked one. /// @@ -259,6 +295,53 @@ mod tests { assert_eq!(invalidated, vec![key_a]); } + fn bucketed_key(bucket: &str, object: &str) -> ObjectDataCacheKey { + ObjectDataCacheKey::new(bucket, object, Some("v1"), "etag", 1, ObjectDataCacheBodyVariant::FullObjectPlainV1) + } + + #[tokio::test] + async fn remove_matching_returns_only_predicate_matches() { + let index = StarshardIdentityIndex::new(4); + let id_photos = ObjectDataCacheIdentity::new("bucket", "photos/a.jpg"); + let id_photos2 = ObjectDataCacheIdentity::new("bucket", "photos/b.jpg"); + let id_videos = ObjectDataCacheIdentity::new("bucket", "videos/c.mp4"); + + let _ = index + .insert(id_photos.clone(), bucketed_key("bucket", "photos/a.jpg"), 1) + .await; + let _ = index + .insert(id_photos2.clone(), bucketed_key("bucket", "photos/b.jpg"), 2) + .await; + let _ = index + .insert(id_videos.clone(), bucketed_key("bucket", "videos/c.mp4"), 3) + .await; + + let removed = index + .remove_matching(|identity| identity.bucket.as_ref() == "bucket" && identity.object.starts_with("photos/")) + .await; + + assert_eq!(removed.len(), 2, "only the two photos/ identities match the prefix"); + // The unmatched identity is still tracked; the matched ones are gone. + assert!(index.remove_identity(&id_videos).await.len() == 1); + assert!(index.remove_identity(&id_photos).await.is_empty()); + } + + #[tokio::test] + async fn remove_matching_true_predicate_clears_index() { + let index = StarshardIdentityIndex::new(4); + let _ = index + .insert(ObjectDataCacheIdentity::new("b1", "o1"), bucketed_key("b1", "o1"), 1) + .await; + let _ = index + .insert(ObjectDataCacheIdentity::new("b2", "o2"), bucketed_key("b2", "o2"), 2) + .await; + + let removed = index.remove_matching(|_| true).await; + + assert_eq!(removed.len(), 2); + assert_eq!(index.identity_count().await, 0, "a true predicate clears every identity"); + } + #[tokio::test] async fn identity_index_matching_eviction_removes_key() { let index = StarshardIdentityIndex::new(4); diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index e0cc3d447..7ad4723cb 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -36,6 +36,7 @@ pub mod kms_management; pub mod metrics; pub mod module_switch; mod notify_runtime_access; +pub mod object_data_cache; pub mod object_zip_download; pub mod oidc; pub mod plugins_catalog; diff --git a/rustfs/src/admin/handlers/object_data_cache.rs b/rustfs/src/admin/handlers/object_data_cache.rs new file mode 100644 index 000000000..d470e8b56 --- /dev/null +++ b/rustfs/src/admin/handlers/object_data_cache.rs @@ -0,0 +1,219 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Admin surface for the object data cache (ODC-C2, backlog#1143). +//! +//! `GET {ADMIN_PREFIX}/v3/object-data-cache/stats` returns the current stats +//! snapshot plus mode, the only production window into the cache short of a +//! metrics scrape. `POST {ADMIN_PREFIX}/v3/object-data-cache/flush` drops +//! cached bodies; with no query it clears everything, with `bucket` it flushes +//! that bucket, and with `bucket`+`object` it flushes that one identity — the +//! only remediation for a poisoned entry short of a node restart. + +use crate::admin::auth::validate_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::runtime_sources::current_object_data_cache; +use crate::app::object_data_cache::ObjectDataCacheAdapter; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue}; +use hyper::{Method, StatusCode}; +use matchit::Params; +use rustfs_object_data_cache::{ObjectDataCacheIdentity, ObjectDataCacheInvalidationReason, ObjectDataCacheInvalidationResult}; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::header::CONTENT_TYPE; +use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use serde::Serialize; +use std::sync::Arc; + +const JSON_CONTENT_TYPE: &str = "application/json"; + +#[derive(Debug, Serialize)] +struct ObjectDataCacheStatsResponse { + mode: &'static str, + disabled: bool, + entries: u64, + lookups: u64, + hits: u64, + fills: u64, + invalidations: u64, + inflight_fills: u64, + singleflight_joins: u64, + memory_pressure_events: u64, +} + +#[derive(Debug, Serialize)] +struct ObjectDataCacheFlushResponse { + scope: &'static str, + bucket: Option, + object: Option, + outcome: &'static str, + removed_keys: usize, +} + +pub fn register_object_data_cache_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/object-data-cache/stats").as_str(), + AdminOperation(&ObjectDataCacheStatsHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/object-data-cache/flush").as_str(), + AdminOperation(&ObjectDataCacheFlushHandler {}), + )?; + Ok(()) +} + +async fn authorize(req: &S3Request, action: AdminAction) -> S3Result<()> { + let Some(input_cred) = req.credentials.as_ref() else { + return Err(s3_error!(InvalidRequest, "missing credentials")); + }; + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + let remote_addr = req + .extensions + .get::>() + .and_then(|opt| opt.map(|addr| addr.0)); + validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await +} + +fn json_response(body: &T) -> S3Result> { + let data = serde_json::to_vec(body) + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to encode response: {err}")))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static(JSON_CONTENT_TYPE)); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers)) +} + +fn query_value(req: &S3Request, key: &str) -> Option { + req.uri.query().and_then(|query| { + url::form_urlencoded::parse(query.as_bytes()) + .find(|(k, _)| k == key) + .map(|(_, v)| v.into_owned()) + .filter(|v| !v.is_empty()) + }) +} + +fn invalidation_outcome(result: &ObjectDataCacheInvalidationResult) -> (&'static str, usize) { + match result { + ObjectDataCacheInvalidationResult::Removed { keys } => ("removed", *keys), + ObjectDataCacheInvalidationResult::NoOp => ("noop", 0), + } +} + +pub struct ObjectDataCacheStatsHandler {} + +#[async_trait::async_trait] +impl Operation for ObjectDataCacheStatsHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::ServerInfoAdminAction).await?; + + let response = match current_object_data_cache() { + Some(adapter) => { + let snapshot = adapter.stats(); + ObjectDataCacheStatsResponse { + mode: adapter.mode().as_str(), + disabled: adapter.is_disabled(), + entries: snapshot.entries, + lookups: snapshot.lookups, + hits: snapshot.hits, + fills: snapshot.fills, + invalidations: snapshot.invalidations, + inflight_fills: snapshot.inflight_fills, + singleflight_joins: snapshot.singleflight_joins, + memory_pressure_events: snapshot.memory_pressure_events, + } + } + None => ObjectDataCacheStatsResponse { + mode: "disabled", + disabled: true, + entries: 0, + lookups: 0, + hits: 0, + fills: 0, + invalidations: 0, + inflight_fills: 0, + singleflight_joins: 0, + memory_pressure_events: 0, + }, + }; + + json_response(&response) + } +} + +pub struct ObjectDataCacheFlushHandler {} + +#[async_trait::async_trait] +impl Operation for ObjectDataCacheFlushHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::ConfigUpdateAdminAction).await?; + + let bucket = query_value(&req, "bucket"); + let object = query_value(&req, "object"); + if object.is_some() && bucket.is_none() { + return Err(s3_error!(InvalidRequest, "object flush requires a bucket query parameter")); + } + + let adapter: Arc = + current_object_data_cache().ok_or_else(|| s3_error!(InternalError, "object data cache is not initialized"))?; + + // Manual is the existing-but-uncalled reason reserved for operator-driven + // invalidation; every flush scope reports under it. + let reason = ObjectDataCacheInvalidationReason::Manual; + let (scope, result) = match (bucket.as_deref(), object.as_deref()) { + (Some(bucket), Some(object)) => ( + "object", + adapter + .invalidate_object(ObjectDataCacheIdentity::new(bucket, object), reason) + .await, + ), + (Some(bucket), None) => ("bucket", adapter.invalidate_bucket(bucket, reason).await), + (None, _) => ("all", adapter.clear(reason).await), + }; + + let (outcome, removed_keys) = invalidation_outcome(&result); + json_response(&ObjectDataCacheFlushResponse { + scope, + bucket, + object, + outcome, + removed_keys, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn flush_outcome_maps_removed_and_noop() { + assert_eq!( + invalidation_outcome(&ObjectDataCacheInvalidationResult::Removed { keys: 3 }), + ("removed", 3) + ); + assert_eq!(invalidation_outcome(&ObjectDataCacheInvalidationResult::NoOp), ("noop", 0)); + } + + #[test] + fn stats_handler_requires_server_info_action() { + // Guard the auth contract: the stats endpoint is a read, the flush + // endpoint mutates, so they must not share one action. + let src = include_str!("object_data_cache.rs"); + assert!(src.contains("authorize(&req, AdminAction::ServerInfoAdminAction).await?;")); + assert!(src.contains("authorize(&req, AdminAction::ConfigUpdateAdminAction).await?;")); + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 775f99e93..222ff169b 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -34,9 +34,9 @@ mod route_registration_test; use handlers::{ audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, durability as durability_handler, extensions, - heal, health, idp_compat, kms, module_switch, object_zip_download, oidc, plugins_catalog, plugins_instances, pools, - profile_admin, quota as quota_handler, rebalance, replication as replication_handler, scanner, site_replication, sts, system, - table_catalog, tier, tls_debug, user, + heal, health, idp_compat, kms, module_switch, object_data_cache, object_zip_download, oidc, plugins_catalog, + plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, replication as replication_handler, scanner, + site_replication, sts, system, table_catalog, tier, tls_debug, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -76,6 +76,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() bucket_meta::register_bucket_meta_route(r)?; config_admin::register_config_route(r)?; scanner::register_scanner_route(r)?; + object_data_cache::register_object_data_cache_route(r)?; audit::register_audit_target_route(r)?; module_switch::register_module_switch_route(r)?; cluster_snapshot::register_cluster_snapshot_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 711ac9714..525bd8573 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -291,6 +291,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ admin(HttpMethod::Get, "/rustfs/admin/v3/info", SERVER_INFO, RouteRiskLevel::Sensitive), admin(HttpMethod::Get, "/rustfs/admin/v3/storageinfo", STORAGE_INFO, RouteRiskLevel::Sensitive), admin(HttpMethod::Get, "/rustfs/admin/v3/metrics", GET_METRICS, RouteRiskLevel::Sensitive), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/object-data-cache/stats", + SERVER_INFO, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/rustfs/admin/v3/object-data-cache/flush", + CONFIG_UPDATE, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/rustfs/admin/v3/pools/decommission", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index e94157b6c..e6b08cbcf 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -172,6 +172,8 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::GET, "/v3/storageinfo"), admin_route(Method::GET, "/v3/datausageinfo"), admin_route(Method::GET, "/v3/metrics"), + admin_route(Method::GET, "/v3/object-data-cache/stats"), + admin_route(Method::POST, "/v3/object-data-cache/flush"), admin_route(Method::GET, "/v3/pools/list"), admin_route(Method::GET, "/v3/pools/status"), admin_route(Method::GET, "/v3/decommission/status"), diff --git a/rustfs/src/admin/runtime_sources.rs b/rustfs/src/admin/runtime_sources.rs index c3e552924..8c824b451 100644 --- a/rustfs/src/admin/runtime_sources.rs +++ b/rustfs/src/admin/runtime_sources.rs @@ -18,14 +18,16 @@ use crate::admin::storage_api::runtime_sources::{ pub(crate) use crate::app::admin_usecase::{ AdminPoolStatus, DefaultAdminUsecase, QueryPoolStatusRequest, QueryServerInfoRequest, }; +use crate::app::object_data_cache::ObjectDataCacheAdapter; use crate::app::object_usecase::DefaultObjectUsecase; use crate::runtime_sources as root_runtime_sources; pub(crate) use crate::runtime_sources::{ AppContext, ServerContextSlot, current_action_credentials, current_boot_time, current_bucket_metadata_handle, current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_iam_handle, - current_kms_runtime_service_manager, current_notification_system_for_context, current_object_store_handle_for_context, - current_oidc_handle, current_ready_iam_handle, current_region, current_replication_pool_handle, - current_replication_stats_handle, current_server_config_for_context, current_token_signing_key, + current_kms_runtime_service_manager, current_notification_system_for_context, current_object_data_cache_handle_for_context, + current_object_store_handle_for_context, current_oidc_handle, current_ready_iam_handle, current_region, + current_replication_pool_handle, current_replication_stats_handle, current_server_config_for_context, + current_token_signing_key, }; use rustfs_config::server_config::Config; use rustfs_kms::KmsServiceManager; @@ -52,6 +54,14 @@ pub(crate) fn current_object_store_handle() -> Option> { current_object_store_handle_for_context(context.as_deref()) } +/// Resolve the object data cache adapter for an admin request through the +/// process AppContext. `None` when no context is initialised (the admin +/// stats/flush handlers then report the cache as unavailable). +pub(crate) fn current_object_data_cache() -> Option> { + let context = current_app_context(); + current_object_data_cache_handle_for_context(context.as_deref()) +} + /// Resolve the object store for an admin request through the server's context /// slot injected at router dispatch (backlog#1052 S2). Falls back to the /// ambient process context when no slot was injected (direct handler tests, diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index c4a5a8d88..bcc69b0da 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -63,9 +63,10 @@ use super::storage_api::bucket_usecase::{ use crate::admin::handlers::site_replication::{ site_replication_bucket_meta_hook, site_replication_delete_bucket_hook, site_replication_make_bucket_hook, }; +use crate::app::object_data_cache::invalidate_object_data_cache_bucket_after_delete; use crate::app::runtime_sources::{ AppContext, current_app_context, current_encryption_service, current_notification_system, - current_notify_interface_for_context, current_object_store_handle_for_context, + current_notify_interface_for_context, current_object_data_cache_for_context, current_object_store_handle_for_context, }; use crate::auth::get_condition_values_with_client_info; use crate::error::ApiError; @@ -1193,6 +1194,13 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + // Drop every cached object body for the now-deleted bucket so dead + // bytes do not sit resident until TTL. Covers both the normal and the + // force-delete path, which share this single delete_bucket call + // (ODC-28, backlog#1133). + let cache_adapter = current_object_data_cache_for_context(self.context.as_deref()); + let _ = invalidate_object_data_cache_bucket_after_delete(&cache_adapter, &input.bucket).await; + // Invalidate bucket validation cache crate::storage::invalidate_bucket_validation_cache(&input.bucket); diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 4f6eebf7c..dd0ca2977 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -131,7 +131,7 @@ pub fn resolve_object_store_handle_for_context(context: Option<&AppContext>) -> /// Resolve object data cache adapter using AppContext-first precedence. #[expect( dead_code, - reason = "ST-05 exposes the global resolver; app use sites start in later cache phases" + reason = "admin/app read sites resolve through the _for_context variant re-exported by runtime_sources" )] pub(crate) fn resolve_object_data_cache_handle() -> Option> { let context = get_global_app_context(); diff --git a/rustfs/src/app/context/global.rs b/rustfs/src/app/context/global.rs index 7903b4414..9275a00f8 100644 --- a/rustfs/src/app/context/global.rs +++ b/rustfs/src/app/context/global.rs @@ -82,6 +82,11 @@ impl AppContext { // Let ecstore probe this cache inside get_object_reader, after // metadata resolution but before the erasure data read (backlog#802). crate::app::object_data_cache::register_object_data_cache_body_hook(Arc::clone(&object_data_cache)); + // Let ecstore's internal delete paths (lifecycle/scanner expiry, + // noncurrent-version cleanup, restored-copy expiry) drop the removed + // object's cached body instead of leaving it resident until TTL + // (ODC-26, backlog#1131). + crate::app::object_data_cache::register_object_data_cache_mutation_hook(Arc::clone(&object_data_cache)); Self { object_store, diff --git a/rustfs/src/app/object_data_cache/adapter.rs b/rustfs/src/app/object_data_cache/adapter.rs index 57455545b..a5691ac72 100644 --- a/rustfs/src/app/object_data_cache/adapter.rs +++ b/rustfs/src/app/object_data_cache/adapter.rs @@ -16,7 +16,7 @@ use bytes::Bytes; use rustfs_object_data_cache::{ ObjectDataCache, ObjectDataCacheConfig, ObjectDataCacheConfigError, ObjectDataCacheFillResult, ObjectDataCacheGetPlan, ObjectDataCacheGetRequest, ObjectDataCacheIdentity, ObjectDataCacheInvalidationReason, ObjectDataCacheInvalidationResult, - ObjectDataCacheLookup, ObjectDataCacheMode, + ObjectDataCacheLookup, ObjectDataCacheMode, ObjectDataCacheStatsSnapshot, }; use std::{sync::Arc, time::Duration}; use tracing::warn; @@ -125,6 +125,40 @@ impl ObjectDataCacheAdapter { ) -> ObjectDataCacheInvalidationResult { self.cache.invalidate_object(identity, reason).await } + + /// Executes an engine-level prefix invalidation (ODC-27). + pub(crate) async fn invalidate_prefix( + &self, + bucket: &str, + prefix: &str, + reason: ObjectDataCacheInvalidationReason, + ) -> ObjectDataCacheInvalidationResult { + self.cache.invalidate_prefix(bucket, prefix, reason).await + } + + /// Executes an engine-level bucket invalidation (ODC-28). + pub(crate) async fn invalidate_bucket( + &self, + bucket: &str, + reason: ObjectDataCacheInvalidationReason, + ) -> ObjectDataCacheInvalidationResult { + self.cache.invalidate_bucket(bucket, reason).await + } + + /// Drops every cached body and resets the identity index (ODC-C2 flush). + pub(crate) async fn clear(&self, reason: ObjectDataCacheInvalidationReason) -> ObjectDataCacheInvalidationResult { + self.cache.clear(reason).await + } + + /// Current cache statistics snapshot (ODC-C2 admin stats). + pub(crate) fn stats(&self) -> ObjectDataCacheStatsSnapshot { + self.cache.stats() + } + + /// Configured runtime mode (ODC-C2 admin stats). + pub(crate) fn mode(&self) -> ObjectDataCacheMode { + self.cache.mode() + } } impl ObjectDataCacheAdapter { diff --git a/rustfs/src/app/object_data_cache/invalidation.rs b/rustfs/src/app/object_data_cache/invalidation.rs index 42dd7ea48..0c2a621b6 100644 --- a/rustfs/src/app/object_data_cache/invalidation.rs +++ b/rustfs/src/app/object_data_cache/invalidation.rs @@ -63,6 +63,42 @@ pub(crate) async fn invalidate_object_data_cache_after_complete_multipart_succes .await } +/// Invalidates every cached body under a prefix before a force-prefix delete +/// begins (ODC-27). The exact-key `..._before_mutation` helper only covers the +/// prefix string itself, leaving every object beneath it cached. +pub(crate) async fn invalidate_object_data_cache_prefix_before_mutation( + adapter: &ObjectDataCacheAdapter, + bucket: &str, + prefix: &str, +) -> ObjectDataCacheInvalidationResult { + adapter + .invalidate_prefix(bucket, prefix, ObjectDataCacheInvalidationReason::BeforeMutation) + .await +} + +/// Invalidates every cached body under a prefix after a successful force-prefix +/// delete (ODC-27). +pub(crate) async fn invalidate_object_data_cache_prefix_after_delete( + adapter: &ObjectDataCacheAdapter, + bucket: &str, + prefix: &str, +) -> ObjectDataCacheInvalidationResult { + adapter + .invalidate_prefix(bucket, prefix, ObjectDataCacheInvalidationReason::AfterPrefixDelete) + .await +} + +/// Invalidates every cached body in a bucket after the bucket is deleted +/// (ODC-28). +pub(crate) async fn invalidate_object_data_cache_bucket_after_delete( + adapter: &ObjectDataCacheAdapter, + bucket: &str, +) -> ObjectDataCacheInvalidationResult { + adapter + .invalidate_bucket(bucket, ObjectDataCacheInvalidationReason::AfterBucketDelete) + .await +} + /// Invalidates a single object identity through the app-layer cache adapter. pub(crate) async fn invalidate_object_data_cache_object( adapter: &ObjectDataCacheAdapter, diff --git a/rustfs/src/app/object_data_cache/mod.rs b/rustfs/src/app/object_data_cache/mod.rs index 006d2f4ad..b05feaffe 100644 --- a/rustfs/src/app/object_data_cache/mod.rs +++ b/rustfs/src/app/object_data_cache/mod.rs @@ -18,6 +18,7 @@ mod adapter; mod body; mod hook; mod invalidation; +mod mutation_hook; mod planner; pub(crate) use adapter::ObjectDataCacheAdapter; @@ -29,7 +30,9 @@ pub(crate) use hook::register_object_data_cache_body_hook; pub(crate) use invalidation::{ invalidate_object_data_cache_after_complete_multipart_success, invalidate_object_data_cache_after_copy_success, invalidate_object_data_cache_after_delete_success, invalidate_object_data_cache_after_put_success, - invalidate_object_data_cache_before_mutation, invalidate_object_data_cache_objects_after_delete_success, - invalidate_object_data_cache_objects_before_mutation, + invalidate_object_data_cache_before_mutation, invalidate_object_data_cache_bucket_after_delete, + invalidate_object_data_cache_objects_after_delete_success, invalidate_object_data_cache_objects_before_mutation, + invalidate_object_data_cache_prefix_after_delete, invalidate_object_data_cache_prefix_before_mutation, }; +pub(crate) use mutation_hook::register_object_data_cache_mutation_hook; pub(crate) use planner::{GetObjectBodyCachePlan, GetObjectBodyCacheRequest, build_get_object_body_cache_plan}; diff --git a/rustfs/src/app/object_data_cache/mutation_hook.rs b/rustfs/src/app/object_data_cache/mutation_hook.rs new file mode 100644 index 000000000..a07af5e97 --- /dev/null +++ b/rustfs/src/app/object_data_cache/mutation_hook.rs @@ -0,0 +1,104 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! ecstore-facing write-side cache hook. +//! +//! Registered into ecstore so its internal delete paths (lifecycle/scanner +//! expiry, noncurrent-version cleanup, restored-copy expiry) can drop the +//! object's cached body the moment the object is removed, instead of leaving +//! dead bytes resident until TTL (ODC-26, backlog#1131). + +use crate::app::object_data_cache::ObjectDataCacheAdapter; +use crate::storage::storage_api::ecstore_object::{ObjectMutationHook, register_object_mutation_hook}; +use rustfs_object_data_cache::{ObjectDataCacheIdentity, ObjectDataCacheInvalidationReason}; +use std::sync::Arc; + +/// Adapter-backed implementation of ecstore's object mutation hook. +pub(crate) struct ObjectDataCacheMutationHook { + adapter: Arc, +} + +/// Registers the mutation hook into ecstore. No-op for a disabled cache so the +/// delete paths keep a single `None` branch when the feature is off, matching +/// [`register_object_data_cache_body_hook`](super::register_object_data_cache_body_hook). +pub(crate) fn register_object_data_cache_mutation_hook(adapter: Arc) { + if adapter.is_disabled() { + return; + } + register_object_mutation_hook(Arc::new(ObjectDataCacheMutationHook { adapter })); +} + +#[async_trait::async_trait] +impl ObjectMutationHook for ObjectDataCacheMutationHook { + async fn after_object_mutation(&self, bucket: &str, object: &str) { + let _ = self + .adapter + .invalidate_object( + ObjectDataCacheIdentity::new(bucket, object), + ObjectDataCacheInvalidationReason::AfterLifecycleExpiry, + ) + .await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use rustfs_object_data_cache::{ + ObjectDataCacheBodyVariant, ObjectDataCacheConfig, ObjectDataCacheFillResult, ObjectDataCacheGetRequest, + ObjectDataCacheLookup, ObjectDataCacheMode, + }; + + fn fill_enabled_adapter() -> Arc { + Arc::new( + ObjectDataCacheAdapter::new(ObjectDataCacheConfig { + mode: ObjectDataCacheMode::FillMaterializeEnabled, + max_bytes: 8_388_608, + // Fill must not depend on the live memory reading (host vs container). + min_free_memory_percent: 0, + ..ObjectDataCacheConfig::default() + }) + .expect("adapter"), + ) + } + + fn request<'a>(bucket: &'a str, object: &'a str) -> ObjectDataCacheGetRequest<'a> { + ObjectDataCacheGetRequest { + bucket, + object, + version_id: None, + etag: "etag", + size: 5, + body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, + } + } + + #[tokio::test] + async fn mutation_hook_invalidates_cached_body() { + let adapter = fill_enabled_adapter(); + let plan = adapter.plan_get(request("b", "k")); + assert_eq!( + adapter.fill_body(&plan, Bytes::from_static(b"hello")).await, + ObjectDataCacheFillResult::Inserted + ); + + let hook = ObjectDataCacheMutationHook { + adapter: Arc::clone(&adapter), + }; + hook.after_object_mutation("b", "k").await; + + assert!(matches!(adapter.lookup_body(&plan).await, ObjectDataCacheLookup::Miss)); + } +} diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 41f38c3a5..27b75dcf8 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -200,7 +200,8 @@ use crate::app::object_data_cache::{ fill_get_object_body_cache_from_materialized_body, invalidate_object_data_cache_after_copy_success, invalidate_object_data_cache_after_delete_success, invalidate_object_data_cache_after_put_success, invalidate_object_data_cache_before_mutation, invalidate_object_data_cache_objects_after_delete_success, - invalidate_object_data_cache_objects_before_mutation, lookup_get_object_body_cache_hit, + invalidate_object_data_cache_objects_before_mutation, invalidate_object_data_cache_prefix_after_delete, + invalidate_object_data_cache_prefix_before_mutation, lookup_get_object_body_cache_hit, }; type S3StdError = Box; @@ -5543,7 +5544,14 @@ impl DefaultObjectUsecase { }; let cache_adapter = self.object_data_cache(); - let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await; + // A force (delete_prefix) delete removes every object under `key` as a + // prefix, so invalidating only the exact key would strand every cached + // body beneath it. Use the prefix primitive in that branch (ODC-27). + if force_delete { + let _ = invalidate_object_data_cache_prefix_before_mutation(&cache_adapter, &bucket, &key).await; + } else { + let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await; + } let obj_info = { match store.delete_object(&bucket, &key, opts.clone()).await { @@ -5574,7 +5582,11 @@ impl DefaultObjectUsecase { "failed to persist transitioned object cleanup journal" ); } - let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; + if force_delete { + let _ = invalidate_object_data_cache_prefix_after_delete(&cache_adapter, &bucket, &key).await; + } else { + let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; + } // Fast in-memory update for immediate quota and admin usage consistency if delete_creates_delete_marker(&opts) { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index f98a04dc7..cc4befc45 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -441,7 +441,9 @@ pub(crate) mod ecstore_rpc { } pub(crate) mod ecstore_object { - pub(crate) use rustfs_ecstore::api::object::{GetObjectBodyCacheHook, register_get_object_body_cache_hook}; + pub(crate) use rustfs_ecstore::api::object::{ + GetObjectBodyCacheHook, ObjectMutationHook, register_get_object_body_cache_hook, register_object_mutation_hook, + }; } pub(crate) mod ecstore_set_disk {