refactor(mt): move input-range commit policy to Rust

Move the ready/wrap branch and synchronization ordering from ZSTDMT_tryGetInputRange into Rust. The Rust policy now waits for the prefix range, invokes the C-owned prefix move, waits for the selected source range, and publishes the buffer; C retains memmove, private buffer fields, round-buffer state, and LDM callbacks behind the projection. Add ABI assertions and focused wrapped/ordinary ordering tests.

Test Plan: git diff --cached --check; focused Rust tests added; full capped Rust/native/original-suite verification follows.

lib/compress/zstdmt_compress.c

@@ -957,6 +957,19 @@ typedef struct {
     size_t bufferCapacity;
     size_t roundBufferPos;
 } ZSTDMT_RustTryGetInputRangeResult;
+typedef char ZSTDMT_rust_try_get_input_range_result_layout[
+        (offsetof(ZSTDMT_RustTryGetInputRangeResult, ready) == 0
+         && offsetof(ZSTDMT_RustTryGetInputRangeResult, movePrefix)
+                == sizeof(unsigned)
+         && offsetof(ZSTDMT_RustTryGetInputRangeResult, bufferStart)
+                == 2 * sizeof(unsigned)
+         && offsetof(ZSTDMT_RustTryGetInputRangeResult, bufferCapacity)
+                == 2 * sizeof(unsigned) + sizeof(void*)
+         && offsetof(ZSTDMT_RustTryGetInputRangeResult, roundBufferPos)
+                == 2 * sizeof(unsigned) + sizeof(void*) + sizeof(size_t)
+         && sizeof(ZSTDMT_RustTryGetInputRangeResult)
+                == 2 * sizeof(unsigned) + sizeof(void*) + 2 * sizeof(size_t))
+        ? 1 : -1];
 typedef void (*ZSTDMT_jobProjectionFn)(void* opaque, unsigned jobID,
                                       ZSTDMT_RustJobProjection* projection);
 ZSTDMT_RustInputRange ZSTDMT_rust_getInputDataInUse(
@@ -965,6 +978,19 @@ ZSTDMT_RustInputRange ZSTDMT_rust_getInputDataInUse(
         void* opaque, ZSTDMT_jobProjectionFn projectJob);
 ZSTDMT_RustTryGetInputRangeResult ZSTDMT_rust_tryGetInputRange(
         const ZSTDMT_RustTryGetInputRangeProjection* projection);
+typedef void (*ZSTDMT_inputRangeWaitFn)(
+        void* opaque, void* bufferStart, size_t bufferCapacity);
+typedef void (*ZSTDMT_inputRangeMovePrefixFn)(
+        void* opaque, void* destination, const void* source, size_t size,
+        size_t roundBufferPos);
+typedef void (*ZSTDMT_inputRangePublishBufferFn)(
+        void* opaque, void* bufferStart, size_t bufferCapacity);
+int ZSTDMT_rust_commitInputRange(
+        const ZSTDMT_RustTryGetInputRangeResult* result,
+        void* roundBufferStart, const void* prefixStart, size_t prefixSize,
+        void* opaque, ZSTDMT_inputRangeWaitFn waitForLdm,
+        ZSTDMT_inputRangeMovePrefixFn movePrefix,
+        ZSTDMT_inputRangePublishBufferFn publishBuffer);
 size_t ZSTDMT_rust_toFlushNow(unsigned doneJobID, unsigned nextJobID,
                               unsigned jobIDMask, void* opaque,
                               ZSTDMT_jobProjectionFn projectJob);
@@ -3296,6 +3322,43 @@ static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, Buffer buffer)
     ZSTDMT_rust_waitForLdmComplete(&state);
 }
 
