diff --git a/crates/object-data-cache/Cargo.toml b/crates/object-data-cache/Cargo.toml
index f8ee22887..f4b6784c7 100644
--- a/crates/object-data-cache/Cargo.toml
+++ b/crates/object-data-cache/Cargo.toml
@@ -38,8 +38,16 @@ tokio = { workspace = true, features = ["sync", "rt", "time"] }
 tracing.workspace = true
 
 [dev-dependencies]
+criterion = { workspace = true }
 metrics-util = { version = "0.20", features = ["debugging"] }
-tokio = { workspace = true, features = ["macros", "rt", "time"] }
+# `rt-multi-thread` lets the concurrency stress tests run tasks on real worker
+# threads, so they exercise true parallelism on the shared singleflight/index
+# state rather than only cooperative interleaving.
+tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "time"] }
+
+[[bench]]
+name = "cache_bench"
+harness = false
 
 [lints]
 workspace = true
diff --git a/crates/object-data-cache/benches/cache_bench.rs b/crates/object-data-cache/benches/cache_bench.rs
new file mode 100644
index 000000000..7a47e4259
--- /dev/null
+++ b/crates/object-data-cache/benches/cache_bench.rs
@@ -0,0 +1,105 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Throughput / latency benchmarks for the object data cache
+//! (backlog#1107 production-readiness follow-up).
+//!
+//! The concurrency correctness was covered by deterministic and stress tests;
+//! these measure the GET-path cost the audit only argued about statically. Run
+//! with `cargo bench -p rustfs-object-data-cache`.
+
+use bytes::Bytes;
+use criterion::{Criterion, criterion_group, criterion_main};
+use rustfs_object_data_cache::{ObjectDataCache, ObjectDataCacheConfig, ObjectDataCacheGetRequest, ObjectDataCacheMode};
+use std::hint::black_box;
+
+fn fill_cache() -> ObjectDataCache {
+    let config = ObjectDataCacheConfig {
+        mode: ObjectDataCacheMode::FillBufferedOnly,
+        max_bytes: 256 * 1024 * 1024,
+        max_entry_bytes: 1024 * 1024,
+        // Opt out of the memory gate so fills are deterministic regardless of
+        // the host's live memory (matches the test suite's convention).
+        min_free_memory_percent: 0,
+        ..ObjectDataCacheConfig::default()
+    };
+    ObjectDataCache::new(config).expect("cache should build")
+}
+
+fn request(object: &str) -> ObjectDataCacheGetRequest<'_> {
+    ObjectDataCacheGetRequest {
+        bucket: "bucket",
+        object,
+        etag: "etag",
+        size: 4096,
+        ..Default::default()
+    }
+}
+
+/// The per-GET planning cost on the hot path: key construction plus the metric
+/// label-set hash and recorder dispatch. `plan_get` is synchronous, so this is
+/// measured directly with no runtime overhead.
+fn bench_plan_get(c: &mut Criterion) {
+    let cache = fill_cache();
+    c.bench_function("plan_get", |b| {
+        b.iter(|| {
+            black_box(cache.plan_get(black_box(request("object"))));
+        });
+    });
+}
+
+/// Hit-path latency: lookup of a resident body. Batches 100 lookups per sample
+/// so the single `block_on` is amortised out of the measurement.
+fn bench_lookup_hit(c: &mut Criterion) {
+    let runtime = tokio::runtime::Builder::new_current_thread().build().expect("runtime");
+    let cache = fill_cache();
+    let plan = cache.plan_get(request("hot-object"));
+    runtime.block_on(async {
+        let _ = cache.fill_body(&plan, Bytes::from(vec![0u8; 4096])).await;
+    });
+
+    c.bench_function("lookup_hit_x100", |b| {
+        b.iter(|| {
+            runtime.block_on(async {
+                for _ in 0..100 {
+                    black_box(cache.lookup_body(black_box(&plan)).await);
+                }
+            });
+        });
+    });
+}
+
+/// Fill throughput for distinct cold keys. Batches 100 fills per sample.
+fn bench_fill(c: &mut Criterion) {
+    let runtime = tokio::runtime::Builder::new_current_thread().build().expect("runtime");
+    let cache = fill_cache();
+    let body = Bytes::from(vec![0u8; 4096]);
+
+    c.bench_function("fill_distinct_x100", |b| {
+        let mut n: u64 = 0;
+        b.iter(|| {
+            runtime.block_on(async {
+                for _ in 0..100 {
+                    n += 1;
+                    let object = format!("fill-{n}");
+                    let plan = cache.plan_get(request(&object));
+                    black_box(cache.fill_body(black_box(&plan), body.clone()).await);
+                }
+            });
+        });
+    });
+}
+
+criterion_group!(benches, bench_plan_get, bench_lookup_hit, bench_fill);
+criterion_main!(benches);
diff --git a/crates/object-data-cache/src/moka_backend.rs b/crates/object-data-cache/src/moka_backend.rs
index e8b8d3524..b6c13d37c 100644
--- a/crates/object-data-cache/src/moka_backend.rs
+++ b/crates/object-data-cache/src/moka_backend.rs
@@ -875,4 +875,125 @@ mod tests {
         }
         assert!(!cached, "a cancelled fill must not leave an orphaned body after a racing invalidation");
     }
