From 85e9457fccb7a192388a03551f72efbb6e747034 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Tue, 21 Jul 2026 06:28:29 +0200 Subject: [PATCH] refactor(compress): move stable stream init policy to Rust ZSTD_compressStream2_c still owns private context mutation, diagnostics, buffer updates, and codec initialization, but its stable-input transparent-initialization decision was a remaining scalar policy island in C. Move the continuation validation, block-size threshold, format-specific progress hint, and wrapping size arithmetic behind a Rust projection. Preserve the original C error messages and return values while keeping all private context state on the C side.\n\nThe bridge uses explicit ABI layout assertions for the mixed pointer/size projection. Focused Rust tests cover buffered and non-continue calls, block-boundary initialization, empty and short stable input, both frame formats, matching continuation state, invalid source/position, and null bridge input.\n\nTest Plan:\n- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check\n- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings\n- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings\n- Full capped native and original-test verification follows after the batch is committed. --- lib/compress/zstd_compress.c | 84 +++++++-- rust/src/zstd_compress.rs | 328 +++++++++++++++++++++++++++++++++++ 2 files changed, 401 insertions(+), 11 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 5a8d621ca..6b1fbb4e5 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -807,6 +807,51 @@ typedef char ZSTD_rust_compress_stream2_policy_state_layout[ (offsetof(ZSTD_rust_compressStream2PolicyState, endOp) == 0 && sizeof(ZSTD_rust_compressStream2PolicyState) == sizeof(int)) ? 1 : -1]; +typedef struct { + int inBufferMode; + int endOp; + size_t inputSize; + size_t stableInNotConsumed; + const void* inputSrc; + size_t inputPos; + const void* expectedInputSrc; + size_t expectedInputSize; + int format; +} ZSTD_rust_compressStream2InitPolicyState; +enum { + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE = 0, + /* These values are the exact progress hints returned by the C API. */ + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS = 2, + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD = 6, + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID = -1, + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID = -2 +}; +int ZSTD_rust_compressStream2InitPolicy( + const ZSTD_rust_compressStream2InitPolicyState* state); +typedef char ZSTD_rust_compress_stream2_init_policy_state_layout[ + (offsetof(ZSTD_rust_compressStream2InitPolicyState, inBufferMode) == 0 + && offsetof(ZSTD_rust_compressStream2InitPolicyState, endOp) + == sizeof(int) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, inputSize) + == 2 * sizeof(int) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, + stableInNotConsumed) + == 2 * sizeof(int) + sizeof(size_t) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, inputSrc) + == 2 * sizeof(int) + 2 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, inputPos) + == 2 * sizeof(int) + 3 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, + expectedInputSrc) + == 2 * sizeof(int) + 4 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, + expectedInputSize) + == 2 * sizeof(int) + 5 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStream2InitPolicyState, format) + == 2 * sizeof(int) + 6 * sizeof(size_t) + && sizeof(ZSTD_rust_compressStream2InitPolicyState) + == 2 * sizeof(int) + 7 * sizeof(size_t)) + ? 1 : -1]; int ZSTD_rust_simpleCompress2Level(const void* cctx, size_t srcSize); typedef int (*ZSTD_rust_simpleCompress2Level_f)(const void* cctx, size_t srcSize); typedef struct { @@ -8351,15 +8396,28 @@ size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx, if (cctx->streamStage == zcss_init) { size_t const inputSize = input->size - input->pos; /* no obligation to start from pos==0 */ - size_t const totalInputSize = inputSize + cctx->stableIn_notConsumed; - if ( (cctx->requestedParams.inBufferMode == ZSTD_bm_stable) /* input is presumed stable, across invocations */ - && (endOp == ZSTD_e_continue) /* no flush requested, more input to come */ - && (totalInputSize < ZSTD_BLOCKSIZE_MAX) ) { /* not even reached one block yet */ - if (cctx->stableIn_notConsumed) { /* not the first time */ - /* check stable source guarantees */ - RETURN_ERROR_IF(input->src != cctx->expectedInBuffer.src, stabilityCondition_notRespected, "stableInBuffer condition not respected: wrong src pointer"); - RETURN_ERROR_IF(input->pos != cctx->expectedInBuffer.size, stabilityCondition_notRespected, "stableInBuffer condition not respected: externally modified pos"); - } + ZSTD_rust_compressStream2InitPolicyState const state = { + (int)cctx->requestedParams.inBufferMode, + (int)endOp, + inputSize, + cctx->stableIn_notConsumed, + input->src, + input->pos, + cctx->expectedInBuffer.src, + cctx->expectedInBuffer.size, + (int)cctx->requestedParams.format + }; + int const initPolicy = ZSTD_rust_compressStream2InitPolicy(&state); + RETURN_ERROR_IF( + initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID, + stabilityCondition_notRespected, + "stableInBuffer condition not respected: wrong src pointer"); + RETURN_ERROR_IF( + initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID, + stabilityCondition_notRespected, + "stableInBuffer condition not respected: externally modified pos"); + if (initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS + || initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD) { /* pretend input was consumed, to give a sense forward progress */ input->pos = input->size; /* save stable inBuffer, for later control, and flush/end */ @@ -8367,9 +8425,13 @@ size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx, /* but actually input wasn't consumed, so keep track of position from where compression shall resume */ cctx->stableIn_notConsumed += inputSize; /* don't initialize yet, wait for the first block of flush() order, for better parameters adaptation */ - return ZSTD_FRAMEHEADERSIZE_MIN(cctx->requestedParams.format); /* at least some header to produce */ + return (size_t)initPolicy; /* at least some header to produce */ } - FORWARD_IF_ERROR(ZSTD_CCtx_init_compressStream2(cctx, endOp, totalInputSize), "compressStream2 initialization failed"); + FORWARD_IF_ERROR( + ZSTD_CCtx_init_compressStream2( + cctx, endOp, + inputSize + cctx->stableIn_notConsumed), + "compressStream2 initialization failed"); ZSTD_setBufferExpectations(cctx, output, input); /* Set initial buffer expectations now that we've initialized */ } /* end of transparent initialization stage */ diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 015ea748b..82e83b61c 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -163,6 +163,7 @@ const ZSTD_CHUNKSIZE_MAX: usize = u32::MAX as usize - ZSTD_CURRENT_MAX; const ZSTD_E_END: c_int = 2; const ZSTD_E_CONTINUE: c_int = 0; const ZSTD_E_FLUSH: c_int = 1; +const ZSTD_F_ZSTD1: c_int = 0; const ZSTD_RUST_COMPRESS_STREAM2_BUFFER_VALID: c_int = 0; const ZSTD_RUST_COMPRESS_STREAM2_BUFFER_OUTPUT_INVALID: c_int = 1; @@ -256,6 +257,110 @@ pub unsafe extern "C" fn ZSTD_rust_compressStream2Policy( compress_stream2_policy(state.end_op) } +/// Projection for the stable-input part of `ZSTD_compressStream2_c`'s +/// transparent initialization stage. +/// +/// Rust owns the scalar decision, including the stable-buffer continuation +/// checks and the exact progress hint. C retains the input-buffer mutation, +/// error reporting, context initialization, and all private context state. +#[repr(C)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ZSTD_rust_compressStream2InitPolicyState { + in_buffer_mode: c_int, + end_op: c_int, + input_size: usize, + stable_in_not_consumed: usize, + input_src: *const c_void, + input_pos: usize, + expected_input_src: *const c_void, + expected_input_size: usize, + format: c_int, +} + +const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE: c_int = 0; +/* These values are the exact progress hints returned by the C API. */ +const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS: c_int = 2; +const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD: c_int = 6; +const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID: c_int = -1; +const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID: c_int = -2; + +const _: () = { + assert!(offset_of!(ZSTD_rust_compressStream2InitPolicyState, in_buffer_mode) == 0); + assert!(offset_of!(ZSTD_rust_compressStream2InitPolicyState, end_op) == size_of::()); + assert!( + offset_of!(ZSTD_rust_compressStream2InitPolicyState, input_size) == 2 * size_of::() + ); + assert!( + offset_of!( + ZSTD_rust_compressStream2InitPolicyState, + stable_in_not_consumed + ) == 2 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2InitPolicyState, input_src) + == 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2InitPolicyState, input_pos) + == 2 * size_of::() + 3 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2InitPolicyState, expected_input_src) + == 2 * size_of::() + 4 * size_of::() + ); + assert!( + offset_of!( + ZSTD_rust_compressStream2InitPolicyState, + expected_input_size + ) == 2 * size_of::() + 5 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2InitPolicyState, format) + == 2 * size_of::() + 6 * size_of::() + ); + assert!( + size_of::() + == 2 * size_of::() + 7 * size_of::() + ); +}; + +#[inline] +fn compress_stream2_init_policy(state: &ZSTD_rust_compressStream2InitPolicyState) -> c_int { + let total_input_size = state.input_size.wrapping_add(state.stable_in_not_consumed); + if state.in_buffer_mode != ZSTD_BM_STABLE + || state.end_op != ZSTD_E_CONTINUE + || total_input_size >= ZSTD_BLOCKSIZE_MAX + { + return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE; + } + + if state.stable_in_not_consumed != 0 { + if state.input_src != state.expected_input_src { + return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID; + } + if state.input_pos != state.expected_input_size { + return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID; + } + } + + if state.format == ZSTD_F_ZSTD1 { + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD + } else { + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS + } +} + +/// Classify whether transparent stream initialization must be deferred. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_compressStream2InitPolicy( + state: *const ZSTD_rust_compressStream2InitPolicyState, +) -> c_int { + let Some(state) = (unsafe { state.as_ref() }) else { + return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE; + }; + compress_stream2_init_policy(state) +} + const ZSTD_C_WINDOW_LOG: c_int = 101; const ZSTD_C_HASH_LOG: c_int = 102; const ZSTD_C_CHAIN_LOG: c_int = 103; @@ -10413,6 +10518,7 @@ mod tests { const ZSTD_BTULTRA: c_int = 8; const ZSTD_BTOPT: c_int = 7; const ZSTD_BTULTRA2: c_int = 9; + const ZSTD_F_ZSTD1_MAGICLESS: c_int = 1; struct EstimateCCtxSizeTestContext { compression_levels: Vec, @@ -14220,6 +14326,228 @@ mod tests { assert_eq!(unsafe { ZSTD_rust_compressStream2Policy(ptr::null()) }, 0); } + fn compress_stream2_init_policy_state( + in_buffer_mode: c_int, + end_op: c_int, + input_size: usize, + stable_in_not_consumed: usize, + input_src: *const c_void, + input_pos: usize, + expected_input_src: *const c_void, + expected_input_size: usize, + format: c_int, + ) -> ZSTD_rust_compressStream2InitPolicyState { + ZSTD_rust_compressStream2InitPolicyState { + in_buffer_mode, + end_op, + input_size, + stable_in_not_consumed, + input_src, + input_pos, + expected_input_src, + expected_input_size, + format, + } + } + + #[test] + fn compress_stream2_init_policy_initializes_for_non_stable_or_flush_calls() { + let source = [0u8; 1]; + for (in_buffer_mode, end_op) in [ + (ZSTD_BM_BUFFERED, ZSTD_E_CONTINUE), + (ZSTD_BM_STABLE, ZSTD_E_FLUSH), + (ZSTD_BM_STABLE, ZSTD_E_END), + ] { + let state = compress_stream2_init_policy_state( + in_buffer_mode, + end_op, + 1, + 0, + source.as_ptr().cast(), + 0, + ptr::null(), + 0, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE + ); + } + } + + #[test] + fn compress_stream2_init_policy_initializes_at_block_boundary() { + let source = [0u8; 1]; + for (input_size, stable_in_not_consumed) in + [(ZSTD_BLOCKSIZE_MAX, 0), (ZSTD_BLOCKSIZE_MAX - 1, 1)] + { + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + input_size, + stable_in_not_consumed, + source.as_ptr().cast(), + 0, + ptr::null(), + 0, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE + ); + } + } + + #[test] + fn compress_stream2_init_policy_returns_zstd_progress_hint_for_first_short_stable_input() { + let source = [0u8; 1]; + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 1, + 0, + source.as_ptr().cast(), + 0, + ptr::null(), + 0, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD + ); + } + + #[test] + fn compress_stream2_init_policy_defers_an_empty_stable_input() { + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 0, + 0, + ptr::null(), + 0, + ptr::null(), + 0, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD + ); + } + + #[test] + fn compress_stream2_init_policy_accepts_matching_stable_continuation() { + let source = [0u8; 1]; + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 3, + 4, + source.as_ptr().cast(), + 8, + source.as_ptr().cast(), + 8, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD + ); + } + + #[test] + fn compress_stream2_init_policy_returns_magicless_progress_hint() { + let source = [0u8; 1]; + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 1, + 0, + source.as_ptr().cast(), + 0, + ptr::null(), + 0, + ZSTD_F_ZSTD1_MAGICLESS, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS + ); + } + + #[test] + fn compress_stream2_init_policy_checks_source_before_position() { + let source = [0u8; 1]; + let other_source = [0u8; 1]; + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 1, + 1, + source.as_ptr().cast(), + 7, + other_source.as_ptr().cast(), + 8, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID + ); + } + + #[test] + fn compress_stream2_init_policy_rejects_modified_stable_position() { + let source = [0u8; 1]; + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 1, + 1, + source.as_ptr().cast(), + 7, + source.as_ptr().cast(), + 8, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID + ); + } + + #[test] + fn compress_stream2_init_policy_wraps_total_input_like_c() { + let source = [0u8; 1]; + let state = compress_stream2_init_policy_state( + ZSTD_BM_STABLE, + ZSTD_E_CONTINUE, + 1, + usize::MAX, + source.as_ptr().cast(), + 0, + source.as_ptr().cast(), + 0, + ZSTD_F_ZSTD1, + ); + assert_eq!( + compress_stream2_init_policy(&state), + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD + ); + } + + #[test] + fn compress_stream2_init_policy_null_state_is_conservative() { + assert_eq!( + unsafe { ZSTD_rust_compressStream2InitPolicy(ptr::null()) }, + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE + ); + } + #[test] fn reduce_table_applies_threshold_and_wrapping_subtraction() { let mut table = [0, 1, 2, 3, 4, 5, 6, u32::MAX, 0, 0, 0, 0, 0, 0, 0, 0];