From ae4aa4f5d4c8c9928c73d6ae4f3ec80c6de1caa3 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 27 Jun 2026 08:35:38 +0800 Subject: [PATCH] feat(get): add rustfs codec decode engine (#3928) --- crates/ecstore/src/erasure_codec/bridge.rs | 322 ++++++++++++++++++ crates/ecstore/src/erasure_codec/workspace.rs | 17 + .../src/erasure_coding/decode_reader.rs | 70 +++- crates/ecstore/src/set_disk/mod.rs | 27 ++ crates/ecstore/src/set_disk/read.rs | 47 ++- 5 files changed, 478 insertions(+), 5 deletions(-) diff --git a/crates/ecstore/src/erasure_codec/bridge.rs b/crates/ecstore/src/erasure_codec/bridge.rs index 133bacaf9..8765f765d 100644 --- a/crates/ecstore/src/erasure_codec/bridge.rs +++ b/crates/ecstore/src/erasure_codec/bridge.rs @@ -12,9 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::erasure_codec::workspace::RustfsCodecDecodeWorkspace; use crate::erasure_coding::Erasure; +use reed_solomon_erasure::galois_8::ReedSolomon; use std::io; +pub(crate) const GET_CODEC_STREAMING_ENGINE_LEGACY: &str = "legacy"; +pub(crate) const GET_CODEC_STREAMING_ENGINE_RUSTFS: &str = "rustfs"; + pub(crate) trait DecodeWorkspace: Send + Sync + 'static { fn shard_len(&self) -> usize; } @@ -38,6 +43,40 @@ fn data_shards_complete(shards: &[Option>], data_shards: usize) -> bool shards.len() >= data_shards && shards.iter().take(data_shards).all(Option::is_some) } +fn recover_empty_payload_data_shards( + shards: &mut [Option>], + data_shards: usize, + parity_shards: usize, +) -> io::Result { + let expected_shards = data_shards + parity_shards; + if shards.len() != expected_shards { + return Err(io::Error::other(format!( + "invalid shard count: got {}, expected {}", + shards.len(), + expected_shards + ))); + } + + let mut present_shards = 0usize; + for shard in shards.iter().filter_map(Option::as_ref) { + present_shards += 1; + if !shard.is_empty() { + return Ok(false); + } + } + if present_shards == 0 || present_shards < data_shards { + return Ok(false); + } + + for shard in shards.iter_mut().take(data_shards) { + if shard.is_none() { + *shard = Some(Vec::new()); + } + } + + Ok(true) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) struct LegacyDecodeWorkspace { shard_len: usize, @@ -49,6 +88,12 @@ impl LegacyDecodeWorkspace { } } +impl DecodeWorkspace for RustfsCodecDecodeWorkspace { + fn shard_len(&self) -> usize { + RustfsCodecDecodeWorkspace::shard_len(self) + } +} + impl DecodeWorkspace for LegacyDecodeWorkspace { fn shard_len(&self) -> usize { self.shard_len @@ -102,10 +147,187 @@ impl ErasureDecodeEngine for LegacyEcDecodeEngine { } } +#[derive(Clone)] +pub(crate) struct RustfsCodecDecodeEngine { + data_shards: usize, + parity_shards: usize, + block_size: usize, + codec: Option, +} + +impl RustfsCodecDecodeEngine { + pub(crate) fn new(erasure: &Erasure) -> io::Result { + let codec = if erasure.parity_shards > 0 { + ReedSolomon::new(erasure.data_shards, erasure.parity_shards) + .map_err(|err| io::Error::other(format!("Failed to create RustFS codec decode engine: {err:?}"))) + .map(Some)? + } else { + None + }; + + Ok(Self { + data_shards: erasure.data_shards, + parity_shards: erasure.parity_shards, + block_size: erasure.block_size, + codec, + }) + } +} + +impl ErasureDecodeEngine for RustfsCodecDecodeEngine { + type Workspace = RustfsCodecDecodeWorkspace; + + fn data_shards(&self) -> usize { + self.data_shards + } + + fn parity_shards(&self) -> usize { + self.parity_shards + } + + fn block_size(&self) -> usize { + self.block_size + } + + fn supports_progressive_decode(&self) -> bool { + false + } + + fn supports_aligned_shards(&self) -> bool { + false + } + + fn prepare_workspace(&self, shard_len: usize) -> io::Result { + Ok(RustfsCodecDecodeWorkspace::new(shard_len)) + } + + fn reconstruct_into(&self, shards: &mut [Option>], _workspace: &mut Self::Workspace) -> io::Result<()> { + if data_shards_complete(shards, self.data_shards) { + return Ok(()); + } + if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? { + return Ok(()); + } + + if let Some(codec) = &self.codec { + codec + .reconstruct_data_opt(shards) + .map_err(|err| io::Error::other(format!("RustFS codec reconstruct failed: {err:?}"))) + } else { + Ok(()) + } + } +} + +#[derive(Clone)] +pub(crate) enum CodecStreamingDecodeEngine { + Legacy(Box), + Rustfs(Box), +} + +impl CodecStreamingDecodeEngine { + pub(crate) fn legacy(erasure: Erasure) -> Self { + Self::Legacy(Box::new(LegacyEcDecodeEngine::new(erasure))) + } + + pub(crate) fn rustfs(erasure: &Erasure) -> io::Result { + RustfsCodecDecodeEngine::new(erasure).map(Box::new).map(Self::Rustfs) + } +} + +pub(crate) enum CodecStreamingDecodeWorkspace { + Legacy(LegacyDecodeWorkspace), + Rustfs(RustfsCodecDecodeWorkspace), +} + +impl DecodeWorkspace for CodecStreamingDecodeWorkspace { + fn shard_len(&self) -> usize { + match self { + Self::Legacy(workspace) => workspace.shard_len(), + Self::Rustfs(workspace) => workspace.shard_len(), + } + } +} + +impl ErasureDecodeEngine for CodecStreamingDecodeEngine { + type Workspace = CodecStreamingDecodeWorkspace; + + fn data_shards(&self) -> usize { + match self { + Self::Legacy(engine) => engine.data_shards(), + Self::Rustfs(engine) => engine.data_shards(), + } + } + + fn parity_shards(&self) -> usize { + match self { + Self::Legacy(engine) => engine.parity_shards(), + Self::Rustfs(engine) => engine.parity_shards(), + } + } + + fn block_size(&self) -> usize { + match self { + Self::Legacy(engine) => engine.block_size(), + Self::Rustfs(engine) => engine.block_size(), + } + } + + fn supports_progressive_decode(&self) -> bool { + match self { + Self::Legacy(engine) => engine.supports_progressive_decode(), + Self::Rustfs(engine) => engine.supports_progressive_decode(), + } + } + + fn supports_aligned_shards(&self) -> bool { + match self { + Self::Legacy(engine) => engine.supports_aligned_shards(), + Self::Rustfs(engine) => engine.supports_aligned_shards(), + } + } + + fn prepare_workspace(&self, shard_len: usize) -> io::Result { + match self { + Self::Legacy(engine) => engine.prepare_workspace(shard_len).map(CodecStreamingDecodeWorkspace::Legacy), + Self::Rustfs(engine) => engine.prepare_workspace(shard_len).map(CodecStreamingDecodeWorkspace::Rustfs), + } + } + + fn reconstruct_into(&self, shards: &mut [Option>], workspace: &mut Self::Workspace) -> io::Result<()> { + match (self, workspace) { + (Self::Legacy(engine), CodecStreamingDecodeWorkspace::Legacy(workspace)) => { + engine.reconstruct_into(shards, workspace) + } + (Self::Rustfs(engine), CodecStreamingDecodeWorkspace::Rustfs(workspace)) => { + engine.reconstruct_into(shards, workspace) + } + _ => Err(io::Error::other("codec streaming decode engine/workspace mismatch")), + } + } +} + #[cfg(test)] mod tests { use super::*; + fn encoded_shards(erasure: &Erasure, data: &[u8]) -> Vec>> { + erasure + .encode_data(data) + .expect("test stripe should encode") + .into_iter() + .map(|shard| Some(shard.to_vec())) + .collect() + } + + fn reconstruct_with(engine: &E, shards: &mut [Option>]) -> io::Result<()> + where + E: ErasureDecodeEngine, + { + let mut workspace = engine.prepare_workspace(4)?; + engine.reconstruct_into(shards, &mut workspace) + } + #[test] fn data_shards_complete_ignores_parity_slots() { let shards = vec![Some(vec![1]), Some(vec![2]), None]; @@ -151,4 +373,104 @@ mod tests { assert_eq!(workspace.shard_len(), 4); assert!(shards.iter().take(engine.data_shards()).all(Option::is_some)); } + + #[test] + fn rustfs_codec_decode_engine_reports_erasure_shape() { + let erasure = Erasure::new(4, 2, 1 << 20); + let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + + assert_eq!(engine.data_shards(), 4); + assert_eq!(engine.parity_shards(), 2); + assert_eq!(engine.block_size(), 1 << 20); + assert!(!engine.supports_progressive_decode()); + assert!(!engine.supports_aligned_shards()); + } + + #[test] + fn rustfs_codec_decode_engine_keeps_complete_data_shards() { + let erasure = Erasure::new(4, 2, 16); + let mut shards = encoded_shards(&erasure, b"all data shards are present"); + let before = shards.clone(); + + let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + reconstruct_with(&engine, &mut shards).expect("complete data shards should not reconstruct"); + + assert_eq!(shards, before); + } + + #[test] + fn rustfs_codec_decode_engine_reconstructs_missing_data_like_legacy() { + let erasure = Erasure::new(4, 2, 16); + let mut legacy_shards = encoded_shards(&erasure, b"missing data shard must match legacy output"); + let mut rustfs_shards = legacy_shards.clone(); + legacy_shards[1] = None; + rustfs_shards[1] = None; + + let legacy = LegacyEcDecodeEngine::new(erasure.clone()); + let rustfs = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + + reconstruct_with(&legacy, &mut legacy_shards).expect("legacy should reconstruct"); + reconstruct_with(&rustfs, &mut rustfs_shards).expect("rustfs codec should reconstruct"); + + assert_eq!(rustfs_shards, legacy_shards); + } + + #[test] + fn rustfs_codec_decode_engine_leaves_missing_parity_unreconstructed() { + let erasure = Erasure::new(4, 2, 16); + let mut shards = encoded_shards(&erasure, b"parity-only missing should not touch output data"); + let before_data = shards.iter().take(erasure.data_shards).cloned().collect::>(); + shards[erasure.data_shards] = None; + + let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + reconstruct_with(&engine, &mut shards).expect("missing parity should not fail"); + + assert_eq!(shards.iter().take(erasure.data_shards).cloned().collect::>(), before_data); + assert!(shards[erasure.data_shards].is_none()); + } + + #[test] + fn rustfs_codec_decode_engine_errors_on_insufficient_shards_like_legacy() { + let erasure = Erasure::new(4, 2, 16); + let mut legacy_shards = encoded_shards(&erasure, b"insufficient shards must fail"); + let mut rustfs_shards = legacy_shards.clone(); + for index in [0, 1, 2] { + legacy_shards[index] = None; + rustfs_shards[index] = None; + } + + let legacy = LegacyEcDecodeEngine::new(erasure.clone()); + let rustfs = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + + assert!(reconstruct_with(&legacy, &mut legacy_shards).is_err()); + assert!(reconstruct_with(&rustfs, &mut rustfs_shards).is_err()); + } + + #[test] + fn rustfs_codec_decode_engine_recovers_empty_data_shard() { + let erasure = Erasure::new(4, 2, 16); + let mut shards = encoded_shards(&erasure, b""); + shards[0] = None; + + let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + reconstruct_with(&engine, &mut shards).expect("empty shard should reconstruct"); + + assert_eq!(shards[0], Some(Vec::new())); + assert!(shards.iter().take(erasure.data_shards).all(Option::is_some)); + } + + #[test] + fn codec_streaming_decode_engine_dispatches_to_rustfs_workspace() { + let erasure = Erasure::new(4, 2, 16); + let mut shards = encoded_shards(&erasure, b"dispatch keeps rustfs engine byte-identical"); + shards[2] = None; + + let engine = CodecStreamingDecodeEngine::rustfs(&erasure).expect("engine should be created"); + let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared"); + engine + .reconstruct_into(&mut shards, &mut workspace) + .expect("enum engine should dispatch reconstruction"); + + assert!(shards.iter().take(engine.data_shards()).all(Option::is_some)); + } } diff --git a/crates/ecstore/src/erasure_codec/workspace.rs b/crates/ecstore/src/erasure_codec/workspace.rs index 405b89e42..7d05700b1 100644 --- a/crates/ecstore/src/erasure_codec/workspace.rs +++ b/crates/ecstore/src/erasure_codec/workspace.rs @@ -12,6 +12,23 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct RustfsCodecDecodeWorkspace { + shard_len: usize, +} + +impl RustfsCodecDecodeWorkspace { + #[inline] + pub(crate) fn new(shard_len: usize) -> Self { + Self { shard_len } + } + + #[inline] + pub(crate) fn shard_len(&self) -> usize { + self.shard_len + } +} + #[derive(Debug, Default)] pub(crate) struct ShardBufferPool { buffers: Vec>>, diff --git a/crates/ecstore/src/erasure_coding/decode_reader.rs b/crates/ecstore/src/erasure_coding/decode_reader.rs index 2b09e37c7..21bb019b0 100644 --- a/crates/ecstore/src/erasure_coding/decode_reader.rs +++ b/crates/ecstore/src/erasure_coding/decode_reader.rs @@ -476,7 +476,9 @@ fn emit_data_shards(state: &StripeReadState, data_shards: usize, block_size: usi #[cfg(test)] mod tests { use super::*; - use crate::erasure_codec::bridge::LegacyEcDecodeEngine; + use crate::erasure_codec::bridge::{ + CodecStreamingDecodeEngine, ErasureDecodeEngine, LegacyEcDecodeEngine, RustfsCodecDecodeEngine, + }; use crate::erasure_coding::Erasure; use crate::set_disk::shard_source::{ShardSlot, StripeReadState}; use std::collections::VecDeque; @@ -533,15 +535,27 @@ mod tests { } } - async fn decode_all(erasure: Erasure, data: &[u8], missing_indexes: &[usize]) -> io::Result> { - let source = source_from_data(&erasure, data, missing_indexes); - let engine = LegacyEcDecodeEngine::new(erasure); + async fn decode_all_with_engine( + erasure: &Erasure, + engine: E, + data: &[u8], + missing_indexes: &[usize], + ) -> io::Result> + where + E: ErasureDecodeEngine + Clone + Send + Sync + 'static, + { + let source = source_from_data(erasure, data, missing_indexes); let mut reader = ErasureDecodeReader::new(source, engine, data.len())?; let mut decoded = Vec::new(); reader.read_to_end(&mut decoded).await?; Ok(decoded) } + async fn decode_all(erasure: Erasure, data: &[u8], missing_indexes: &[usize]) -> io::Result> { + let engine = LegacyEcDecodeEngine::new(erasure.clone()); + decode_all_with_engine(&erasure, engine, data, missing_indexes).await + } + #[tokio::test] async fn erasure_decode_reader_reads_single_stripe() { let erasure = Erasure::new(4, 2, 64); @@ -597,6 +611,54 @@ mod tests { assert_eq!(decoded, data); } + #[tokio::test] + async fn erasure_decode_reader_rustfs_engine_matches_legacy_with_missing_data() { + let erasure = Erasure::new(4, 2, 32); + let data = b"rustfs codec reader output must match legacy reader output exactly"; + let legacy = LegacyEcDecodeEngine::new(erasure.clone()); + let rustfs = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + + let legacy_decoded = decode_all_with_engine(&erasure, legacy, data, &[1]) + .await + .expect("legacy reader should decode"); + let rustfs_decoded = decode_all_with_engine(&erasure, rustfs, data, &[1]) + .await + .expect("rustfs codec reader should decode"); + + assert_eq!(rustfs_decoded, legacy_decoded); + assert_eq!(rustfs_decoded, data); + } + + #[tokio::test] + async fn erasure_decode_reader_rustfs_engine_handles_empty_object() { + let erasure = Erasure::new(4, 2, 32); + let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created"); + + let decoded = decode_all_with_engine(&erasure, engine, b"", &[]) + .await + .expect("empty object should decode"); + + assert!(decoded.is_empty()); + } + + #[tokio::test] + async fn erasure_decode_reader_codec_streaming_engine_enum_matches_legacy() { + let erasure = Erasure::new(4, 2, 32); + let data = b"selected codec streaming engine preserves reader output"; + let legacy = CodecStreamingDecodeEngine::legacy(erasure.clone()); + let rustfs = CodecStreamingDecodeEngine::rustfs(&erasure).expect("engine should be created"); + + let legacy_decoded = decode_all_with_engine(&erasure, legacy, data, &[2]) + .await + .expect("legacy enum reader should decode"); + let rustfs_decoded = decode_all_with_engine(&erasure, rustfs, data, &[2]) + .await + .expect("rustfs enum reader should decode"); + + assert_eq!(rustfs_decoded, legacy_decoded); + assert_eq!(rustfs_decoded, data); + } + #[tokio::test] async fn erasure_decode_reader_reads_when_only_parity_shards_are_missing() { let erasure = Erasure::new(4, 2, 32); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 1be841f40..6718d5b6d 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -33,6 +33,9 @@ use crate::disk::{ conv_part_err_to_int, has_part_err, }; use crate::disk::{STORAGE_FORMAT_FILE, count_part_not_success}; +use crate::erasure_codec::bridge::{ + CodecStreamingDecodeEngine, GET_CODEC_STREAMING_ENGINE_LEGACY, GET_CODEC_STREAMING_ENGINE_RUSTFS, +}; use crate::erasure_coding; use crate::error::{Error, Result, is_err_version_not_found}; use crate::error::{GenericError, ObjectApiError, is_err_object_not_found}; @@ -301,8 +304,10 @@ const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_millis(250); const GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 1024; const ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE"; const ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE"; +const ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = "RUSTFS_GET_CODEC_STREAMING_ENGINE"; const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE: bool = false; const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: usize = MI_B; +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = GET_CODEC_STREAMING_ENGINE_LEGACY; const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE"; const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = false; static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); @@ -383,6 +388,28 @@ fn get_codec_streaming_min_size() -> usize { rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE) } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetCodecStreamingEngine { + Legacy, + Rustfs, +} + +fn get_codec_streaming_engine() -> GetCodecStreamingEngine { + let engine = rustfs_utils::get_env_str(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE); + match engine.trim() { + value if value.eq_ignore_ascii_case(GET_CODEC_STREAMING_ENGINE_RUSTFS) => GetCodecStreamingEngine::Rustfs, + value if value.eq_ignore_ascii_case(GET_CODEC_STREAMING_ENGINE_LEGACY) => GetCodecStreamingEngine::Legacy, + _ => GetCodecStreamingEngine::Legacy, + } +} + +fn build_get_codec_streaming_decode_engine(erasure: erasure_coding::Erasure) -> std::io::Result { + match get_codec_streaming_engine() { + GetCodecStreamingEngine::Legacy => Ok(CodecStreamingDecodeEngine::legacy(erasure)), + GetCodecStreamingEngine::Rustfs => CodecStreamingDecodeEngine::rustfs(&erasure), + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum GetCodecStreamingDecision { Use, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 0e3199f0e..4ec237c49 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2083,7 +2083,7 @@ impl SetDisks { Some(GET_OBJECT_PATH_CODEC_STREAMING), read_costs, ); - let engine = crate::erasure_codec::bridge::LegacyEcDecodeEngine::new(erasure); + let engine = build_get_codec_streaming_decode_engine(erasure)?; let reader = erasure_coding::decode_reader::ErasureDecodeReader::new(source, engine, part_length)?; Ok(Box::new(erasure_coding::decode_reader::SyncErasureDecodeReader::new(reader))) } @@ -2979,6 +2979,51 @@ mod tests { ); } + #[test] + fn codec_streaming_engine_defaults_to_legacy_and_parses_rustfs() { + temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, None::<&str>, || { + assert_eq!(get_codec_streaming_engine(), GetCodecStreamingEngine::Legacy); + }); + + temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || { + assert_eq!(get_codec_streaming_engine(), GetCodecStreamingEngine::Rustfs); + }); + + temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some("unknown"), || { + assert_eq!(get_codec_streaming_engine(), GetCodecStreamingEngine::Legacy); + }); + } + + #[test] + fn codec_streaming_engine_env_is_ignored_when_streaming_is_disabled() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("false")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + ], + || { + let fi = codec_streaming_test_fileinfo(1024, 1); + let object_info = codec_streaming_test_object_info(&fi); + + assert_eq!( + get_codec_streaming_reader_decision(&None, &object_info, &fi, true), + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled) + ); + }, + ); + } + + #[test] + fn codec_streaming_decode_engine_builder_selects_rustfs() { + temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || { + let erasure = erasure_coding::Erasure::new(4, 2, 32); + let engine = build_get_codec_streaming_decode_engine(erasure).expect("engine should be built"); + + assert!(matches!(engine, CodecStreamingDecodeEngine::Rustfs(_))); + }); + } + #[test] fn codec_streaming_fallback_metric_labels_are_stable() { assert_eq!(GetCodecStreamingFallbackReason::Disabled.as_str(), "disabled");