feat(get): add rustfs codec decode engine (#3928)

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<Vec<u8>>], 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<Vec<u8>>],
+    data_shards: usize,
+    parity_shards: usize,
+) -> io::Result<bool> {
+    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<ReedSolomon>,
+}
+
+impl RustfsCodecDecodeEngine {
+    pub(crate) fn new(erasure: &Erasure) -> io::Result<Self> {
+        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<Self::Workspace> {
+        Ok(RustfsCodecDecodeWorkspace::new(shard_len))
+    }
+
+    fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], _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<LegacyEcDecodeEngine>),
+    Rustfs(Box<RustfsCodecDecodeEngine>),
+}
+
+impl CodecStreamingDecodeEngine {
+    pub(crate) fn legacy(erasure: Erasure) -> Self {
+        Self::Legacy(Box::new(LegacyEcDecodeEngine::new(erasure)))
+    }
+
+    pub(crate) fn rustfs(erasure: &Erasure) -> io::Result<Self> {
+        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<Self::Workspace> {
+        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<Vec<u8>>], 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<Option<Vec<u8>>> {
+        erasure
+            .encode_data(data)
+            .expect("test stripe should encode")
+            .into_iter()
+            .map(|shard| Some(shard.to_vec()))
+            .collect()
+    }
+
+    fn reconstruct_with<E>(engine: &E, shards: &mut [Option<Vec<u8>>]) -> 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::<Vec<_>>();
+        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::<Vec<_>>(), 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));
+    }
 }

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<Option<Vec<u8>>>,

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<Vec<u8>> {
-        let source = source_from_data(&erasure, data, missing_indexes);
-        let engine = LegacyEcDecodeEngine::new(erasure);
+    async fn decode_all_with_engine<E>(
+        erasure: &Erasure,
+        engine: E,
+        data: &[u8],
+        missing_indexes: &[usize],
+    ) -> io::Result<Vec<u8>>
+    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<Vec<u8>> {
+        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);

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<bool> = 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<CodecStreamingDecodeEngine> {
+    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,

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");