feat(mt): move compression-job begin policy to Rust

The worker context previously selected the cdict or raw-prefix initialization
path, applied non-first-job parameter updates, and published the frame-header
projection directly in C. That left the branch and failure order intertwined
with private CCtx and parameter layouts.

Project only the first-job and cdict flags into Rust. Rust now validates cdict
placement, selects the initialization path, stops on force-window or
prefix-policy errors, and publishes the header projection only after successful
initialization. C callbacks retain the private cdict, CCtx, parameter mutation,
and frame-header field operations, including the original pledged-size and
force-window behavior.

Focused tests cover cdict ordering, non-first parameter ordering, parameter and
initialization failures, header-publication suppression, and invalid cdict
placement.

Test Plan:
- `rustfmt +nightly --edition 2021 --check rust/src/zstdmt_compress.rs` -- passed.
- Capped GCC syntax-only check for `zstdmt_compress.c` -- passed.
- Capped Clang syntax-only check for `zstdmt_compress.c` -- passed.
- `git diff --cached --check` -- passed.
- Cargo, native builds, fuzzers, and large tests were not run per assignment.

lib/compress/zstdmt_compress.c

@@ -222,6 +222,18 @@ typedef char ZSTDMT_compression_job_projection_layout[
                 == 2 * sizeof(unsigned) + sizeof(void*))
         ? 1 : -1];
 
