feat(compress): move optimal block policy to Rust

Move the frame chunk pre-split policy behind a scalar Rust ABI while retaining C ownership of ZSTD_CCtx and workspace extraction.

Test Plan: cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression; cargo clippy --manifest-path rust/Cargo.toml; cargo clippy --manifest-path rust/Cargo.toml --benches; cargo clippy --manifest-path rust/Cargo.toml --tests; make -B -C lib -j2 lib; make -B -C tests -j2 test-zstream (84 deterministic, 6697 and 9217 fuzzer cases).

rust/src/zstd_compress_frame.rs

@@ -9,8 +9,9 @@
 //! crosses the language boundary.
 
 use crate::errors::{ZstdErrorCode, ERROR};
+use crate::zstd_presplit::ZSTD_splitBlock;
 use std::ffi::c_void;
-use std::os::raw::{c_int, c_uint};
+use std::os::raw::{c_int, c_longlong, c_uint};
 use std::ptr;
 
 const ZSTD_MAGICNUMBER: u32 = 0xFD2F_B528;
@@ -21,6 +22,10 @@ const ZSTD_BLOCKHEADERSIZE: usize = 3;
 const ZSTD_WINDOWLOG_ABSOLUTEMIN: u32 = 10;
 const ZSTD_CONTENTSIZE_UNKNOWN: u64 = u64::MAX;
 const ZSTD_F_ZSTD1: c_int = 0;
+const ZSTD_BLOCK_SIZE: usize = 128 << 10;
+const ZSTD_FAST: c_int = 1;
+const ZSTD_BTULTRA2: c_int = 9;
+const SPLIT_LEVELS: [c_int; 10] = [0, 0, 1, 2, 2, 3, 3, 4, 4, 4];
 
 #[inline]
 unsafe fn write_le16(dst: *mut u8, value: u16) {
@@ -209,6 +214,42 @@ pub unsafe extern "C" fn ZSTD_writeLastEmptyBlock(dst: *mut c_void, dst_capacity
     ZSTD_BLOCKHEADERSIZE
 }
 
+/// Rust implementation of the private `ZSTD_optimalBlockSize()` policy.
+///
+/// The compressor context stays in C.  This leaf receives only the source
+/// block, scalar policy inputs, and the pre-split workspace projected by the C
+/// caller, then delegates the actual split heuristic to `ZSTD_splitBlock`.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTD_rust_optimalBlockSize(
+    src: *const c_void,
+    src_size: usize,
+    block_size_max: usize,
+    split_level: c_int,
+    strategy: c_int,
+    savings: c_longlong,
+    workspace: *mut c_void,
+    workspace_size: usize,
+) -> usize {
+    if src_size < ZSTD_BLOCK_SIZE || block_size_max < ZSTD_BLOCK_SIZE {
+        return src_size.min(block_size_max);
+    }
+    if savings < 3 {
+        return ZSTD_BLOCK_SIZE;
+    }
+
+    let split_level = if split_level == 1 {
+        return ZSTD_BLOCK_SIZE;
+    } else if split_level == 0 {
+        debug_assert!((ZSTD_FAST..=ZSTD_BTULTRA2).contains(&strategy));
+        SPLIT_LEVELS[strategy as usize]
+    } else {
+        debug_assert!((2..=6).contains(&split_level));
+        split_level - 2
+    };
+
+    unsafe { ZSTD_splitBlock(src, block_size_max, split_level, workspace, workspace_size) }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -418,4 +459,75 @@ mod tests {
             ZstdErrorCode::DstSizeTooSmall as i32
         );
     }
+
+    #[test]
+    fn optimal_block_size_keeps_small_blocks_intact() {
+        let result = unsafe {
+            ZSTD_rust_optimalBlockSize(
+                ptr::null(),
+                ZSTD_BLOCK_SIZE - 1,
+                ZSTD_BLOCK_SIZE,
+                0,
+                ZSTD_FAST,
+                3,
+                ptr::null_mut(),
+                0,
+            )
+        };
+        assert_eq!(result, ZSTD_BLOCK_SIZE - 1);
+    }
+
+    #[test]
+    fn optimal_block_size_requires_savings_before_splitting() {
+        let result = unsafe {
+            ZSTD_rust_optimalBlockSize(
+                ptr::null(),
+                ZSTD_BLOCK_SIZE,
+                ZSTD_BLOCK_SIZE,
+                0,
+                ZSTD_FAST,
+                2,
+                ptr::null_mut(),
+                0,
+            )
+        };
+        assert_eq!(result, ZSTD_BLOCK_SIZE);
+    }
+
+    #[test]
+    fn optimal_block_size_honors_explicit_no_split_level() {
+        let result = unsafe {
+            ZSTD_rust_optimalBlockSize(
+                ptr::null(),
+                ZSTD_BLOCK_SIZE,
+                ZSTD_BLOCK_SIZE,
+                1,
+                ZSTD_BTULTRA2,
+                3,
+                ptr::null_mut(),
+                0,
+            )
+        };
+        assert_eq!(result, ZSTD_BLOCK_SIZE);
+    }
+
+    #[test]
+    fn optimal_block_size_delegates_valid_split_requests() {
+        let source = vec![0u8; ZSTD_BLOCK_SIZE];
+        let mut workspace = vec![0usize; 8_208usize.div_ceil(std::mem::size_of::<usize>())];
+        let result = unsafe {
+            ZSTD_rust_optimalBlockSize(
+                source.as_ptr().cast(),
+                source.len(),
+                source.len(),
+                2,
+                ZSTD_FAST,
+                3,
+                workspace.as_mut_ptr().cast(),
+                workspace.len() * std::mem::size_of::<usize>(),
+            )
+        };
+        assert!(result > 0);
+        assert!(result <= ZSTD_BLOCK_SIZE);
+    }
 }