fix(cache): harden object data cache coordination (#5004)

* fix(cache): enforce projected entry capacity

Refs: rustfs/backlog#1335

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(cache): fence identity budget eviction by generation

Refs rustfs/backlog#1334.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(cache): fence clear against concurrent fills

Refs rustfs/backlog#1333

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(cache): linearize memory reservation claims

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(cache): retain allocation memory claims

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(cache): publish memory snapshots by epoch

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(cache): coordinate cold object fills

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): fence metadata cache transition races

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/object_api/body_cache_hook.rs

@@ -20,11 +20,17 @@
 //! probing earlier would require a second metadata fan-out, and probing later
 //! (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 crate::object_api::{ObjectInfo, ObjectOptions};
+use crate::storage_api_contracts::range::HTTPRangeSpec;
 use bytes::Bytes;
+use std::future::Future;
 use std::sync::Arc;
 
+tokio::task_local! {
+    static SKIP_GET_OBJECT_BODY_CACHE_HOOK: bool;
+}
+
 /// Serves full-object GET bodies from a cache keyed by object identity.
 ///
 /// Implementations must validate identity (etag/version/size) against the
@@ -62,14 +68,263 @@ pub fn register_get_object_body_cache_hook(hook: Arc<dyn GetObjectBodyCacheHook>
     );
 }
 
+/// Unregister the process-wide GET body cache hook.
+///
+/// Config reloads use this when body caching becomes disabled so an adapter
+/// retained by the previous configuration cannot continue serving stale hits.
+pub fn unregister_get_object_body_cache_hook() {
+    GET_OBJECT_BODY_CACHE_HOOK.clear();
+}
+
+/// Probes the registered hook against an already resolved metadata snapshot.
+/// Staged GET callers use this once before suppressing the nested reader probe,
+/// preserving the legacy hook contract.
+#[non_exhaustive]
+pub enum GetObjectBodyCacheHookLookup {
+    Ineligible,
+    Absent,
+    Miss,
+    Hit(Bytes),
+}
+
+/// Returns the complete plaintext length only when the request can safely use
+/// the body-cache hook. Callers may use this before conditional decisions so
+/// ineligible reads retain the established reader-path error precedence.
+pub fn get_object_body_cache_plaintext_len(
+    range: &Option<HTTPRangeSpec>,
+    opts: &ObjectOptions,
+    info: &ObjectInfo,
+) -> Option<i64> {
+    crate::set_disk::body_cache_plaintext_len(range, opts, info)
+}
+
+pub async fn lookup_get_object_body_cache_hook(
+    bucket: &str,
+    object: &str,
+    range: &Option<HTTPRangeSpec>,
+    opts: &ObjectOptions,
+    info: &ObjectInfo,
+) -> GetObjectBodyCacheHookLookup {
+    let Some(plaintext_len) = get_object_body_cache_plaintext_len(range, opts, info) else {
+        return GetObjectBodyCacheHookLookup::Ineligible;
+    };
+    let Some(hook) = get_object_body_cache_hook() else {
+        return GetObjectBodyCacheHookLookup::Absent;
+    };
+    match hook.lookup(bucket, object, info).await {
+        Some(body) if i64::try_from(body.len()).is_ok_and(|body_len| body_len == plaintext_len) => {
+            GetObjectBodyCacheHookLookup::Hit(body)
+        }
+        Some(_) | None => GetObjectBodyCacheHookLookup::Miss,
+    }
+}
+
 /// The registered hook, if any.
 pub(crate) fn get_object_body_cache_hook() -> Option<Arc<dyn GetObjectBodyCacheHook>> {
     GET_OBJECT_BODY_CACHE_HOOK.get()
 }
 
