feat(mt): move serial advance orchestration into Rust

Move the serial turn increment, broadcast, and unlock ordering into a Rust
entry point. C retains the pthread condition variable and mutex operations
behind callbacks, while Rust preserves wrapping unsigned job-counter
semantics and the required broadcast-before-unlock contract. Add ABI layout
checks and focused tests for ordinary and wrapping advancement.

Test Plan:
- cargo fmt --manifest-path rust/Cargo.toml -- --check
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml zstdmt_compress::tests::serial_ -- --nocapture
- ulimit -v 41943040 && make -j1

lib/compress/zstdmt_compress.c

@@ -379,6 +379,21 @@ typedef char ZSTDMT_rust_serial_wait_for_turn_state_layout[
          && sizeof(ZSTDMT_RustSerialWaitForTurnState) == 5 * sizeof(void*)) ? 1 : -1];
 int ZSTDMT_rust_serialStateWaitForTurn(
         const ZSTDMT_RustSerialWaitForTurnState* state);
+typedef void (*ZSTDMT_serialStateBroadcastFn)(void* opaque);
+typedef void (*ZSTDMT_serialStateUnlockFn)(void* opaque);
+typedef struct {
+    void* callbackContext;
+    unsigned* nextJobID;
+    ZSTDMT_serialStateBroadcastFn broadcast;
+    ZSTDMT_serialStateUnlockFn unlock;
+} ZSTDMT_RustSerialAdvanceState;
+typedef char ZSTDMT_rust_serial_advance_state_layout[
+        (offsetof(ZSTDMT_RustSerialAdvanceState, callbackContext) == 0
+         && offsetof(ZSTDMT_RustSerialAdvanceState, nextJobID) == sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialAdvanceState, broadcast) == 2 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialAdvanceState, unlock) == 3 * sizeof(void*)
+         && sizeof(ZSTDMT_RustSerialAdvanceState) == 4 * sizeof(void*)) ? 1 : -1];
+void ZSTDMT_rust_serialStateAdvance(const ZSTDMT_RustSerialAdvanceState* state);
 void ZSTDMT_rust_serialStateGenSequences(
         ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize,
         unsigned jobID, int ldmEnabled, int checksumEnabled, void* opaque,
@@ -1120,6 +1135,18 @@ static void ZSTDMT_serialState_wait(void* opaque)
     ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex);
 }
 
+static void ZSTDMT_serialState_broadcast(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    ZSTD_pthread_cond_broadcast(&serialState->cond);
+}
+
+static void ZSTDMT_serialState_unlock(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    ZSTD_pthread_mutex_unlock(&serialState->mutex);
+}
+
 /* Rust owns the serial turn/skip decision and operation ordering.  The wait
  * callback intentionally leaves the main serial mutex locked; the advance
  * callback releases it after Rust has performed the current turn's work. */
