refactor(compress): move stream fast-path policy to Rust

Project the private stream state needed by the complete-input streaming fast
path into a stable C/Rust layout and let Rust own the eligibility predicate.
The C bridge keeps the CCtx private and retains the existing simple-level
selection leaf as a callback, while focused tests cover the accepted and
rejected stream states.

All heavy verification was run serially with a 40 GiB virtual-memory cap and
one build job.

Test Plan:
- git diff --cached --check
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --all-targets (788 passed)
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets (181 passed)
- ulimit -v 41943040; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check
- ulimit -v 41943040; make -j1
- ulimit -v 41943040; make -j1 -C tests test

rust/src/zstd_compress.rs

@@ -117,6 +117,40 @@ const MIN_CBLOCK_SIZE: usize = 2;
 const MIN_COMPRESSIBLE_BLOCK_SIZE: usize = MIN_CBLOCK_SIZE + ZSTD_BLOCK_HEADER_SIZE + 1 + 1;
 const ZSTD_ROWSIZE: usize = 16;
 const ZSTD_CSTREAM_STAGE_INIT: c_int = 0;
+
+/// Scalar stream state used to select the complete-input fast path. The C
+/// context remains opaque; its private fields are projected before Rust
+/// evaluates the eligibility predicate.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTD_rust_simpleCompressStream2Projection {
+    stream_stage: c_int,
+    pledged_src_size_plus_one: u64,
+    rust_simple_compress2_completed: c_uint,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTD_rust_simpleCompressStream2Projection, stream_stage) == 0);
+    assert!(
+        offset_of!(
+            ZSTD_rust_simpleCompressStream2Projection,
+            pledged_src_size_plus_one
+        ) == if size_of::<usize>() == 8 { 8 } else { 4 }
+    );
+    assert!(
+        offset_of!(
+            ZSTD_rust_simpleCompressStream2Projection,
+            rust_simple_compress2_completed
+        ) == if size_of::<usize>() == 8 { 16 } else { 12 }
+    );
+    assert!(
+        size_of::<ZSTD_rust_simpleCompressStream2Projection>()
+            == if size_of::<usize>() == 8 { 24 } else { 16 }
+    );
+};
+
+pub type ZSTD_rust_simpleCompress2LevelFn = unsafe extern "C" fn(*const c_void, usize) -> c_int;
+
 const ZSTD_WINDOW_START_INDEX: u32 = 2;
 const ZSTD_DUBT_UNSORTED_MARK: u32 = 1;
 const ZSTD_INDEXOVERFLOW_MARGIN: usize = 16usize << 20;
@@ -9825,6 +9859,32 @@ unsafe fn compress_frame(
     output_offset
 }
 
+/// Select the complete-input streaming fast path before invoking the C-owned
+/// simple-level predicate. Rust owns the stream-state branch; C keeps the
+/// private context and level-selection leaf behind a callback.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTD_rust_simpleCompressStream2Policy(
+    projection: *const ZSTD_rust_simpleCompressStream2Projection,
+    cctx: *const c_void,
+    src_size: usize,
+    simple_compress2_level: Option<ZSTD_rust_simpleCompress2LevelFn>,
+) -> c_int {
+    let Some(projection) = (unsafe { projection.as_ref() }) else {
+        return c_int::MIN;
+    };
+    if cctx.is_null()
+        || projection.stream_stage != ZSTD_CSTREAM_STAGE_INIT
+        || projection.pledged_src_size_plus_one != 0
+        || projection.rust_simple_compress2_completed != 0
+    {
+        return c_int::MIN;
+    }
+    let Some(simple_compress2_level) = simple_compress2_level else {
+        return c_int::MIN;
+    };
+    unsafe { simple_compress2_level(cctx, src_size) }
+}
+
 /// Simple one-shot compression entry point.
 #[no_mangle]
 pub unsafe extern "C" fn ZSTD_compress(
@@ -13674,6 +13734,77 @@ mod tests {
         );
     }
 
+    #[derive(Default)]
+    struct SimpleStreamPolicyTestContext {
+        calls: usize,
+        src_size: usize,
+    }
+
+    unsafe extern "C" fn simple_stream_policy_test_level(
+        context: *const c_void,
+        src_size: usize,
+    ) -> c_int {
+        let context = unsafe { &mut *context.cast_mut().cast::<SimpleStreamPolicyTestContext>() };
+        context.calls += 1;
+        context.src_size = src_size;
+        7
+    }
+
+    #[test]
+    fn simple_stream_policy_calls_the_level_leaf_only_for_a_new_frame() {
+        let mut context = SimpleStreamPolicyTestContext::default();
+        let cctx = (&mut context as *mut SimpleStreamPolicyTestContext).cast();
+        let projection = ZSTD_rust_simpleCompressStream2Projection {
+            stream_stage: ZSTD_CSTREAM_STAGE_INIT,
+            pledged_src_size_plus_one: 0,
+            rust_simple_compress2_completed: 0,
+        };
+
+        let result = unsafe {
+            ZSTD_rust_simpleCompressStream2Policy(
+                &projection,
+                cctx,
+                123,
+                Some(simple_stream_policy_test_level),
+            )
+        };
+
+        assert_eq!(result, 7);
+        assert_eq!(context.calls, 1);
+        assert_eq!(context.src_size, 123);
+    }
+
+    #[test]
+    fn simple_stream_policy_rejects_non_initial_or_completed_frames() {
+        let mut context = SimpleStreamPolicyTestContext::default();
+        let cctx = (&mut context as *mut SimpleStreamPolicyTestContext).cast();
+        for projection in [
+            ZSTD_rust_simpleCompressStream2Projection {
+                stream_stage: ZSTD_CSTREAM_STAGE_LOAD,
+                ..Default::default()
+            },
+            ZSTD_rust_simpleCompressStream2Projection {
+                pledged_src_size_plus_one: 1,
+                ..Default::default()
+            },
+            ZSTD_rust_simpleCompressStream2Projection {
+                rust_simple_compress2_completed: 1,
+                ..Default::default()
+            },
+        ] {
+            let result = unsafe {
+                ZSTD_rust_simpleCompressStream2Policy(
+                    &projection,
+                    cctx,
+                    123,
+                    Some(simple_stream_policy_test_level),
+                )
+            };
+            assert_eq!(result, c_int::MIN);
+        }
+        assert_eq!(context.calls, 0);
+    }
+
     #[test]
     fn reduce_table_applies_threshold_and_wrapping_subtraction() {
         let mut table = [0, 1, 2, 3, 4, 5, 6, u32::MAX, 0, 0, 0, 0, 0, 0, 0, 0];