linuxpi
diff --git a/‎sandbox/libs/tiered-storage/src/main/rust/Cargo.toml‎
Lines changed: 1 addition & 0 deletions b/‎sandbox/libs/tiered-storage/src/main/rust/Cargo.toml‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎sandbox/libs/tiered-storage/src/main/rust/src/ffm.rs‎
Lines changed: 22 additions & 6 deletions b/‎sandbox/libs/tiered-storage/src/main/rust/src/ffm.rs‎
Lines changed: 22 additions & 6 deletions
diff --git a/‎sandbox/libs/tiered-storage/src/main/rust/src/ffm_tests.rs‎
Lines changed: 66 additions & 7 deletions b/‎sandbox/libs/tiered-storage/src/main/rust/src/ffm_tests.rs‎
Lines changed: 66 additions & 7 deletions
@@ -18,6 +18,7 @@ thiserror = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 native-bridge-common = { workspace = true }
+opensearch-block-cache = { path = "../../../../../plugins/block-cache-foyer/src/main/rust" }
 
 [dev-dependencies]
 tempfile = { workspace = true }
 
@@ -18,6 +18,8 @@ use std::sync::Arc;
 use native_bridge_common::ffm_safe;
 use object_store::ObjectStore;
 
+use opensearch_block_cache::traits::BlockCache;
+
 use crate::registry::TieredStorageRegistry;
 use crate::registry::FileRegistry;
 use crate::tiered_object_store::TieredObjectStore;
@@ -72,11 +74,15 @@ unsafe fn arc_from_ptr(ptr: i64) -> Result<Arc<TieredObjectStore>, String> {
 /// - `remote_store_box_ptr`: if non-zero, a `Box<Arc<dyn ObjectStore>>` pointer from a repository
 ///   plugin. The Arc is cloned (ownership is NOT taken — the pointer remains valid for other
 ///   shards). If 0, no remote store.
+/// - `cache_box_ptr`: if non-zero, a `Box<Arc<dyn BlockCache>>` pointer from a block cache
+///   plugin. The Arc is cloned (ownership is NOT taken — the pointer remains valid for the
+///   node lifetime). If 0, no block cache is attached.
 #[ffm_safe]
 #[no_mangle]
 pub extern "C" fn ts_create_tiered_object_store(
     local_store_box_ptr: i64,
     remote_store_box_ptr: i64,
+    cache_box_ptr: i64,
 ) -> i64 {
     let file_registry = Arc::new(TieredStorageRegistry::new());
 
@@ -86,7 +92,7 @@ pub extern "C" fn ts_create_tiered_object_store(
         Arc::new(object_store::local::LocalFileSystem::new())
     };
 
-    let store = Arc::new(TieredObjectStore::new(file_registry, local));
+    let mut store = TieredObjectStore::new(file_registry, local);
 
     if remote_store_box_ptr != NULL_PTR {
         // IMPORTANT: Do NOT consume the Box — the pointer is node-level and shared
@@ -96,8 +102,17 @@ pub extern "C" fn ts_create_tiered_object_store(
         store.set_remote(remote_arc);
     }
 
-    let ptr = Arc::into_raw(store) as i64;
-    native_bridge_common::log_info!("ffm: ts_create_tiered_object_store ptr={}", ptr);
+    if cache_box_ptr != NULL_PTR {
+        let cache_box = unsafe { &*(cache_box_ptr as *const Arc<dyn BlockCache>) };
+        store = store.with_cache(Arc::clone(cache_box));
+        native_bridge_common::log_info!("ffm: ts_create_tiered_object_store: block cache wired");
+    }
+
+    let ptr = Arc::into_raw(Arc::new(store)) as i64;
+    native_bridge_common::log_info!(
+        "ffm: ts_create_tiered_object_store cache_wired={}",
+        cache_box_ptr != NULL_PTR
+    );
     Ok(ptr)
 }
 
@@ -117,7 +132,7 @@ pub extern "C" fn ts_get_object_store_box_ptr(tiered_store_ptr: i64) -> i64 {
     // Coerce to trait object and box it
     let boxed: Box<Arc<dyn ObjectStore>> = Box::new(arc as Arc<dyn ObjectStore>);
     let ptr = Box::into_raw(boxed) as i64;
-    native_bridge_common::log_info!("ffm: ts_get_object_store_box_ptr input={}, output={}", tiered_store_ptr, ptr);
+    native_bridge_common::log_info!("ffm: ts_get_object_store_box_ptr: ok");
     Ok(ptr)
 }
 
@@ -130,7 +145,7 @@ pub extern "C" fn ts_destroy_object_store_box_ptr(ptr: i64) -> i64 {
         return Err("ts_destroy_object_store_box_ptr: null pointer (0)".to_string());
     }
     let _boxed = unsafe { Box::from_raw(ptr as *mut Arc<dyn ObjectStore>) };
-    native_bridge_common::log_info!("ffm: ts_destroy_object_store_box_ptr ptr={}", ptr);
+    native_bridge_common::log_info!("ffm: ts_destroy_object_store_box_ptr: ok");
     Ok(0)
 }
 
@@ -144,7 +159,7 @@ pub extern "C" fn ts_destroy_tiered_object_store(ptr: i64) -> i64 {
         return Err("ts_destroy_tiered_object_store: null pointer (0)".to_string());
     }
     let _store = unsafe { Arc::from_raw(ptr as *const TieredObjectStore) };
-    native_bridge_common::log_info!("ffm: ts_destroy_tiered_object_store ptr={}", ptr);
+    native_bridge_common::log_info!("ffm: ts_destroy_tiered_object_store: ok");
     Ok(0)
 }
 