@@ -1172,9 +1199,12 @@ static void ZSTDMT_serialState_updateChecksum(
 static void ZSTDMT_serialState_advance(void* opaque)
 {
     SerialState* const serialState = (SerialState*)opaque;
-    serialState->nextJobID++;
+    ZSTDMT_RustSerialAdvanceState state;
-    ZSTD_pthread_cond_broadcast(&serialState->cond);
+    state.callbackContext = serialState;
-    ZSTD_pthread_mutex_unlock(&serialState->mutex);
+    state.nextJobID = &serialState->nextJobID;
+    state.broadcast = ZSTDMT_serialState_broadcast;
+    state.unlock = ZSTDMT_serialState_unlock;
+    ZSTDMT_rust_serialStateAdvance(&state);
 }
 
 static void

rust/src/zstdmt_compress.rs

@@ -91,6 +91,8 @@ pub type ZSTDMT_serialUpdateChecksumFn = unsafe extern "C" fn(*mut c_void, *cons
 pub type ZSTDMT_serialAdvanceFn = unsafe extern "C" fn(*mut c_void);
 pub type ZSTDMT_serialStateLockFn = unsafe extern "C" fn(*mut c_void);
 pub type ZSTDMT_serialStateWaitFn = unsafe extern "C" fn(*mut c_void);
+pub type ZSTDMT_serialStateBroadcastFn = unsafe extern "C" fn(*mut c_void);
+pub type ZSTDMT_serialStateUnlockFn = unsafe extern "C" fn(*mut c_void);
 
 /// Projection for the MT serial turn wait.  Rust owns the lock/wait loop and
 /// comparison; C retains the pthread mutex and condition variable behind the
@@ -143,6 +145,49 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateWaitForTurn(
     }
 }
 
+/// Projection for advancing the MT serial turn.  Rust owns the increment and
+/// callback order; C retains the condition variable broadcast and mutex
+/// unlock operations behind the callbacks.
+#[repr(C)]
+pub struct ZSTDMT_RustSerialAdvanceState {
+    callback_context: *mut c_void,
+    next_job_id: *mut c_uint,
+    broadcast: Option<ZSTDMT_serialStateBroadcastFn>,
+    unlock: Option<ZSTDMT_serialStateUnlockFn>,
+}
+
+const _: () = {
+    assert!(size_of::<ZSTDMT_serialStateBroadcastFn>() == size_of::<usize>());
+    assert!(size_of::<ZSTDMT_serialStateUnlockFn>() == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, callback_context) == 0);
+    assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, next_job_id) == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, broadcast) == 2 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, unlock) == 3 * size_of::<usize>());
+    assert!(size_of::<ZSTDMT_RustSerialAdvanceState>() == size_of::<[usize; 4]>());
+};
+
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_serialStateAdvance(
+    state: *const ZSTDMT_RustSerialAdvanceState,
+) {
+    if state.is_null() {
+        return;
+    }
+    let state = unsafe { &*state };
+    if state.callback_context.is_null() || state.next_job_id.is_null() {
+        return;
+    }
+    let (Some(broadcast), Some(unlock)) = (state.broadcast, state.unlock) else {
+        return;
+    };
+
+    unsafe {
+        *state.next_job_id = (*state.next_job_id).wrapping_add(1);
+        broadcast(state.callback_context);
+        unlock(state.callback_context);
+    }
+}
+
 #[repr(C)]
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub struct ZSTDMT_serialStateEnsureFinishedResult {
@@ -3540,6 +3585,38 @@ mod tests {
         }
     }
 
+    struct SerialAdvanceTestContext {
+        events: Vec<&'static str>,
+    }
+
+    unsafe extern "C" fn serial_advance_test_broadcast(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialAdvanceTestContext>())
+                .events
+                .push("broadcast")
+        };
+    }
+
+    unsafe extern "C" fn serial_advance_test_unlock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialAdvanceTestContext>())
+                .events
+                .push("unlock")
+        };
+    }
+
+    fn serial_advance_test_state(
+        context: &mut SerialAdvanceTestContext,
+        next_job_id: &mut c_uint,
+    ) -> ZSTDMT_RustSerialAdvanceState {
+        ZSTDMT_RustSerialAdvanceState {
+            callback_context: context as *mut _ as *mut c_void,
+            next_job_id,
+            broadcast: Some(serial_advance_test_broadcast),
+            unlock: Some(serial_advance_test_unlock),
+        }
+    }
+
     unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) {
         unsafe {
             (*context.cast::<WaitForLdmTestContext>())
@@ -3661,6 +3738,30 @@ mod tests {
         assert_eq!(context.next_job_id, 5);
     }
 
+    #[test]
+    fn serial_advance_increments_before_broadcast_and_unlock() {
+        let mut context = SerialAdvanceTestContext { events: Vec::new() };
+        let mut next_job_id = 7;
+        let state = serial_advance_test_state(&mut context, &mut next_job_id);
+
+        unsafe { ZSTDMT_rust_serialStateAdvance(&state) };
+
+        assert_eq!(next_job_id, 8);
+        assert_eq!(context.events, vec!["broadcast", "unlock"]);
+    }
+
+    #[test]
+    fn serial_advance_wraps_the_job_counter_like_c_unsigned_arithmetic() {
+        let mut context = SerialAdvanceTestContext { events: Vec::new() };
+        let mut next_job_id = c_uint::MAX;
+        let state = serial_advance_test_state(&mut context, &mut next_job_id);
+
+        unsafe { ZSTDMT_rust_serialStateAdvance(&state) };
+
+        assert_eq!(next_job_id, 0);
+        assert_eq!(context.events, vec!["broadcast", "unlock"]);
+    }
+
     fn record_compression_job_event(state: &Rc<RefCell<MockCompressionJob>>, event: &'static str) {
         state.borrow_mut().events.push(event);
     }