feat(mt): move job resource acquisition policy to Rust

MT worker execution used to combine CCtx acquisition, raw-sequence-store
acquisition, destination-buffer allocation, frame-header publication, and the
LDM resource check in one C callback. That made the resource order and
short-circuit policy another C-owned implementation boundary even though the
operations themselves must remain private to C.

Project the scalar LDM and destination-presence inputs into Rust and let Rust
own the ordering and error normalization. The C callbacks now only acquire
private resources, report readiness, allocate the destination buffer, and
publish the frame-header destination. The ordering remains compatible with the
worker cleanup path: both pool gets happen first, CCtx failure stops before
destination work, destination failure stops before publication, and the LDM
sequence-store check runs after publication.

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 +nightly fmt --manifest-path rust/Cargo.toml --all -- --check` -- passed
- Capped GCC and Clang syntax-only checks for `lib/compress/zstdmt_compress.c` -- passed by the worker
- `git diff --cached --check` -- passed
- Native build and full upstream tests are deferred to the post-batch serial verification.

lib/compress/zstdmt_compress.c

@@ -294,9 +294,33 @@ typedef char ZSTDMT_compression_job_parameters_layout[
                 == 2 * sizeof(int) + sizeof(unsigned))
         ? 1 : -1];
 
+typedef struct {
+    int ldmEnabled;
+    int hasDestination;
+} ZSTDMT_RustCompressionJobResourceProjection;
+typedef char ZSTDMT_compression_job_resource_projection_layout[
+        (offsetof(ZSTDMT_RustCompressionJobResourceProjection, ldmEnabled) == 0
+         && offsetof(ZSTDMT_RustCompressionJobResourceProjection, hasDestination)
+                == sizeof(int)
+         && sizeof(ZSTDMT_RustCompressionJobResourceProjection)
+                == 2 * sizeof(int))
+        ? 1 : -1];
+
+typedef int (*ZSTDMT_compressionJobResourceReadyFn)(void* opaque);
+
 ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters(
         unsigned jobID, int checksumFlag, int ldmEnable, unsigned nbWorkers);
 
+size_t ZSTDMT_rust_compressionJobAcquireResources(
+        const ZSTDMT_RustCompressionJobResourceProjection* projection,
+        void* opaque,
+        ZSTDMT_compressionJobVoidFn acquireCCtx,
+        ZSTDMT_compressionJobVoidFn acquireRawSeqStore,
+        ZSTDMT_compressionJobResourceReadyFn hasCCtx,
+        ZSTDMT_compressionJobStepFn acquireDestinationBuffer,
+        ZSTDMT_compressionJobResourceReadyFn hasRawSeqStore,
+        ZSTDMT_compressionJobVoidFn publishDestination);
+
 size_t ZSTDMT_rust_compressionJobBegin(
         const ZSTDMT_RustCompressionJobBeginProjection* projection,
         void* opaque,
@@ -1664,28 +1688,73 @@ typedef struct {
     ZSTDMT_RustCompressionJobFrameHeaderState frameHeaderState;
 } ZSTDMT_compressionJobState;
 
+static void ZSTDMT_compressionJobAcquireCCtx(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    state->cctx = ZSTDMT_getCCtx(state->job->cctxPool);
+}
+
+static void ZSTDMT_compressionJobAcquireRawSeqStore(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    state->rawSeqStore = ZSTDMT_getSeq(state->job->seqPool);
+}
+
+static int ZSTDMT_compressionJobHasCCtx(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    return state->cctx != NULL;
+}
+
+static size_t ZSTDMT_compressionJobAcquireDestinationBuffer(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    ZSTDMT_jobDescription* const job = state->job;
+    state->dstBuff = ZSTDMT_getBuffer(job->bufPool);
+    if (state->dstBuff.start == NULL) return ERROR(memory_allocation);
+    job->dstBuff = state->dstBuff;
+    return 0;
+}
+
+static int ZSTDMT_compressionJobHasRawSeqStore(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    return state->rawSeqStore.seq != NULL;
+}
+
+static void ZSTDMT_compressionJobPublishDestination(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    state->frameHeaderState.dst = state->dstBuff.start;
+    state->frameHeaderState.dstCapacity = state->dstBuff.capacity;
+}
+
 static size_t ZSTDMT_compressionJobAcquireResources(void* opaque)
 {
     ZSTDMT_compressionJobState* const state =
         (ZSTDMT_compressionJobState*)opaque;
     ZSTDMT_jobDescription* const job = state->job;
 
-    state->cctx = ZSTDMT_getCCtx(job->cctxPool);
-    state->rawSeqStore = ZSTDMT_getSeq(job->seqPool);
     state->dstBuff = job->dstBuff;
     DEBUGLOG(5, "ZSTDMT_compressionJob: job %u", job->jobID);
-    if (state->cctx == NULL) return ERROR(memory_allocation);
+    ZSTDMT_RustCompressionJobResourceProjection const projection = {
-    if (state->dstBuff.start == NULL) {
+        state->jobParams.ldmParams.enableLdm,
-        state->dstBuff = ZSTDMT_getBuffer(job->bufPool);
+        state->dstBuff.start != NULL
-        if (state->dstBuff.start == NULL) return ERROR(memory_allocation);
+    };
-        job->dstBuff = state->dstBuff;
+    return ZSTDMT_rust_compressionJobAcquireResources(
-    }
+            &projection, state,
-    state->frameHeaderState.dst = state->dstBuff.start;
+            ZSTDMT_compressionJobAcquireCCtx,
-    state->frameHeaderState.dstCapacity = state->dstBuff.capacity;
+            ZSTDMT_compressionJobAcquireRawSeqStore,
-    if (state->jobParams.ldmParams.enableLdm == ZSTD_ps_enable &&
+            ZSTDMT_compressionJobHasCCtx,
-        state->rawSeqStore.seq == NULL)
+            ZSTDMT_compressionJobAcquireDestinationBuffer,
-        return ERROR(memory_allocation);
+            ZSTDMT_compressionJobHasRawSeqStore,
-    return 0;
+            ZSTDMT_compressionJobPublishDestination);
 }
 
 static void ZSTDMT_compressionJobPrepareParameters(void* opaque)

rust/src/zstdmt_compress.rs

@@ -87,6 +87,28 @@ const _: () = {
     );
 };
 
+/// Scalar inputs for worker-resource acquisition. The worker descriptor,
+/// synchronization, pools, and destination storage remain private to C;
+/// Rust owns the acquisition/publication order and error short-circuiting.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_compressionJobResourceProjection {
+    pub ldmEnabled: c_int,
+    pub hasDestination: c_int,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTDMT_compressionJobResourceProjection, ldmEnabled) == 0);
+    assert!(
+        offset_of!(ZSTDMT_compressionJobResourceProjection, hasDestination) == size_of::<c_int>()
+    );
+    assert!(size_of::<ZSTDMT_compressionJobResourceProjection>() == 2 * size_of::<c_int>());
+};
+
+pub type ZSTDMT_compressionJobResourceVoidFn = unsafe extern "C" fn(*mut c_void);
+pub type ZSTDMT_compressionJobResourceStepFn = unsafe extern "C" fn(*mut c_void) -> usize;
+pub type ZSTDMT_compressionJobResourceReadyFn = unsafe extern "C" fn(*mut c_void) -> c_int;
+
 #[inline]
 fn prepare_compression_job_parameters(
     job_id: c_uint,
@@ -117,6 +139,100 @@ pub extern "C" fn ZSTDMT_rust_prepareCompressionJobParameters(
     prepare_compression_job_parameters(jobID, checksumFlag, ldmEnable, nbWorkers)
 }
 
+#[inline]
+fn compression_job_acquire_resources_with<A, S, C, B, R, P>(
+    projection: ZSTDMT_compressionJobResourceProjection,
+    mut acquire_cctx: A,
+    mut acquire_raw_seq_store: S,
+    mut has_cctx: C,
+    mut acquire_destination_buffer: B,
+    mut has_raw_seq_store: R,
+    mut publish_destination: P,
+) -> usize
+where
+    A: FnMut(),
+    S: FnMut(),
+    C: FnMut() -> bool,
+    B: FnMut() -> usize,
+    R: FnMut() -> bool,
+    P: FnMut(),
+{
+    /* Keep both pool gets before checking the CCtx, matching the C worker
+     * cleanup path that releases every resource acquired for the job. */
+    acquire_cctx();
+    acquire_raw_seq_store();
+    if !has_cctx() {
+        return ERROR(ZstdErrorCode::MemoryAllocation);
+    }
+
+    if projection.hasDestination == 0 {
+        let error = acquire_destination_buffer();
+        if ERR_isError(error) {
+            return error;
+        }
+    }
+    publish_destination();
+
+    if projection.ldmEnabled != 0 && !has_raw_seq_store() {
+        return ERROR(ZstdErrorCode::MemoryAllocation);
+    }
+    0
+}
+
+/// Own worker-resource acquisition policy while C retains pool, descriptor,
+/// synchronization, and allocator operations behind callbacks. The order is
+/// intentionally: acquire both worker resources, validate the CCtx, acquire a
+/// destination only when absent, publish its scalar destination projection,
+/// then validate the LDM sequence store.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_compressionJobAcquireResources(
+    projection: *const ZSTDMT_compressionJobResourceProjection,
+    opaque: *mut c_void,
+    acquire_cctx: Option<ZSTDMT_compressionJobResourceVoidFn>,
+    acquire_raw_seq_store: Option<ZSTDMT_compressionJobResourceVoidFn>,
+    has_cctx: Option<ZSTDMT_compressionJobResourceReadyFn>,
+    acquire_destination_buffer: Option<ZSTDMT_compressionJobResourceStepFn>,
+    has_raw_seq_store: Option<ZSTDMT_compressionJobResourceReadyFn>,
+    publish_destination: Option<ZSTDMT_compressionJobResourceVoidFn>,
+) -> usize {
+    let Some(projection) = (unsafe { projection.as_ref() }).copied() else {
+        return ERROR(ZstdErrorCode::Generic);
+    };
+    let (
+        Some(acquire_cctx),
+        Some(acquire_raw_seq_store),
+        Some(has_cctx),
+        Some(acquire_destination_buffer),
+        Some(has_raw_seq_store),
+        Some(publish_destination),
+    ) = (
+        acquire_cctx,
+        acquire_raw_seq_store,
+        has_cctx,
+        acquire_destination_buffer,
+        has_raw_seq_store,
+        publish_destination,
+    )
+    else {
+        return ERROR(ZstdErrorCode::Generic);
+    };
+    if opaque.is_null() {
+        return ERROR(ZstdErrorCode::Generic);
+    }
+
+    unsafe {
+        compression_job_acquire_resources_with(
+            projection,
+            || acquire_cctx(opaque),
+            || acquire_raw_seq_store(opaque),
+            || has_cctx(opaque) != 0,
+            || acquire_destination_buffer(opaque),
+            || has_raw_seq_store(opaque) != 0,
+            || publish_destination(opaque),
+        )
+    }
+}
+
 /// Scalar job state used by the Rust compression-job scheduler.
 ///
 /// The job descriptor, pools, synchronization, and codec state remain
@@ -5510,6 +5626,151 @@ mod tests {
         assert_eq!(&output[result..], &[0xa5; 12]);
     }
 
+    #[test]
+    fn compression_job_resources_order_pool_gets_destination_and_ldm_check() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let acquire_cctx_events = Rc::clone(&events);
+        let acquire_seq_events = Rc::clone(&events);
+        let cctx_ready_events = Rc::clone(&events);
+        let destination_events = Rc::clone(&events);
+        let seq_ready_events = Rc::clone(&events);
+        let publish_events = Rc::clone(&events);
+
+        let result = compression_job_acquire_resources_with(
+            ZSTDMT_compressionJobResourceProjection {
+                ldmEnabled: 1,
+                hasDestination: 0,
+            },
+            move || acquire_cctx_events.borrow_mut().push("cctx"),
+            move || acquire_seq_events.borrow_mut().push("seq"),
+            move || {
+                cctx_ready_events.borrow_mut().push("cctx-ready");
+                true
+            },
+            move || {
+                destination_events.borrow_mut().push("destination");
+                0
+            },
+            move || {
+                seq_ready_events.borrow_mut().push("seq-ready");
+                true
+            },
+            move || publish_events.borrow_mut().push("publish"),
+        );
+
+        assert_eq!(result, 0);
+        assert_eq!(
+            &*events.borrow(),
+            &[
+                "cctx",
+                "seq",
+                "cctx-ready",
+                "destination",
+                "publish",
+                "seq-ready",
+            ]
+        );
+    }
+
+    #[test]
+    fn compression_job_resources_reject_missing_cctx_after_both_pool_gets() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let acquire_cctx_events = Rc::clone(&events);
+        let acquire_seq_events = Rc::clone(&events);
+        let cctx_ready_events = Rc::clone(&events);
+        let expected_error = ERROR(ZstdErrorCode::MemoryAllocation);
+
+        let result = compression_job_acquire_resources_with(
+            ZSTDMT_compressionJobResourceProjection {
+                ldmEnabled: 1,
+                hasDestination: 0,
+            },
+            move || acquire_cctx_events.borrow_mut().push("cctx"),
+            move || acquire_seq_events.borrow_mut().push("seq"),
+            move || {
+                cctx_ready_events.borrow_mut().push("cctx-ready");
+                false
+            },
+            || panic!("CCtx failure must stop before destination acquisition"),
+            || panic!("CCtx failure must stop before LDM validation"),
+            || panic!("CCtx failure must stop before destination publication"),
+        );
+
+        assert_eq!(result, expected_error);
+        assert_eq!(&*events.borrow(), &["cctx", "seq", "cctx-ready"]);
+    }
+
+    #[test]
+    fn compression_job_resources_stops_before_publication_on_destination_error() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let acquire_cctx_events = Rc::clone(&events);
+        let acquire_seq_events = Rc::clone(&events);
+        let cctx_ready_events = Rc::clone(&events);
+        let destination_events = Rc::clone(&events);
+        let expected_error = ERROR(ZstdErrorCode::MemoryAllocation);
+
+        let result = compression_job_acquire_resources_with(
+            ZSTDMT_compressionJobResourceProjection {
+                ldmEnabled: 1,
+                hasDestination: 0,
+            },
+            move || acquire_cctx_events.borrow_mut().push("cctx"),
+            move || acquire_seq_events.borrow_mut().push("seq"),
+            move || {
+                cctx_ready_events.borrow_mut().push("cctx-ready");
+                true
+            },
+            move || {
+                destination_events.borrow_mut().push("destination");
+                expected_error
+            },
+            || panic!("destination failure must stop before LDM validation"),
+            || panic!("destination failure must stop before publication"),
+        );
+
+        assert_eq!(result, expected_error);
+        assert_eq!(
+            &*events.borrow(),
+            &["cctx", "seq", "cctx-ready", "destination"]
+        );
+    }
+
+    #[test]
+    fn compression_job_resources_publishes_before_missing_ldm_store_error() {
+        let events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let acquire_cctx_events = Rc::clone(&events);
+        let acquire_seq_events = Rc::clone(&events);
+        let cctx_ready_events = Rc::clone(&events);
+        let seq_ready_events = Rc::clone(&events);
+        let publish_events = Rc::clone(&events);
+        let expected_error = ERROR(ZstdErrorCode::MemoryAllocation);
+
+        let result = compression_job_acquire_resources_with(
+            ZSTDMT_compressionJobResourceProjection {
+                ldmEnabled: 1,
+                hasDestination: 1,
+            },
+            move || acquire_cctx_events.borrow_mut().push("cctx"),
+            move || acquire_seq_events.borrow_mut().push("seq"),
+            move || {
+                cctx_ready_events.borrow_mut().push("cctx-ready");
+                true
+            },
+            || panic!("an existing destination must skip buffer acquisition"),
+            move || {
+                seq_ready_events.borrow_mut().push("seq-ready");
+                false
+            },
+            move || publish_events.borrow_mut().push("publish"),
+        );
+
+        assert_eq!(result, expected_error);
+        assert_eq!(
+            &*events.borrow(),
+            &["cctx", "seq", "cctx-ready", "publish", "seq-ready"]
+        );
+    }
+
     #[test]
     fn compression_job_runs_first_job_stages_and_reports_final_block() {
         let state = Rc::new(RefCell::new(MockCompressionJob::default()));