fix(cache): stop the GET body-cache hook from bypassing ReadPlan (#4654)

* fix(cache): gate GET body-cache hook to preserve ReadPlan output

The ecstore GET body-cache hook serves cached full-object plaintext
directly, bypassing ReadPlan/ReadTransform. That is only sound when the
normal read path returns that same plaintext byte-for-byte. Two probe
conditions were missing, plus two usecase-layer planner gaps.

ODC-01 (backlog#1108): raw/data-movement reads. ReadPlan::build returns
the STORED representation for raw_data_movement_read (e.g. compressed
bytes, length = oi.size), but the cache holds the post-decompression
body. Decommission (raw_data_movement_read: true) would receive
decompressed plaintext where raw compressed bytes are required, silently
corrupting the destination pool.

ODC-02 (backlog#1109): compressed objects. ReadTransform::Compressed
rewrites object_info.size to the decompressed length; on a hook hit
object_info is returned unchanged, so object_info.size is the compressed
size while the stream carries the decompressed body. UploadPartCopy then
uses src_info.size as the copy length and truncates the part.

Fix: gate the hook probe with should_probe_body_cache_hook, refusing
raw_data_movement_read, data_movement, and compressed objects, mirroring
the conditions get_small_object_direct_memory_decision already applies.

ODC-33 (backlog#1138): build_get_object_body_cache_plan lacked the
is_remote() exclusion the ecstore hook enforces; add it so transitioned
(remote-tier) objects are excluded uniformly.

ODC-C1 (backlog#1142): zero-length bodies save no I/O (ecstore returns an
empty body before the hook probe) yet the planner admitted them; change
the guard to response_content_length <= 0 so they plan Skip, mirroring
should_buffer_get_object_in_memory_with_threshold.

Tests: body_cache_hook_gate_tests (4) cover plain-probe plus
raw/data-movement/compressed skips; planner gains
plan_skips_remote_transitioned_objects and plan_skips_zero_length_objects.

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

* fix(cache): allow compressed bodies via a fail-closed read allow-list

The body-cache hook probe was gated by a deny-list that refused compressed
objects outright, which cost the cache every compressed body — a growing
share of stored data. Replace it with an allow-list that returns the exact
plaintext length a hit may serve, or None.

full_object_plaintext_len() answers a single question: would the normal
ReadPlan produce this object's complete plaintext, and under which size?
Compressed objects now qualify, and the hit site publishes the returned
length as object_info.size, reproducing the contract ReadTransform::
Compressed establishes. A hit whose body length disagrees is refused and
falls through to the erasure read.

This also closes a gate the deny-list only covered by accident: a restore
read forces ReadPlan down the Plain branch, so a compressed object yields
STORED bytes under its compressed size. Refusing compressed objects hid
that; admitting them exposes it, so restore reads are refused explicitly.

Being fail-closed, a newly added ReadPlan branch bypasses the cache by
default rather than silently serving the wrong representation — the
structural defect behind both backlog#1108 and backlog#1109.

Refs: backlog#1108, backlog#1109

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

---------

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

crates/ecstore/src/object_api/readers.rs

@@ -130,7 +130,9 @@ fn http_range_spec_from_object_info(oi: &ObjectInfo, part_number: usize) -> Opti
     HTTPRangeSpec::from_part_sizes(oi.size, part_number, oi.parts.iter().map(part_plaintext_size))
 }
 
-fn restore_request_active(opts: &ObjectOptions) -> bool {
+/// A restore read forces `ReadPlan::build` down the `Plain` branch, so it
+/// yields the STORED representation even for compressed or encrypted objects.
+pub(crate) fn restore_request_active(opts: &ObjectOptions) -> bool {
     let restore = &opts.transition.restore_request;
     restore.type_.is_some() || restore.days.is_some() || restore.output_location.is_some() || restore.select_parameters.is_some()
 }

crates/ecstore/src/set_disk/ops/object.rs

@@ -23,6 +23,49 @@ use super::super::*;
 
 use crate::disk::OldCurrentSize;
 
+/// Length of the full plaintext body when — and only when — this read's output
+/// is exactly the object's complete plaintext, so the app-layer body cache may
+/// serve it in place of the erasure read.
+///
+/// A hook hit bypasses `ReadPlan`/`ReadTransform` entirely, so it is sound only
+/// where the normal read path would produce that same plaintext byte-for-byte
+/// AND expose the same `object_info.size`. This is a fail-closed allow-list, not
+/// a deny-list: every read whose `ReadPlan` applies some other transform returns
+/// `None`, so a newly added `ReadPlan` branch bypasses the cache by default
+/// instead of silently serving bytes in the wrong representation.
+///
+/// Refused reads, each mapping to a `ReadPlan::build` branch:
+/// - ranged / part-number reads — the cache only holds whole objects;
+/// - `raw_data_movement_read` / `data_movement` — yields the STORED
+///   representation, e.g. compressed bytes (backlog#1108);
+/// - restore reads — `restore_request_active` forces the `Plain` branch, so a
+///   compressed object yields STORED bytes under its compressed `size`;
+/// - encrypted objects — `ReadTransform::Encrypted` rewrites size and the cache
+///   must never hold their plaintext;
+/// - remote (transitioned) objects — served from the warm tier.
+///
+/// Compressed objects ARE eligible: `ReadTransform::Compressed` returns the full
+/// plaintext and rewrites `object_info.size` to the decompressed length, which
+/// the caller must replicate with the returned length (backlog#1109).
+fn full_object_plaintext_len(range: &Option<HTTPRangeSpec>, opts: &ObjectOptions, object_info: &ObjectInfo) -> Option<i64> {
+    if range.is_some()
+        || opts.part_number.is_some()
+        || opts.raw_data_movement_read
+        || opts.data_movement
+        || crate::object_api::restore_request_active(opts)
+        || object_info.is_encrypted()
+        || object_info.is_remote()
+    {
+        return None;
+    }
+
+    if object_info.is_compressed() {
+        return object_info.get_actual_size().ok();
+    }
+
+    Some(object_info.size)
+}
+
 #[async_trait::async_trait]
 impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
     type Error = Error;
@@ -340,12 +383,20 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         // but no data shards have been read yet, so a hit skips the erasure
         // read, bitrot verify and decode entirely. The hook validates object
         // identity and rejects anything it cannot serve byte-identically.
-        if range.is_none()
-            && opts.part_number.is_none()
+        //
+        // `plaintext_len` carries the size the normal `ReadPlan` would have
+        // published, which the reader below must reproduce: for a compressed
+        // object `ReadTransform::Compressed` sets `object_info.size` to the
+        // decompressed length, and consumers such as UploadPartCopy read the
+        // copy length straight off that field (backlog#1109).
+        if let Some(plaintext_len) = full_object_plaintext_len(&range, opts, &object_info)
             && let Some(hook) = get_object_body_cache_hook()
             && let Some(body) = hook.lookup(bucket, object, &object_info).await
+            && i64::try_from(body.len()).is_ok_and(|len| len == plaintext_len)
         {
             record_get_object_reader_path_observation(GET_OBJECT_PATH_BODY_CACHE, object_class, size_bucket);
+            let mut object_info = object_info;
+            object_info.size = plaintext_len;
             let reader = GetObjectReader {
                 stream: Box::new(Cursor::new(body.clone())),
                 object_info,
@@ -2936,3 +2987,124 @@ mod delete_objects_lock_gating_tests {
         assert!(deleted[0].found, "existing object must still be deleted");
     }
 }
+
+#[cfg(test)]
+mod body_cache_hook_gate_tests {
+    //! Regression coverage for backlog#1108 (raw data-movement reads) and
+    //! backlog#1109 (compressed objects): the app-layer body-cache hook serves
+    //! cached full-object plaintext directly, bypassing `ReadPlan`. It must not
+    //! be probed when the normal read path would return a different byte stream.
+
+    use super::full_object_plaintext_len;
+    use crate::object_api::{ObjectInfo, ObjectOptions};
+    use std::collections::HashMap;
+    use std::sync::Arc;
+
+    const STORED_SIZE: i64 = 1024;
+    const PLAINTEXT_SIZE: i64 = 4096;
+
+    fn plain_object_info() -> ObjectInfo {
+        ObjectInfo {
+            size: STORED_SIZE,
+            ..Default::default()
+        }
+    }
+
+    fn compressed_object_info() -> ObjectInfo {
+        let mut user_defined = HashMap::new();
+        // is_compressed() checks for the internal compression suffix key.
+        user_defined.insert("x-rustfs-internal-compression".to_string(), "klauspost/compress/s2".to_string());
+        user_defined.insert("x-rustfs-internal-actual-size".to_string(), PLAINTEXT_SIZE.to_string());
+        ObjectInfo {
+            size: STORED_SIZE,
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        }
+    }
+
+    fn encrypted_object_info() -> ObjectInfo {
+        let mut user_defined = HashMap::new();
+        user_defined.insert("x-minio-encryption-key".to_string(), "opaque".to_string());
+        ObjectInfo {
+            size: STORED_SIZE,
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        }
+    }
+
+    fn restore_opts() -> ObjectOptions {
+        let mut opts = ObjectOptions::default();
+        opts.transition.restore_request.days = Some(1);
+        opts
+    }
+
+    #[test]
+    fn plain_full_object_read_yields_its_stored_size() {
+        let len = full_object_plaintext_len(&None, &ObjectOptions::default(), &plain_object_info());
+        assert_eq!(len, Some(STORED_SIZE));
+    }
+
+    #[test]
+    fn compressed_object_yields_decompressed_size() {
+        // ReadTransform::Compressed publishes the decompressed length as
+        // object_info.size; the hit site must reproduce exactly this value or
+        // UploadPartCopy truncates the copy (backlog#1109).
+        let len = full_object_plaintext_len(&None, &ObjectOptions::default(), &compressed_object_info());
+        assert_eq!(len, Some(PLAINTEXT_SIZE));
+    }
+
+    #[test]
+    fn raw_data_movement_read_is_refused() {
+        // Decommission reads set raw_data_movement_read: ReadPlan returns the
+        // STORED bytes, so the cached decompressed body must not be served
+        // (backlog#1108 — silent data corruption on the destination pool).
+        let opts = ObjectOptions {
+            raw_data_movement_read: true,
+            ..Default::default()
+        };
+        assert_eq!(full_object_plaintext_len(&None, &opts, &plain_object_info()), None);
+    }
+
+    #[test]
+    fn data_movement_read_is_refused() {
+        let opts = ObjectOptions {
+            data_movement: true,
+            ..Default::default()
+        };
+        assert_eq!(full_object_plaintext_len(&None, &opts, &plain_object_info()), None);
+    }
+
+    #[test]
+    fn restore_read_of_compressed_object_is_refused() {
+        // restore_request_active forces ReadPlan down the Plain branch, so the
+        // read yields STORED (compressed) bytes under the compressed size.
+        assert_eq!(full_object_plaintext_len(&None, &restore_opts(), &compressed_object_info()), None);
+    }
+
+    #[test]
+    fn restore_read_of_plain_object_is_refused() {
+        assert_eq!(full_object_plaintext_len(&None, &restore_opts(), &plain_object_info()), None);
+    }
+
+    #[test]
+    fn encrypted_object_is_refused() {
+        assert_eq!(
+            full_object_plaintext_len(&None, &ObjectOptions::default(), &encrypted_object_info()),
+            None
+        );
+    }
+
+    #[test]
+    fn compressed_object_without_actual_size_is_refused() {
+        // A compressed object whose actual size cannot be resolved must not be
+        // served from cache: there is no length to publish as object_info.size.
+        let mut user_defined = HashMap::new();
+        user_defined.insert("x-rustfs-internal-compression".to_string(), "klauspost/compress/s2".to_string());
+        let info = ObjectInfo {
+            size: STORED_SIZE,
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        };
+        assert_eq!(full_object_plaintext_len(&None, &ObjectOptions::default(), &info), None);
+    }
+}

rustfs/src/app/object_data_cache/planner.rs

@@ -16,6 +16,7 @@
 
 use crate::app::object_data_cache::ObjectDataCacheAdapter;
 use crate::app::storage_api::object_usecase::StorageObjectInfo;
+use crate::storage::storage_api::ecstore_bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps as _;
 use rustfs_object_data_cache::{ObjectDataCacheBodyVariant, ObjectDataCacheGetPlan, ObjectDataCacheGetRequest};
 
 /// App-layer GET request snapshot used for cache planning.
@@ -53,7 +54,14 @@ pub(crate) fn build_get_object_body_cache_plan(
         || request.info.delete_marker
         || request.info.version_only
         || request.info.metadata_only
-        || request.response_content_length < 0
+        // Remote (transitioned) objects are served from the warm tier; the
+        // ecstore hook already refuses them (hook.rs is_remote()), so the
+        // usecase-layer planner must exclude them too for a uniform contract.
+        || request.info.is_remote()
+        // Zero-length bodies save no I/O — ecstore returns an empty body before
+        // the hook probe — so admitting them only creates useless entries and
+        // inflates hit metrics. Mirrors should_buffer_get_object_in_memory_with_threshold.
+        || request.response_content_length <= 0
     {
         return GetObjectBodyCachePlan::Skip;
     }
@@ -178,6 +186,62 @@ mod tests {
         assert!(matches!(plan, GetObjectBodyCachePlan::Skip));
     }
 
+    #[test]
+    fn plan_skips_remote_transitioned_objects() {
+        // backlog#1138: transitioned (remote-tier) objects are served from the
+        // warm backend; the ecstore hook already refuses them, so the
+        // usecase-layer planner must exclude them too for a uniform contract.
+        let adapter = enabled_adapter();
+        let mut info = crate::storage::storage_api::StorageObjectInfo {
+            etag: Some("etag".to_string()),
+            size: 4,
+            ..Default::default()
+        };
+        info.transitioned_object.status = "complete".to_string();
+
+        let plan = build_get_object_body_cache_plan(
+            &adapter,
+            GetObjectBodyCacheRequest {
+                bucket: "bucket",
+                key: "object",
+                info: &info,
+                response_content_length: 4,
+                has_range: false,
+                part_number: None,
+                encryption_applied: false,
+            },
+        );
+
+        assert!(matches!(plan, GetObjectBodyCachePlan::Skip));
+    }
+
+    #[test]
+    fn plan_skips_zero_length_objects() {
+        // backlog#1142: ecstore returns an empty body before the hook probe, so
+        // a zero-length GET saves no I/O; admitting it only inflates hit metrics.
+        let adapter = enabled_adapter();
+        let info = crate::storage::storage_api::StorageObjectInfo {
+            etag: Some("etag".to_string()),
+            size: 0,
+            ..Default::default()
+        };
+
+        let plan = build_get_object_body_cache_plan(
+            &adapter,
+            GetObjectBodyCacheRequest {
+                bucket: "bucket",
+                key: "object",
+                info: &info,
+                response_content_length: 0,
+                has_range: false,
+                part_number: None,
+                encryption_applied: false,
+            },
+        );
+
+        assert!(matches!(plan, GetObjectBodyCachePlan::Skip));
+    }
+
     #[test]
     fn plan_is_cacheable_for_plain_full_object() {
         let adapter = enabled_adapter();