perf(ecstore): bound cached legacy workspaces

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/ecstore/src/erasure/coding/erasure.rs

@@ -71,7 +71,8 @@ impl EncodedBlock {
 
 const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::ORDER;
 const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64;
-const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64;
+const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 16;
+const LEGACY_REED_SOLOMON_CACHE_MAX_WORKSPACE_BYTES: usize = 1024 * 1024;
 
 type ModernReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<ReedSolomon>>>;
 type LegacyReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<LegacyReedSolomonEncoder>>>;
@@ -158,6 +159,12 @@ impl LegacyReedSolomonEncoder {
         })
     }
 
+    fn should_cache_workspace(&self, shard_len: usize) -> bool {
+        shard_len
+            .checked_mul(self.data_shards + self.parity_shards)
+            .is_some_and(|bytes| bytes <= LEGACY_REED_SOLOMON_CACHE_MAX_WORKSPACE_BYTES)
+    }
+
     fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
         let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
         if shards_vec.is_empty() {
@@ -197,12 +204,14 @@ impl LegacyReedSolomonEncoder {
             }
         }
         drop(result);
-        let mut cache = self
-            .encoder_cache
-            .write()
-            .map_err(|_| io::Error::other("Failed to return encoder to cache"))?;
-        if cache.is_none() {
-            *cache = Some(encoder);
+        if self.should_cache_workspace(shard_len) {
+            let mut cache = self
+                .encoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to return encoder to cache"))?;
+            if cache.is_none() {
+                *cache = Some(encoder);
+            }
         }
         Ok(())
     }
@@ -270,12 +279,14 @@ impl LegacyReedSolomonEncoder {
 
         drop(result);
 
-        let mut cache = self
-            .decoder_cache
-            .write()
-            .map_err(|_| io::Error::other("Failed to return decoder to cache"))?;
-        if cache.is_none() {
-            *cache = Some(decoder);
+        if self.should_cache_workspace(shard_len) {
+            let mut cache = self
+                .decoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to return decoder to cache"))?;
+            if cache.is_none() {
+                *cache = Some(decoder);
+            }
         }
 
         Ok(())
@@ -1448,6 +1459,24 @@ mod tests {
         assert!(Arc::ptr_eq(&first, &second));
     }
 
+    #[test]
+    fn legacy_workspace_cache_rejects_oversize_buffers_and_isolates_layouts() {
+        let four_plus_two = Erasure::new_with_options(4, 2, 64, true)
+            .legacy_encoder
+            .expect("legacy codec should be initialized");
+        let four_plus_one = Erasure::new_with_options(4, 1, 64, true)
+            .legacy_encoder
+            .expect("distinct parity layout should be initialized");
+        let three_plus_two = Erasure::new_with_options(3, 2, 64, true)
+            .legacy_encoder
+            .expect("distinct data layout should be initialized");
+
+        assert!(!Arc::ptr_eq(&four_plus_two, &four_plus_one));
+        assert!(!Arc::ptr_eq(&four_plus_two, &three_plus_two));
+        assert!(four_plus_two.should_cache_workspace(LEGACY_REED_SOLOMON_CACHE_MAX_WORKSPACE_BYTES / 6));
+        assert!(!four_plus_two.should_cache_workspace(LEGACY_REED_SOLOMON_CACHE_MAX_WORKSPACE_BYTES / 6 + 1));
+    }
+
     #[test]
     fn concurrent_legacy_codecs_preserve_byte_exact_results() {
         let barrier = Arc::new(std::sync::Barrier::new(2));

crates/ecstore/src/set_disk/ops/object.rs

@@ -5993,6 +5993,37 @@ mod inline_put_commit_path_tests {
         assert_eq!(restored, payload);
     }
 
+    #[tokio::test]
+    async fn repeated_gets_reuse_the_set_erasure_shell() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "get-erasure-shell-cache";
+        let object = "object.bin";
+        let payload = vec![0x4d; 1024 * 1024];
+        make_bucket(&disk_stores, bucket).await;
+
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("non-inline object should commit");
+        assert!(set_disks.erasure_cache.entries.read().is_empty());
+
+        for _ in 0..2 {
+            let mut object_reader = set_disks
+                .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                .await
+                .expect("cached-shell GET should succeed");
+            let mut restored = Vec::new();
+            object_reader
+                .stream
+                .read_to_end(&mut restored)
+                .await
+                .expect("cached-shell GET should stream");
+            assert_eq!(restored, payload);
+            assert_eq!(set_disks.erasure_cache.entries.read().len(), 1);
+        }
+    }
+
     #[tokio::test]
     async fn inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;