refactor(mt): move job preparation order to Rust

Make Rust the owner of the compression-job preparation sequence: publish the descriptor, reset the reusable input state, and publish terminal frame state last. Keep all MT job and stream layouts in C behind three focused callbacks, and preserve the nonterminal and terminal branches exactly.

Test Plan: ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check; GCC and Clang syntax-only checks; make -j1 -C tests test (all 41 shell tests, fuzzer, zstd tester, and zstream tester passed).

lib/compress/zstdmt_compress.c

@@ -547,6 +547,12 @@ typedef void (*ZSTDMT_prepareJobFn)(void* opaque, unsigned jobID,
                                    const ZSTDMT_RustJobInitialization* init);
 typedef void (*ZSTDMT_writeEmptyJobFn)(void* opaque, unsigned jobID);
 typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID);
+void ZSTDMT_rust_prepareCompressionJob(
+        void* opaque, unsigned jobID,
+        const ZSTDMT_RustJobInitialization* initialization,
+        ZSTDMT_prepareJobFn initializeJob,
+        ZSTDMT_prepareJobFn resetInputState,
+        ZSTDMT_prepareJobFn publishFrameState);
 ZSTDMT_RustCreateJobResult ZSTDMT_rust_createCompressionJob(
         const ZSTDMT_RustCreateJobProjection* projection,
         void* opaque, ZSTDMT_prepareJobFn prepareJob,
@@ -2877,7 +2883,7 @@ static void ZSTDMT_writeLastEmptyBlock(ZSTDMT_jobDescription* job)
     assert(!ZSTD_isError(job->cSize));
 }
 
