feat(compress): move local dictionary init policy into Rust

The transparent stream initializer previously called a C helper that both
inspected local-dictionary state and chose the no-dictionary, already-created,
and create/publish branches.  Project that private state through a checked C/Rust
ABI so Rust owns the branch and callback order while C retains the requested
parameters, custom allocator, and CDict implementation details.  Keep creation
and publication split so allocation failure cannot publish a partial dictionary.

Test Plan:
- cargo fmt --manifest-path rust/Cargo.toml -- --check
- CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml compress_stream_init_
- CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- make -j1
- make -j1 -C tests test-zstream ZSTREAM_TESTTIME=-T2s
- make -j1 -C tests test-fuzzer FUZZERTEST=-T3s FUZZER_FLAGS=--no-big-tests
- CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml

lib/compress/zstd_compress.c

@@ -1033,7 +1033,22 @@ typedef struct {
     int cdictCompressionLevel;
     size_t cdictDictContentSize;
 } ZSTD_rust_compressStreamInitDictionaryState;
-typedef size_t (*ZSTD_rust_compressStreamInitLocalDict_f)(void* context);
+typedef struct {
+    const void* dict;
+    void* dictBuffer;
+    size_t dictSize;
+    int dictContentType;
+    void* localCDict;
+    const void* cdict;
+    const void* prefixDict;
+    void* createdCDict;
+} ZSTD_rust_compressStreamInitLocalDictState;
+typedef void (*ZSTD_rust_compressStreamInitGetLocalDict_f)(
+        void* context, ZSTD_rust_compressStreamInitLocalDictState* state);
+typedef size_t (*ZSTD_rust_compressStreamInitCreateLocalDict_f)(
+        void* context, ZSTD_rust_compressStreamInitLocalDictState* state);
+typedef void (*ZSTD_rust_compressStreamInitPublishLocalDict_f)(
+        void* context, const ZSTD_rust_compressStreamInitLocalDictState* state);
 typedef void (*ZSTD_rust_compressStreamInitRefreshCDict_f)(
         void* context, ZSTD_rust_compressStreamInitDictionaryState* state);
 typedef void (*ZSTD_rust_compressStreamInitClearPrefix_f)(void* context);
@@ -1083,7 +1098,9 @@ typedef struct {
     size_t inSize;
     int multithreaded;
     size_t mtJobSizeMin;
-    ZSTD_rust_compressStreamInitLocalDict_f initLocalDict;
+    ZSTD_rust_compressStreamInitGetLocalDict_f getLocalDict;
+    ZSTD_rust_compressStreamInitCreateLocalDict_f createLocalDict;
+    ZSTD_rust_compressStreamInitPublishLocalDict_f publishLocalDict;
     ZSTD_rust_compressStreamInitRefreshCDict_f refreshCDict;
     ZSTD_rust_compressStreamInitClearPrefix_f clearPrefix;
     ZSTD_rust_compressStreamInitAssertDictionaries_f assertDictionaries;
@@ -1109,6 +1126,25 @@ typedef struct {
     ZSTD_rust_compressStreamInitGetBlockSize_f getBlockSize;
     ZSTD_rust_compressStreamInitCommitOrdinary_f commitOrdinary;
 } ZSTD_rust_compressStreamInitState;
+typedef char ZSTD_rust_compress_stream_init_local_dict_layout[
+        (offsetof(ZSTD_rust_compressStreamInitLocalDictState, dict) == 0
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, dictBuffer)
+                == sizeof(void*)
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, dictSize)
+                == 2 * sizeof(void*)
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, dictContentType)
+                == 3 * sizeof(void*)
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, localCDict)
+                == 4 * sizeof(void*)
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, cdict)
+                == 5 * sizeof(void*)
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, prefixDict)
+                == 6 * sizeof(void*)
+         && offsetof(ZSTD_rust_compressStreamInitLocalDictState, createdCDict)
+                == 7 * sizeof(void*)
+         && sizeof(ZSTD_rust_compressStreamInitLocalDictState)
+                == 8 * sizeof(void*))
+        ? 1 : -1];
 typedef char ZSTD_rust_compress_stream_init_dictionary_layout[
         (offsetof(ZSTD_rust_compressStreamInitDictionaryState, prefixDict) == 0
          && offsetof(ZSTD_rust_compressStreamInitDictionaryState, prefixDictSize)
@@ -1140,7 +1176,7 @@ typedef char ZSTD_rust_compress_stream_init_state_layout[
          && offsetof(ZSTD_rust_compressStreamInitState, mtJobSizeMin)
                 == 5 * sizeof(void*) + sizeof(size_t)
          && sizeof(ZSTD_rust_compressStreamInitState)
-                == (sizeof(void*) == 8 ? 256 : 128))
+                == (sizeof(void*) == 8 ? 272 : 144))
         ? 1 : -1];
 size_t ZSTD_rust_compressStreamInit(
         const ZSTD_rust_compressStreamInitState* state);
