feat(mt): move worker trace policy to Rust

Move the first-job versus continuation-job trace policy out of the C worker
callback. Rust now decides when the external-dictionary assertion is required
and orders it before the trace, while C retains the private window assertion
and trace-recording leaves.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo check --tests
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --tests -- -A clippy::manual-bits -D warnings
- ulimit -v 41943040; make -j1
- ulimit -v 41943040; make -j1 -C tests test

lib/compress/zstdmt_compress.c

@@ -362,6 +362,24 @@ typedef char ZSTDMT_compression_job_finish_projection_layout[
                    + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection))
         ? 1 : -1];
 
+typedef struct {
+    void* callbackContext;
+    unsigned firstJob;
+    ZSTDMT_compressionJobVoidFn assertNoExtDict;
+    ZSTDMT_compressionJobVoidFn trace;
+} ZSTDMT_RustCompressionJobTraceProjection;
+typedef char ZSTDMT_compression_job_trace_projection_layout[
+        (offsetof(ZSTDMT_RustCompressionJobTraceProjection, callbackContext) == 0
+         && offsetof(ZSTDMT_RustCompressionJobTraceProjection, firstJob)
+                == sizeof(void*)
+         && offsetof(ZSTDMT_RustCompressionJobTraceProjection, assertNoExtDict)
+                == 2 * sizeof(void*)
+         && offsetof(ZSTDMT_RustCompressionJobTraceProjection, trace)
+                == 3 * sizeof(void*)
+         && sizeof(ZSTDMT_RustCompressionJobTraceProjection)
+                == 4 * sizeof(void*))
+        ? 1 : -1];
+
 typedef struct {
     int checksumFlag;
     int ldmEnable;
@@ -422,6 +440,8 @@ void ZSTDMT_rust_compressionJob(
         ZSTDMT_compressionJobApplySequencesFn applySequences,
         ZSTDMT_compressionJobCompressFn compressJob,
         ZSTDMT_compressionJobVoidFn traceJob);
+void ZSTDMT_rust_compressionJobTrace(
+        const ZSTDMT_RustCompressionJobTraceProjection* projection);
 
 ZSTDMT_chunkProcessResult ZSTDMT_rust_compressJobChunks(
         ZSTD_CCtx* cctx, const void* src, size_t srcSize,
@@ -2247,18 +2267,34 @@ static ZSTDMT_chunkProcessResult ZSTDMT_compressionJobCompress(
         &progress, ZSTDMT_rust_compressionJobProgress);
 }
 
+static void ZSTDMT_compressionJobTraceAssertNoExtDict(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+
+    /* Double check that we don't have an ext-dict, because then our repcode
+     * invalidation doesn't work. */
+    assert(!ZSTD_window_hasExtDict(state->cctx->blockState.matchState.window));
+}
+
+static void ZSTDMT_compressionJobTraceRecord(void* opaque)
+{
+    ZSTDMT_compressionJobState* const state =
+        (ZSTDMT_compressionJobState*)opaque;
+    ZSTD_CCtx_trace(state->cctx, 0);
+}
+
 static void ZSTDMT_compressionJobTrace(void* opaque)
 {
     ZSTDMT_compressionJobState* const state =
         (ZSTDMT_compressionJobState*)opaque;
-    ZSTDMT_jobDescription* const job = state->job;
+    ZSTDMT_RustCompressionJobTraceProjection const projection = {
-
+        state,
-    if (!job->firstJob) {
+        state->job->firstJob,
-        /* Double check that we don't have an ext-dict, because then our
+        ZSTDMT_compressionJobTraceAssertNoExtDict,
-         * repcode invalidation doesn't work. */
+        ZSTDMT_compressionJobTraceRecord
-        assert(!ZSTD_window_hasExtDict(state->cctx->blockState.matchState.window));
+    };
-    }
+    ZSTDMT_rust_compressionJobTrace(&projection);
-    ZSTD_CCtx_trace(state->cctx, 0);
 }
 
 static void ZSTDMT_compressionJobEnsureFinished(void* opaque)

rust/src/zstdmt_compress.rs

@@ -840,6 +840,67 @@ const _: () = {
             + size_of::<ZSTDMT_compressionJobFinishPublicationProjection>());
 };
 
