diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index f946c0f63..906134c98 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/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() } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 469ad1976..7294509db 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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, opts: &ObjectOptions, object_info: &ObjectInfo) -> Option { + 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); + } +} diff --git a/rustfs/src/app/object_data_cache/planner.rs b/rustfs/src/app/object_data_cache/planner.rs index 534edd507..d1e6434e0 100644 --- a/rustfs/src/app/object_data_cache/planner.rs +++ b/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();