+
+    // ---- Concurrency stress (backlog#1107 production-readiness follow-up) ----
+    //
+    // The deterministic race tests above pin specific interleavings with an
+    // injected barrier. These drive sustained, real-parallelism contention on a
+    // small key space to catch leaks and state corruption the pinned tests
+    // cannot: a leaked singleflight leader, a stranded semaphore permit, an
+    // index/cache divergence that only shows up under volume.
+
+    /// Hammer fill / lookup / invalidate concurrently on a few hot identities,
+    /// then assert the shared machinery is neither leaked nor corrupted.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn moka_backend_concurrency_storm_leaves_no_leaked_state() {
+        let stats = Arc::new(ObjectDataCacheStats::default());
+        // Long TTL so entries survive the storm — we want fill-vs-invalidate
+        // races, not just expiry churn.
+        let mut config = enabled_config();
+        config.ttl = Duration::from_secs(30);
+        config.time_to_idle = Duration::from_secs(30);
+        let backend = Arc::new(MokaBackend::new(&config, Arc::clone(&stats)).expect("moka backend should build"));
+
+        const TASKS: usize = 48;
+        const ITERS: usize = 400;
+        const KEYS: usize = 6; // small pool → heavy contention on shared identities
+
+        let mut handles = Vec::new();
+        for t in 0..TASKS {
+            let backend = Arc::clone(&backend);
+            handles.push(tokio::spawn(async move {
+                for i in 0..ITERS {
+                    let k = (t + i) % KEYS;
+                    let object = format!("object-{}", k % 3);
+                    let plan = versioned_plan(&object, &format!("v{k}"), &format!("etag-{k}"));
+                    // Deterministic per-(task,iter) mix: mostly fills, plus
+                    // lookups and invalidations racing against them.
+                    match (t + i) % 4 {
+                        0 | 1 => {
+                            let _ = backend.fill_body(&plan, Bytes::from_static(b"hello")).await;
+                        }
+                        2 => {
+                            let _ = backend.lookup_body(&plan).await;
+                        }
+                        _ => {
+                            let identity = ObjectDataCacheIdentity::new("bucket", object.as_str());
+                            let _ = backend.invalidate_object(&identity).await;
+                        }
+                    }
+                }
+            }));
+        }
+        for handle in handles {
+            handle.await.expect("stress task must not panic or deadlock");
+        }
+
+        // Every fill_body call returned, so no singleflight leader may still be
+        // registered: a non-zero count means a fill leaked its in-flight slot.
+        assert_eq!(
+            stats.snapshot().inflight_fills,
+            0,
+            "the concurrency storm leaked an in-flight fill (singleflight/semaphore not released)"
+        );
+
+        // The shared state is not corrupted: a fresh, uncontended sequence must
+        // still fill, hit, and invalidate correctly.
+        let clean = versioned_plan("fresh-object", "v1", "fresh-etag");
+        assert_eq!(
+            backend.fill_body(&clean, Bytes::from_static(b"hello")).await,
+            ObjectDataCacheFillResult::Inserted,
+            "a clean fill after the storm must still insert"
+        );
+        assert!(matches!(backend.lookup_body(&clean).await, ObjectDataCacheLookup::Hit(_)));
+        let _ = backend
+            .invalidate_object(&ObjectDataCacheIdentity::new("bucket", "fresh-object"))
+            .await;
+        assert!(
+            matches!(backend.lookup_body(&clean).await, ObjectDataCacheLookup::Miss),
+            "invalidation must still win after a concurrency storm"
+        );
+
+        // clear() plus a moka drain empties both the cache and the identity index.
+        let _ = backend.clear().await;
+        backend.cache.run_pending_tasks().await;
+        assert_eq!(backend.entry_count(), 0, "clear() must drop every entry");
+    }
+
+    /// A memory-constrained container (low snapshot) must reject an entire
+    /// concurrent fill burst — admission is bounded, not a check-then-act race
+    /// that lets the burst through. This is the CI-runnable proxy for the
+    /// OOM-under-burst concern; it does not cover the separate 5s-stale-snapshot
+    /// overshoot, which needs byte-based reservation in the gate.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn moka_backend_gate_bounds_admission_under_concurrent_burst() {
+        let stats = Arc::new(ObjectDataCacheStats::default());
+        let backend = Arc::new(MokaBackend::new(&memory_gated_config(), Arc::clone(&stats)).expect("moka backend should build"));
+        // Simulate a pod whose available memory is far below the free floor.
+        backend
+            .memory_gate
+            .set_test_snapshot(Some(crate::memory::ObjectDataCacheMemorySnapshot {
+                total_bytes: 1_000,
+                available_bytes: 50,
+            }));
+
+        const TASKS: usize = 32;
+        let mut handles = Vec::new();
+        for t in 0..TASKS {
+            let backend = Arc::clone(&backend);
+            handles.push(tokio::spawn(async move {
+                let plan = versioned_plan(&format!("object-{t}"), "v1", &format!("etag-{t}"));
+                backend.fill_body(&plan, Bytes::from_static(b"hello")).await
+            }));
+        }
+        let mut rejected = 0;
+        for handle in handles {
+            if handle.await.expect("burst task should complete") == ObjectDataCacheFillResult::SkippedMemoryPressure {
+                rejected += 1;
+            }
+        }
+
+        assert_eq!(rejected, TASKS, "the gate must reject every fill in the burst when memory is constrained");
+        assert_eq!(backend.entry_count(), 0, "no body may be admitted under memory pressure");
+    }
 }