+/// C retains the private window assertion and trace-recording leaves for an
+/// MT worker. Rust owns the first-job policy that determines whether the
+/// external-dictionary assertion is required before tracing.
+#[repr(C)]
+#[derive(Clone, Copy)]
+pub struct ZSTDMT_compressionJobTraceProjection {
+    pub callbackContext: *mut c_void,
+    pub firstJob: c_uint,
+    pub assertNoExtDict: Option<ZSTDMT_compressionJobVoidFn>,
+    pub trace: Option<ZSTDMT_compressionJobVoidFn>,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTDMT_compressionJobTraceProjection, callbackContext) == 0);
+    assert!(offset_of!(ZSTDMT_compressionJobTraceProjection, firstJob) == size_of::<usize>());
+    assert!(
+        offset_of!(ZSTDMT_compressionJobTraceProjection, assertNoExtDict)
+            == 2 * size_of::<usize>()
+    );
+    assert!(offset_of!(ZSTDMT_compressionJobTraceProjection, trace) == 3 * size_of::<usize>());
+    assert!(size_of::<ZSTDMT_compressionJobTraceProjection>() == 4 * size_of::<usize>());
+};
+
+#[inline]
+fn compression_job_trace_with<A, T>(first_job: c_uint, mut assert_no_ext_dict: A, mut trace: T)
+where
+    A: FnMut(),
+    T: FnMut(),
+{
+    if first_job == 0 {
+        assert_no_ext_dict();
+    }
+    trace();
+}
+
+/// Apply the worker trace policy while C retains the opaque trace leaves.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_compressionJobTrace(
+    projection: *const ZSTDMT_compressionJobTraceProjection,
+) {
+    let Some(projection) = (unsafe { projection.as_ref() }) else {
+        return;
+    };
+    let Some(trace) = projection.trace else {
+        return;
+    };
+    if projection.firstJob == 0 && projection.assertNoExtDict.is_none() {
+        return;
+    }
+    let assert_no_ext_dict = projection.assertNoExtDict;
+    compression_job_trace_with(
+        projection.firstJob,
+        || {
+            if let Some(assert_no_ext_dict) = assert_no_ext_dict {
+                unsafe { assert_no_ext_dict(projection.callbackContext) };
+            }
+        },
+        || unsafe { trace(projection.callbackContext) },
+    );
+}
+
 pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int;
 pub type ZSTDMT_serialLdmWindowUpdateFn =
     unsafe extern "C" fn(*mut c_void, *mut ZstdMtRawSeqStore, *const c_void, usize);
@@ -7041,6 +7102,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn compression_job_trace_guards_only_continuation_jobs() {
+        let first_job_events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let first_job_assert_events = Rc::clone(&first_job_events);
+        let first_job_trace_events = Rc::clone(&first_job_events);
+        compression_job_trace_with(
+            1,
+            move || first_job_assert_events.borrow_mut().push("assert"),
+            move || first_job_trace_events.borrow_mut().push("trace"),
+        );
+        assert_eq!(&*first_job_events.borrow(), &["trace"]);
+
+        let continuation_events = Rc::new(RefCell::new(Vec::<&'static str>::new()));
+        let continuation_assert_events = Rc::clone(&continuation_events);
+        let continuation_trace_events = Rc::clone(&continuation_events);
+        compression_job_trace_with(
+            0,
+            move || continuation_assert_events.borrow_mut().push("assert"),
+            move || continuation_trace_events.borrow_mut().push("trace"),
+        );
+        assert_eq!(&*continuation_events.borrow(), &["assert", "trace"]);
+    }
+
     #[test]
     fn compression_job_runs_first_job_stages_and_reports_final_block() {
         let state = Rc::new(RefCell::new(MockCompressionJob::default()));