+typedef struct {
+    unsigned firstJob;
+    unsigned hasCDict;
+} ZSTDMT_RustCompressionJobBeginProjection;
+typedef char ZSTDMT_compression_job_begin_projection_layout[
+        (offsetof(ZSTDMT_RustCompressionJobBeginProjection, firstJob) == 0
+         && offsetof(ZSTDMT_RustCompressionJobBeginProjection, hasCDict)
+                == sizeof(unsigned)
+         && sizeof(ZSTDMT_RustCompressionJobBeginProjection)
+                == 2 * sizeof(unsigned))
+        ? 1 : -1];
+
 typedef struct {
     int status;
     size_t toFlush;
@@ -235,6 +247,7 @@ typedef void (*ZSTDMT_compressionJobVoidFn)(void* opaque);
 typedef ZSTDMT_chunkProcessResult (*ZSTDMT_compressionJobCompressFn)(
         void* opaque, unsigned lastJob);
 typedef void (*ZSTDMT_compressionJobSizeFn)(void* opaque, size_t size);
+typedef size_t (*ZSTDMT_compressionJobSetParameterFn)(void* opaque, int value);
 
 typedef struct {
     void* callbackContext;
@@ -284,6 +297,15 @@ typedef char ZSTDMT_compression_job_parameters_layout[
 ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters(
         unsigned jobID, int checksumFlag, int ldmEnable, unsigned nbWorkers);
 
+size_t ZSTDMT_rust_compressionJobBegin(
+        const ZSTDMT_RustCompressionJobBeginProjection* projection,
+        void* opaque,
+        ZSTDMT_compressionJobStepFn beginWithCDict,
+        ZSTDMT_compressionJobSetParameterFn setForceMaxWindow,
+        ZSTDMT_compressionJobStepFn setDeterministicRefPrefix,
+        ZSTDMT_compressionJobStepFn beginWithPrefix,
+        ZSTDMT_compressionJobVoidFn publishFrameHeader);
+
 void ZSTDMT_rust_compressionJob(
         const ZSTDMT_RustCompressionJobProjection* projection,
         const ZSTDMT_RustCompressionJobFinishProjection* finishProjection,
@@ -1702,36 +1724,79 @@ static void ZSTDMT_compressionJobGenerateSequences(void* opaque)
             ZSTDMT_serialState_advance);
 }
 
+static size_t ZSTDMT_compressionJobBeginWithCDict(void* opaque);
+static size_t ZSTDMT_compressionJobSetForceMaxWindow(void* opaque, int value);
+static size_t ZSTDMT_compressionJobSetDeterministicRefPrefix(void* opaque);
+static size_t ZSTDMT_compressionJobBeginWithPrefix(void* opaque);
+static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque);
+
 static size_t ZSTDMT_compressionJobBegin(void* opaque)
 {
     ZSTDMT_compressionJobState* const state =
         (ZSTDMT_compressionJobState*)opaque;
     ZSTDMT_jobDescription* const job = state->job;
-    size_t initError;
 
-    if (job->cdict) {
+    ZSTDMT_RustCompressionJobBeginProjection const projection = {
-        initError = ZSTD_compressBegin_advanced_internal(
+        job->firstJob,
-            state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict,
+        (unsigned)(job->cdict != NULL)
-            &state->jobParams, job->fullFrameSize);
+    };
-        assert(job->firstJob);  /* only allowed for first job */
+    return ZSTDMT_rust_compressionJobBegin(
-    } else {
+            &projection, state,
-        U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size;
+            ZSTDMT_compressionJobBeginWithCDict,
-        size_t const forceWindowError = ZSTD_CCtxParams_setParameter(
+            ZSTDMT_compressionJobSetForceMaxWindow,
-                &state->jobParams, ZSTD_c_forceMaxWindow, !job->firstJob);
+            ZSTDMT_compressionJobSetDeterministicRefPrefix,
-        if (ZSTD_isError(forceWindowError)) return forceWindowError;
+            ZSTDMT_compressionJobBeginWithPrefix,
-        if (!job->firstJob) {
+            ZSTDMT_compressionJobPublishFrameHeader);
-            size_t const err = ZSTD_CCtxParams_setParameter(
+}
-                &state->jobParams, ZSTD_c_deterministicRefPrefix, 0);
+
-            if (ZSTD_isError(err)) return err;
+static size_t ZSTDMT_compressionJobBeginWithCDict(void* opaque)
-        }
+{
-        DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size);
+    ZSTDMT_compressionJobState* const state =
-        initError = ZSTD_compressBegin_advanced_internal(
+        (ZSTDMT_compressionJobState*)opaque;
-            state->cctx, job->prefix.start, job->prefix.size,
+    ZSTDMT_jobDescription* const job = state->job;
-            ZSTD_dct_rawContent, ZSTD_dtlm_fast, NULL, /*cdict*/
+
-            &state->jobParams, pledgedSrcSize);
+    return ZSTD_compressBegin_advanced_internal(
-    }
+        state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict,
+        &state->jobParams, job->fullFrameSize);
+}
+
+static size_t ZSTDMT_compressionJobSetForceMaxWindow(void* opaque, int value)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+
+    return ZSTD_CCtxParams_setParameter(
+            &state->jobParams, ZSTD_c_forceMaxWindow, value);
+}
+
+static size_t ZSTDMT_compressionJobSetDeterministicRefPrefix(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+
+    return ZSTD_CCtxParams_setParameter(
+            &state->jobParams, ZSTD_c_deterministicRefPrefix, 0);
+}
+
+static size_t ZSTDMT_compressionJobBeginWithPrefix(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    ZSTDMT_jobDescription* const job = state->job;
+    U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size;
+
+    DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size);
+    return ZSTD_compressBegin_advanced_internal(
+        state->cctx, job->prefix.start, job->prefix.size,
+        ZSTD_dct_rawContent, ZSTD_dtlm_fast, NULL, /*cdict*/
+        &state->jobParams, pledgedSrcSize);
+}
+
+static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
 
-    if (ZSTD_isError(initError)) return initError;
     state->frameHeaderState.stage = (int*)&state->cctx->stage;
     state->frameHeaderState.noDictIDFlag =
             state->cctx->appliedParams.fParams.noDictIDFlag;
@@ -1747,7 +1812,6 @@ static size_t ZSTDMT_compressionJobBegin(void* opaque)
     state->frameHeaderState.dictID = state->cctx->dictID;
     state->frameHeaderState.repCodes = state->cctx->blockState.prevCBlock == NULL
             ? NULL : state->cctx->blockState.prevCBlock->rep;
-    return initError;
 }
 
 static void ZSTDMT_compressionJobApplySequences(void* opaque)

rust/src/zstdmt_compress.rs

@@ -266,6 +266,107 @@ pub type ZSTDMT_compressionJobVoidFn = unsafe extern "C" fn(*mut c_void);
 pub type ZSTDMT_compressionJobCompressFn =
     unsafe extern "C" fn(*mut c_void, c_uint) -> ZSTDMT_chunkProcessResult;
 pub type ZSTDMT_compressionJobSizeFn = unsafe extern "C" fn(*mut c_void, usize);
