refactor(mt): move terminal frame action policy to Rust

Classify the MT terminal-job flags in Rust while retaining the two private C
mutations as narrow callbacks. Descriptor setup, reusable-input reset, and
terminal frame publication keep their original order; nonterminal jobs remain
untouched, and checksum clearing still occurs only for the original flag
combination.

Test Plan:
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings` -- passed
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings` -- passed
- `ulimit -v 41943040; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check` -- passed
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets` -- passed, 190 tests
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 make -j1` -- passed
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 ./tests/rustLibSmoke` -- passed
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 make -j1 -C tests test` -- passed, including large streaming, native, fuzzer, and zstream phases

lib/compress/zstdmt_compress.c

@@ -545,6 +545,7 @@ enum {
 };
 typedef void (*ZSTDMT_prepareJobFn)(void* opaque, unsigned jobID,
                                    const ZSTDMT_RustJobInitialization* init);
+typedef void (*ZSTDMT_prepareFrameStateFn)(void* opaque);
 typedef void (*ZSTDMT_writeEmptyJobFn)(void* opaque, unsigned jobID);
 typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID);
 void ZSTDMT_rust_prepareCompressionJob(
@@ -552,7 +553,8 @@ void ZSTDMT_rust_prepareCompressionJob(
         const ZSTDMT_RustJobInitialization* initialization,
         ZSTDMT_prepareJobFn initializeJob,
         ZSTDMT_prepareJobFn resetInputState,
-        ZSTDMT_prepareJobFn publishFrameState);
+        ZSTDMT_prepareFrameStateFn setFrameEnded,
+        ZSTDMT_prepareFrameStateFn clearChecksumFlag);
 ZSTDMT_RustCreateJobResult ZSTDMT_rust_createCompressionJob(
         const ZSTDMT_RustCreateJobProjection* projection,
         void* opaque, ZSTDMT_prepareJobFn prepareJob,
@@ -2931,17 +2933,16 @@ static void ZSTDMT_prepareCompressionJobInputState(
     };
 }
 
-static void ZSTDMT_prepareCompressionJobFrameState(
+static void ZSTDMT_prepareCompressionJobSetFrameEnded(void* opaque)
-        void* opaque, unsigned jobID,
-        const ZSTDMT_RustJobInitialization* initialization)
 {
     ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
-    (void)jobID;
+    mtctx->frameEnded = 1;
-    if (initialization->lastJob) {
+}
-        mtctx->frameEnded = 1;
+
-        if (initialization->clearChecksumFlag)
+static void ZSTDMT_prepareCompressionJobClearChecksumFlag(void* opaque)
-            mtctx->params.fParams.checksumFlag = 0;
+{
-    }
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    mtctx->params.fParams.checksumFlag = 0;
 }
 
 static void ZSTDMT_prepareCompressionJob(
@@ -2952,7 +2953,8 @@ static void ZSTDMT_prepareCompressionJob(
             opaque, jobID, initialization,
             ZSTDMT_prepareCompressionJobDescriptor,
             ZSTDMT_prepareCompressionJobInputState,
-            ZSTDMT_prepareCompressionJobFrameState);
+            ZSTDMT_prepareCompressionJobSetFrameEnded,
+            ZSTDMT_prepareCompressionJobClearChecksumFlag);
 }
 
 static void ZSTDMT_writeEmptyCompressionJob(void* opaque, unsigned jobID)

rust/src/zstdmt_compress.rs

@@ -1544,30 +1544,65 @@ pub type ZSTDMT_prepareJobFn = unsafe extern "C" fn(
     job_id: c_uint,
     initialization: *const ZSTDMT_jobInitialization,
 );
+pub type ZSTDMT_prepareFrameStateFn = unsafe extern "C" fn(opaque: *mut c_void);
 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;
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum CompressionJobFrameStateAction {
+    None,
+    SetFrameEnded,
+    SetFrameEndedAndClearChecksum,
+}
+
+#[inline]
+fn classify_compression_job_frame_state(
+    last_job: c_uint,
+    clear_checksum_flag: c_uint,
+) -> CompressionJobFrameStateAction {
+    if last_job == 0 {
+        CompressionJobFrameStateAction::None
+    } else if clear_checksum_flag == 0 {
+        CompressionJobFrameStateAction::SetFrameEnded
+    } else {
+        CompressionJobFrameStateAction::SetFrameEndedAndClearChecksum
+    }
+}
+
 /// 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
+/// cleared, and terminal frame state is published last. `lastJob` and
-/// boolean scalar, so any non-zero value selects the terminal publication.
+/// `clearChecksumFlag` are classified in Rust; C callbacks retain the private
+/// context mutations.
 #[inline]
