feat(mt): move job-resource slot iteration into Rust

Move the deterministic inclusive job-table cleanup loop into the Rust MT seam. C retains each descriptor's mutex and condition variables, buffer-pool release, descriptor clearing, and the containing context's final state reset behind a per-slot callback.

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

@@ -379,6 +379,9 @@ typedef void (*ZSTDMT_waitForJobCompleteFn)(
 unsigned ZSTDMT_rust_waitForAllJobsCompleted(
         unsigned doneJobID, unsigned nextJobID, unsigned jobIDMask,
         void* opaque, ZSTDMT_waitForJobCompleteFn waitForJob);
+typedef void (*ZSTDMT_releaseJobResourceFn)(void* opaque, unsigned jobID);
+void ZSTDMT_rust_releaseAllJobResources(
+        unsigned jobIDMask, void* opaque, ZSTDMT_releaseJobResourceFn releaseJob);
 typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque);
 typedef int (*ZSTDMT_waitForLdmOverlapFn)(
         void* opaque, void* bufferStart, size_t bufferCapacity);
@@ -1483,25 +1486,29 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbWorkers, ZSTD_customMem cMem,
 }
 
 
+static void ZSTDMT_releaseJobResource(void* opaque, unsigned jobID)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    /* Copy the mutex/cond out */
+    ZSTD_pthread_mutex_t const mutex = mtctx->jobs[jobID].job_mutex;
+    ZSTD_pthread_cond_t const cond = mtctx->jobs[jobID].job_cond;
+
+    DEBUGLOG(4, "job%02u: release dst address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].dstBuff.start);
+    ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff);
+
+    /* Clear the job description, but keep the mutex/cond */
+    ZSTD_memset(&mtctx->jobs[jobID], 0, sizeof(mtctx->jobs[jobID]));
+    mtctx->jobs[jobID].job_mutex = mutex;
+    mtctx->jobs[jobID].job_cond = cond;
+}
+
 /* ZSTDMT_releaseAllJobResources() :
  * note : ensure all workers are killed first ! */
 static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx)
 {
-    unsigned jobID;
     DEBUGLOG(3, "ZSTDMT_releaseAllJobResources");
-    for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) {
+    ZSTDMT_rust_releaseAllJobResources(
-        /* Copy the mutex/cond out */
+            mtctx->jobIDMask, mtctx, ZSTDMT_releaseJobResource);
-        ZSTD_pthread_mutex_t const mutex = mtctx->jobs[jobID].job_mutex;
-        ZSTD_pthread_cond_t const cond = mtctx->jobs[jobID].job_cond;
-
-        DEBUGLOG(4, "job%02u: release dst address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].dstBuff.start);
-        ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff);
-
-        /* Clear the job description, but keep the mutex/cond */
-        ZSTD_memset(&mtctx->jobs[jobID], 0, sizeof(mtctx->jobs[jobID]));
-        mtctx->jobs[jobID].job_mutex = mutex;
-        mtctx->jobs[jobID].job_cond = cond;
-    }
     mtctx->inBuff.buffer = g_nullBuffer;
     mtctx->inBuff.filled = 0;
     mtctx->allJobsCompleted = 1;

rust/src/zstdmt_compress.rs

@@ -97,6 +97,7 @@ pub struct ZSTDMT_serialStateEnsureFinishedResult {
     pub nextJobID: c_uint,
 }
 pub type ZSTDMT_waitForJobCompleteFn = unsafe extern "C" fn(*mut c_void, c_uint, c_uint);
+pub type ZSTDMT_releaseJobResourceFn = unsafe extern "C" fn(*mut c_void, 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;
@@ -886,6 +887,23 @@ pub unsafe extern "C" fn ZSTDMT_rust_waitForAllJobsCompleted(
     done_job_id
 }
 
+/// Visit every MT job slot so C can release its private descriptor resources.
+/// Rust owns the deterministic inclusive ring-table iteration; C retains the
+/// buffer-pool, mutex/condition, and descriptor cleanup side effects.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_releaseAllJobResources(
+    job_id_mask: c_uint,
+    opaque: *mut c_void,
+    release_job: Option<ZSTDMT_releaseJobResourceFn>,
+) {
+    let Some(release_job) = release_job else {
+        return;
+    };
+    for job_id in 0..=job_id_mask {
+        unsafe { release_job(opaque, 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.
@@ -3597,6 +3615,35 @@ mod tests {
         assert_eq!(context.events, vec![(3, 3), (0, 4), (1, 5)]);
     }
 
+    struct ReleaseAllJobsTestContext {
+        job_ids: Vec<c_uint>,
+    }
+
+    unsafe extern "C" fn release_all_jobs_test_callback(context: *mut c_void, job_id: c_uint) {
+        unsafe {
+            (*context.cast::<ReleaseAllJobsTestContext>())
+                .job_ids
+                .push(job_id);
+        }
+    }
+
+    #[test]
+    fn release_all_jobs_visits_each_slot_in_order() {
+        let mut context = ReleaseAllJobsTestContext {
+            job_ids: Vec::new(),
+        };
+
+        unsafe {
+            ZSTDMT_rust_releaseAllJobResources(
+                3,
+                (&mut context as *mut ReleaseAllJobsTestContext).cast(),
+                Some(release_all_jobs_test_callback),
+            )
+        };
+
+        assert_eq!(context.job_ids, vec![0, 1, 2, 3]);
+    }
+
     #[test]
     fn compression_job_stops_on_non_first_chunk_error_and_cleans_up() {
         let state = Rc::new(RefCell::new(MockCompressionJob::default()));