feat(compress): move resetCStream policy into Rust

Move the public CStream reset wrapper's pledge normalization, callback order, and error short-circuiting into Rust. Keep the C-owned context reset and pledged-size mutation behind opaque callbacks with an explicit ABI projection. Test Plan: - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml reset_cstream -- --test-threads=1 - ulimit -v 41943040; make -B -C programs -j1 zstd - git diff --cached --check
2026-07-19 11:46:18 +02:00
parent df6ecd0e2f
commit fe6cf3bccc
2 changed files with 175 additions and 9 deletions
@@ -51,6 +51,23 @@ int ZSTD_rust_updateFrameProgression(unsigned long long* consumedSrcSize,
 size_t ZSTD_rust_setPledgedSrcSize(int streamStage,
                                   unsigned long long pledgedSrcSize,
                                   unsigned long long* pledgedSrcSizePlusOne);
+typedef size_t (*ZSTD_rust_resetCStreamReset_f)(void* context);
+typedef size_t (*ZSTD_rust_resetCStreamSetPledgedSrcSize_f)(
+        void* context, unsigned long long pledgedSrcSize);
+typedef struct {
+    void* callbackContext;
+    ZSTD_rust_resetCStreamReset_f resetSession;
+    ZSTD_rust_resetCStreamSetPledgedSrcSize_f setPledgedSrcSize;
+} ZSTD_rust_resetCStreamState;
+size_t ZSTD_rust_resetCStream(const ZSTD_rust_resetCStreamState* state,
+                              unsigned long long pss);
+typedef char ZSTD_rust_reset_cstream_state_layout[
+        (offsetof(ZSTD_rust_resetCStreamState, callbackContext) == 0
+         && offsetof(ZSTD_rust_resetCStreamState, resetSession) == sizeof(void*)
+         && offsetof(ZSTD_rust_resetCStreamState, setPledgedSrcSize)
+                == 2 * sizeof(void*)
+         && sizeof(ZSTD_rust_resetCStreamState) == 3 * sizeof(void*))
+        ? 1 : -1];
 ZSTD_frameProgression ZSTD_rust_frameProgression(U64 consumedSrcSize,
                                                  size_t buffered,
                                                  U64 producedCSize);
@@ -4770,17 +4787,24 @@ static ZSTD_CParamMode_e ZSTD_getCParamMode(ZSTD_CDict const* cdict, ZSTD_CCtx_p

 /* ZSTD_resetCStream():
 * pledgedSrcSize == 0 means "unknown" */
+static size_t ZSTD_rust_resetCStream_resetSession(void* context)
+{
+    return ZSTD_CCtx_reset((ZSTD_CCtx*)context, ZSTD_reset_session_only);
+}
+
+static size_t ZSTD_rust_resetCStream_setPledgedSrcSize(
+        void* context, unsigned long long pledgedSrcSize)
+{
+    return ZSTD_CCtx_setPledgedSrcSize((ZSTD_CCtx*)context, pledgedSrcSize);
+}
+
 size_t ZSTD_resetCStream(ZSTD_CStream* zcs, unsigned long long pss)
 {
-    /* temporary : 0 interpreted as "unknown" during transition period.
-     * Users willing to specify "unknown" **must** use ZSTD_CONTENTSIZE_UNKNOWN.
-     * 0 will be interpreted as "empty" in the future.
-     */
-    U64 const pledgedSrcSize = (pss==0) ? ZSTD_CONTENTSIZE_UNKNOWN : pss;
-    DEBUGLOG(4, "ZSTD_resetCStream: pledgedSrcSize = %u", (unsigned)pledgedSrcSize);
-    FORWARD_IF_ERROR( ZSTD_CCtx_reset(zcs, ZSTD_reset_session_only) , "");
-    FORWARD_IF_ERROR( ZSTD_CCtx_setPledgedSrcSize(zcs, pledgedSrcSize) , "");
-    return 0;
+    ZSTD_rust_resetCStreamState state;
+    state.callbackContext = zcs;
+    state.resetSession = ZSTD_rust_resetCStream_resetSession;
+    state.setPledgedSrcSize = ZSTD_rust_resetCStream_setPledgedSrcSize;
+    return ZSTD_rust_resetCStream(&state, pss);
 }

 /*! ZSTD_initCStream_internal() :
@@ -803,6 +803,60 @@ pub unsafe extern "C" fn ZSTD_rust_compress2(
    unsafe { compress2_body_with(&*state, dst, dst_capacity, src, src_size) }
 }

+type ResetCStreamResetFn = unsafe extern "C" fn(*mut c_void) -> usize;
+type ResetCStreamSetPledgedSrcSizeFn = unsafe extern "C" fn(*mut c_void, u64) -> usize;
+
+/// Explicit projection for the public `ZSTD_resetCStream` wrapper.
+///
+/// Rust owns the zero-to-unknown conversion, callback ordering, and error
+/// propagation. The opaque callback context stays in C, where the callbacks
+/// retain access to the private `ZSTD_CCtx` layout.
+#[repr(C)]
+pub struct ZSTD_rust_resetCStreamState {
+    callback_context: *mut c_void,
+    reset_session: ResetCStreamResetFn,
+    set_pledged_src_size: ResetCStreamSetPledgedSrcSizeFn,
+}
+
+const _: () = {
+    assert!(size_of::<ResetCStreamResetFn>() == size_of::<usize>());
+    assert!(size_of::<ResetCStreamSetPledgedSrcSizeFn>() == size_of::<usize>());
+    assert!(offset_of!(ZSTD_rust_resetCStreamState, callback_context) == 0);
+    assert!(offset_of!(ZSTD_rust_resetCStreamState, reset_session) == size_of::<usize>());
+    assert!(
+        offset_of!(ZSTD_rust_resetCStreamState, set_pledged_src_size) == 2 * size_of::<usize>()
+    );
+    assert!(size_of::<ZSTD_rust_resetCStreamState>() == 3 * size_of::<usize>());
+};
+
+/// Drive the public `ZSTD_resetCStream` policy through C-owned callbacks.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTD_rust_resetCStream(
+    state: *const ZSTD_rust_resetCStreamState,
+    pss: u64,
+) -> usize {
+    if state.is_null() {
+        return ERROR(ZstdErrorCode::Generic);
+    }
+    let state = unsafe { &*state };
+    let pledged_src_size = if pss == 0 {
+        ZSTD_CONTENTSIZE_UNKNOWN
+    } else {
+        pss
+    };
+
+    let reset_result = unsafe { (state.reset_session)(state.callback_context) };
+    if ERR_isError(reset_result) {
+        return reset_result;
+    }
+    let pledged_size_result =
+        unsafe { (state.set_pledged_src_size)(state.callback_context, pledged_src_size) };
+    if ERR_isError(pledged_size_result) {
+        return pledged_size_result;
+    }
+    0
+}
+
 type CompressStreamBlockFn =
    unsafe extern "C" fn(*mut c_void, *mut c_void, usize, *const c_void, usize) -> usize;
 type CompressStreamResetFn = unsafe extern "C" fn(*mut c_void) -> usize;
@@ -8559,6 +8613,94 @@ mod tests {
        assert_eq!(dict_limit, 0x89ab_cdef);
    }

+    #[derive(Default)]
+    struct ResetCStreamTestContext {
+        events: Vec<&'static str>,
+        reset_result: usize,
+        pledged_result: usize,
+        pledged_src_size: u64,
+    }
+
+    unsafe extern "C" fn reset_cstream_test_reset(context: *mut c_void) -> usize {
+        let context = unsafe { &mut *context.cast::<ResetCStreamTestContext>() };
+        context.events.push("reset");
+        context.reset_result
+    }
+
+    unsafe extern "C" fn reset_cstream_test_set_pledged_src_size(
+        context: *mut c_void,
+        pledged_src_size: u64,
+    ) -> usize {
+        let context = unsafe { &mut *context.cast::<ResetCStreamTestContext>() };
+        context.events.push("pledged");
+        context.pledged_src_size = pledged_src_size;
+        context.pledged_result
+    }
+
+    fn reset_cstream_test_state(
+        context: &mut ResetCStreamTestContext,
+    ) -> ZSTD_rust_resetCStreamState {
+        ZSTD_rust_resetCStreamState {
+            callback_context: (context as *mut ResetCStreamTestContext).cast(),
+            reset_session: reset_cstream_test_reset,
+            set_pledged_src_size: reset_cstream_test_set_pledged_src_size,
+        }
+    }
+
+    #[test]
+    fn reset_cstream_converts_zero_pledge_to_unknown_after_reset() {
+        let mut context = ResetCStreamTestContext::default();
+        let state = reset_cstream_test_state(&mut context);
+
+        let result = unsafe { ZSTD_rust_resetCStream(&state, 0) };
+
+        assert_eq!(result, 0);
+        assert_eq!(context.events, ["reset", "pledged"]);
+        assert_eq!(context.pledged_src_size, ZSTD_CONTENTSIZE_UNKNOWN);
+    }
+
+    #[test]
+    fn reset_cstream_forwards_nonzero_pledge() {
+        let mut context = ResetCStreamTestContext::default();
+        let state = reset_cstream_test_state(&mut context);
+
+        let result = unsafe { ZSTD_rust_resetCStream(&state, 123) };
+
+        assert_eq!(result, 0);
+        assert_eq!(context.events, ["reset", "pledged"]);
+        assert_eq!(context.pledged_src_size, 123);
+    }
+
+    #[test]
+    fn reset_cstream_stops_before_pledge_after_reset_error() {
+        let mut context = ResetCStreamTestContext {
+            reset_result: ERROR(ZstdErrorCode::MemoryAllocation),
+            ..ResetCStreamTestContext::default()
+        };
+        let state = reset_cstream_test_state(&mut context);
+
+        let result = unsafe { ZSTD_rust_resetCStream(&state, 123) };
+
+        assert_eq!(result, ERROR(ZstdErrorCode::MemoryAllocation));
+        assert_eq!(context.events, ["reset"]);
+        assert_eq!(context.pledged_src_size, 0);
+    }
+
+    #[test]
+    fn reset_cstream_propagates_pledge_error_after_reset() {
+        let mut context = ResetCStreamTestContext {
+            pledged_result: ERROR(ZstdErrorCode::StageWrong),
+            ..ResetCStreamTestContext::default()
+        };
+        let state = reset_cstream_test_state(&mut context);
+
+        let result = unsafe { ZSTD_rust_resetCStream(&state, 123) };
+
+        assert_eq!(result, ERROR(ZstdErrorCode::StageWrong));
+        assert_eq!(context.events, ["reset", "pledged"]);
+        assert_eq!(context.pledged_src_size, 123);
+    }
+
    #[test]
    fn pledged_src_size_writes_the_init_stage_value_plus_one() {
        let mut pledged_src_size_plus_one = 0;