-fn prepare_compression_job_with<P, I, F>(
+fn prepare_compression_job_with<P, I, S, C>(
     job_id: c_uint,
     initialization: &ZSTDMT_jobInitialization,
     mut initialize_job: P,
     mut reset_input_state: I,
-    mut publish_frame_state: F,
+    mut set_frame_ended: S,
+    mut clear_checksum_flag: C,
 ) where
     P: FnMut(c_uint, &ZSTDMT_jobInitialization),
     I: FnMut(c_uint, &ZSTDMT_jobInitialization),
-    F: FnMut(c_uint, &ZSTDMT_jobInitialization),
+    S: FnMut(),
+    C: FnMut(),
 {
+    let frame_state_action = classify_compression_job_frame_state(
+        initialization.lastJob,
+        initialization.clearChecksumFlag,
+    );
+
     initialize_job(job_id, initialization);
     reset_input_state(job_id, initialization);
-    if initialization.lastJob != 0 {
+    match frame_state_action {
-        publish_frame_state(job_id, initialization);
+        CompressionJobFrameStateAction::None => {}
+        CompressionJobFrameStateAction::SetFrameEnded => set_frame_ended(),
+        CompressionJobFrameStateAction::SetFrameEndedAndClearChecksum => {
+            set_frame_ended();
+            clear_checksum_flag();
+        }
     }
 }
 
@@ -1580,13 +1615,23 @@ pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob(
     initialization: *const ZSTDMT_jobInitialization,
     initialize_job: Option<ZSTDMT_prepareJobFn>,
     reset_input_state: Option<ZSTDMT_prepareJobFn>,
-    publish_frame_state: Option<ZSTDMT_prepareJobFn>,
+    set_frame_ended: Option<ZSTDMT_prepareFrameStateFn>,
+    clear_checksum_flag: Option<ZSTDMT_prepareFrameStateFn>,
 ) {
     let Some(initialization) = (unsafe { initialization.as_ref() }) else {
         return;
     };
-    let (Some(initialize_job), Some(reset_input_state), Some(publish_frame_state)) =
+    let (
-        (initialize_job, reset_input_state, publish_frame_state)
+        Some(initialize_job),
+        Some(reset_input_state),
+        Some(set_frame_ended),
+        Some(clear_checksum_flag),
+    ) = (
+        initialize_job,
+        reset_input_state,
+        set_frame_ended,
+        clear_checksum_flag,
+    )
     else {
         return;
     };
@@ -1603,8 +1648,11 @@ pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob(
         |job_id, initialization| unsafe {
             reset_input_state(opaque, job_id, initialization);
         },
-        |job_id, initialization| unsafe {
+        || unsafe {
-            publish_frame_state(opaque, job_id, initialization);
+            set_frame_ended(opaque);
+        },
+        || unsafe {
+            clear_checksum_flag(opaque);
         },
     );
 }
@@ -7125,6 +7173,25 @@ mod tests {
         assert_eq!(result.jobReady, 0);
     }
 
+    #[test]
+    fn compression_job_frame_state_classifies_all_flag_combinations() {
+        for (last_job, clear_checksum_flag, expected) in [
+            (0, 0, CompressionJobFrameStateAction::None),
+            (0, 1, CompressionJobFrameStateAction::None),
+            (1, 0, CompressionJobFrameStateAction::SetFrameEnded),
+            (
+                1,
+                1,
+                CompressionJobFrameStateAction::SetFrameEndedAndClearChecksum,
+            ),
+        ] {
+            assert_eq!(
+                classify_compression_job_frame_state(last_job, clear_checksum_flag),
+                expected
+            );
+        }
+    }
+
     #[test]
     fn job_preparation_orders_descriptor_input_and_terminal_state() {
         let initialization = ZSTDMT_jobInitialization {
@@ -7148,14 +7215,14 @@ mod tests {
                 assert_eq!(value.roundBuffPosDelta, 0);
                 events.borrow_mut().push("input");
             },
-            |job_id, value| {
+            || events.borrow_mut().push("frame-ended"),
-                assert_eq!(job_id, 3);
+            || events.borrow_mut().push("checksum"),
-                assert_eq!(value.clearChecksumFlag, 1);
-                events.borrow_mut().push("frame");
-            },
         );
 
-        assert_eq!(events.into_inner(), vec!["descriptor", "input", "frame"]);
+        assert_eq!(
+            events.into_inner(),
+            vec!["descriptor", "input", "frame-ended", "checksum"]
+        );
     }
 
     #[test]
@@ -7168,7 +7235,8 @@ mod tests {
             &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"),
+            || panic!("nonterminal job must not set frame-ended state"),
+            || panic!("nonterminal job must not clear checksum state"),
         );
 
         assert_eq!(events.into_inner(), vec!["descriptor", "input"]);