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 <heihutu@gmail.com>

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -1625,6 +1625,7 @@ mod tests {
                     ..Default::default()
                 },
                 buffered_body: None,
+                body_source: Default::default(),
             })
         }
 

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
             //})

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(),
             })
         }
 

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")

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<dyn AsyncRead + Unpin + Send + Sync>,
     pub object_info: ObjectInfo,
     pub buffered_body: Option<Bytes>,
+    /// 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,

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(),
         }
     }
 }

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<SetDisks>, 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

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(),
             })
         }
 

crates/ecstore/src/store/object.rs

@@ -2195,6 +2195,7 @@ mod tests {
                 stream: Box::new(Cursor::new(Vec::<u8>::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));

crates/object-data-cache/src/cache.rs

@@ -45,6 +45,16 @@ pub struct ObjectDataCache {
     backend: ObjectDataCacheBackendKind,
     config: Arc<ObjectDataCacheConfig>,
     stats: Arc<ObjectDataCacheStats>,
+    /// 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);
+    }
 }

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<str>,
     /// 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<Arc<str>>,
         object: impl Into<Arc<str>>,
         version_id: Option<&str>,
         etag: impl Into<Arc<str>>,
         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::<str>::from(NULL_VERSION_ID), Arc::<str>::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<Arc<str>>,
+        object: impl Into<Arc<str>>,
+        version_id: Option<&str>,
+        etag: impl Into<Arc<str>>,
+        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");

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(),
             })
         }