diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index ca735020b..53e00dd06 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -71,7 +71,8 @@ impl EncodedBlock { const MODERN_MAX_TOTAL_SHARDS: usize = ::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>>; type LegacyReedSolomonCache = RwLock>>; @@ -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)); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index e96188db4..ff92752b1 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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;