feat(mt): move failed-job serial cleanup into Rust

Move the failed-job serial-state orchestration out of zstdmt_compress.c.
Rust now owns the lock, skip decision, serial-counter publication,
broadcast, LDM cleanup ordering, and final unlock. C retains the pthread
objects, private LDM window, and error/debug leaves behind callbacks. Add
ABI layout checks and focused tests for both skipped predecessors and later
jobs that only need the lock/unlock pair.

Test Plan:
- cargo fmt --manifest-path rust/Cargo.toml -- --check
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml zstdmt_compress::tests::serial_ensure_finished -- --nocapture
- ulimit -v 41943040 && make -j1
- ulimit -v 41943040 && make -j1 -C tests test-fuzzer FUZZERTEST=-T3s FUZZER_FLAGS=--no-big-tests

lib/compress/zstdmt_compress.c

@@ -407,6 +407,40 @@ typedef struct {
 } ZSTDMT_RustSerialEnsureFinishedResult;
 ZSTDMT_RustSerialEnsureFinishedResult ZSTDMT_rust_serialStateEnsureFinished(
         unsigned nextJobID, unsigned jobID);
+typedef void (*ZSTDMT_serialStateCallbackFn)(void* opaque);
+typedef void (*ZSTDMT_serialStateSkipFn)(
+        void* opaque, unsigned jobID, size_t cSize);
+typedef struct {
+    void* callbackContext;
+    unsigned* nextJobID;
+    ZSTDMT_serialStateCallbackFn lock;
+    ZSTDMT_serialStateCallbackFn broadcast;
+    ZSTDMT_serialStateCallbackFn ldmLock;
+    ZSTDMT_serialStateCallbackFn clearLdmWindow;
+    ZSTDMT_serialStateCallbackFn ldmSignal;
+    ZSTDMT_serialStateCallbackFn ldmUnlock;
+    ZSTDMT_serialStateCallbackFn unlock;
+    ZSTDMT_serialStateSkipFn onSkip;
+    size_t cSize;
+    unsigned jobID;
+} ZSTDMT_RustSerialEnsureFinishedState;
+typedef char ZSTDMT_rust_serial_ensure_finished_state_layout[
+        (offsetof(ZSTDMT_RustSerialEnsureFinishedState, callbackContext) == 0
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, nextJobID) == sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, lock) == 2 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, broadcast) == 3 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmLock) == 4 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, clearLdmWindow) == 5 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmSignal) == 6 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmUnlock) == 7 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, unlock) == 8 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, onSkip) == 9 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, cSize) == 10 * sizeof(void*)
+         && offsetof(ZSTDMT_RustSerialEnsureFinishedState, jobID)
+                == 10 * sizeof(void*) + sizeof(size_t)
+         && sizeof(ZSTDMT_RustSerialEnsureFinishedState) == 12 * sizeof(void*)) ? 1 : -1];
+void ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(
+        const ZSTDMT_RustSerialEnsureFinishedState* state);
 typedef void (*ZSTDMT_waitForJobCompleteFn)(
         void* opaque, unsigned jobID, unsigned doneJobID);
 unsigned ZSTDMT_rust_waitForAllJobsCompleted(
@@ -1207,6 +1241,40 @@ static void ZSTDMT_serialState_advance(void* opaque)
     ZSTDMT_rust_serialStateAdvance(&state);
 }
 
+static void ZSTDMT_serialState_ldmLock(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex);
+}
+
+static void ZSTDMT_serialState_clearLdmWindow(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    ZSTD_rust_windowClear((size_t)(serialState->ldmWindow.nextSrc -
+                                   serialState->ldmWindow.base),
+                          &serialState->ldmWindow.lowLimit,
+                          &serialState->ldmWindow.dictLimit);
+}
+
+static void ZSTDMT_serialState_ldmSignal(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    ZSTD_pthread_cond_signal(&serialState->ldmWindowCond);
+}
+
+static void ZSTDMT_serialState_ldmUnlock(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex);
+}
+
+static void ZSTDMT_serialState_onSkip(void* opaque, unsigned jobID, size_t cSize)
+{
+    (void)opaque;
+    assert(ZSTD_isError(cSize)); (void)cSize;
+    DEBUGLOG(5, "Skipping past job %u because of error", jobID);
+}
+
 static void
 ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an assert() check */
                                   ZSTD_CCtx* jobCCtx,