@@ -223,6 +238,7 @@ pub extern "C" fn ts_remove_file(
     let path = path.strip_prefix('/').unwrap_or(path);
 
     store.registry().remove(path, false);
+    store.evict_path(path);
 
     native_bridge_common::log_debug!("ffm: ts_remove_file path='{}'", path);
     Ok(0)
 
@@ -7,7 +7,7 @@ fn test_destroy_null_returns_error() {
 
 #[test]
 fn test_create_and_destroy_no_leak() {
-    let store_ptr = ts_create_tiered_object_store(0, 0);
+    let store_ptr = ts_create_tiered_object_store(0, 0, 0);
     assert!(store_ptr > 0);
     assert_eq!(ts_destroy_tiered_object_store(store_ptr), 0);
 }
@@ -27,7 +27,7 @@ fn test_remove_file_null_store_returns_error() {
 
 #[test]
 fn test_register_files_and_remove_round_trip() {
-    let store_ptr = ts_create_tiered_object_store(0, 0);
+    let store_ptr = ts_create_tiered_object_store(0, 0, 0);
     assert!(store_ptr > 0);
 
     // Batch register: two files as Remote (triplets: path\nremotePath\nsize\n...)
@@ -44,7 +44,7 @@ fn test_register_files_and_remove_round_trip() {
 
 #[test]
 fn test_register_files_invalid_location_returns_error() {
-    let store_ptr = ts_create_tiered_object_store(0, 0);
+    let store_ptr = ts_create_tiered_object_store(0, 0, 0);
     assert!(store_ptr > 0);
 
     let entries = b"test.parquet\nremote/test.parquet\n2048";
@@ -66,7 +66,7 @@ fn test_destroy_object_store_box_ptr_null_returns_error() {
 
 #[test]
 fn test_get_and_destroy_object_store_box_ptr_round_trip() {
-    let store_ptr = ts_create_tiered_object_store(0, 0);
+    let store_ptr = ts_create_tiered_object_store(0, 0, 0);
     assert!(store_ptr > 0);
 
     // Get a boxed pointer — this increments the Arc refcount
@@ -83,7 +83,7 @@ fn test_get_and_destroy_object_store_box_ptr_round_trip() {
 
 #[test]
 fn test_get_object_store_box_ptr_multiple_calls() {
-    let store_ptr = ts_create_tiered_object_store(0, 0);
+    let store_ptr = ts_create_tiered_object_store(0, 0, 0);
     assert!(store_ptr > 0);
 
     // Multiple box pointers can coexist (simulates multiple reader managers)
@@ -109,8 +109,8 @@ fn test_create_with_remote_does_not_consume_pointer() {
     let remote_ptr = Box::into_raw(remote_box) as i64;
 
     // Create two TieredObjectStores sharing the same remote pointer
-    let store1 = ts_create_tiered_object_store(0, remote_ptr);
-    let store2 = ts_create_tiered_object_store(0, remote_ptr);
+    let store1 = ts_create_tiered_object_store(0, remote_ptr, 0);
+    let store2 = ts_create_tiered_object_store(0, remote_ptr, 0);
     assert!(store1 > 0);
     assert!(store2 > 0);
 
@@ -121,3 +121,62 @@ fn test_create_with_remote_does_not_consume_pointer() {
     // Clean up the remote Box (simulates repository.doClose())
     let _remote_box = unsafe { Box::from_raw(remote_ptr as *mut Arc<dyn ObjectStore>) };
 }
+
+// -- cache_box_ptr tests ------------------------------------------------
+
+/// ts_create_tiered_object_store with 0 cache_box_ptr creates a store without
+/// a cache. The store must still be functional for all registry and I/O operations.
+#[test]
+fn test_create_with_zero_cache_ptr_creates_uncached_store() {
+    let store_ptr = ts_create_tiered_object_store(0, 0, 0);
+    assert!(store_ptr > 0);
+
+    // Store is functional — register a file without error.
+    let entries = b"a.parquet\nremote/a.parquet\n512";
+    let result = ts_register_files(store_ptr, entries.as_ptr(), entries.len() as i64, 1, 1);
+    assert_eq!(result, 0);
+
+    assert_eq!(ts_destroy_tiered_object_store(store_ptr), 0);
+}
+
+/// ts_create_tiered_object_store with a valid cache_box_ptr wires the cache
+/// without consuming the Box — two stores can share the same cache pointer.
+#[test]
+fn test_create_with_cache_does_not_consume_pointer() {
+    use opensearch_block_cache::range_cache::CacheKey;
+    use opensearch_block_cache::traits::BlockCache;
+    use bytes::Bytes;
+
+    // Minimal no-op cache used only to construct a valid Box<Arc<dyn BlockCache>> pointer.
+    struct NoopCache;
+    impl BlockCache for NoopCache {
+        fn as_any(&self) -> &dyn std::any::Any { self }
+        fn get<'a>(&'a self, _key: &'a CacheKey)
+            -> std::pin::Pin<Box<dyn std::future::Future<Output = Option<Bytes>> + Send + 'a>>
+        {
+            Box::pin(std::future::ready(None))
+        }
+        fn put(&self, _key: &CacheKey, _data: Bytes) {}
+        fn evict_prefix(&self, _prefix: &str) {}
+        fn clear(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + '_>> {
+            Box::pin(std::future::ready(()))
+        }
+    }
+
+    let cache: Arc<dyn BlockCache> = Arc::new(NoopCache);
+    let cache_box = Box::new(Arc::clone(&cache));
+    let cache_ptr = Box::into_raw(cache_box) as i64;
+
+    // Create two stores sharing the same cache pointer.
+    let store1 = ts_create_tiered_object_store(0, 0, cache_ptr);
+    let store2 = ts_create_tiered_object_store(0, 0, cache_ptr);
+    assert!(store1 > 0);
+    assert!(store2 > 0);
+
+    // Both stores created — cache pointer not consumed (Arc strong count still valid).
+    assert_eq!(ts_destroy_tiered_object_store(store1), 0);
+    assert_eq!(ts_destroy_tiered_object_store(store2), 0);
+
+    // Clean up the cache Box (simulates FoyerBlockCache.close()).
+    let _cache_box = unsafe { Box::from_raw(cache_ptr as *mut Arc<dyn BlockCache>) };
+}