refactor(compress): move toFlushNow policy to Rust

Move the worker-count branch of ZSTD_toFlushNow into the Rust policy layer.
The C adapter retains the private CCtx and multithread context, exposing only
the worker count and a callback for the MT flush query.  Single-threaded
contexts keep the existing zero result and the ABI projection is checked on
both sides.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml to_flush_now --lib
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; make -j1

lib/compress/zstd_compress.c

@@ -4020,18 +4020,40 @@ ZSTD_frameProgression ZSTD_getFrameProgression(const ZSTD_CCtx* cctx)
                                           cctx->producedCSize);
     }   }
 
+typedef size_t (*ZSTD_rust_toFlushNow_f)(void* context);
+typedef struct {
+    void* callbackContext;
+    int nbWorkers;
+    ZSTD_rust_toFlushNow_f toFlushNow;
+} ZSTD_rust_toFlushNowState;
+typedef char ZSTD_rust_to_flush_now_state_layout[
+        (offsetof(ZSTD_rust_toFlushNowState, callbackContext) == 0
+         && offsetof(ZSTD_rust_toFlushNowState, nbWorkers) == sizeof(void*)
+         && offsetof(ZSTD_rust_toFlushNowState, toFlushNow) == 2 * sizeof(void*)
+         && sizeof(ZSTD_rust_toFlushNowState) == 3 * sizeof(void*))
+        ? 1 : -1];
+size_t ZSTD_rust_toFlushNow(const ZSTD_rust_toFlushNowState* state);
+
+static size_t ZSTD_rust_toFlushNow_callback(void* context)
+{
+#ifdef ZSTD_MULTITHREAD
+    return ZSTDMT_toFlushNow(((ZSTD_CCtx*)context)->mtctx);
+#else
+    (void)context;
+    return 0;
+#endif
+}
+
 /*! ZSTD_toFlushNow()
  *  Only useful for multithreading scenarios currently (nbWorkers >= 1).
  */
 size_t ZSTD_toFlushNow(ZSTD_CCtx* cctx)
 {
-#ifdef ZSTD_MULTITHREAD
+    ZSTD_rust_toFlushNowState state;
-    if (cctx->appliedParams.nbWorkers > 0) {
+    state.callbackContext = cctx;
-        return ZSTDMT_toFlushNow(cctx->mtctx);
+    state.nbWorkers = (int)cctx->appliedParams.nbWorkers;
-    }
+    state.toFlushNow = ZSTD_rust_toFlushNow_callback;
-#endif
+    return ZSTD_rust_toFlushNow(&state);
-    (void)cctx;
-    return 0;   /* over-simplification; could also check if context is currently running in streaming mode, and in which case, report how many bytes are left to be flushed within output buffer */
 }
 
 static void ZSTD_assertEqualCParams(ZSTD_compressionParameters cParams1,

rust/src/zstd_compress.rs

@@ -6275,6 +6275,38 @@ pub extern "C" fn ZSTD_rust_frameProgression(
     frame_progression(consumed_src_size, buffered, produced_c_size)
 }
 
+type ToFlushNowFn = unsafe extern "C" fn(*mut c_void) -> usize;
+
+/// Explicit projection for the `ZSTD_toFlushNow` multithread policy.
+///
+/// Rust owns the worker-count branch; C retains the private CCtx and MT
+/// context behind the callback.
+#[repr(C)]
+pub struct ZSTD_rust_toFlushNowState {
+    callback_context: *mut c_void,
+    nb_workers: c_int,
+    to_flush_now: ToFlushNowFn,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTD_rust_toFlushNowState, callback_context) == 0);
+    assert!(offset_of!(ZSTD_rust_toFlushNowState, nb_workers) == size_of::<usize>());
+    assert!(offset_of!(ZSTD_rust_toFlushNowState, to_flush_now) == 2 * size_of::<usize>());
+    assert!(size_of::<ZSTD_rust_toFlushNowState>() == 3 * size_of::<usize>());
+};
+
+#[no_mangle]
+pub unsafe extern "C" fn ZSTD_rust_toFlushNow(state: *const ZSTD_rust_toFlushNowState) -> usize {
+    if state.is_null() {
+        return ERROR(ZstdErrorCode::Generic);
+    }
+    let state = unsafe { &*state };
+    if state.nb_workers <= 0 {
+        return 0;
+    }
+    unsafe { (state.to_flush_now)(state.callback_context) }
+}
+
 #[inline]
 fn window_correct_overflow(curr: u32, cycle_log: u32, max_dist: u32) -> u32 {
     let cycle_size = 1u32.wrapping_shl(cycle_log);
@@ -14272,6 +14304,37 @@ mod tests {
         assert_eq!(progression.nbActiveWorkers, 0);
     }
 
+    unsafe extern "C" fn to_flush_now_test_callback(context: *mut c_void) -> usize {
+        let calls = unsafe { &mut *context.cast::<usize>() };
+        *calls += 1;
+        17
+    }
+
+    #[test]
+    fn to_flush_now_only_calls_mt_callback_for_workers() {
+        let mut calls: usize = 0;
+        let mut state = ZSTD_rust_toFlushNowState {
+            callback_context: &mut calls as *mut usize as *mut c_void,
+            nb_workers: 0,
+            to_flush_now: to_flush_now_test_callback,
+        };
+
+        assert_eq!(unsafe { ZSTD_rust_toFlushNow(&state) }, 0);
+        assert_eq!(calls, 0);
+
+        state.nb_workers = 1;
+        assert_eq!(unsafe { ZSTD_rust_toFlushNow(&state) }, 17);
+        assert_eq!(calls, 1);
+    }
+
+    #[test]
+    fn to_flush_now_rejects_a_null_state() {
+        assert_eq!(
+            unsafe { ZSTD_rust_toFlushNow(ptr::null()) },
+            ERROR(ZstdErrorCode::Generic)
+        );
+    }
+
     #[test]
     fn next_input_size_hint_uses_remaining_stable_block_capacity() {
         let hint = ZSTD_rust_nextInputSizeHint(ZSTD_BM_STABLE, 256, 37, 99, 12);