+pub type ZSTDMT_compressionJobSetParameterFn = unsafe extern "C" fn(*mut c_void, c_int) -> usize;
+
+/// Scalar view used to choose the private worker-context initialization path.
+/// C retains the cdict, parameter, and CCtx objects; Rust owns the branch and
+/// the rule that the frame-header projection is published only after init.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_compressionJobBeginProjection {
+    pub firstJob: c_uint,
+    pub hasCDict: c_uint,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTDMT_compressionJobBeginProjection, firstJob) == 0);
+    assert!(offset_of!(ZSTDMT_compressionJobBeginProjection, hasCDict) == size_of::<c_uint>());
+    assert!(size_of::<ZSTDMT_compressionJobBeginProjection>() == 2 * size_of::<c_uint>());
+};
+
+#[inline]
+fn begin_compression_job_with<B, F, D, P, H>(
+    projection: ZSTDMT_compressionJobBeginProjection,
+    mut begin_with_cdict: B,
+    mut set_force_max_window: F,
+    mut set_deterministic_ref_prefix: D,
+    mut begin_with_prefix: P,
+    mut publish_frame_header: H,
+) -> usize
+where
+    B: FnMut() -> usize,
+    F: FnMut(c_int) -> usize,
+    D: FnMut() -> usize,
+    P: FnMut() -> usize,
+    H: FnMut(),
+{
+    let init_error = if projection.hasCDict != 0 {
+        if projection.firstJob == 0 {
+            return ERROR(ZstdErrorCode::StageWrong);
+        }
+        begin_with_cdict()
+    } else {
+        let force_window_error = set_force_max_window((projection.firstJob == 0) as c_int);
+        if ERR_isError(force_window_error) {
+            return force_window_error;
+        }
+        if projection.firstJob == 0 {
+            let error = set_deterministic_ref_prefix();
+            if ERR_isError(error) {
+                return error;
+            }
+        }
+        begin_with_prefix()
+    };
+
+    if ERR_isError(init_error) {
+        return init_error;
+    }
+    publish_frame_header();
+    init_error
+}
+
+/// Own the worker-context initialization branch and failure/publication order
+/// while C keeps all private codec operations behind callbacks.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_compressionJobBegin(
+    projection: *const ZSTDMT_compressionJobBeginProjection,
+    opaque: *mut c_void,
+    beginWithCDict: Option<ZSTDMT_compressionJobStepFn>,
+    setForceMaxWindow: Option<ZSTDMT_compressionJobSetParameterFn>,
+    setDeterministicRefPrefix: Option<ZSTDMT_compressionJobStepFn>,
+    beginWithPrefix: Option<ZSTDMT_compressionJobStepFn>,
+    publishFrameHeader: Option<ZSTDMT_compressionJobVoidFn>,
+) -> usize {
+    let Some(projection) = (unsafe { projection.as_ref() }).copied() else {
+        return ERROR(ZstdErrorCode::Generic);
+    };
+    let (
+        Some(begin_with_cdict),
+        Some(set_force_max_window),
+        Some(set_deterministic_ref_prefix),
+        Some(begin_with_prefix),
+        Some(publish_frame_header),
+    ) = (
+        beginWithCDict,
+        setForceMaxWindow,
+        setDeterministicRefPrefix,
+        beginWithPrefix,
+        publishFrameHeader,
+    )
+    else {
+        return ERROR(ZstdErrorCode::Generic);
+    };
+
+    begin_compression_job_with(
+        *projection,
+        || unsafe { begin_with_cdict(opaque) },
+        |value| unsafe { set_force_max_window(opaque, value) },
+        || unsafe { set_deterministic_ref_prefix(opaque) },
+        || unsafe { begin_with_prefix(opaque) },
+        || unsafe { publish_frame_header(opaque) },
+    )
+}
 
 /// C-owned leaves for the worker-job finish path.
 ///
@@ -7312,6 +7413,143 @@ mod tests {
         assert_eq!(result.clearSource, 0);
     }
 