@@ -1223,25 +1291,20 @@ ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an
 static void ZSTDMT_serialState_ensureFinished(SerialState* serialState,
                                               unsigned jobID, size_t cSize)
 {
-    ZSTDMT_RustSerialEnsureFinishedResult result;
+    ZSTDMT_RustSerialEnsureFinishedState state;
-    ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex);
+    state.callbackContext = serialState;
-    result = ZSTDMT_rust_serialStateEnsureFinished(
+    state.nextJobID = &serialState->nextJobID;
-            serialState->nextJobID, jobID);
+    state.lock = ZSTDMT_serialState_lock;
-    if (result.skip) {
+    state.broadcast = ZSTDMT_serialState_broadcast;
-        assert(ZSTD_isError(cSize)); (void)cSize;
+    state.ldmLock = ZSTDMT_serialState_ldmLock;
-        DEBUGLOG(5, "Skipping past job %u because of error", jobID);
+    state.clearLdmWindow = ZSTDMT_serialState_clearLdmWindow;
-        serialState->nextJobID = result.nextJobID;
+    state.ldmSignal = ZSTDMT_serialState_ldmSignal;
-        ZSTD_pthread_cond_broadcast(&serialState->cond);
+    state.ldmUnlock = ZSTDMT_serialState_ldmUnlock;
-
+    state.unlock = ZSTDMT_serialState_unlock;
-        ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex);
+    state.onSkip = ZSTDMT_serialState_onSkip;
-        ZSTD_rust_windowClear((size_t)(serialState->ldmWindow.nextSrc -
+    state.cSize = cSize;
-                                       serialState->ldmWindow.base),
+    state.jobID = jobID;
-                              &serialState->ldmWindow.lowLimit,
+    ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state);
-                              &serialState->ldmWindow.dictLimit);
-        ZSTD_pthread_cond_signal(&serialState->ldmWindowCond);
-        ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex);
-    }
-    ZSTD_pthread_mutex_unlock(&serialState->mutex);
 
 }
 

rust/src/zstdmt_compress.rs

@@ -93,6 +93,8 @@ 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);
+pub type ZSTDMT_serialStateCallbackFn = unsafe extern "C" fn(*mut c_void);
+pub type ZSTDMT_serialStateSkipFn = unsafe extern "C" fn(*mut c_void, c_uint, usize);
 
 /// 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
@@ -188,6 +190,50 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateAdvance(
     }
 }
 
+/// Projection for the failed-job serial cleanup path.  Rust owns the lock,
+/// skip decision, serial-counter publication, and callback ordering; C keeps
+/// the mutex, condition variables, and private LDM-window cleanup behind the
+/// callbacks.
+#[repr(C)]
+pub struct ZSTDMT_RustSerialEnsureFinishedState {
+    callback_context: *mut c_void,
+    next_job_id: *mut c_uint,
+    lock: Option<ZSTDMT_serialStateCallbackFn>,
+    broadcast: Option<ZSTDMT_serialStateCallbackFn>,
+    ldm_lock: Option<ZSTDMT_serialStateCallbackFn>,
+    clear_ldm_window: Option<ZSTDMT_serialStateCallbackFn>,
+    ldm_signal: Option<ZSTDMT_serialStateCallbackFn>,
+    ldm_unlock: Option<ZSTDMT_serialStateCallbackFn>,
+    unlock: Option<ZSTDMT_serialStateCallbackFn>,
+    on_skip: Option<ZSTDMT_serialStateSkipFn>,
+    c_size: usize,
+    job_id: c_uint,
+}
+
+const _: () = {
+    assert!(size_of::<ZSTDMT_serialStateCallbackFn>() == size_of::<usize>());
+    assert!(size_of::<ZSTDMT_serialStateSkipFn>() == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, callback_context) == 0);
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, next_job_id) == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, lock) == 2 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, broadcast) == 3 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, ldm_lock) == 4 * size_of::<usize>());
+    assert!(
+        offset_of!(ZSTDMT_RustSerialEnsureFinishedState, clear_ldm_window)
+            == 5 * size_of::<usize>()
+    );
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, ldm_signal) == 6 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, ldm_unlock) == 7 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, unlock) == size_of::<[usize; 8]>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, on_skip) == 9 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, c_size) == 10 * size_of::<usize>());
+    assert!(
+        offset_of!(ZSTDMT_RustSerialEnsureFinishedState, job_id)
+            == 10 * size_of::<usize>() + size_of::<usize>()
+    );
+    assert!(size_of::<ZSTDMT_RustSerialEnsureFinishedState>() == size_of::<[usize; 12]>());
+};
+
 #[repr(C)]
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub struct ZSTDMT_serialStateEnsureFinishedResult {
@@ -948,9 +994,7 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateGenSequences(
     }
 }
 
-/// Decide whether a failed job must advance the shared serial turn.  C keeps
+/// Decide whether a failed job must advance the shared serial turn.
-/// the mutex, error assertion, condition broadcast, and LDM-window cleanup;
-/// Rust owns the scalar comparison and wrapping job-id transition.
 #[no_mangle]
 pub extern "C" fn ZSTDMT_rust_serialStateEnsureFinished(
     next_job_id: c_uint,
@@ -969,6 +1013,58 @@ pub extern "C" fn ZSTDMT_rust_serialStateEnsureFinished(
     }
 }
 
