feat(compress): route complete stream frames through Rust

Route ordinary complete-input ZSTD_compressStream2 calls through the Rust frame compressor while retaining the C state machine for partial, advanced, and dictionary-backed streams. Track explicit maxBlockSize requests so the migrated path preserves the original byte-identical default behavior, and extend the C archive smoke test across context reuse and fallback cases.

Test Plan:

- cargo clippy --manifest-path rust/Cargo.toml

- cargo clippy --manifest-path rust/Cargo.toml --benches

- cargo clippy --manifest-path rust/Cargo.toml --tests

- cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression (133 passed)

- make -B -C tests -j2 test-rust-lib-smoke and ./tests/rustLibSmoke

- make -B -C tests -j2 test-zstream (passed, including both APIs and maxBlockSize case 78)

lib/compress/zstd_compress.c

@@ -46,7 +46,12 @@ void ZSTD_rust_markSimpleCompression2Complete(void* cctx);
 size_t ZSTD_compress2_c(ZSTD_CCtx* cctx,
                         void* dst, size_t dstCapacity,
                         const void* src, size_t srcSize);
+size_t ZSTD_compressStream2_c(ZSTD_CCtx* cctx,
+                              ZSTD_outBuffer* output,
+                              ZSTD_inBuffer* input,
+                              ZSTD_EndDirective endOp);
 int ZSTD_rust_simpleCompress2Level(const void* cctx);
+int ZSTD_rust_simpleCompressStream2Level(const void* cctx);
 
 /* Context-free compression-parameter selection and sizing leaves live in
  * Rust (rust/src/zstd_compress_params.rs).  This file retains
@@ -949,7 +954,12 @@ size_t ZSTD_CCtx_setParameter(ZSTD_CCtx* cctx, ZSTD_cParameter param, int value)
 
     default: RETURN_ERROR(parameter_unsupported, "unknown parameter");
     }
-    return ZSTD_CCtxParams_setParameter(&cctx->requestedParams, param, value);
+    {   size_t const result = ZSTD_CCtxParams_setParameter(&cctx->requestedParams, param, value);
+        if (!ZSTD_isError(result) && param == ZSTD_c_maxBlockSize) {
+            cctx->rustSimpleCompress2MaxBlockSizeSet = 1;
+        }
+        return result;
+    }
 }
 
 size_t ZSTD_CCtx_getParameter(ZSTD_CCtx const* cctx, ZSTD_cParameter param, int* value)
@@ -1166,6 +1176,7 @@ size_t ZSTD_CCtx_reset(ZSTD_CCtx* cctx, ZSTD_ResetDirective reset)
         RETURN_ERROR_IF(cctx->streamStage != zcss_init, stage_wrong,
                         "Reset parameters is only possible during init stage.");
         ZSTD_clearAllDicts(cctx);
+        cctx->rustSimpleCompress2MaxBlockSizeSet = 0;
         return ZSTD_CCtxParams_reset(&cctx->requestedParams);
     }
     return 0;
@@ -4532,7 +4543,7 @@ int ZSTD_rust_simpleCompress2Level(const void* opaqueCctx)
         || params->targetCBlockSize != 0
         || params->srcSizeHint != 0
         || params->attachDictPref != ZSTD_dictDefaultAttach
-        || params->literalCompressionMode != ZSTD_lcm_auto
+        || params->literalCompressionMode != (ZSTD_ParamSwitch_e)ZSTD_lcm_auto
         || params->nbWorkers != 0
         || params->jobSize != 0
         || params->overlapLog != 0
@@ -4553,6 +4564,7 @@ int ZSTD_rust_simpleCompress2Level(const void* opaqueCctx)
         || params->postBlockSplitter != ZSTD_ps_auto
         || params->preBlockSplitter_level != 0
         || params->maxBlockSize != 0
+        || cctx->rustSimpleCompress2MaxBlockSizeSet != 0
         || params->useRowMatchFinder != ZSTD_ps_auto
         || params->deterministicRefPrefix != 0
         || params->prefetchCDictTables != ZSTD_ps_auto
@@ -4570,6 +4582,22 @@ int ZSTD_rust_simpleCompress2Level(const void* opaqueCctx)
     return params->compressionLevel;
 }
 
+/* Return the simple level only while a new stream frame can be initialized.
+ * Rust handles this complete-input case; all other stream states stay on the
+ * original implementation so partial output and advanced buffering remain
+ * governed by the C state machine. */
+int ZSTD_rust_simpleCompressStream2Level(const void* opaqueCctx)
+{
+    ZSTD_CCtx const* const cctx = (ZSTD_CCtx const*)opaqueCctx;
+    if (cctx == NULL
+        || cctx->streamStage != zcss_init
+        || cctx->pledgedSrcSizePlusOne != 0
+        || cctx->rustSimpleCompress2Completed != 0) {
+        return (-2147483647 - 1);
+    }
+    return ZSTD_rust_simpleCompress2Level(opaqueCctx);
+}
+
 /* ZSTD_compress() is implemented by rust/src/zstd_compress.rs. */
 
 