+    #[test]
+    fn compression_job_begin_uses_cdict_and_publishes_header_after_init() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let projection = ZSTDMT_compressionJobBeginProjection {
+            firstJob: 1,
+            hasCDict: 1,
+        };
+
+        let cdict_events = Rc::clone(&events);
+        let header_events = Rc::clone(&events);
+        let result = begin_compression_job_with(
+            projection,
+            move || {
+                cdict_events.borrow_mut().push("cdict");
+                0
+            },
+            |_| panic!("cdict jobs must not update force-window parameters"),
+            || panic!("cdict jobs must not update deterministic-prefix parameters"),
+            || panic!("cdict jobs must not initialize from a raw prefix"),
+            move || header_events.borrow_mut().push("header"),
+        );
+
+        assert_eq!(result, 0);
+        assert_eq!(&*events.borrow(), &["cdict", "header"]);
+    }
+
+    #[test]
+    fn compression_job_begin_orders_nonfirst_parameter_policy_before_prefix_init() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let projection = ZSTDMT_compressionJobBeginProjection {
+            firstJob: 0,
+            hasCDict: 0,
+        };
+
+        let force_events = Rc::clone(&events);
+        let deterministic_events = Rc::clone(&events);
+        let prefix_events = Rc::clone(&events);
+        let header_events = Rc::clone(&events);
+        let result = begin_compression_job_with(
+            projection,
+            || panic!("raw-prefix jobs must not use a cdict"),
+            move |value| {
+                assert_eq!(value, 1);
+                force_events.borrow_mut().push("force-window");
+                0
+            },
+            move || {
+                deterministic_events
+                    .borrow_mut()
+                    .push("deterministic-prefix");
+                0
+            },
+            move || {
+                prefix_events.borrow_mut().push("prefix");
+                0
+            },
+            move || header_events.borrow_mut().push("header"),
+        );
+
+        assert_eq!(result, 0);
+        assert_eq!(
+            &*events.borrow(),
+            &["force-window", "deterministic-prefix", "prefix", "header"]
+        );
+    }
+
+    #[test]
+    fn compression_job_begin_stops_after_force_window_error() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let expected_error = ERROR(ZstdErrorCode::MemoryAllocation);
+        let projection = ZSTDMT_compressionJobBeginProjection {
+            firstJob: 0,
+            hasCDict: 0,
+        };
+        let force_events = Rc::clone(&events);
+
+        let result = begin_compression_job_with(
+            projection,
+            || panic!("parameter failure must stop before cdict initialization"),
+            move |_| {
+                force_events.borrow_mut().push("force-window");
+                expected_error
+            },
+            || panic!("parameter failure must stop before deterministic-prefix update"),
+            || panic!("parameter failure must stop before prefix initialization"),
+            || panic!("parameter failure must stop before header publication"),
+        );
+
+        assert_eq!(result, expected_error);
+        assert_eq!(&*events.borrow(), &["force-window"]);
+    }
+
+    #[test]
+    fn compression_job_begin_does_not_publish_header_after_init_error() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let expected_error = ERROR(ZstdErrorCode::Generic);
+        let projection = ZSTDMT_compressionJobBeginProjection {
+            firstJob: 1,
+            hasCDict: 1,
+        };
+        let cdict_events = Rc::clone(&events);
+
+        let result = begin_compression_job_with(
+            projection,
+            move || {
+                cdict_events.borrow_mut().push("cdict");
+                expected_error
+            },
+            |_| panic!("cdict initialization failure must not update parameters"),
+            || panic!("cdict initialization failure must not update prefix policy"),
+            || panic!("cdict initialization failure must not initialize a raw prefix"),
+            || panic!("cdict initialization failure must not publish a header"),
+        );
+
+        assert_eq!(result, expected_error);
+        assert_eq!(&*events.borrow(), &["cdict"]);
+    }
+
+    #[test]
+    fn compression_job_begin_rejects_cdict_on_nonfirst_job() {
+        let projection = ZSTDMT_compressionJobBeginProjection {
+            firstJob: 0,
+            hasCDict: 1,
+        };
+
+        let result = begin_compression_job_with(
+            projection,
+            || panic!("invalid cdict placement must not initialize"),
+            |_| panic!("invalid cdict placement must not update parameters"),
+            || panic!("invalid cdict placement must not update prefix policy"),
+            || panic!("invalid cdict placement must not initialize a raw prefix"),
+            || panic!("invalid cdict placement must not publish a header"),
+        );
+
+        assert_eq!(result, ERROR(ZstdErrorCode::StageWrong));
+    }
+
     #[test]
     fn chunk_loop_handles_empty_jobs_without_compression() {
         let mut compressor = MockChunkCompressor {