+static void ZSTDMT_inputRangeWaitForLdm(
+        void* opaque, void* bufferStart, size_t bufferCapacity)
+{
+    ZSTDMT_waitForLdmComplete((ZSTDMT_CCtx*)opaque,
+                              (Buffer){ bufferStart, bufferCapacity });
+}
+
+static void ZSTDMT_inputRangeMovePrefix(
+        void* opaque, void* destination, const void* source, size_t size,
+        size_t roundBufferPos)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    /* ZSTD_invalidateRepCodes() doesn't work for extDict variants.
+     * Simply copy the prefix to the beginning in that case. */
+    ZSTD_memmove(destination, source, size);
+    mtctx->inBuff.prefix.start = (const BYTE*)destination;
+    mtctx->roundBuff.pos = roundBufferPos;
+}
+
+static void ZSTDMT_inputRangePublishBuffer(
+        void* opaque, void* bufferStart, size_t bufferCapacity)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    Buffer const buffer = { bufferStart, bufferCapacity };
+
+    DEBUGLOG(5, "Using prefix range [%zx, %zx)",
+                (size_t)mtctx->inBuff.prefix.start,
+                (size_t)mtctx->inBuff.prefix.start + mtctx->inBuff.prefix.size);
+    DEBUGLOG(5, "Using source range [%zx, %zx)",
+                (size_t)buffer.start,
+                (size_t)buffer.start + buffer.capacity);
+
+    mtctx->inBuff.buffer = buffer;
+    mtctx->inBuff.filled = 0;
+    assert(mtctx->roundBuff.pos + buffer.capacity <= mtctx->roundBuff.capacity);
+}
+
 /**
  * Attempts to set the inBuff to the next section to fill.
  * If any part of the new section is still in use we give up.
@@ -3316,47 +3379,22 @@ static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx)
     };
     ZSTDMT_RustTryGetInputRangeResult const result =
         ZSTDMT_rust_tryGetInputRange(&projection);
-    Buffer buffer;
 
     DEBUGLOG(5, "ZSTDMT_tryGetInputRange");
     assert(mtctx->inBuff.buffer.start == NULL);
 
-    if (!result.ready) {
+    if (!ZSTDMT_rust_commitInputRange(
+                &result,
+                mtctx->roundBuff.buffer,
+                mtctx->inBuff.prefix.start,
+                mtctx->inBuff.prefix.size,
+                mtctx,
+                ZSTDMT_inputRangeWaitForLdm,
+                ZSTDMT_inputRangeMovePrefix,
+                ZSTDMT_inputRangePublishBuffer)) {
         DEBUGLOG(5, "Waiting for buffer...");
         return 0;
     }
-
-    if (result.movePrefix) {
-        /* ZSTD_invalidateRepCodes() doesn't work for extDict variants.
-         * Simply copy the prefix to the beginning in that case.
-         */
-        BYTE* const start = (BYTE*)mtctx->roundBuff.buffer;
-        size_t const prefixSize = mtctx->inBuff.prefix.size;
-
-        buffer.start = start;
-        buffer.capacity = prefixSize;
-        ZSTDMT_waitForLdmComplete(mtctx, buffer);
-        ZSTD_memmove(start, mtctx->inBuff.prefix.start, prefixSize);
-        mtctx->inBuff.prefix.start = start;
-        mtctx->roundBuff.pos = result.roundBufferPos;
-    }
-
-    buffer.start = (BYTE*)result.bufferStart;
-    buffer.capacity = result.bufferCapacity;
-
-    ZSTDMT_waitForLdmComplete(mtctx, buffer);
-
-    DEBUGLOG(5, "Using prefix range [%zx, %zx)",
-                (size_t)mtctx->inBuff.prefix.start,
-                (size_t)mtctx->inBuff.prefix.start + mtctx->inBuff.prefix.size);
-    DEBUGLOG(5, "Using source range [%zx, %zx)",
-                (size_t)buffer.start,
-                (size_t)buffer.start + buffer.capacity);
-
-
-    mtctx->inBuff.buffer = buffer;
-    mtctx->inBuff.filled = 0;
-    assert(mtctx->roundBuff.pos + buffer.capacity <= mtctx->roundBuff.capacity);
     return 1;
 }
 

rust/src/zstdmt_compress.rs

@@ -1689,9 +1689,9 @@ pub struct ZSTDMT_tryGetInputRangeProjection {
     pub inUseSize: usize,
 }
 