@@ -5503,10 +5531,10 @@ static size_t ZSTD_CCtx_init_compressStream2(ZSTD_CCtx* cctx,
 
 /* @return provides a minimum amount of data remaining to be flushed from internal buffers
  */
-size_t ZSTD_compressStream2( ZSTD_CCtx* cctx,
-                             ZSTD_outBuffer* output,
-                             ZSTD_inBuffer* input,
-                             ZSTD_EndDirective endOp)
+size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
+                               ZSTD_outBuffer* output,
+                               ZSTD_inBuffer* input,
+                               ZSTD_EndDirective endOp)
 {
     DEBUGLOG(5, "ZSTD_compressStream2, endOp=%u ", (unsigned)endOp);
     /* check conditions */

lib/compress/zstd_compress_internal.h

@@ -518,6 +518,10 @@ struct ZSTD_CCtx_s {
     U32    frameEnded;
     /* The next one-shot stream call may reuse a Rust-compressed frame. */
     unsigned rustSimpleCompress2Completed;
+    /* A caller may explicitly set maxBlockSize to zero.  Keep that request
+     * on the C path so it remains byte-identical to the explicit default
+     * maxBlockSize value. */
+    unsigned rustSimpleCompress2MaxBlockSizeSet;
 
     /* Stable in/out buffer verification */
     ZSTD_inBuffer expectedInBuffer;

rust/src/zstd_compress.rs

@@ -8,14 +8,15 @@
 //! The layout-independent one-shot entry point in this module drives the
 //! already migrated compression leaves.  `ZSTD_compressCCtx` uses the same
 //! path after a narrow C-owned context reset; the public context lifecycle
-//! and streaming APIs remain C-owned because they share the
-//! configuration-dependent private `ZSTD_CCtx_s` layout.  `ZSTD_compress2`
-//! dispatches simple, dictionary-free contexts through the same Rust frame
-//! path and retains the C implementation for advanced contexts.  A following
-//! one-shot `ZSTD_compressStream2(..., ZSTD_e_end)` call can reuse that path.
+//! remains C-owned because it shares the configuration-dependent private
+//! `ZSTD_CCtx_s` layout.  `ZSTD_compress2` and the complete-input simple
+//! `ZSTD_compressStream2(..., ZSTD_e_end)` path dispatch through Rust while
+//! retaining the C implementation for advanced and partial-stream cases.
 
 use crate::errors::{ERR_isError, ZstdErrorCode, ERROR};
 use crate::mem::MEM_writeLE24;
+#[cfg(not(test))]
+use crate::zstd_compress_api::ZSTD_compressBound;
 use crate::zstd_compress_frame::{ZSTD_rust_writeFrameHeader, ZSTD_writeLastEmptyBlock};
 use crate::zstd_compress_params::{
     ZSTD_rust_params_adjustCParams, ZSTD_rust_params_maxNbSeq, ZSTD_rust_params_selectCParams,
@@ -40,6 +41,7 @@ unsafe extern "C" {
     fn ZSTD_rust_resetCCtxForSimpleCompressionSession(cctx: *mut c_void) -> usize;
     fn ZSTD_rust_markSimpleCompression2Complete(cctx: *mut c_void);
     fn ZSTD_rust_simpleCompress2Level(cctx: *const c_void) -> c_int;
+    fn ZSTD_rust_simpleCompressStream2Level(cctx: *const c_void) -> c_int;
     fn ZSTD_compress2_c(
         cctx: *mut c_void,
         dst: *mut c_void,
@@ -47,12 +49,36 @@ unsafe extern "C" {
         src: *const c_void,
         src_size: usize,
     ) -> usize;
+    fn ZSTD_compressStream2_c(
+        cctx: *mut c_void,
+        output: *mut ZSTD_outBuffer,
+        input: *mut ZSTD_inBuffer,
+        end_op: c_int,
+    ) -> usize;
 }
 
 const ZSTD_FAST: c_int = 1;
 const ZSTD_DFAST: c_int = 2;
 const ZSTD_BLOCKSIZE_MAX: usize = 1 << 17;
 const ZSTD_CONTENTSIZE_UNKNOWN: u64 = u64::MAX;
+#[cfg(not(test))]
+const ZSTD_E_END: c_int = 2;
+
+#[cfg(not(test))]
+#[repr(C)]
+pub struct ZSTD_inBuffer {
+    src: *const c_void,
+    size: usize,
+    pos: usize,
+}
+
+#[cfg(not(test))]
+#[repr(C)]
+pub struct ZSTD_outBuffer {
+    dst: *mut c_void,
+    size: usize,
+    pos: usize,
+}
 
 /* HUF_WORKSPACE_SIZE + (MaxSeq + 2) * sizeof(unsigned), rounded up.  The
  * superblock leaf also accepts the larger pre-split workspace, so a fixed
@@ -433,6 +459,70 @@ pub unsafe extern "C" fn ZSTD_compress2(
     }
 }
 
+/// Compress a streaming call when the complete input and a full output bound
+/// are already available for the ordinary context configuration.
+///
+/// The C implementation remains the fallback for partial-output streaming,
+/// `ZSTD_e_continue`/`ZSTD_e_flush`, dictionaries, and every advanced context
+/// configuration.  The Rust path resets the C-owned session after emitting a
+/// complete frame so the same context can immediately start another frame.
+#[cfg(not(test))]
+#[no_mangle]
+pub unsafe extern "C" fn ZSTD_compressStream2(
+    cctx: *mut c_void,
+    output: *mut ZSTD_outBuffer,
+    input: *mut ZSTD_inBuffer,
+    end_op: c_int,
+) -> usize {
+    if cctx.is_null() || output.is_null() || input.is_null() {
+        return ERROR(ZstdErrorCode::Generic);
+    }
+
+    let output_ref = unsafe { &mut *output };
+    let input_ref = unsafe { &mut *input };
+    if output_ref.pos > output_ref.size {
+        return ERROR(ZstdErrorCode::DstSizeTooSmall);
+    }
+    if input_ref.pos > input_ref.size {
+        return ERROR(ZstdErrorCode::SrcSizeWrong);
+    }
+
+    if end_op == ZSTD_E_END {
+        let level = unsafe { ZSTD_rust_simpleCompressStream2Level(cctx.cast_const()) };
+        if level != c_int::MIN {
+            let src_size = input_ref.size - input_ref.pos;
+            let dst_capacity = output_ref.size - output_ref.pos;
+            let bound = ZSTD_compressBound(src_size);
+            if !ERR_isError(bound) && dst_capacity >= bound {
+                if dst_capacity != 0 && output_ref.dst.is_null() {
+                    return ERROR(ZstdErrorCode::DstBufferNull);
+                }
+                if src_size != 0 && input_ref.src.is_null() {
+                    return ERROR(ZstdErrorCode::SrcSizeWrong);
+                }
+                let dst = if output_ref.dst.is_null() {
+                    ptr::null_mut()
+                } else {
+                    unsafe { output_ref.dst.cast::<u8>().add(output_ref.pos).cast() }
+                };
+                let src = if input_ref.src.is_null() {
+                    ptr::null()
+                } else {
+                    unsafe { input_ref.src.cast::<u8>().add(input_ref.pos).cast() }
+                };
+                let result = unsafe { compress_frame(dst, dst_capacity, src, src_size, level) };
+                if !ERR_isError(result) {
+                    output_ref.pos += result;
+                    input_ref.pos = input_ref.size;
+                    return unsafe { ZSTD_rust_resetCCtxForSimpleCompressionSession(cctx) };
+                }
+            }
+        }
+    }
+
+    unsafe { ZSTD_compressStream2_c(cctx, output, input, end_op) }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

tests/rustLibSmoke.c

@@ -24,6 +24,7 @@ int main(void)
     void* restored = malloc(sourceSize);
     size_t compressedSize;
     size_t restoredSize;
+    ZSTD_CCtx* cctx = NULL;
     ZSTD_DDict* ddict;
     size_t index;
 
@@ -48,6 +49,103 @@ int main(void)
         goto cleanup;
     }
 
+    cctx = ZSTD_createCCtx();
+    if (cctx == NULL) {
+        fputs("CCtx creation failed\n", stderr);
+        goto cleanup;
+    }
+    {
+        ZSTD_inBuffer input = { source, sourceSize, 0 };
+        ZSTD_outBuffer output = { compressed, compressedCapacity, 0 };
+        size_t const remaining = ZSTD_compressStream2(
+                cctx, &output, &input, ZSTD_e_end);
+        if (checkError(remaining, "ZSTD_compressStream2")
+            || remaining != 0
+            || input.pos != input.size) {
+            fputs("streaming end call did not complete\n", stderr);
+            goto cleanup;
+        }
+        restoredSize = ZSTD_decompress(
+                restored, sourceSize, compressed, output.pos);
+        if (checkError(restoredSize, "ZSTD_decompress(stream)")
+            || restoredSize != sourceSize
+            || memcmp(source, restored, sourceSize) != 0) {
+            fputs("streaming round trip mismatch\n", stderr);
+            goto cleanup;
+        }
+    }
+    /* The Rust end path resets the C-owned session before returning. */
+    {
+        ZSTD_inBuffer input = { source, sourceSize, 0 };
+        ZSTD_outBuffer output = { compressed, compressedCapacity, 0 };
+        size_t const remaining = ZSTD_compressStream2(
+                cctx, &output, &input, ZSTD_e_end);
+        if (checkError(remaining, "ZSTD_compressStream2(reuse)")
+            || remaining != 0
+            || input.pos != input.size) {
+            fputs("streaming context reuse failed\n", stderr);
+            goto cleanup;
+        }
+    }
+    if (ZSTD_isError(ZSTD_CCtx_setParameter(cctx, ZSTD_c_checksumFlag, 1))) {
+        fputs("advanced parameter setup failed\n", stderr);
+        goto cleanup;
+    }
+    {
+        ZSTD_inBuffer input = { source, sourceSize, 0 };
+        ZSTD_outBuffer output = { compressed, compressedCapacity, 0 };
+        size_t const remaining = ZSTD_compressStream2(
+                cctx, &output, &input, ZSTD_e_end);
+        if (checkError(remaining, "ZSTD_compressStream2(checksum)")
+            || remaining != 0
+            || input.pos != input.size) {
+            fputs("advanced stream fallback did not complete\n", stderr);
+            goto cleanup;
+        }
+        restoredSize = ZSTD_decompress(
+                restored, sourceSize, compressed, output.pos);
+        if (checkError(restoredSize, "ZSTD_decompress(checksum)")
+            || restoredSize != sourceSize
+            || memcmp(source, restored, sourceSize) != 0) {
+            fputs("advanced stream fallback mismatch\n", stderr);
+            goto cleanup;
+        }
+    }
+    if (ZSTD_CCtx_reset(cctx, ZSTD_reset_session_and_parameters) != 0) {
+        fputs("CCtx parameter reset failed\n", stderr);
+        goto cleanup;
+    }
+    if (ZSTD_CCtx_reset(cctx, ZSTD_reset_session_and_parameters) != 0) {
+        fputs("CCtx parameter reset failed\n", stderr);
+        goto cleanup;
+    }
+    {
+        ZSTD_inBuffer input = { source, sourceSize, 0 };
+        ZSTD_outBuffer output = { compressed, compressedCapacity - 1, 0 };
+        size_t const remaining = ZSTD_compressStream2(
+                cctx, &output, &input, ZSTD_e_end);
+        if (checkError(remaining, "ZSTD_compressStream2(short output)")
+            || remaining != 0
+            || input.pos != input.size) {
+            fputs("short-output stream fallback did not complete\n", stderr);
+            goto cleanup;
+        }
+        restoredSize = ZSTD_decompress(
+                restored, sourceSize, compressed, output.pos);
+        if (checkError(restoredSize, "ZSTD_decompress(short output)")
+            || restoredSize != sourceSize
+            || memcmp(source, restored, sourceSize) != 0) {
+            fputs("short-output stream fallback mismatch\n", stderr);
+            goto cleanup;
+        }
+    }
+    if (ZSTD_freeCCtx(cctx) != 0) {
+        fputs("CCtx cleanup failed\n", stderr);
+        cctx = NULL;
+        goto cleanup;
+    }
+    cctx = NULL;
+
     /* This must pull the Rust DDict object and resolve its C helper cycle. */
     ddict = ZSTD_createDDict(NULL, 0);
     if (ddict == NULL || ZSTD_freeDDict(ddict) != 0) {
@@ -60,6 +158,7 @@ int main(void)
     return 0;
 
 cleanup:
+    ZSTD_freeCCtx(cctx);
     free(restored);
     free(compressed);
     return 1;