From 6780140318cd9e5224f3eff2c3e3adf00789a397 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 11 Jul 2026 02:22:20 +0800 Subject: [PATCH] fix(object-data-cache): make the GET key write-unique and dedup the lookup (#4693) fix(object-data-cache): make GET body cache key write-unique and dedup lookups Address four object-data-cache GET-path findings (backlog#1107 batch): ODC-06 (backlog#1111): the cache key was content-unique, not write-unique. Extend ObjectDataCacheKey with the resolved version's modification time (i128 unix nanoseconds, None -> 0), derived once in the shared planner so the ecstore hook and the usecase layer produce an identical key. An unversioned overwrite advances mod_time, so a stale node can no longer serve old bytes for up to the TTL under an MD5 collision; etag + size stay as belt-and-braces. ODC-16 (backlog#1121): every cacheable GET planned and looked up twice (once in the ecstore hook, once in the usecase layer), double-counting hits, hit_bytes and lookups. GetObjectReader now carries a GetObjectBodySource marker (Unprobed / HookMissed / HookServed); the hook stamps it, and build_get_object_body_with_cache serves a hook-served body directly and skips its lookup whenever the hook already probed. One hook-served GET now records exactly one lookup. ODC-19 (backlog#1124): ENABLE=true with no explicit mode defaulted to HitOnly, which never fills and keeps a permanent 0% hit rate. Default to FillBufferedOnly, log the resolved mode at startup, and warn when HitOnly is selected explicitly. ODC-24 (backlog#1129): max_entry_bytes above the in-memory GET fill limits was silently inert. Clamp the planner's size eligibility to min(max_entry_bytes, seek-support threshold, 64 MiB buffer cap) so ineligible sizes plan SkipTooLarge instead of being reported eligible, and warn at startup when the excess is inert. Co-authored-by: heihutu --- .../bucket/replication/replication_pool.rs | 1 + crates/ecstore/src/client/object_api_utils.rs | 1 + crates/ecstore/src/config/com.rs | 2 + crates/ecstore/src/data_movement/mod.rs | 1 + crates/ecstore/src/object_api/readers.rs | 42 ++++ .../rebalance/rebalance_unit_tests.rs | 1 + crates/ecstore/src/set_disk/ops/object.rs | 145 +++++++++-- crates/ecstore/src/store/init.rs | 1 + crates/ecstore/src/store/object.rs | 4 + crates/object-data-cache/src/cache.rs | 99 +++++++- crates/object-data-cache/src/key.rs | 83 ++++++- crates/scanner/src/scanner.rs | 1 + rustfs/src/app/object_data_cache/adapter.rs | 118 ++++++++- rustfs/src/app/object_data_cache/body.rs | 1 + .../src/app/object_data_cache/invalidation.rs | 3 + rustfs/src/app/object_data_cache/planner.rs | 92 +++++++ rustfs/src/app/object_usecase.rs | 225 +++++++++++++++++- 17 files changed, 781 insertions(+), 39 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 6107f50ae..5bf13b5be 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -1625,6 +1625,7 @@ mod tests { ..Default::default() }, buffered_body: None, + body_source: Default::default(), }) } diff --git a/crates/ecstore/src/client/object_api_utils.rs b/crates/ecstore/src/client/object_api_utils.rs index 944610198..b3a32c5b7 100644 --- a/crates/ecstore/src/client/object_api_utils.rs +++ b/crates/ecstore/src/client/object_api_utils.rs @@ -150,6 +150,7 @@ pub fn new_getobjectreader<'a>( object_info: oi.clone(), stream: Box::new(input_reader), buffered_body: None, + body_source: Default::default(), }; r //}) diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 6527b4cbb..1c84b6b60 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -1824,6 +1824,7 @@ mod tests { }), object_info: self.object_info(bucket, object), buffered_body: None, + body_source: Default::default(), }) } @@ -3042,6 +3043,7 @@ mod tests { stream: Box::new(Cursor::new(data)), object_info, buffered_body: None, + body_source: Default::default(), }) } diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 66a58c3c1..f52820393 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -1240,6 +1240,7 @@ mod tests { stream: Box::new(Cursor::new(raw_payload.clone())), object_info: object_info.clone(), buffered_body: None, + body_source: Default::default(), }; let mut data = data_movement_put_object_reader("bucket-a", &object_info, rd, "test_migration") diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index 906134c98..e8baa6545 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/crates/ecstore/src/object_api/readers.rs @@ -275,10 +275,49 @@ impl PutObjReader { } } +/// Provenance of a [`GetObjectReader`] with respect to the app-layer object +/// data cache hook, so the app layer can avoid repeating the cache lookup the +/// ecstore GET probe already ran after fresh metadata resolution (backlog#1121 +/// / ODC-16). One hook-served GET must record exactly one lookup, not two. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum GetObjectBodySource { + /// The cache hook did not probe this read: the hook is unregistered, or the + /// read is ineligible under the fail-closed allow-list. The app layer runs + /// its own cache lookup, as before. + #[default] + Unprobed, + /// The hook probed after fresh metadata resolution and did not serve a body + /// (a genuine miss, or a length-defensive rejection). Its miss is + /// authoritative, so the app layer must skip the lookup and only build a + /// plan to fill. + HookMissed, + /// `buffered_body` is exactly the body the hook served from the cache. The + /// app layer serves it directly as the object-data-cache source, with no + /// second lookup and no re-fill. + HookServed, +} + pub struct GetObjectReader { pub stream: Box, pub object_info: ObjectInfo, pub buffered_body: Option, + /// Cache-hook provenance; defaults to [`GetObjectBodySource::Unprobed`] for + /// every reader that never passed through the app-layer cache probe. + pub body_source: GetObjectBodySource, +} + +impl GetObjectReader { + /// 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 { + matches!(self.body_source, GetObjectBodySource::HookServed) + } + + /// True when the cache hook probed this read (whether it served a body or + /// missed). The app layer must not repeat the lookup in either case. + pub fn cache_hook_probed(&self) -> bool { + !matches!(self.body_source, GetObjectBodySource::Unprobed) + } } #[derive(Debug, Clone, Copy)] @@ -532,6 +571,7 @@ impl ReadPlan { stream: reader, object_info: oi.clone(), buffered_body: None, + body_source: GetObjectBodySource::Unprobed, }, self.storage_offset, self.storage_length, @@ -586,6 +626,7 @@ impl ReadPlan { stream: final_reader, object_info, buffered_body: None, + body_source: GetObjectBodySource::Unprobed, }, self.storage_offset, self.storage_length, @@ -687,6 +728,7 @@ impl ReadPlan { stream: final_reader, object_info, buffered_body: None, + body_source: GetObjectBodySource::Unprobed, }, self.storage_offset, self.storage_length, diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index 97716e80c..bc6006907 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -141,6 +141,7 @@ impl MigrationBackendSpy { stream: Box::new(Cursor::new(vec![0_u8; 3])), object_info: ObjectInfo::default(), buffered_body: None, + body_source: Default::default(), } } } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 74dec731e..88e7346d1 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -22,6 +22,7 @@ use super::super::*; use crate::disk::OldCurrentSize; +use crate::object_api::GetObjectBodySource; /// 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 @@ -170,6 +171,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { stream: Box::new(Cursor::new(Vec::new())), object_info, buffered_body: Some(Bytes::new()), + body_source: GetObjectBodySource::Unprobed, }; return Ok(reader); } @@ -256,6 +258,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), + body_source: GetObjectBodySource::Unprobed, }; return Ok(reader); } @@ -338,6 +341,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), + body_source: GetObjectBodySource::Unprobed, }; return Ok(reader); } @@ -389,23 +393,39 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // 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). + // Records whether the app-layer cache probe ran for this read, so the + // app layer does not repeat the lookup it already performed after fresh + // metadata resolution (backlog#1121 / ODC-16). It stays `Unprobed` when + // the read is ineligible under the allow-list or the hook is not + // registered; the direct-memory and streaming readers built below carry + // it forward. + let mut body_source = GetObjectBodySource::Unprobed; 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, - buffered_body: Some(body), - }; - if lock_optimization_enabled { - release_materialized_read_lock(bucket, object, read_lock_guard.take()); + match hook.lookup(bucket, object, &object_info).await { + Some(body) if 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, + buffered_body: Some(body), + body_source: GetObjectBodySource::HookServed, + }; + if lock_optimization_enabled { + release_materialized_read_lock(bucket, object, read_lock_guard.take()); + } + return Ok(reader); + } + // Probed after fresh metadata resolution but no usable body: a + // genuine miss, or a length-defensive rejection. The miss is + // authoritative, so the app layer must not look up again. + _ => { + body_source = GetObjectBodySource::HookMissed; + } } - return Ok(reader); } let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, &fi, opts); @@ -435,6 +455,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), + body_source, }; if lock_optimization_enabled { release_materialized_read_lock(bucket, object, read_lock_guard.take()); @@ -476,6 +497,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), + body_source, }; if lock_optimization_enabled { release_materialized_read_lock(bucket, object, read_lock_guard.take()); @@ -508,7 +530,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { size_bucket, ); record_get_object_reader_path_observation(GET_OBJECT_PATH_CODEC_STREAMING, object_class, size_bucket); - let (reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?; + let (mut reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?; + // Carry the hook probe result so the app layer skips its + // now-redundant lookup on the streaming miss path (ODC-16). + reader.body_source = body_source; return Ok(finish_set_disk_read_lock( reader, read_lock_guard.take(), @@ -543,7 +568,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let (rd, wd) = tokio::io::duplex(duplex_buffer_size); debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer"); - let (reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h).await?; + let (mut reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h).await?; + // Carry the hook probe result so the app layer skips its now-redundant + // lookup on the streaming miss path (ODC-16). + reader.body_source = body_source; // let disks = disks.clone(); let bucket = bucket.to_owned(); @@ -2232,6 +2260,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { stream: Box::new(pr), object_info: oi, buffered_body: None, + body_source: GetObjectBodySource::Unprobed, }); let cloned_bucket = bucket.to_string(); @@ -3145,7 +3174,7 @@ mod body_cache_hook_e2e_tests { use crate::ecstore_validation_blackbox::make_local_set_disks; use crate::io_support::rio::{HashReader, compression_metadata_value, compression_reader}; use crate::object_api::{ - GetObjectBodyCacheHook, ObjectInfo, ObjectOptions, PutObjReader, clear_get_object_body_cache_hook, + GetObjectBodyCacheHook, GetObjectBodySource, ObjectInfo, ObjectOptions, PutObjReader, clear_get_object_body_cache_hook, register_get_object_body_cache_hook, }; use crate::set_disk::SetDisks; @@ -3257,6 +3286,90 @@ mod body_cache_hook_e2e_tests { b"rustfs-body-cache-e2e-regression-".repeat(20_000) } + /// Writes a plain (uncompressed, unencrypted) object of `data`, large enough + /// to keep it off the inline fast path so the cache hook is actually probed. + async fn put_plain_object(set_disks: &Arc, bucket: &str, object: &str, data: &[u8]) { + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + let stream = HashReader::from_stream(Cursor::new(data.to_vec()), data.len() as i64, data.len() as i64, None, None, false) + .expect("hash reader over plain bytes"); + let mut reader = PutObjReader::new(stream); + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + set_disks + .put_object(bucket, object, &mut reader, &opts) + .await + .expect("plain object should be written"); + } + + /// ODC-16: a cache-hook hit must mark the reader `HookServed` so the app + /// layer serves the buffered body without a second lookup. A large plain + /// object stays off the inline fast path, so the hook is genuinely probed. + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn plain_cache_hit_marks_reader_hook_served() { + let (_dirs, set_disks) = make_local_set_disks(4, 2).await; + let bucket = "e2e-body-cache-hook-served"; + let object = "plain.bin"; + let payload = b"rustfs-hook-served-payload-".repeat(40_000); + put_plain_object(&set_disks, bucket, object, &payload).await; + + let _guard = HookGuard::install(bucket, object, Bytes::from(payload.clone())); + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &opts) + .await + .expect("object reader should open"); + + assert_eq!( + reader.body_source, + GetObjectBodySource::HookServed, + "a hook hit must mark the reader HookServed" + ); + assert!(reader.buffered_body.is_some(), "a hook-served reader carries the cache body"); + let mut body = Vec::new(); + reader.stream.read_to_end(&mut body).await.expect("object should stream"); + assert_eq!(body, payload, "the hook-served body must be the primed plaintext"); + } + + /// ODC-16: when the hook is registered but misses this object, the reader + /// must be marked `HookMissed` so the app layer skips its now-redundant + /// lookup (the hook's miss ran after fresh metadata resolution). + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn plain_cache_miss_marks_reader_hook_missed() { + let (_dirs, set_disks) = make_local_set_disks(4, 2).await; + let bucket = "e2e-body-cache-hook-missed"; + let object = "plain.bin"; + let payload = b"rustfs-hook-missed-payload-".repeat(40_000); + put_plain_object(&set_disks, bucket, object, &payload).await; + + // Register a hook primed for a DIFFERENT object, so this read is probed + // (hook registered + eligible) but the probe misses. + let _guard = HookGuard::install(bucket, "other-object", Bytes::from_static(b"unrelated")); + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + let reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &opts) + .await + .expect("object reader should open"); + + assert_eq!( + reader.body_source, + GetObjectBodySource::HookMissed, + "a probed miss must mark the reader HookMissed" + ); + } + /// backlog#1108: a raw data-movement read (decommission/rebalance copy) must /// yield the STORED (compressed) representation, never the cached plaintext. /// Serving the cache here writes decompressed bytes into the destination diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 302913d94..1ade03736 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -572,6 +572,7 @@ mod tests { stream: Box::new(Cursor::new(self.read_payload.clone())), object_info: self.object_info(bucket, object, self.read_payload.len()), buffered_body: None, + body_source: Default::default(), }) } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 6b75f25c7..4ae00a48b 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -2195,6 +2195,7 @@ mod tests { stream: Box::new(Cursor::new(Vec::::new())), object_info: ObjectInfo::default(), buffered_body: None, + body_source: Default::default(), }; let reader = ECStore::attach_read_lock_guard(reader, Some(read_guard)); @@ -2234,6 +2235,7 @@ mod tests { stream: Box::new(Cursor::new(vec![1, 2, 3])), object_info: ObjectInfo::default(), buffered_body: None, + body_source: Default::default(), }; let reader = ECStore::attach_read_lock_guard(reader, Some(read_guard)); @@ -2270,6 +2272,7 @@ mod tests { stream: Box::new(Cursor::new(vec![1, 2, 3])), object_info: ObjectInfo::default(), buffered_body: Some(Bytes::from_static(b"123")), + body_source: Default::default(), }; let reader = ECStore::attach_read_lock_guard(reader, Some(read_guard)); @@ -2306,6 +2309,7 @@ mod tests { stream: Box::new(Cursor::new(vec![1, 2, 3])), object_info: ObjectInfo::default(), buffered_body: None, + body_source: Default::default(), }; let mut reader = ECStore::attach_read_lock_guard(reader, Some(read_guard)); diff --git a/crates/object-data-cache/src/cache.rs b/crates/object-data-cache/src/cache.rs index 5aa276c50..9e4f1ce01 100644 --- a/crates/object-data-cache/src/cache.rs +++ b/crates/object-data-cache/src/cache.rs @@ -45,6 +45,16 @@ pub struct ObjectDataCache { backend: ObjectDataCacheBackendKind, config: Arc, stats: Arc, + /// Effective fill ceiling in bytes: a body larger than this is planned + /// `SkipTooLarge` even when it fits `max_entry_bytes`. The app layer sets it + /// to `min(max_entry_bytes, seek-support threshold, 64 MiB buffer cap)` so a + /// `max_entry_bytes` above the in-memory GET fill limits is not reported as + /// eligible while fill could never materialize it (backlog#1129 / ODC-24). + /// `0` means "no additional clamp" — eligibility rests on `max_entry_bytes` + /// alone (the default, used by engine-level tests). The concurrency-driven + /// shrink of the fill threshold is dynamic and cannot be captured at plan + /// time; this static ceiling is the conservative floor. + fill_ceiling_bytes: u64, /// Monotonic origin for the cache-state publish debounce. created_at: Instant, /// Millis since `created_at` of the last cache-state gauge publish, or `0` @@ -62,6 +72,7 @@ impl ObjectDataCache { backend: ObjectDataCacheBackendKind::Noop(NoopBackend), config, stats, + fill_ceiling_bytes: 0, created_at: Instant::now(), last_entry_publish_ms: AtomicU64::new(0), } @@ -82,11 +93,32 @@ impl ObjectDataCache { backend, config: Arc::new(config), stats, + fill_ceiling_bytes: 0, created_at: Instant::now(), last_entry_publish_ms: AtomicU64::new(0), }) } + /// Sets the effective fill ceiling (backlog#1129 / ODC-24). The app layer + /// applies this at startup so a body above the in-memory GET fill limits is + /// planned `SkipTooLarge` instead of being reported eligible. `0` disables + /// the extra clamp. + #[must_use] + pub fn with_fill_ceiling_bytes(mut self, fill_ceiling_bytes: u64) -> Self { + self.fill_ceiling_bytes = fill_ceiling_bytes; + self + } + + /// Effective size above which a body is planned `SkipTooLarge`: + /// `max_entry_bytes` clamped by the fill ceiling when one is set. + fn effective_size_ceiling(&self) -> u64 { + if self.fill_ceiling_bytes == 0 { + self.config.max_entry_bytes + } else { + self.config.max_entry_bytes.min(self.fill_ceiling_bytes) + } + } + /// Produces a lightweight GET plan from request metadata. pub fn plan_get(&self, request: ObjectDataCacheGetRequest<'_>) -> ObjectDataCacheGetPlan { if self.config.is_disabled() { @@ -100,7 +132,11 @@ impl ObjectDataCache { return ObjectDataCacheGetPlan::Disabled; } - if request.size > self.config.max_entry_bytes { + // ODC-24: clamp eligibility to the effective fill ceiling, not just + // `max_entry_bytes`. A body in the gap between `max_entry_bytes` and the + // in-memory GET fill limits could never fill, so admitting it here would + // report it "eligible" while it kept a permanent 0% hit rate. + if request.size > self.effective_size_ceiling() { record_plan_decision(self.backend.as_metric_label(), self.config.mode, "skip", "too_large", request.size); return ObjectDataCacheGetPlan::SkipTooLarge; } @@ -108,12 +144,13 @@ impl ObjectDataCache { record_plan_decision(self.backend.as_metric_label(), self.config.mode, "cacheable", "eligible", request.size); ObjectDataCacheGetPlan::Cacheable { - key: ObjectDataCacheKey::new( + key: ObjectDataCacheKey::with_mod_time( request.bucket, request.object, request.version_id.as_deref(), request.etag, request.size, + request.mod_time_unix_nanos, request.body_variant, ), } @@ -308,6 +345,10 @@ pub struct ObjectDataCacheGetRequest<'a> { pub etag: &'a str, /// Object size in bytes. pub size: u64, + /// Resolved version's modification time as Unix nanoseconds, or `0` when + /// absent. Carried into the key so it is write-unique (backlog#1111 / + /// ODC-06). + pub mod_time_unix_nanos: i128, /// Supported response body variant. pub body_variant: ObjectDataCacheBodyVariant, } @@ -466,6 +507,7 @@ mod tests { version_id: None, etag, size, + mod_time_unix_nanos: 0, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, } } @@ -645,6 +687,7 @@ mod tests { version_id: None, etag: "etag", size: 5, + mod_time_unix_nanos: 0, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }); @@ -654,4 +697,56 @@ mod tests { assert_eq!(fill, ObjectDataCacheFillResult::SkippedSizeMismatch); assert!(matches!(lookup, ObjectDataCacheLookup::Miss)); } + + #[test] + fn plan_clamps_size_eligibility_to_fill_ceiling() { + // ODC-24 (backlog#1129): a body in the gap between `max_entry_bytes` and + // the effective fill ceiling must plan `SkipTooLarge`, not `Cacheable`, + // so it stops being reported as eligible while it can never fill. + let config = ObjectDataCacheConfig { + mode: ObjectDataCacheMode::HitOnly, + max_bytes: 32 * 1024 * 1024, + max_memory_percent: 0, + max_entry_bytes: 16 * 1024 * 1024, + ..ObjectDataCacheConfig::default() + }; + let cache = ObjectDataCache::new(config) + .expect("clamped cache config should initialize") + .with_fill_ceiling_bytes(4 * 1024 * 1024); + + // Within the ceiling: eligible. + assert!(matches!( + cache.plan_get(plain_request("bucket", "object", "etag", 4 * 1024 * 1024)), + ObjectDataCacheGetPlan::Cacheable { .. } + )); + // In the gap (ceiling < size <= max_entry_bytes): skipped as too large. + assert_eq!( + cache.plan_get(plain_request("bucket", "object", "etag", 4 * 1024 * 1024 + 1)), + ObjectDataCacheGetPlan::SkipTooLarge + ); + assert_eq!( + cache.plan_get(plain_request("bucket", "object", "etag", 16 * 1024 * 1024)), + ObjectDataCacheGetPlan::SkipTooLarge + ); + } + + #[test] + fn plan_carries_mod_time_into_key() { + // ODC-06: the resolved modification time must reach the key so two + // writes with identical etag + size derive different keys. + let cache = hit_only_cache(); + let request = ObjectDataCacheGetRequest { + bucket: "bucket", + object: "object", + version_id: None, + etag: "etag", + size: 5, + mod_time_unix_nanos: 42, + body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, + }; + let ObjectDataCacheGetPlan::Cacheable { key } = cache.plan_get(request) else { + panic!("plan should be cacheable"); + }; + assert_eq!(key.mod_time_unix_nanos, 42); + } } diff --git a/crates/object-data-cache/src/key.rs b/crates/object-data-cache/src/key.rs index d288e3f6c..0c0116514 100644 --- a/crates/object-data-cache/src/key.rs +++ b/crates/object-data-cache/src/key.rs @@ -26,6 +26,16 @@ pub enum ObjectDataCacheBodyVariant { } /// Stable cache key for a reusable object body. +/// +/// The key is *write-unique*, not merely *content-unique* (backlog#1111 / +/// ODC-06). `etag + size` alone identify content: for an unversioned overwrite +/// two same-length payloads that collide on MD5 would derive the identical key, +/// so a GET on a node that never observed the overwrite could serve the old +/// bytes for up to the TTL, and the same collision turns the +/// fill-after-invalidation race (backlog#1118) into a serving bug. Including +/// the resolved version's modification time distinguishes two writes even under +/// an MD5 collision, because an overwrite advances `mod_time`. `etag + size` +/// stay in the key as belt-and-braces. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct ObjectDataCacheKey { /// Bucket name. @@ -38,6 +48,11 @@ pub struct ObjectDataCacheKey { pub etag: Arc, /// Object size in bytes. pub size: u64, + /// Resolved version's modification time as Unix nanoseconds + /// (`OffsetDateTime::unix_timestamp_nanos`), or `0` when absent. This is the + /// write-unique component: an overwrite advances `mod_time`, so the key + /// changes even when `etag + size` are unchanged (an MD5 collision). + pub mod_time_unix_nanos: i128, /// Cached body semantics. pub body_variant: ObjectDataCacheBodyVariant, } @@ -52,13 +67,17 @@ pub struct ObjectDataCacheIdentity { } impl ObjectDataCacheKey { - /// Creates a new stable object data cache key. - pub fn new( + /// Creates a stable object data cache key with an explicit modification + /// time. This is the full constructor; the production GET planner uses it so + /// the key is write-unique (backlog#1111 / ODC-06). + #[allow(clippy::too_many_arguments)] + pub fn with_mod_time( bucket: impl Into>, object: impl Into>, version_id: Option<&str>, etag: impl Into>, size: u64, + mod_time_unix_nanos: i128, body_variant: ObjectDataCacheBodyVariant, ) -> Self { Self { @@ -67,10 +86,25 @@ impl ObjectDataCacheKey { version_id: version_id.map_or_else(|| Arc::::from(NULL_VERSION_ID), Arc::::from), etag: etag.into(), size, + mod_time_unix_nanos, body_variant, } } + /// Creates a stable object data cache key with no modification time + /// (`mod_time_unix_nanos == 0`). Retained for callers that key purely by + /// content identity (index/backend tests). + pub fn new( + bucket: impl Into>, + object: impl Into>, + version_id: Option<&str>, + etag: impl Into>, + size: u64, + body_variant: ObjectDataCacheBodyVariant, + ) -> Self { + Self::with_mod_time(bucket, object, version_id, etag, size, 0, body_variant) + } + /// Returns true when this key targets the canonical unversioned body variant. /// /// Only exercised by tests; gated so it is not compiled into the shipping @@ -112,6 +146,51 @@ mod tests { assert_ne!(latest, versioned); } + #[test] + fn key_distinguishes_writes_by_mod_time() { + // ODC-06 (backlog#1111): two writes with identical etag + size (an MD5 + // collision on an unversioned overwrite) must derive different keys once + // the modification time differs, so a stale node cannot serve old bytes. + let old = ObjectDataCacheKey::with_mod_time( + "bucket", + "object", + None, + "etag", + 42, + 1_000, + ObjectDataCacheBodyVariant::FullObjectPlainV1, + ); + let new = ObjectDataCacheKey::with_mod_time( + "bucket", + "object", + None, + "etag", + 42, + 2_000, + ObjectDataCacheBodyVariant::FullObjectPlainV1, + ); + + assert_ne!(old, new, "keys differing only by mod_time must not collide"); + // Same mod_time still collapses to one key (a true re-read of one write). + let new_again = ObjectDataCacheKey::with_mod_time( + "bucket", + "object", + None, + "etag", + 42, + 2_000, + ObjectDataCacheBodyVariant::FullObjectPlainV1, + ); + assert_eq!(new, new_again); + } + + #[test] + fn key_new_defaults_mod_time_to_zero() { + let key = ObjectDataCacheKey::new("bucket", "object", None, "etag", 42, ObjectDataCacheBodyVariant::FullObjectPlainV1); + + assert_eq!(key.mod_time_unix_nanos, 0); + } + #[test] fn identity_new_preserves_bucket_and_object() { let identity = ObjectDataCacheIdentity::new("bucket", "object"); diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 85d54fa75..cdcaabcde 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1237,6 +1237,7 @@ mod tests { stream: Box::new(Cursor::new(data)), object_info: ObjectInfo::default(), buffered_body: None, + body_source: Default::default(), }) } diff --git a/rustfs/src/app/object_data_cache/adapter.rs b/rustfs/src/app/object_data_cache/adapter.rs index 87b843b89..57455545b 100644 --- a/rustfs/src/app/object_data_cache/adapter.rs +++ b/rustfs/src/app/object_data_cache/adapter.rs @@ -44,9 +44,14 @@ pub(crate) struct ObjectDataCacheAdapter { impl ObjectDataCacheAdapter { /// Creates an adapter from validated cache configuration. + /// + /// ODC-24 (backlog#1129): the engine's size eligibility is clamped to the + /// in-memory GET fill limits so a `max_entry_bytes` above them is not + /// reported as eligible while fill could never materialize the body. pub(crate) fn new(config: ObjectDataCacheConfig) -> Result { + let fill_ceiling = resolve_fill_ceiling_bytes(config.max_entry_bytes); Ok(Self { - cache: Arc::new(ObjectDataCache::new(config)?), + cache: Arc::new(ObjectDataCache::new(config)?.with_fill_ceiling_bytes(fill_ceiling)), }) } @@ -124,8 +129,37 @@ impl ObjectDataCacheAdapter { impl ObjectDataCacheAdapter { fn from_config_or_disabled(config: ObjectDataCacheConfig, source: &str) -> Arc { + let mode = config.mode; + let max_entry_bytes = config.max_entry_bytes; + // `Self::new` applies the ODC-24 fill-ceiling clamp; this startup path + // only adds the resolved-config logging on top. match Self::new(config) { - Ok(adapter) => Arc::new(adapter), + Ok(adapter) => { + if !adapter.is_disabled() { + let ceiling = resolve_fill_ceiling_bytes(max_entry_bytes); + // ODC-19: log the resolved mode at startup, and warn when the + // operator explicitly selected the never-filling HitOnly mode. + tracing::info!(source, ?mode, "object data cache enabled"); + // ODC-24: warn when the excess above the effective ceiling is inert. + if max_entry_bytes > ceiling { + warn!( + source, + max_entry_bytes, + effective_fill_ceiling = ceiling, + "RUSTFS_OBJECT_DATA_CACHE_MAX_ENTRY_BYTES exceeds the in-memory GET fill limit; \ + bodies above the effective cap are planned SkipTooLarge" + ); + } + if mode == ObjectDataCacheMode::HitOnly { + warn!( + source, + "object data cache mode 'hit_only' never populates the cache; it only \ + serves as a runtime-downgrade target and keeps a permanent 0% hit rate" + ); + } + } + Arc::new(adapter) + } Err(err) => { warn!( error = %err, @@ -138,6 +172,17 @@ impl ObjectDataCacheAdapter { } } +/// Resolves the effective fill ceiling from the app-layer in-memory GET fill +/// limits: `min(max_entry_bytes, seek-support threshold, 64 MiB buffer cap)` +/// (backlog#1129 / ODC-24). The concurrency-driven shrink of the fill threshold +/// is dynamic and cannot be captured here, so this static ceiling is the +/// conservative floor. +fn resolve_fill_ceiling_bytes(max_entry_bytes: u64) -> u64 { + let seek_threshold = crate::app::object_usecase::object_seek_support_threshold() as u64; + let hard_cap = u64::try_from(crate::app::object_usecase::MAX_GET_OBJECT_MEMORY_BUFFER_BYTES).unwrap_or(u64::MAX); + max_entry_bytes.min(seek_threshold).min(hard_cap) +} + impl Default for ObjectDataCacheAdapter { fn default() -> Self { Self::disabled() @@ -210,9 +255,13 @@ fn object_data_cache_config_from_values(values: ObjectDataCacheEnvValues) -> Obj config.mode = ObjectDataCacheMode::Disabled; } Some(true) if !mode_explicit => { - // Enable flag without an explicit mode: start at the safest - // enabled stage instead of silently staying disabled. - config.mode = ObjectDataCacheMode::HitOnly; + // Enable flag without an explicit mode: start at the safest stage + // that actually populates the cache. `HitOnly` never fills, so it + // would keep a permanent 0% hit rate and pay per-GET overhead for + // nothing (backlog#1124 / ODC-19). `FillBufferedOnly` fills only + // from bodies the GET path already buffered — no extra + // materialization — so it is both safe and effective. + config.mode = ObjectDataCacheMode::FillBufferedOnly; } _ => {} } @@ -382,13 +431,66 @@ mod tests { } #[test] - fn object_data_cache_enable_true_without_mode_defaults_to_hit_only() { + fn object_data_cache_enable_true_without_mode_defaults_to_fill_buffered_only() { + // ODC-19 (backlog#1124): ENABLE=true with no explicit mode must default + // to a mode that actually populates the cache. HitOnly never fills. let config = object_data_cache_config_from_values(ObjectDataCacheEnvValues { enabled: Some(true), ..ObjectDataCacheEnvValues::default() }); - assert_eq!(config.mode, ObjectDataCacheMode::HitOnly); + assert_eq!(config.mode, ObjectDataCacheMode::FillBufferedOnly); + } + + #[test] + fn resolve_fill_ceiling_clamps_large_max_entry_bytes() { + // ODC-24 (backlog#1129): a max_entry_bytes above the in-memory GET fill + // limits is clamped down; the ceiling never exceeds the 64 MiB hard cap. + let huge = 512 * 1024 * 1024; + let ceiling = super::resolve_fill_ceiling_bytes(huge); + assert!(ceiling < huge, "a huge max_entry_bytes must be clamped down"); + assert!( + ceiling <= u64::try_from(crate::app::object_usecase::MAX_GET_OBJECT_MEMORY_BUFFER_BYTES).unwrap(), + "the ceiling must not exceed the 64 MiB hard cap" + ); + } + + #[test] + fn resolve_fill_ceiling_leaves_small_max_entry_bytes_untouched() { + // A small max_entry_bytes is already below the fill limits, so it is the + // binding constraint and passes through unchanged. + assert_eq!(super::resolve_fill_ceiling_bytes(4096), 4096); + } + + #[test] + fn startup_adapter_clamps_plan_eligibility_to_fill_ceiling() { + // ODC-24: an adapter built on the startup path with a `max_entry_bytes` + // above the in-memory GET fill limits must plan a gap-sized object + // `SkipTooLarge`, not `Cacheable` — otherwise it reports eligible while + // it can never fill. + let adapter = ObjectDataCacheAdapter::from_config_or_disabled( + ObjectDataCacheConfig { + mode: ObjectDataCacheMode::HitOnly, + max_bytes: 200 * 1024 * 1024, + max_memory_percent: 0, + max_entry_bytes: 128 * 1024 * 1024, + ..ObjectDataCacheConfig::default() + }, + "test", + ); + + // 32 MiB is within max_entry_bytes (128 MiB) but above the effective + // ceiling (<= 64 MiB hard cap, and the 10 MiB default seek threshold). + let plan = adapter.plan_get(rustfs_object_data_cache::ObjectDataCacheGetRequest { + bucket: "bucket", + object: "object", + version_id: None, + etag: "etag", + size: 32 * 1024 * 1024, + mod_time_unix_nanos: 0, + body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, + }); + assert_eq!(plan, rustfs_object_data_cache::ObjectDataCacheGetPlan::SkipTooLarge); } #[test] @@ -453,7 +555,7 @@ mod tests { temp_env::with_vars(vars, || { let config = object_data_cache_config_from_env().expect("valid numeric env should be applied"); - assert_eq!(config.mode, ObjectDataCacheMode::HitOnly); + assert_eq!(config.mode, ObjectDataCacheMode::FillBufferedOnly); assert_eq!(config.max_entry_bytes, 2_097_152); }); } diff --git a/rustfs/src/app/object_data_cache/body.rs b/rustfs/src/app/object_data_cache/body.rs index 13c0c4d0e..f54c96a77 100644 --- a/rustfs/src/app/object_data_cache/body.rs +++ b/rustfs/src/app/object_data_cache/body.rs @@ -124,6 +124,7 @@ mod tests { version_id: None, etag: "etag", size: 5, + mod_time_unix_nanos: 0, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let fill = adapter.cache().fill_body(&plan, Bytes::from_static(b"hello")).await; diff --git a/rustfs/src/app/object_data_cache/invalidation.rs b/rustfs/src/app/object_data_cache/invalidation.rs index 138157be2..42dd7ea48 100644 --- a/rustfs/src/app/object_data_cache/invalidation.rs +++ b/rustfs/src/app/object_data_cache/invalidation.rs @@ -144,6 +144,7 @@ mod tests { version_id: None, etag: "etag", size: 5, + mod_time_unix_nanos: 0, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let fill = adapter.fill_body(&plan, Bytes::from_static(b"hello")).await; @@ -170,6 +171,7 @@ mod tests { version_id: None, etag: "etag-a", size: 5, + mod_time_unix_nanos: 0, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let plan_b = adapter.plan_get(rustfs_object_data_cache::ObjectDataCacheGetRequest { @@ -178,6 +180,7 @@ mod tests { version_id: None, etag: "etag-b", size: 5, + mod_time_unix_nanos: 0, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let _ = adapter.fill_body(&plan_a, Bytes::from_static(b"aaaaa")).await; diff --git a/rustfs/src/app/object_data_cache/planner.rs b/rustfs/src/app/object_data_cache/planner.rs index d1e6434e0..6a273ac2d 100644 --- a/rustfs/src/app/object_data_cache/planner.rs +++ b/rustfs/src/app/object_data_cache/planner.rs @@ -81,12 +81,21 @@ pub(crate) fn build_get_object_body_cache_plan( .version_id .filter(|version_id| !version_id.is_nil()) .map(|version_id| version_id.to_string()); + // ODC-06 (backlog#1111): carry the resolved version's modification time into + // the key so an unversioned overwrite (which advances mod_time) cannot be + // served the stale body under an MD5 collision. Absent mod_time maps to 0. + let mod_time_unix_nanos = request + .info + .mod_time + .map(|mod_time| mod_time.unix_timestamp_nanos()) + .unwrap_or(0); let engine_request = ObjectDataCacheGetRequest { bucket: request.bucket, object: request.key, version_id, etag, size, + mod_time_unix_nanos, body_variant: ObjectDataCacheBodyVariant::FullObjectPlainV1, }; @@ -242,6 +251,89 @@ mod tests { assert!(matches!(plan, GetObjectBodyCachePlan::Skip)); } + fn cacheable_key(plan: &GetObjectBodyCachePlan) -> rustfs_object_data_cache::ObjectDataCacheKey { + match plan { + GetObjectBodyCachePlan::Cacheable(rustfs_object_data_cache::ObjectDataCacheGetPlan::Cacheable { key }) => key.clone(), + other => panic!("expected a cacheable plan, got {other:?}"), + } + } + + #[test] + fn hook_and_planner_derive_identical_keys() { + // ODC-16 correctness guard: the ecstore hook and the usecase planner both + // route through build_get_object_body_cache_plan, so for the same object + // they must derive byte-identical keys — otherwise every GET misses. The + // hook builds its request with response_content_length = get_actual_size; + // the usecase builds the same for a plain, non-range GET. + let adapter = enabled_adapter(); + let info = crate::storage::storage_api::StorageObjectInfo { + etag: Some("etag".to_string()), + size: 4, + actual_size: 4, + mod_time: Some(time::OffsetDateTime::from_unix_timestamp_nanos(1_234_567_890).unwrap()), + ..Default::default() + }; + + let hook_request = GetObjectBodyCacheRequest { + bucket: "bucket", + key: "object", + info: &info, + response_content_length: info.get_actual_size().expect("actual size"), + has_range: false, + part_number: None, + encryption_applied: false, + }; + let usecase_request = GetObjectBodyCacheRequest { + bucket: "bucket", + key: "object", + info: &info, + response_content_length: 4, + has_range: false, + part_number: None, + encryption_applied: false, + }; + + let hook_key = cacheable_key(&build_get_object_body_cache_plan(&adapter, hook_request)); + let usecase_key = cacheable_key(&build_get_object_body_cache_plan(&adapter, usecase_request)); + + assert_eq!(hook_key, usecase_key, "hook and planner must derive the same key"); + // ODC-06: the modification time flows into the key. + assert_eq!(hook_key.mod_time_unix_nanos, 1_234_567_890); + } + + #[test] + fn planner_key_changes_with_mod_time() { + // ODC-06 (backlog#1111): an unversioned overwrite advances mod_time, so + // the same etag + size must derive a different key. + let adapter = enabled_adapter(); + let mut info = crate::storage::storage_api::StorageObjectInfo { + etag: Some("etag".to_string()), + size: 4, + actual_size: 4, + mod_time: Some(time::OffsetDateTime::from_unix_timestamp_nanos(1_000).unwrap()), + ..Default::default() + }; + let make_request = |info: &crate::storage::storage_api::StorageObjectInfo| { + build_get_object_body_cache_plan( + &adapter, + GetObjectBodyCacheRequest { + bucket: "bucket", + key: "object", + info, + response_content_length: 4, + has_range: false, + part_number: None, + encryption_applied: false, + }, + ) + }; + let old_key = cacheable_key(&make_request(&info)); + info.mod_time = Some(time::OffsetDateTime::from_unix_timestamp_nanos(2_000).unwrap()); + let new_key = cacheable_key(&make_request(&info)); + + assert_ne!(old_key, new_key, "keys differing only by mod_time must not collide"); + } + #[test] fn plan_is_cacheable_for_plain_full_object() { let adapter = enabled_adapter(); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 8745a5c7d..41f38c3a5 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -206,7 +206,7 @@ use crate::app::object_data_cache::{ type S3StdError = Box; const ACCEPT_RANGES_BYTES: &str = "bytes"; -const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024; +pub(crate) const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024; const MEDIUM_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 8 * 1024 * 1024; const HIGH_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 4 * 1024 * 1024; const VERY_HIGH_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 1024 * 1024; @@ -337,6 +337,12 @@ struct GetObjectReadSetup { info: ObjectInfo, final_stream: DynReader, buffered_body: Option, + /// ODC-16: `buffered_body` is the body the ecstore cache hook served, so the + /// app layer serves it as the object-data-cache source without a re-lookup. + cache_hook_served: bool, + /// ODC-16: the cache hook probed this read (served or missed), so the app + /// layer must skip its own lookup. + cache_hook_probed: bool, rs: Option, content_type: Option, last_modified: Option, @@ -1497,7 +1503,7 @@ where Ok(ChunkedBytesReader::new(chunks)) } -fn object_seek_support_threshold() -> usize { +pub(crate) fn object_seek_support_threshold() -> usize { static OBJECT_SEEK_SUPPORT_THRESHOLD: OnceLock = OnceLock::new(); *OBJECT_SEEK_SUPPORT_THRESHOLD.get_or_init(|| { rustfs_utils::get_env_usize( @@ -2730,6 +2736,11 @@ impl DefaultObjectUsecase { ); } + // ODC-16: capture whether the ecstore cache hook already probed this + // read, so the app layer does not repeat the lookup it ran after fresh + // metadata resolution. + let cache_hook_served = reader.is_cache_hook_served(); + let cache_hook_probed = reader.cache_hook_probed(); let info = reader.object_info; let stream = reader.stream; let buffered_body = reader.buffered_body; @@ -2843,6 +2854,8 @@ impl DefaultObjectUsecase { info, final_stream, buffered_body, + cache_hook_served, + cache_hook_probed, rs, content_type, last_modified, @@ -3140,6 +3153,8 @@ impl DefaultObjectUsecase { has_range: bool, encryption_applied: bool, buffered_body: Option, + cache_hook_served: bool, + cache_hook_probed: bool, bucket: &str, key: &str, mut lifecycle: GetObjectBodyLifecycle, @@ -3158,16 +3173,35 @@ impl DefaultObjectUsecase { }; let cache_plan = build_get_object_body_cache_plan(cache_adapter, cache_request); - match lookup_get_object_body_cache_hit(cache_adapter, &cache_plan).await { - GetObjectBodyCacheLookup::Hit(bytes) => { - return Ok(Self::build_memory_bytes_blob( - bytes, - response_content_length, - GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE, - lifecycle, - )); + // ODC-16 (backlog#1121): when the ecstore hook already served this body + // from the cache, serve it straight through as the object-data-cache + // source. Re-running the lookup here would record a second hit, double + // the hit_bytes, and do redundant moka work for one hook-served GET. + if cache_hook_served && let Some(bytes) = buffered_body.clone() { + return Ok(Self::build_memory_bytes_blob( + bytes, + response_content_length, + GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE, + lifecycle, + )); + } + + // ODC-16: only look up when the hook did not probe this read. When it did + // probe (a served body handled above, or a miss), its result is + // authoritative because it ran after fresh metadata resolution, so the + // app layer skips its own lookup and only uses the plan to fill. + if !cache_hook_probed { + match lookup_get_object_body_cache_hit(cache_adapter, &cache_plan).await { + GetObjectBodyCacheLookup::Hit(bytes) => { + return Ok(Self::build_memory_bytes_blob( + bytes, + response_content_length, + GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE, + lifecycle, + )); + } + GetObjectBodyCacheLookup::Disabled | GetObjectBodyCacheLookup::Skip | GetObjectBodyCacheLookup::Miss => {} } - GetObjectBodyCacheLookup::Disabled | GetObjectBodyCacheLookup::Skip | GetObjectBodyCacheLookup::Miss => {} } if let Some(buffered_body) = buffered_body { @@ -4059,6 +4093,8 @@ impl DefaultObjectUsecase { event_info: Option, final_stream: DynReader, buffered_body: Option, + cache_hook_served: bool, + cache_hook_probed: bool, rs: Option, content_type: Option, last_modified: Option, @@ -4111,6 +4147,8 @@ impl DefaultObjectUsecase { rs.is_some(), encryption_applied, buffered_body, + cache_hook_served, + cache_hook_probed, bucket, key, lifecycle, @@ -4277,6 +4315,8 @@ impl DefaultObjectUsecase { info, final_stream, buffered_body, + cache_hook_served, + cache_hook_probed, rs, content_type, last_modified, @@ -4309,6 +4349,8 @@ impl DefaultObjectUsecase { event_info, final_stream, buffered_body, + cache_hook_served, + cache_hook_probed, rs, content_type, last_modified, @@ -7124,6 +7166,7 @@ mod tests { version_id: None, etag, size, + mod_time_unix_nanos: 0, body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, }); for _ in 0..400 { @@ -7534,6 +7577,7 @@ mod tests { version_id: None, etag: "etag", size: 5, + mod_time_unix_nanos: 0, body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let fill = adapter.cache().fill_body(&plan, Bytes::from_static(b"hello")).await; @@ -7552,6 +7596,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "cached-object", GetObjectBodyLifecycle::disabled(), @@ -7592,6 +7638,7 @@ mod tests { version_id: None, etag: "etag", size: 5, + mod_time_unix_nanos: 0, body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, }); let fill = adapter.cache().fill_body(&plan, Bytes::from_static(b"oops")).await; @@ -7608,6 +7655,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "cached-object", GetObjectBodyLifecycle::disabled(), @@ -7665,6 +7714,8 @@ mod tests { false, false, Some(Bytes::from_static(b"hello")), + false, + false, "test-bucket", "cached-object", GetObjectBodyLifecycle::disabled(), @@ -7688,6 +7739,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "cached-object", GetObjectBodyLifecycle::disabled(), @@ -7733,6 +7786,7 @@ mod tests { version_id: None, etag: "etag", size: 5, + mod_time_unix_nanos: 0, body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, }); @@ -7748,6 +7802,8 @@ mod tests { false, false, Some(Bytes::from_static(b"oops")), + false, + false, "test-bucket", "cached-object", GetObjectBodyLifecycle::disabled(), @@ -7767,6 +7823,143 @@ mod tests { ); } + #[tokio::test] + async fn build_get_object_body_with_cache_hook_served_records_no_second_lookup() { + // ODC-16 (backlog#1121): a hook-served GET must record exactly one + // lookup — the ecstore hook's. The app layer, handed the cache body as + // buffered_body with cache_hook_served=true, must serve it directly + // without a second lookup (which would double the hits and hit_bytes). + let reads = Arc::new(AtomicUsize::new(0)); + let reader = ReadProbeReader { + reads: Arc::clone(&reads), + }; + let info = ObjectInfo { + size: 5, + etag: Some("etag".to_string()), + ..Default::default() + }; + let adapter = + crate::app::object_data_cache::ObjectDataCacheAdapter::new(rustfs_object_data_cache::ObjectDataCacheConfig { + mode: rustfs_object_data_cache::ObjectDataCacheMode::FillBufferedOnly, + max_bytes: 8_388_608, + min_free_memory_percent: 0, + ..rustfs_object_data_cache::ObjectDataCacheConfig::default() + }) + .expect("fill-enabled cache adapter should initialize"); + let plan = adapter.plan_get(rustfs_object_data_cache::ObjectDataCacheGetRequest { + bucket: "test-bucket", + object: "hook-served", + version_id: None, + etag: "etag", + size: 5, + mod_time_unix_nanos: 0, + body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, + }); + let hit_body = Bytes::from_static(b"hello"); + assert_eq!( + adapter.cache().fill_body(&plan, hit_body.clone()).await, + rustfs_object_data_cache::ObjectDataCacheFillResult::Inserted + ); + + // Simulate the ecstore hook: it performs exactly one lookup after fresh + // metadata resolution, hits, and hands the body forward as buffered_body. + assert!(matches!( + adapter.lookup_body(&plan).await, + rustfs_object_data_cache::ObjectDataCacheLookup::Hit(_) + )); + let lookups_after_hook = adapter.cache().stats().lookups; + assert_eq!(lookups_after_hook, 1, "the hook performs exactly one lookup"); + + let _body = DefaultObjectUsecase::build_get_object_body_with_cache( + &adapter, + reader, + &info, + 5, + 128 * 1024, + false, + 1, + None, + false, + false, + Some(hit_body), + /* cache_hook_served */ true, + /* cache_hook_probed */ true, + "test-bucket", + "hook-served", + GetObjectBodyLifecycle::disabled(), + ) + .await + .expect("hook-served body handoff should succeed"); + + assert_eq!( + adapter.cache().stats().lookups, + lookups_after_hook, + "a hook-served GET must not record a second lookup in the app layer" + ); + assert_eq!( + reads.load(AtomicOrdering::Relaxed), + 0, + "hook-served body handoff must not read from the fallback reader" + ); + } + + #[tokio::test] + async fn build_get_object_body_with_cache_hook_miss_skips_app_lookup() { + // ODC-16: when the hook probed and missed, its miss is authoritative + // (it ran after fresh metadata resolution), so the app layer must not + // run a second lookup — it only fills from the buffered body. + let reads = Arc::new(AtomicUsize::new(0)); + let reader = ReadProbeReader { + reads: Arc::clone(&reads), + }; + let info = ObjectInfo { + size: 5, + etag: Some("etag".to_string()), + ..Default::default() + }; + let adapter = + crate::app::object_data_cache::ObjectDataCacheAdapter::new(rustfs_object_data_cache::ObjectDataCacheConfig { + mode: rustfs_object_data_cache::ObjectDataCacheMode::FillBufferedOnly, + max_bytes: 8_388_608, + min_free_memory_percent: 0, + ..rustfs_object_data_cache::ObjectDataCacheConfig::default() + }) + .expect("fill-enabled cache adapter should initialize"); + + let lookups_before = adapter.cache().stats().lookups; + let _body = DefaultObjectUsecase::build_get_object_body_with_cache( + &adapter, + reader, + &info, + 5, + 128 * 1024, + false, + 1, + None, + false, + false, + Some(Bytes::from_static(b"hello")), + /* cache_hook_served */ false, + /* cache_hook_probed */ true, + "test-bucket", + "hook-missed", + GetObjectBodyLifecycle::disabled(), + ) + .await + .expect("hook-miss buffered-body handoff should succeed"); + + assert_eq!( + adapter.cache().stats().lookups, + lookups_before, + "a hook-probed miss must not trigger an app-layer lookup" + ); + assert_eq!( + reads.load(AtomicOrdering::Relaxed), + 0, + "buffered-body handoff must not read from the fallback reader" + ); + } + #[tokio::test] async fn build_get_object_body_with_cache_materializes_once_and_hits_later() { let first_reads = Arc::new(AtomicUsize::new(0)); @@ -7805,6 +7998,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "materialized-object", GetObjectBodyLifecycle::disabled(), @@ -7828,6 +8023,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "materialized-object", GetObjectBodyLifecycle::disabled(), @@ -7885,6 +8082,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "mismatch-object", GetObjectBodyLifecycle::disabled(), @@ -7932,6 +8131,8 @@ mod tests { false, false, None, + false, + false, "test-bucket", "too-large-object", GetObjectBodyLifecycle::disabled(), @@ -8679,6 +8880,8 @@ mod tests { Some(info), wrap_reader(tokio::io::empty()), None, + false, + false, None, None, None,