+pub(crate) fn get_object_body_cache_hook_suppressed() -> bool {
+    SKIP_GET_OBJECT_BODY_CACHE_HOOK.try_with(|skip| *skip).unwrap_or(false)
+}
+
+pub(crate) async fn without_get_object_body_cache_hook<F>(future: F) -> F::Output
+where
+    F: Future,
+{
+    SKIP_GET_OBJECT_BODY_CACHE_HOOK.scope(true, future).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_get_object_body_cache_hook() {
-    GET_OBJECT_BODY_CACHE_HOOK.clear();
+    unregister_get_object_body_cache_hook();
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    struct LegacyHook {
+        calls: AtomicUsize,
+        body: Option<Bytes>,
+    }
+
+    #[async_trait::async_trait]
+    impl GetObjectBodyCacheHook for LegacyHook {
+        async fn lookup(&self, _bucket: &str, _object: &str, _info: &ObjectInfo) -> Option<Bytes> {
+            self.calls.fetch_add(1, Ordering::Relaxed);
+            self.body.clone()
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn staged_probe_preserves_legacy_hook_hit_once() {
+        clear_get_object_body_cache_hook();
+        let hook = Arc::new(LegacyHook {
+            calls: AtomicUsize::new(0),
+            body: Some(Bytes::from_static(b"legacy")),
+        });
+        register_get_object_body_cache_hook(Arc::clone(&hook) as Arc<dyn GetObjectBodyCacheHook>);
+
+        let info = ObjectInfo {
+            size: 6,
+            actual_size: 6,
+            ..Default::default()
+        };
+        let GetObjectBodyCacheHookLookup::Hit(body) =
+            lookup_get_object_body_cache_hook("bucket", "object", &None, &ObjectOptions::default(), &info).await
+        else {
+            panic!("legacy hook hit must be returned");
+        };
+        assert_eq!(body, Bytes::from_static(b"legacy"));
+        assert_eq!(hook.calls.load(Ordering::Relaxed), 1);
+        clear_get_object_body_cache_hook();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn staged_probe_treats_wrong_length_legacy_body_as_authoritative_miss() {
+        clear_get_object_body_cache_hook();
+        let hook = Arc::new(LegacyHook {
+            calls: AtomicUsize::new(0),
+            body: Some(Bytes::from_static(b"short")),
+        });
+        register_get_object_body_cache_hook(Arc::clone(&hook) as Arc<dyn GetObjectBodyCacheHook>);
+        let info = ObjectInfo {
+            size: 6,
+            actual_size: 6,
+            ..Default::default()
+        };
+
+        assert!(matches!(
+            lookup_get_object_body_cache_hook("bucket", "object", &None, &ObjectOptions::default(), &info).await,
+            GetObjectBodyCacheHookLookup::Miss
+        ));
+        assert_eq!(hook.calls.load(Ordering::Relaxed), 1);
+        clear_get_object_body_cache_hook();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn staged_probe_preserves_legacy_hook_miss_once() {
+        clear_get_object_body_cache_hook();
+        let hook = Arc::new(LegacyHook {
+            calls: AtomicUsize::new(0),
+            body: None,
+        });
+        register_get_object_body_cache_hook(Arc::clone(&hook) as Arc<dyn GetObjectBodyCacheHook>);
+
+        let info = ObjectInfo {
+            size: 4,
+            actual_size: 4,
+            ..Default::default()
+        };
+        assert!(matches!(
+            lookup_get_object_body_cache_hook("bucket", "object", &None, &ObjectOptions::default(), &info).await,
+            GetObjectBodyCacheHookLookup::Miss
+        ));
+        assert_eq!(hook.calls.load(Ordering::Relaxed), 1);
+        clear_get_object_body_cache_hook();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn staged_probe_bypasses_raw_movement_and_restore_reads() {
+        clear_get_object_body_cache_hook();
+        let hook = Arc::new(LegacyHook {
+            calls: AtomicUsize::new(0),
+            body: Some(Bytes::from_static(b"body")),
+        });
+        register_get_object_body_cache_hook(Arc::clone(&hook) as Arc<dyn GetObjectBodyCacheHook>);
+        let info = ObjectInfo {
+            size: 4,
+            actual_size: 4,
+            ..Default::default()
+        };
+        let mut restore = ObjectOptions::default();
+        restore.transition.restore_request.days = Some(1);
+        let cases = [
+            ObjectOptions {
+                raw_data_movement_read: true,
+                ..Default::default()
+            },
+            ObjectOptions {
+                data_movement: true,
+                ..Default::default()
+            },
+            restore,
+        ];
+
+        for opts in &cases {
+            assert!(matches!(
+                lookup_get_object_body_cache_hook("bucket", "object", &None, opts, &info).await,
+                GetObjectBodyCacheHookLookup::Ineligible
+            ));
+        }
+        assert_eq!(hook.calls.load(Ordering::Relaxed), 0);
+        clear_get_object_body_cache_hook();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn staged_probe_bypasses_pre_hook_early_return_objects() {
+        clear_get_object_body_cache_hook();
+        let hook = Arc::new(LegacyHook {
+            calls: AtomicUsize::new(0),
+            body: Some(Bytes::from_static(b"body")),
+        });
+        register_get_object_body_cache_hook(Arc::clone(&hook) as Arc<dyn GetObjectBodyCacheHook>);
+        let delete_marker = ObjectInfo {
+            delete_marker: true,
+            size: 4,
+            actual_size: 4,
+            ..Default::default()
+        };
+        let zero = ObjectInfo::default();
+        let inline = ObjectInfo {
+            inlined: true,
+            size: 4,
+            actual_size: 4,
+            parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
+                number: 1,
+                ..Default::default()
+            }]),
+            ..Default::default()
+        };
+        let version_only = ObjectInfo {
+            version_only: true,
+            size: 4,
+            actual_size: 4,
+            ..Default::default()
+        };
+        let metadata_only = ObjectInfo {
+            metadata_only: true,
+            size: 4,
+            actual_size: 4,
+            ..Default::default()
+        };
+
+        for info in [&delete_marker, &zero, &inline, &version_only, &metadata_only] {
+            assert!(matches!(
+                lookup_get_object_body_cache_hook("bucket", "object", &None, &ObjectOptions::default(), info).await,
+                GetObjectBodyCacheHookLookup::Ineligible
+            ));
+        }
+        assert_eq!(hook.calls.load(Ordering::Relaxed), 0);
+        clear_get_object_body_cache_hook();
+    }
+
+    #[tokio::test]
+    async fn staged_reader_scope_suppresses_only_the_nested_probe() {
+        assert!(!get_object_body_cache_hook_suppressed());
+        without_get_object_body_cache_hook(async {
+            assert!(get_object_body_cache_hook_suppressed());
+        })
+        .await;
+        assert!(!get_object_body_cache_hook_suppressed());
+    }
 }

crates/ecstore/src/object_api/hook_slot.rs

@@ -61,8 +61,7 @@ impl<T: ?Sized> HookSlot<T> {
         self.inner.read().unwrap_or_else(|poisoned| poisoned.into_inner()).clone()
     }
 
-    /// Clears the slot. Test-only: production never unregisters a hook.
-    #[cfg(test)]
+    /// Clears the slot during feature disable or test cleanup.
     pub(crate) fn clear(&self) {
         *self.inner.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
     }

crates/ecstore/src/object_api/mod.rs

@@ -60,9 +60,14 @@ mod types;
 
 #[cfg(test)]
 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 use body_cache_hook::{
+    GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook,
+    register_get_object_body_cache_hook, unregister_get_object_body_cache_hook,
+};
+pub(crate) use body_cache_hook::{
+    get_object_body_cache_hook, get_object_body_cache_hook_suppressed, without_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 object_mutation_hook::{ObjectMutationHook, register_object_mutation_hook, unregister_object_mutation_hook};
 pub use readers::*;
 pub use types::*;

crates/ecstore/src/object_api/object_mutation_hook.rs

@@ -56,6 +56,14 @@ pub fn register_object_mutation_hook(hook: Arc<dyn ObjectMutationHook>) {
     );
 }
 
+/// Unregister the process-wide object mutation hook.
+///
+/// Config reloads use this when body caching becomes disabled so the previous
+/// adapter and its cached plaintext bodies are not retained until their TTL.
+pub fn unregister_object_mutation_hook() {
+    OBJECT_MUTATION_HOOK.clear();
+}
+
 /// The registered hook, if any.
 fn object_mutation_hook() -> Option<Arc<dyn ObjectMutationHook>> {
     OBJECT_MUTATION_HOOK.get()
@@ -71,13 +79,6 @@ pub(crate) async fn notify_object_mutation(bucket: &str, object: &str) {
     }
 }
 
-/// 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::*;
@@ -97,7 +98,7 @@ mod tests {
     #[tokio::test]
     #[serial_test::serial(object_mutation_hook)]
     async fn notify_invokes_registered_hook_with_identity() {
-        clear_object_mutation_hook();
+        unregister_object_mutation_hook();
         let calls = Arc::new(Mutex::new(Vec::new()));
         register_object_mutation_hook(Arc::new(RecordingHook {
             calls: Arc::clone(&calls),
@@ -106,14 +107,35 @@ mod tests {
         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();
+        unregister_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).
+    async fn unregister_prevents_later_notifications() {
+        unregister_object_mutation_hook();
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        register_object_mutation_hook(Arc::new(RecordingHook {
+            calls: Arc::clone(&calls),
+        }));
+        unregister_object_mutation_hook();
+
         notify_object_mutation("bucket", "object").await;
+
+        assert!(calls.lock().unwrap().is_empty(), "an unregistered hook must receive no mutation callback");
+    }
+
+    #[test]
+    #[serial_test::serial(object_mutation_hook)]
+    fn unregister_releases_the_previous_hook() {
+        unregister_object_mutation_hook();
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        let hook = Arc::new(RecordingHook { calls });
+        let weak = Arc::downgrade(&hook);
+        register_object_mutation_hook(hook);
+
+        assert!(weak.upgrade().is_some());
+        unregister_object_mutation_hook();
+        assert!(weak.upgrade().is_none());
     }
 }

crates/ecstore/src/object_api/readers.rs

@@ -307,6 +307,17 @@ pub struct GetObjectReader {
 }
 
 impl GetObjectReader {
+    /// Builds a fully materialized reader from a cache-coordinated body.
+    pub fn from_cache_body(mut object_info: ObjectInfo, body: Bytes) -> Result<Self> {
+        object_info.size = i64::try_from(body.len()).map_err(|_| Error::other("cached GET body length exceeds i64::MAX"))?;
+        Ok(Self {
+            stream: Box::new(std::io::Cursor::new(body.clone())),
+            object_info,
+            buffered_body: Some(body),
+            body_source: GetObjectBodySource::HookServed,
+        })
+    }
+
     /// True when `buffered_body` is the body the cache hook served. The app
     /// layer serves it as the object-data-cache source without a second lookup.
     pub fn is_cache_hook_served(&self) -> bool {
@@ -1674,6 +1685,40 @@ mod tests {
         bytes
     }
 
+    #[tokio::test]
+    async fn cache_body_uses_plaintext_length_for_compressed_metadata() {
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            rustfs_utils::http::SUFFIX_COMPRESSION,
+            "klauspost/compress/s2".to_string(),
+        );
+        let object_info = ObjectInfo {
+            size: 3,
+            actual_size: 11,
+            user_defined: Arc::new(metadata),
+            ..Default::default()
+        };
+        assert!(object_info.is_compressed());
+
+        let body = Bytes::from_static(b"hello world");
+        let mut reader =
+            GetObjectReader::from_cache_body(object_info, body.clone()).expect("cache body length must fit in object metadata");
+
+        assert_eq!(reader.body_source, GetObjectBodySource::HookServed);
+        assert_eq!(reader.buffered_body.as_ref(), Some(&body));
+        assert_eq!(reader.object_info.size, 11);
+        assert_eq!(reader.object_info.actual_size, 11);
+        assert!(reader.object_info.is_compressed());
+        let mut restored = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("cache body should stream");
+        assert_eq!(restored, body);
+    }
+
     /// Regression for the #4576 fallout: the encrypt side persists a random
     /// SSE-C nonce, and this reader-side resolver must read it back — falling
     /// back to the deterministic legacy nonce only when no IV was stored.