mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
feat(object-data-cache): close write-side invalidation gaps and add an admin surface (#4694)
* feat(object-data-cache): close write/delete-side invalidation gaps The object data cache exposed only a single per-(bucket,object) invalidation primitive and no write-side ecstore hook, so several delete paths left dead bodies resident until TTL (hygiene/capacity, not stale-serving: lookups follow a fresh metadata quorum and cannot serve a gone object). This adds the missing primitives and wires them in. ODC-26 (backlog#1131): add an `ObjectMutationHook` trait beside the GET body hook, registered next to it at startup, and call it from the ecstore-internal delete paths (`apply_expiry_on_non_transitioned_objects`, `expire_transitioned_object` including the restored-copy branch, and `delete_object_versions`). The app impl is one `invalidate_object` call under a new `AfterLifecycleExpiry` reason. ODC-27 (backlog#1132): force prefix delete now invalidates the whole prefix, not just the prefix string. `store.delete_object(delete_prefix)` returns no deleted-name list, so this uses a new prefix primitive rather than the batch path. ODC-28 (backlog#1133): DeleteBucket now flushes the bucket via a new bucket-scope primitive (covers force and non-force, which share the delete_bucket call). ODC-C2 (backlog#1143): add `ObjectDataCache::clear()` and two admin handlers (GET stats, POST flush) routed through admin runtime_sources. The starshard identity index gains a single `remove_matching` full-scan API backing prefix/bucket/clear; it is documented as admin/delete-path only and never runs on the GET or fill hot path. New invalidation reasons and metric labels added; outcome (removed/noop) labelling kept correct for every new primitive. Also fixes a pre-existing broken intra-doc link in memory.rs. Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(ecstore): extract the shared HookSlot behind both cache hooks This PR introduced object_mutation_hook.rs by mirroring body_cache_hook.rs, which left two process-global registration slots whose register/get/clear bodies were line-for-line identical except the trait type and the WARN string: a RwLock<Option<Arc<dyn _>>>, an Arc::ptr_eq "different instance" warning, the poison-recovery closure, and the same read-lock-and-clone read. Two copies of the same swap-vs-warn logic can drift apart under maintenance. Hoist it into a generic HookSlot<T: ?Sized> that owns the logic once. Each hook module keeps its `static HOOK: HookSlot<dyn XxxHook>` and its thin, unchanged public wrappers (register_/get_/clear_), so the crate's public surface and every call site are untouched — this is an internal consolidation, not a contract change. The load-bearing #1126 guarantee (newest registration wins, so a rebuilt AppContext is never stranded on a first-wins slot) previously had no direct test — the hook tests only covered register-then-notify. HookSlot now has its own unit tests including re_registration_swaps_to_the_latest_instance; mutation-testing confirms a first-wins regression fails exactly that test. No behavior change: the two hooks' existing tests, the P0 body_cache_hook_e2e regressions, and the app-layer mutation-hook tests all pass unchanged. Refs: backlog#1126, backlog#1131 Co-Authored-By: heihutu <heihutu@gmail.com> * fix(admin): register the object-data-cache routes in the policy inventory This PR added GET /object-data-cache/stats and POST /object-data-cache/flush but did not list them in the two registries that must account for every admin route: the route-policy inventory (route_policy.rs) and the route matrix (route_registration_test.rs). Their coverage tests — route_policy_inventory_covers_registered_routes and test_admin_route_matrix_matches_registered_routes — failed on CI because a registered route had no policy/matrix entry. These two tests are not part of `make pre-commit` (which runs fmt + arch + quick-check, not the full suite), so the gap passed local pre-commit and only surfaced in the CI Test-and-Lint lane. stats is a read (ServerInfoAdminAction, Sensitive); flush mutates (ConfigUpdateAdminAction, High) — matching the actions the handlers already enforce. The MinIO-alias matrix test is unaffected: these are native rustfs endpoints with no MinIO equivalent. Refs: backlog#1143 Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -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,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -103,6 +103,11 @@ pub async fn delete_object_versions(api: &Arc<ECStore>, 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;
|
||||
};
|
||||
|
||||
@@ -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<Bytes>;
|
||||
}
|
||||
|
||||
// `RwLock<Option<Arc<dyn ...>>>` rather than `ArcSwapOption`: arc-swap's
|
||||
// `RefCnt` is implemented only for the sized `Arc<T>` (it stores a thin
|
||||
// `*mut T`), so it cannot hold an `Arc<dyn GetObjectBodyCacheHook>` 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<Option<Arc<dyn GetObjectBodyCacheHook>>> = RwLock::new(None);
|
||||
// A `HookSlot` (RwLock<Option<Arc<dyn ...>>>) rather than `ArcSwapOption`:
|
||||
// arc-swap's `RefCnt` is implemented only for the sized `Arc<T>` (it stores a
|
||||
// thin `*mut T`), so it cannot hold an `Arc<dyn GetObjectBodyCacheHook>`
|
||||
// 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<dyn GetObjectBodyCacheHook> = HookSlot::new();
|
||||
|
||||
/// Register (or re-register) the process-wide GET body cache hook.
|
||||
///
|
||||
@@ -52,31 +54,22 @@ static GET_OBJECT_BODY_CACHE_HOOK: RwLock<Option<Arc<dyn GetObjectBodyCacheHook>
|
||||
/// 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<dyn GetObjectBodyCacheHook>) {
|
||||
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<Arc<dyn GetObjectBodyCacheHook>> {
|
||||
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();
|
||||
}
|
||||
|
||||
@@ -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<T>` hook.
|
||||
///
|
||||
/// Registration atomically swaps the stored hook. Poisoned locks recover in
|
||||
/// place: a panicked writer cannot leave an `Option<Arc<_>>` in an unsound
|
||||
/// state, so there is nothing to salvage.
|
||||
pub(crate) struct HookSlot<T: ?Sized> {
|
||||
inner: RwLock<Option<Arc<T>>>,
|
||||
}
|
||||
|
||||
impl<T: ?Sized> HookSlot<T> {
|
||||
/// 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<T>, 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<Arc<T>> {
|
||||
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<dyn Marker> = HookSlot::new();
|
||||
assert!(slot.get().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn register_then_get_returns_the_hook() {
|
||||
let slot: HookSlot<dyn Marker> = 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<dyn Marker> = 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<dyn Marker> = HookSlot::new();
|
||||
let hook: Arc<dyn Marker> = 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<dyn Marker> = HookSlot::new();
|
||||
slot.register(Arc::new(Impl(3)), "unused");
|
||||
slot.clear();
|
||||
assert!(slot.get().is_none());
|
||||
}
|
||||
}
|
||||
@@ -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::*;
|
||||
|
||||
@@ -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<dyn ObjectMutationHook>`. 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<dyn ObjectMutationHook> = 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<dyn ObjectMutationHook>) {
|
||||
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<Arc<dyn ObjectMutationHook>> {
|
||||
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<Mutex<Vec<(String, String)>>>,
|
||||
}
|
||||
|
||||
#[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;
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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<F>(&self, predicate: F) -> Vec<ObjectDataCacheKey>
|
||||
where
|
||||
F: Fn(&ObjectDataCacheIdentity) -> bool,
|
||||
{
|
||||
let identities: Vec<ObjectDataCacheIdentity> = 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);
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String>,
|
||||
object: Option<String>,
|
||||
outcome: &'static str,
|
||||
removed_keys: usize,
|
||||
}
|
||||
|
||||
pub fn register_object_data_cache_route(r: &mut S3Router<AdminOperation>) -> 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<Body>, 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::<Option<RemoteAddr>>()
|
||||
.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<T: Serialize>(body: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
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<Body>, key: &str) -> Option<String> {
|
||||
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
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<ObjectDataCacheAdapter> =
|
||||
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?;"));
|
||||
}
|
||||
}
|
||||
@@ -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<AdminOperation>) -> 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)?;
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -172,6 +172,8 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
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"),
|
||||
|
||||
@@ -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<Arc<ECStore>> {
|
||||
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<Arc<ObjectDataCacheAdapter>> {
|
||||
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,
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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<Arc<ObjectDataCacheAdapter>> {
|
||||
let context = get_global_app_context();
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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};
|
||||
|
||||
@@ -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<ObjectDataCacheAdapter>,
|
||||
}
|
||||
|
||||
/// 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<ObjectDataCacheAdapter>) {
|
||||
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<ObjectDataCacheAdapter> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
@@ -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<dyn std::error::Error + Send + Sync + 'static>;
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user