From a6b58a7dfc81754658bd21a29e18b6ce41b937e9 Mon Sep 17 00:00:00 2001
From: ddidderr <ddidderr@paul.network>
Date: Sun, 19 Jul 2026 19:44:19 +0200
Subject: [PATCH] feat(mt): move serial failure turn policy into Rust

Move the scalar failed-job skip decision and wrapping next-job transition into
Rust. C keeps the serial mutex, error assertion, condition broadcast, and LDM
window cleanup around the projected result.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1; cargo test --manifest-path rust/Cargo.toml
- ulimit -v 41943040; CARGO_BUILD_JOBS=1; cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; make -B -C programs -j1 zstd
- ulimit -v 41943040; make -C tests -j1 test-zstream ZSTREAM_TESTTIME=-T1s
---
 lib/compress/zstdmt_compress.c | 13 ++++++--
 rust/src/zstdmt_compress.rs    | 60 ++++++++++++++++++++++++++++++++++
 2 files changed, 71 insertions(+), 2 deletions(-)

diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c
index e2700d690..df1999e23 100644
--- a/lib/compress/zstdmt_compress.c
+++ b/lib/compress/zstdmt_compress.c
@@ -368,6 +368,12 @@ void ZSTDMT_rust_serialStateGenSequences(
         ZSTDMT_serialGenerateLdmFn generateLdm,
         ZSTDMT_serialUpdateChecksumFn updateChecksum,
         ZSTDMT_serialAdvanceFn advance);
+typedef struct {
+    unsigned skip;
+    unsigned nextJobID;
+} ZSTDMT_RustSerialEnsureFinishedResult;
+ZSTDMT_RustSerialEnsureFinishedResult ZSTDMT_rust_serialStateEnsureFinished(
+        unsigned nextJobID, unsigned jobID);
 typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque);
 typedef int (*ZSTDMT_waitForLdmOverlapFn)(
         void* opaque, void* bufferStart, size_t bufferCapacity);
@@ -975,11 +981,14 @@ ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an
 static void ZSTDMT_serialState_ensureFinished(SerialState* serialState,
                                               unsigned jobID, size_t cSize)
 {
+    ZSTDMT_RustSerialEnsureFinishedResult result;
     ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex);
-    if (serialState->nextJobID <= jobID) {
+    result = ZSTDMT_rust_serialStateEnsureFinished(
+            serialState->nextJobID, jobID);
+    if (result.skip) {
         assert(ZSTD_isError(cSize)); (void)cSize;
         DEBUGLOG(5, "Skipping past job %u because of error", jobID);
-        serialState->nextJobID = jobID + 1;
+        serialState->nextJobID = result.nextJobID;
         ZSTD_pthread_cond_broadcast(&serialState->cond);
 
         ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex);
diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs
index dcdc8eee9..c8d308a6e 100644
--- a/rust/src/zstdmt_compress.rs
+++ b/rust/src/zstdmt_compress.rs
@@ -90,6 +90,13 @@ pub type ZSTDMT_serialGenerateLdmFn =
 pub type ZSTDMT_serialUpdateChecksumFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize);
 pub type ZSTDMT_serialAdvanceFn = unsafe extern "C" fn(*mut c_void);
 
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_serialStateEnsureFinishedResult {
+    pub skip: c_uint,
+    pub nextJobID: c_uint,
+}
+
 type ZSTDMT_waitForLdmLockFn = unsafe extern "C" fn(*mut c_void);
 type ZSTDMT_waitForLdmOverlapFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> c_int;
 type ZSTDMT_waitForLdmWaitFn = unsafe extern "C" fn(*mut c_void);
@@ -836,6 +843,27 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateGenSequences(
     }
 }
 
+/// Decide whether a failed job must advance the shared serial turn.  C keeps
+/// 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,
+    job_id: c_uint,
+) -> ZSTDMT_serialStateEnsureFinishedResult {
+    if next_job_id <= job_id {
+        ZSTDMT_serialStateEnsureFinishedResult {
+            skip: 1,
+            nextJobID: job_id.wrapping_add(1),
+        }
+    } else {
+        ZSTDMT_serialStateEnsureFinishedResult {
+            skip: 0,
+            nextJobID: next_job_id,
+        }
+    }
+}
+
 /// C ABI entry point for the worker-job orchestration. C supplies callbacks
 /// that keep the private descriptor, pools, mutexes, and codec operations on
 /// the C side of this narrow projection.
@@ -3481,6 +3509,38 @@ mod tests {
         assert_eq!(&*empty_events.borrow(), &["ldm", "advance"]);
     }
 
+    #[test]
+    fn serial_state_ensure_finished_owns_skip_and_wrapping_policy() {
+        assert_eq!(
+            ZSTDMT_rust_serialStateEnsureFinished(4, 4),
+            ZSTDMT_serialStateEnsureFinishedResult {
+                skip: 1,
+                nextJobID: 5,
+            }
+        );
+        assert_eq!(
+            ZSTDMT_rust_serialStateEnsureFinished(3, 9),
+            ZSTDMT_serialStateEnsureFinishedResult {
+                skip: 1,
+                nextJobID: 10,
+            }
+        );
+        assert_eq!(
+            ZSTDMT_rust_serialStateEnsureFinished(11, 9),
+            ZSTDMT_serialStateEnsureFinishedResult {
+                skip: 0,
+                nextJobID: 11,
+            }
+        );
+        assert_eq!(
+            ZSTDMT_rust_serialStateEnsureFinished(c_uint::MAX, c_uint::MAX),
+            ZSTDMT_serialStateEnsureFinishedResult {
+                skip: 1,
+                nextJobID: 0,
+            }
+        );
+    }
+
     #[test]
     fn compression_job_stops_on_non_first_chunk_error_and_cleans_up() {
         let state = Rc::new(RefCell::new(MockCompressionJob::default()));