-static void ZSTDMT_prepareCompressionJob(
+static void ZSTDMT_prepareCompressionJobDescriptor(
         void* opaque, unsigned jobID,
         const ZSTDMT_RustJobInitialization* initialization)
 {
@@ -2908,7 +2914,14 @@ static void ZSTDMT_prepareCompressionJob(
     job->lastJob = initialization->lastJob;
     job->frameChecksumNeeded = initialization->frameChecksumNeeded;
     job->dstFlushed = 0;
+}
 
+static void ZSTDMT_prepareCompressionJobInputState(
+        void* opaque, unsigned jobID,
+        const ZSTDMT_RustJobInitialization* initialization)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    (void)jobID;
     /* Update the round buffer position and clear the input buffer to be reset. */
     mtctx->roundBuff.pos += initialization->roundBuffPosDelta;
     mtctx->inBuff.buffer = g_nullBuffer;
@@ -2917,6 +2930,14 @@ static void ZSTDMT_prepareCompressionJob(
         (const BYTE*)initialization->nextPrefixStart,
         initialization->nextPrefixSize
     };
+}
+
+static void ZSTDMT_prepareCompressionJobFrameState(
+        void* opaque, unsigned jobID,
+        const ZSTDMT_RustJobInitialization* initialization)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    (void)jobID;
     if (initialization->lastJob) {
         mtctx->frameEnded = 1;
         if (initialization->clearChecksumFlag)
@@ -2924,6 +2945,17 @@ static void ZSTDMT_prepareCompressionJob(
     }
 }
 
+static void ZSTDMT_prepareCompressionJob(
+        void* opaque, unsigned jobID,
+        const ZSTDMT_RustJobInitialization* initialization)
+{
+    ZSTDMT_rust_prepareCompressionJob(
+            opaque, jobID, initialization,
+            ZSTDMT_prepareCompressionJobDescriptor,
+            ZSTDMT_prepareCompressionJobInputState,
+            ZSTDMT_prepareCompressionJobFrameState);
+}
+
 static void ZSTDMT_writeEmptyCompressionJob(void* opaque, unsigned jobID)
 {
     ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;

rust/src/zstdmt_compress.rs

@@ -1547,6 +1547,68 @@ pub type ZSTDMT_prepareJobFn = unsafe extern "C" fn(
 pub type ZSTDMT_writeEmptyJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint);
 pub type ZSTDMT_tryAddJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint) -> c_int;
 
+/// Order the C-owned leaves that publish a newly prepared MT job.
+///
+/// The descriptor must become complete before the reusable input range is
+/// cleared, and terminal frame state is published last. `lastJob` is a C
+/// boolean scalar, so any non-zero value selects the terminal publication.
+#[inline]
+fn prepare_compression_job_with<P, I, F>(
+    job_id: c_uint,
+    initialization: &ZSTDMT_jobInitialization,
+    mut initialize_job: P,
+    mut reset_input_state: I,
+    mut publish_frame_state: F,
+) where
+    P: FnMut(c_uint, &ZSTDMT_jobInitialization),
+    I: FnMut(c_uint, &ZSTDMT_jobInitialization),
+    F: FnMut(c_uint, &ZSTDMT_jobInitialization),
+{
+    initialize_job(job_id, initialization);
+    reset_input_state(job_id, initialization);
+    if initialization.lastJob != 0 {
+        publish_frame_state(job_id, initialization);
+    }
+}
+
+/// Apply the MT job-preparation ordering while C retains the private job
+/// descriptor and stream-state fields behind callbacks.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob(
+    opaque: *mut c_void,
+    job_id: c_uint,
+    initialization: *const ZSTDMT_jobInitialization,
+    initialize_job: Option<ZSTDMT_prepareJobFn>,
+    reset_input_state: Option<ZSTDMT_prepareJobFn>,
+    publish_frame_state: Option<ZSTDMT_prepareJobFn>,
+) {
+    let Some(initialization) = (unsafe { initialization.as_ref() }) else {
+        return;
+    };
+    let (Some(initialize_job), Some(reset_input_state), Some(publish_frame_state)) =
+        (initialize_job, reset_input_state, publish_frame_state)
+    else {
+        return;
+    };
+    if opaque.is_null() {
+        return;
+    }
+
+    prepare_compression_job_with(
+        job_id,
+        initialization,
+        |job_id, initialization| unsafe {
+            initialize_job(opaque, job_id, initialization);
+        },
+        |job_id, initialization| unsafe {
+            reset_input_state(opaque, job_id, initialization);
+        },
+        |job_id, initialization| unsafe {
+            publish_frame_state(opaque, job_id, initialization);
+        },
+    );
+}
+
 #[inline]
 fn create_job_table_full(projection: ZSTDMT_createJobProjection) -> ZSTDMT_createJobResult {
     ZSTDMT_createJobResult {
@@ -7034,6 +7096,55 @@ mod tests {
         assert_eq!(result.jobReady, 0);
     }
 
+    #[test]
+    fn job_preparation_orders_descriptor_input_and_terminal_state() {
+        let initialization = ZSTDMT_jobInitialization {
+            jobNumber: 7,
+            lastJob: 2,
+            clearChecksumFlag: 1,
+            ..ZSTDMT_jobInitialization::default()
+        };
+        let events = RefCell::new(Vec::new());
+
+        prepare_compression_job_with(
+            3,
+            &initialization,
+            |job_id, value| {
+                assert_eq!(job_id, 3);
+                assert_eq!(value.jobNumber, 7);
+                events.borrow_mut().push("descriptor");
+            },
+            |job_id, value| {
+                assert_eq!(job_id, 3);
+                assert_eq!(value.roundBuffPosDelta, 0);
+                events.borrow_mut().push("input");
+            },
+            |job_id, value| {
+                assert_eq!(job_id, 3);
+                assert_eq!(value.clearChecksumFlag, 1);
+                events.borrow_mut().push("frame");
+            },
+        );
+
+        assert_eq!(events.into_inner(), vec!["descriptor", "input", "frame"]);
+    }
+
+    #[test]
+    fn job_preparation_skips_frame_state_for_nonterminal_job() {
+        let initialization = ZSTDMT_jobInitialization::default();
+        let events = RefCell::new(Vec::new());
+
+        prepare_compression_job_with(
+            4,
+            &initialization,
+            |_job_id, _value| events.borrow_mut().push("descriptor"),
+            |_job_id, _value| events.borrow_mut().push("input"),
+            |_job_id, _value| panic!("nonterminal job must not publish frame state"),
+        );
+
+        assert_eq!(events.into_inner(), vec!["descriptor", "input"]);
+    }
+
     #[test]
     fn create_job_preserves_pool_success_and_failure_state() {
         let projection = ZSTDMT_createJobProjection {