@@ -3263,42 +3299,6 @@ size_t ZSTD_CCtx_setPledgedSrcSize(ZSTD_CCtx* cctx, unsigned long long pledgedSr
 static void ZSTD_dedicatedDictSearch_revertCParams(
         ZSTD_compressionParameters* cParams);
 
-/**
- * Initializes the local dictionary using requested parameters.
- * NOTE: Initialization does not employ the pledged src size,
- * because the dictionary may be used for multiple compressions.
- */
-static size_t ZSTD_initLocalDict(ZSTD_CCtx* cctx)
-{
-    ZSTD_localDict* const dl = &cctx->localDict;
-    if (dl->dict == NULL) {
-        /* No local dictionary. */
-        assert(dl->dictBuffer == NULL);
-        assert(dl->cdict == NULL);
-        assert(dl->dictSize == 0);
-        return 0;
-    }
-    if (dl->cdict != NULL) {
-        /* Local dictionary already initialized. */
-        assert(cctx->cdict == dl->cdict);
-        return 0;
-    }
-    assert(dl->dictSize > 0);
-    assert(cctx->cdict == NULL);
-    assert(cctx->prefixDict.dict == NULL);
-
-    dl->cdict = ZSTD_createCDict_advanced2(
-            dl->dict,
-            dl->dictSize,
-            ZSTD_dlm_byRef,
-            dl->dictContentType,
-            &cctx->requestedParams,
-            cctx->customMem);
-    RETURN_ERROR_IF(!dl->cdict, memory_allocation, "ZSTD_createCDict_advanced failed");
-    cctx->cdict = dl->cdict;
-    return 0;
-}
-
 size_t ZSTD_CCtx_loadDictionary_advanced(
         ZSTD_CCtx* cctx,
         const void* dict, size_t dictSize,
@@ -7353,9 +7353,43 @@ enum {
     ZSTD_RUST_INIT_RESOLVE_EXTERNAL_REPCODE_SEARCH = 5
 };
 
-static size_t ZSTD_rust_compressStreamInit_localDict(void* context)
+static void ZSTD_rust_compressStreamInit_getLocalDict(
+        void* context, ZSTD_rust_compressStreamInitLocalDictState* state)
 {
-    return ZSTD_initLocalDict((ZSTD_CCtx*)context);
+    ZSTD_CCtx const* const cctx = (ZSTD_CCtx const*)context;
+    ZSTD_localDict const* const localDict = &cctx->localDict;
+    state->dict = localDict->dict;
+    state->dictBuffer = localDict->dictBuffer;
+    state->dictSize = localDict->dictSize;
+    state->dictContentType = (int)localDict->dictContentType;
+    state->localCDict = (void*)localDict->cdict;
+    state->cdict = cctx->cdict;
+    state->prefixDict = cctx->prefixDict.dict;
+    state->createdCDict = NULL;
+}
+
+static size_t ZSTD_rust_compressStreamInit_createLocalDict(
+        void* context, ZSTD_rust_compressStreamInitLocalDictState* state)
+{
+    ZSTD_CCtx const* const cctx = (ZSTD_CCtx const*)context;
+    ZSTD_CDict* const cdict = ZSTD_createCDict_advanced2(
+            state->dict,
+            state->dictSize,
+            ZSTD_dlm_byRef,
+            (ZSTD_dictContentType_e)state->dictContentType,
+            &cctx->requestedParams,
+            cctx->customMem);
+    RETURN_ERROR_IF(!cdict, memory_allocation, "ZSTD_createCDict_advanced failed");
+    state->createdCDict = cdict;
+    return 0;
+}
+
+static void ZSTD_rust_compressStreamInit_publishLocalDict(
+        void* context, const ZSTD_rust_compressStreamInitLocalDictState* state)
+{
+    ZSTD_CCtx* const cctx = (ZSTD_CCtx*)context;
+    cctx->localDict.cdict = (ZSTD_CDict*)state->createdCDict;
+    cctx->cdict = cctx->localDict.cdict;
 }
 
 static void ZSTD_rust_compressStreamInit_refreshCDict(
@@ -7628,7 +7662,9 @@ static size_t ZSTD_CCtx_init_compressStream2(ZSTD_CCtx* cctx,
         0,
         0,
 #endif
-        ZSTD_rust_compressStreamInit_localDict,
+        ZSTD_rust_compressStreamInit_getLocalDict,
+        ZSTD_rust_compressStreamInit_createLocalDict,
+        ZSTD_rust_compressStreamInit_publishLocalDict,
         ZSTD_rust_compressStreamInit_refreshCDict,
         ZSTD_rust_compressStreamInit_clearPrefix,
         ZSTD_rust_compressStreamInit_assertDictionaries,

rust/src/zstd_compress.rs

@@ -2661,7 +2661,12 @@ const ZSTD_RUST_INIT_RESOLVE_VALIDATE_SEQUENCES: c_int = 3;
 const ZSTD_RUST_INIT_RESOLVE_MAX_BLOCK_SIZE: c_int = 4;
 const ZSTD_RUST_INIT_RESOLVE_EXTERNAL_REPCODE_SEARCH: c_int = 5;
 
-type CompressStreamInitLocalDictFn = unsafe extern "C" fn(*mut c_void) -> usize;
+type CompressStreamInitGetLocalDictFn =
+    unsafe extern "C" fn(*mut c_void, *mut ZSTD_rust_compressStreamInitLocalDictState);
+type CompressStreamInitCreateLocalDictFn =
+    unsafe extern "C" fn(*mut c_void, *mut ZSTD_rust_compressStreamInitLocalDictState) -> usize;
+type CompressStreamInitPublishLocalDictFn =
+    unsafe extern "C" fn(*mut c_void, *const ZSTD_rust_compressStreamInitLocalDictState);
 type CompressStreamInitRefreshCDictFn =
     unsafe extern "C" fn(*mut c_void, *mut ZSTD_rust_compressStreamInitDictionaryState);
 type CompressStreamInitClearPrefixFn = unsafe extern "C" fn(*mut c_void);
@@ -2704,6 +2709,54 @@ type CompressStreamInitGetBufferModeFn = unsafe extern "C" fn(*mut c_void) -> c_
 type CompressStreamInitGetBlockSizeFn = unsafe extern "C" fn(*mut c_void) -> usize;
 type CompressStreamInitCommitOrdinaryFn = unsafe extern "C" fn(*mut c_void, usize);
 
+/// Private local-dictionary projection used by transparent stream initialization.
+///
+/// Rust owns the no-dictionary, already-initialized, create, and publish order;
+/// C retains the private dictionary storage and allocator-sensitive callbacks.
+#[repr(C)]
+pub struct ZSTD_rust_compressStreamInitLocalDictState {
+    dict: *const c_void,
+    dict_buffer: *mut c_void,
+    dict_size: usize,
+    dict_content_type: c_int,
+    local_cdict: *mut c_void,
+    cdict: *const c_void,
+    prefix_dict: *const c_void,
+    created_cdict: *mut c_void,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTD_rust_compressStreamInitLocalDictState, dict) == 0);
+    assert!(
+        offset_of!(ZSTD_rust_compressStreamInitLocalDictState, dict_buffer) == size_of::<usize>()
+    );
+    assert!(
+        offset_of!(ZSTD_rust_compressStreamInitLocalDictState, dict_size) == 2 * size_of::<usize>()
+    );
+    assert!(
+        offset_of!(
+            ZSTD_rust_compressStreamInitLocalDictState,
+            dict_content_type
+        ) == 3 * size_of::<usize>()
+    );
+    assert!(
+        offset_of!(ZSTD_rust_compressStreamInitLocalDictState, local_cdict)
+            == 4 * size_of::<usize>()
+    );
+    assert!(
+        offset_of!(ZSTD_rust_compressStreamInitLocalDictState, cdict) == 5 * size_of::<usize>()
+    );
+    assert!(
+        offset_of!(ZSTD_rust_compressStreamInitLocalDictState, prefix_dict)
+            == 6 * size_of::<usize>()
+    );
+    assert!(
+        offset_of!(ZSTD_rust_compressStreamInitLocalDictState, created_cdict)
+            == 7 * size_of::<usize>()
+    );
+    assert!(size_of::<ZSTD_rust_compressStreamInitLocalDictState>() == size_of::<[usize; 8]>());
+};
+
 /// Scalar dictionary snapshot used by transparent stream initialization.
 ///
 /// The prefix is populated by C before local-dictionary initialization, which
@@ -2774,7 +2827,9 @@ pub struct ZSTD_rust_compressStreamInitState {
     in_size: usize,
     multithreaded: c_int,
     mt_job_size_min: usize,
-    init_local_dict: CompressStreamInitLocalDictFn,
+    get_local_dict: CompressStreamInitGetLocalDictFn,
+    create_local_dict: CompressStreamInitCreateLocalDictFn,
+    publish_local_dict: CompressStreamInitPublishLocalDictFn,
     refresh_cdict: CompressStreamInitRefreshCDictFn,
     clear_prefix: CompressStreamInitClearPrefixFn,
     assert_dictionaries: CompressStreamInitAssertDictionariesFn,
@@ -2817,17 +2872,57 @@ const _: () = {
     );
     assert!(
         size_of::<ZSTD_rust_compressStreamInitState>()
-            == if size_of::<usize>() == 8 { 256 } else { 128 }
+            == if size_of::<usize>() == 8 { 272 } else { 144 }
     );
 };
 
+#[inline]
+unsafe fn compress_stream_init_local_dict_with(state: &ZSTD_rust_compressStreamInitState) -> usize {
+    let mut local_dict = ZSTD_rust_compressStreamInitLocalDictState {
+        dict: ptr::null(),
+        dict_buffer: ptr::null_mut(),
+        dict_size: 0,
+        dict_content_type: 0,
+        local_cdict: ptr::null_mut(),
+        cdict: ptr::null(),
+        prefix_dict: ptr::null(),
+        created_cdict: ptr::null_mut(),
+    };
+    unsafe {
+        (state.get_local_dict)(state.callback_context, &mut local_dict);
+    }
+    if local_dict.dict.is_null() {
+        debug_assert!(local_dict.dict_buffer.is_null());
+        debug_assert!(local_dict.local_cdict.is_null());
+        debug_assert_eq!(local_dict.dict_size, 0);
+        return 0;
+    }
+    if !local_dict.local_cdict.is_null() {
+        debug_assert_eq!(local_dict.cdict, local_dict.local_cdict.cast_const(),);
+        return 0;
+    }
+    debug_assert!(local_dict.dict_size > 0);
+    debug_assert!(local_dict.cdict.is_null());
+    debug_assert!(local_dict.prefix_dict.is_null());
+
+    let result = unsafe { (state.create_local_dict)(state.callback_context, &mut local_dict) };
+    if ERR_isError(result) {
+        return result;
+    }
+    debug_assert!(!local_dict.created_cdict.is_null());
+    unsafe {
+        (state.publish_local_dict)(state.callback_context, &local_dict);
+    }
+    0
+}
+
 #[inline]
 unsafe fn compress_stream_init_body_with(state: &ZSTD_rust_compressStreamInitState) -> usize {
     if state.params.is_null() || state.dictionaries.is_null() {
         return ERROR(ZstdErrorCode::Generic);
     }
 
-    let result = unsafe { (state.init_local_dict)(state.callback_context) };
+    let result = unsafe { compress_stream_init_local_dict_with(state) };
     if ERR_isError(result) {
         return result;
     }
@@ -9572,11 +9667,19 @@ mod tests {
     #[derive(Default)]
     struct CompressStreamInitTestContext {
         events: Vec<&'static str>,
-        local_result: usize,
+        local_create_result: usize,
         create_result: usize,
         mt_result: usize,
         begin_result: usize,
         pledged: u64,
+        local_dict: *const c_void,
+        local_dict_buffer: *mut c_void,
+        local_dict_size: usize,
+        local_dict_content_type: c_int,
+        local_cdict: *mut c_void,
+        context_cdict: *const c_void,
+        prefix_dict: *const c_void,
+        created_local_cdict: *mut c_void,
         cdict: *const c_void,
         cdict_is_local: c_int,
         cdict_compression_level: c_int,
@@ -9597,10 +9700,43 @@ mod tests {
         unsafe { &mut *context.cast::<CompressStreamInitTestContext>() }
     }
 
-    unsafe extern "C" fn compress_stream_init_test_local_dict(context: *mut c_void) -> usize {
+    unsafe extern "C" fn compress_stream_init_test_get_local_dict(
+        context: *mut c_void,
+        local_dict: *mut ZSTD_rust_compressStreamInitLocalDictState,
+    ) {
         let context = unsafe { compress_stream_init_test_context(context) };
         context.events.push("local-dict");
-        context.local_result
+        let local_dict = unsafe { &mut *local_dict };
+        local_dict.dict = context.local_dict;
+        local_dict.dict_buffer = context.local_dict_buffer;
+        local_dict.dict_size = context.local_dict_size;
+        local_dict.dict_content_type = context.local_dict_content_type;
+        local_dict.local_cdict = context.local_cdict;
+        local_dict.cdict = context.context_cdict;
+        local_dict.prefix_dict = context.prefix_dict;
+        local_dict.created_cdict = ptr::null_mut();
+    }
+
+    unsafe extern "C" fn compress_stream_init_test_create_local_dict(
+        context: *mut c_void,
+        local_dict: *mut ZSTD_rust_compressStreamInitLocalDictState,
+    ) -> usize {
+        let context = unsafe { compress_stream_init_test_context(context) };
+        context.events.push("create-local-dict");
+        unsafe { &mut *local_dict }.created_cdict = context.created_local_cdict;
+        context.local_create_result
+    }
+
+    unsafe extern "C" fn compress_stream_init_test_publish_local_dict(
+        context: *mut c_void,
+        local_dict: *const ZSTD_rust_compressStreamInitLocalDictState,
+    ) {
+        let context = unsafe { compress_stream_init_test_context(context) };
+        context.events.push("publish-local-dict");
+        let local_dict = unsafe { &*local_dict };
+        context.local_cdict = local_dict.created_cdict;
+        context.context_cdict = local_dict.created_cdict.cast_const();
+        context.cdict = local_dict.created_cdict.cast_const();
     }
 
     unsafe extern "C" fn compress_stream_init_test_refresh_cdict(
@@ -9824,7 +9960,9 @@ mod tests {
             in_size,
             multithreaded,
             mt_job_size_min: 10,
-            init_local_dict: compress_stream_init_test_local_dict,
+            get_local_dict: compress_stream_init_test_get_local_dict,
+            create_local_dict: compress_stream_init_test_create_local_dict,
+            publish_local_dict: compress_stream_init_test_publish_local_dict,
             refresh_cdict: compress_stream_init_test_refresh_cdict,
             clear_prefix: compress_stream_init_test_clear_prefix,
             assert_dictionaries: compress_stream_init_test_assert_dictionaries,
@@ -9911,10 +10049,53 @@ mod tests {
         );
     }
 
+    #[test]
+    fn compress_stream_init_orchestrates_local_dictionary_creation_and_publish() {
+        let mut context = CompressStreamInitTestContext {
+            local_dict: ptr::dangling::<c_void>(),
+            local_dict_size: 3,
+            created_local_cdict: ptr::dangling_mut::<c_void>(),
+            cdict_is_local: 1,
+            cdict_dict_content_size: 3,
+            buffer_mode: ZSTD_BM_BUFFERED,
+            block_size: 4,
+            ..CompressStreamInitTestContext::default()
+        };
+        let mut dictionaries = ZSTD_rust_compressStreamInitDictionaryState {
+            prefix_dict: ptr::null(),
+            prefix_dict_size: 0,
+            prefix_dict_content_type: 0,
+            cdict: ptr::null(),
+            cdict_is_local: 0,
+            cdict_compression_level: 0,
+            cdict_dict_content_size: 0,
+        };
+        let state =
+            compress_stream_init_test_state(&mut context, &mut dictionaries, ZSTD_E_CONTINUE, 0, 0);
+
+        let result = unsafe { ZSTD_rust_compressStreamInit(&state) };
+
+        assert_eq!(result, 0);
+        assert_eq!(context.local_cdict, context.created_local_cdict);
+        assert_eq!(
+            context.context_cdict,
+            context.created_local_cdict.cast_const()
+        );
+        assert_eq!(
+            &context.events[..3],
+            ["local-dict", "create-local-dict", "publish-local-dict"]
+        );
+        assert_eq!(context.events[3], "refresh-cdict");
+        assert_eq!(dictionaries.cdict, context.created_local_cdict.cast_const());
+        assert_eq!(dictionaries.cdict_is_local, 1);
+    }
+
     #[test]
     fn compress_stream_init_stops_before_later_callbacks_on_error() {
         let mut context = CompressStreamInitTestContext {
-            local_result: ERROR(ZstdErrorCode::MemoryAllocation),
+            local_dict: ptr::dangling::<c_void>(),
+            local_dict_size: 3,
+            local_create_result: ERROR(ZstdErrorCode::MemoryAllocation),
             ..CompressStreamInitTestContext::default()
         };
         let mut dictionaries = ZSTD_rust_compressStreamInitDictionaryState {
@@ -9932,7 +10113,7 @@ mod tests {
         let result = unsafe { ZSTD_rust_compressStreamInit(&state) };
 
         assert_eq!(result, ERROR(ZstdErrorCode::MemoryAllocation));
-        assert_eq!(context.events, ["local-dict"]);
+        assert_eq!(context.events, ["local-dict", "create-local-dict"]);
     }
 
     #[test]