+/// Own the failed-job serial cleanup order while C retains synchronization and
+/// private LDM-window operations behind callbacks.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(
+    state: *const ZSTDMT_RustSerialEnsureFinishedState,
+) {
+    if state.is_null() {
+        return;
+    }
+    let state = unsafe { &*state };
+    if state.callback_context.is_null() || state.next_job_id.is_null() {
+        return;
+    }
+    let (
+        Some(lock),
+        Some(broadcast),
+        Some(ldm_lock),
+        Some(clear_ldm_window),
+        Some(ldm_signal),
+        Some(ldm_unlock),
+        Some(unlock),
+        Some(on_skip),
+    ) = (
+        state.lock,
+        state.broadcast,
+        state.ldm_lock,
+        state.clear_ldm_window,
+        state.ldm_signal,
+        state.ldm_unlock,
+        state.unlock,
+        state.on_skip,
+    )
+    else {
+        return;
+    };
+
+    unsafe {
+        lock(state.callback_context);
+        let result = ZSTDMT_rust_serialStateEnsureFinished(*state.next_job_id, state.job_id);
+        if result.skip != 0 {
+            on_skip(state.callback_context, state.job_id, state.c_size);
+            *state.next_job_id = result.nextJobID;
+            broadcast(state.callback_context);
+            ldm_lock(state.callback_context);
+            clear_ldm_window(state.callback_context);
+            ldm_signal(state.callback_context);
+            ldm_unlock(state.callback_context);
+        }
+        unlock(state.callback_context);
+    }
+}
+
 /// Wait for each submitted MT job in ring order.  C retains the per-job
 /// mutex, condition variable, and consumed/source counters behind one
 /// callback; Rust owns the ring-slot and completion progression policy.
@@ -3617,6 +3713,102 @@ mod tests {
         }
     }
 
+    struct SerialEnsureFinishedTestContext {
+        events: Vec<&'static str>,
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_lock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("lock")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_broadcast(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("broadcast")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_ldm_lock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("ldm_lock")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_clear_ldm(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("clear_ldm")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_ldm_signal(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("ldm_signal")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_ldm_unlock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("ldm_unlock")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_unlock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("unlock")
+        };
+    }
+
+    unsafe extern "C" fn serial_ensure_finished_test_on_skip(
+        context: *mut c_void,
+        job_id: c_uint,
+        c_size: usize,
+    ) {
+        assert_eq!(job_id, 4);
+        assert_eq!(c_size, 123);
+        unsafe {
+            (*context.cast::<SerialEnsureFinishedTestContext>())
+                .events
+                .push("skip")
+        };
+    }
+
+    fn serial_ensure_finished_test_state(
+        context: &mut SerialEnsureFinishedTestContext,
+        next_job_id: &mut c_uint,
+        job_id: c_uint,
+        c_size: usize,
+    ) -> ZSTDMT_RustSerialEnsureFinishedState {
+        ZSTDMT_RustSerialEnsureFinishedState {
+            callback_context: context as *mut _ as *mut c_void,
+            next_job_id,
+            lock: Some(serial_ensure_finished_test_lock),
+            broadcast: Some(serial_ensure_finished_test_broadcast),
+            ldm_lock: Some(serial_ensure_finished_test_ldm_lock),
+            clear_ldm_window: Some(serial_ensure_finished_test_clear_ldm),
+            ldm_signal: Some(serial_ensure_finished_test_ldm_signal),
+            ldm_unlock: Some(serial_ensure_finished_test_ldm_unlock),
+            unlock: Some(serial_ensure_finished_test_unlock),
+            on_skip: Some(serial_ensure_finished_test_on_skip),
+            c_size,
+            job_id,
+        }
+    }
+
     unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) {
         unsafe {
             (*context.cast::<WaitForLdmTestContext>())
@@ -3762,6 +3954,42 @@ mod tests {
         assert_eq!(context.events, vec!["broadcast", "unlock"]);
     }
 
+    #[test]
+    fn serial_ensure_finished_runs_failed_job_cleanup_in_order() {
+        let mut context = SerialEnsureFinishedTestContext { events: Vec::new() };
+        let mut next_job_id = 4;
+        let state = serial_ensure_finished_test_state(&mut context, &mut next_job_id, 4, 123);
+
+        unsafe { ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state) };
+
+        assert_eq!(next_job_id, 5);
+        assert_eq!(
+            context.events,
+            vec![
+                "lock",
+                "skip",
+                "broadcast",
+                "ldm_lock",
+                "clear_ldm",
+                "ldm_signal",
+                "ldm_unlock",
+                "unlock"
+            ]
+        );
+    }
+
+    #[test]
+    fn serial_ensure_finished_unlocks_without_cleanup_for_later_job() {
+        let mut context = SerialEnsureFinishedTestContext { events: Vec::new() };
+        let mut next_job_id = 6;
+        let state = serial_ensure_finished_test_state(&mut context, &mut next_job_id, 4, 123);
+
+        unsafe { ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state) };
+
+        assert_eq!(next_job_id, 6);
+        assert_eq!(context.events, vec!["lock", "unlock"]);
+    }
+
     fn record_compression_job_event(state: &Rc<RefCell<MockCompressionJob>>, event: &'static str) {
         state.borrow_mut().events.push(event);
     }