-/// Result of the MT input-range selection.  C applies the returned decision,
+/// Result of the MT input-range selection. Rust also owns the ordering of the
-/// including synchronization waits, prefix copying, and private field
+/// synchronization waits and publication callbacks that commit this result;
-/// updates.
+/// C retains the private buffer mutations and synchronization state.
 #[repr(C)]
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub struct ZSTDMT_tryGetInputRangeResult {
@@ -1702,6 +1702,126 @@ pub struct ZSTDMT_tryGetInputRangeResult {
     pub roundBufferPos: usize,
 }
 
+const _: () = {
+    assert!(offset_of!(ZSTDMT_tryGetInputRangeResult, ready) == 0);
+    assert!(offset_of!(ZSTDMT_tryGetInputRangeResult, movePrefix) == size_of::<c_uint>());
+    assert!(
+        offset_of!(ZSTDMT_tryGetInputRangeResult, bufferStart)
+            == 2 * size_of::<c_uint>()
+    );
+    assert!(
+        offset_of!(ZSTDMT_tryGetInputRangeResult, bufferCapacity)
+            == 2 * size_of::<c_uint>() + size_of::<*mut c_void>()
+    );
+    assert!(
+        offset_of!(ZSTDMT_tryGetInputRangeResult, roundBufferPos)
+            == 2 * size_of::<c_uint>() + size_of::<*mut c_void>() + size_of::<usize>()
+    );
+    assert!(
+        size_of::<ZSTDMT_tryGetInputRangeResult>()
+            == 2 * size_of::<c_uint>() + size_of::<*mut c_void>() + 2 * size_of::<usize>()
+    );
+};
+
+pub type ZSTDMT_inputRangeWaitFn = unsafe extern "C" fn(
+    opaque: *mut c_void,
+    buffer_start: *mut c_void,
+    buffer_capacity: usize,
+);
+pub type ZSTDMT_inputRangeMovePrefixFn = unsafe extern "C" fn(
+    opaque: *mut c_void,
+    destination: *mut c_void,
+    source: *const c_void,
+    size: usize,
+    round_buffer_pos: usize,
+);
+pub type ZSTDMT_inputRangePublishBufferFn = unsafe extern "C" fn(
+    opaque: *mut c_void,
+    buffer_start: *mut c_void,
+    buffer_capacity: usize,
+);
+
+#[inline]
+fn commit_input_range_with<W, M, P>(
+    result: ZSTDMT_tryGetInputRangeResult,
+    round_buffer_start: *mut c_void,
+    prefix_start: *const c_void,
+    prefix_size: usize,
+    opaque: *mut c_void,
+    mut wait_for_ldm: W,
+    mut move_prefix: M,
+    mut publish_buffer: P,
+) -> c_int
+where
+    W: FnMut(*mut c_void, *mut c_void, usize),
+    M: FnMut(*mut c_void, *mut c_void, *const c_void, usize, usize),
+    P: FnMut(*mut c_void, *mut c_void, usize),
+{
+    if result.ready == 0 {
+        return 0;
+    }
+
+    if result.movePrefix != 0 {
+        wait_for_ldm(opaque, round_buffer_start, prefix_size);
+        move_prefix(
+            opaque,
+            round_buffer_start,
+            prefix_start,
+            prefix_size,
+            result.roundBufferPos,
+        );
+    }
+
+    wait_for_ldm(opaque, result.bufferStart, result.bufferCapacity);
+    publish_buffer(opaque, result.bufferStart, result.bufferCapacity);
+    1
+}
+
+/// Commit the selected MT input range in the original order. Rust owns the
+/// prefix-wrap branch and the wait/copy/wait/publish sequencing; C callbacks
+/// retain the private round-buffer, input-buffer, and LDM synchronization
+/// operations.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_commitInputRange(
+    result: *const ZSTDMT_tryGetInputRangeResult,
+    roundBufferStart: *mut c_void,
+    prefixStart: *const c_void,
+    prefixSize: usize,
+    opaque: *mut c_void,
+    waitForLdm: Option<ZSTDMT_inputRangeWaitFn>,
+    movePrefix: Option<ZSTDMT_inputRangeMovePrefixFn>,
+    publishBuffer: Option<ZSTDMT_inputRangePublishBufferFn>,
+) -> c_int {
+    let Some(result) = (unsafe { result.as_ref() }).copied() else {
+        return 0;
+    };
+    let (Some(wait_for_ldm), Some(move_prefix), Some(publish_buffer)) =
+        (waitForLdm, movePrefix, publishBuffer)
+    else {
+        return 0;
+    };
+    if opaque.is_null() {
+        return 0;
+    }
+
+    commit_input_range_with(
+        result,
+        roundBufferStart,
+        prefixStart,
+        prefixSize,
+        opaque,
+        |opaque, buffer_start, buffer_capacity| unsafe {
+            wait_for_ldm(opaque, buffer_start, buffer_capacity);
+        },
+        |opaque, destination, source, size, round_buffer_pos| unsafe {
+            move_prefix(opaque, destination, source, size, round_buffer_pos);
+        },
+        |opaque, buffer_start, buffer_capacity| unsafe {
+            publish_buffer(opaque, buffer_start, buffer_capacity);
+        },
+    )
+}
+
 pub type ZSTDMT_jobProjectionFn = unsafe extern "C" fn(
     opaque: *mut c_void,
     job_id: c_uint,
@@ -8944,6 +9064,115 @@ mod tests {
         assert_eq!(try_get_input_range_with(blocked_source).ready, 0);
     }
 
+    #[test]
+    fn commit_input_range_wraps_prefix_before_waiting_for_source() {
+        let mut round_buffer = [0u8; 32];
+        let prefix = [1u8; 4];
+        let round_buffer_start = round_buffer.as_mut_ptr();
+        let selected_buffer = round_buffer_start.wrapping_add(prefix.len());
+        let result = ZSTDMT_tryGetInputRangeResult {
+            ready: 1,
+            movePrefix: 1,
+            bufferStart: selected_buffer.cast(),
+            bufferCapacity: 8,
+            roundBufferPos: prefix.len(),
+        };
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let wait_events = Rc::clone(&events);
+        let move_events = Rc::clone(&events);
+        let publish_events = Rc::clone(&events);
+        let wait_calls = Rc::new(RefCell::new(Vec::new()));
+        let wait_call_records = Rc::clone(&wait_calls);
+
+        let committed = commit_input_range_with(
+            result,
+            round_buffer_start.cast(),
+            prefix.as_ptr().cast(),
+            prefix.len(),
+            ptr::null_mut(),
+            move |_, buffer_start, buffer_capacity| {
+                wait_call_records
+                    .borrow_mut()
+                    .push((buffer_start, buffer_capacity));
+                wait_events.borrow_mut().push("wait");
+            },
+            move |_, destination, source, size, round_buffer_pos| {
+                assert_eq!(destination, round_buffer_start.cast());
+                assert_eq!(source, prefix.as_ptr().cast());
+                assert_eq!(size, prefix.len());
+                assert_eq!(round_buffer_pos, prefix.len());
+                unsafe {
+                    ptr::copy_nonoverlapping(
+                        source.cast::<u8>(),
+                        destination.cast::<u8>(),
+                        size,
+                    );
+                }
+                move_events.borrow_mut().push("move");
+            },
+            move |_, buffer_start, buffer_capacity| {
+                assert_eq!(buffer_start, selected_buffer.cast());
+                assert_eq!(buffer_capacity, 8);
+                publish_events.borrow_mut().push("publish");
+            },
+        );
+
+        assert_eq!(committed, 1);
+        assert_eq!(&*events.borrow(), &["wait", "move", "wait", "publish"]);
+        assert_eq!(
+            &*wait_calls.borrow(),
+            &[
+                (round_buffer_start.cast(), prefix.len()),
+                (selected_buffer.cast(), 8),
+            ]
+        );
+        assert_eq!(&round_buffer[..prefix.len()], &prefix);
+    }
+
+    #[test]
+    fn commit_input_range_without_wrap_waits_once_before_publish() {
+        let mut round_buffer = [0u8; 32];
+        let round_buffer_start = round_buffer.as_mut_ptr();
+        let selected_buffer = round_buffer_start.wrapping_add(8);
+        let result = ZSTDMT_tryGetInputRangeResult {
+            ready: 1,
+            bufferStart: selected_buffer.cast(),
+            bufferCapacity: 8,
+            roundBufferPos: 8,
+            ..ZSTDMT_tryGetInputRangeResult::default()
+        };
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let wait_events = Rc::clone(&events);
+        let move_called = Rc::new(RefCell::new(false));
+        let move_called_by_callback = Rc::clone(&move_called);
+        let publish_events = Rc::clone(&events);
+
+        let committed = commit_input_range_with(
+            result,
+            round_buffer_start.cast(),
+            ptr::null(),
+            0,
+            ptr::null_mut(),
+            move |_, buffer_start, buffer_capacity| {
+                assert_eq!(buffer_start, selected_buffer.cast());
+                assert_eq!(buffer_capacity, 8);
+                wait_events.borrow_mut().push("wait");
+            },
+            move |_, _, _, _, _| {
+                *move_called_by_callback.borrow_mut() = true;
+            },
+            move |_, buffer_start, buffer_capacity| {
+                assert_eq!(buffer_start, selected_buffer.cast());
+                assert_eq!(buffer_capacity, 8);
+                publish_events.borrow_mut().push("publish");
+            },
+        );
+
+        assert_eq!(committed, 1);
+        assert_eq!(&*events.borrow(), &["wait", "publish"]);
+        assert!(!*move_called.borrow());
+    }
+
     #[test]
     fn to_flush_now_projects_oldest_slot_and_normalizes_errors() {
         let projection = ZSTDMT_jobProjection {