feat(cli): move shared-destination scheduling into Rust

Move the shared-output arm of FIO_compressMultipleFilenames into a Rust
scheduler. Rust now owns input-order iteration, processed-file accounting,
index advancement, and non-short-circuiting error aggregation while C retains
file validation, private resources, callbacks, diagnostics, and destination
lifetime.

Test Plan:
- cargo +nightly fmt --manifest-path rust/Cargo.toml
- cargo test --manifest-path rust/Cargo.toml compression_multiple_shared_destination -- --test-threads=1 (2 passed)
- cargo clippy --manifest-path rust/Cargo.toml --tests -- -D warnings
- make -B -C lib -j1 lib
- make -B -C programs -j1 zstd
- make -B -C tests -j1 test-cli-tests (41 passed)
- all commands used CARGO_BUILD_JOBS=1 where applicable and
  ulimit -v 41943040; no heavyweight jobs ran concurrently

Commit is intentionally unsigned because GPG pinentry hangs in this
non-interactive environment.

programs/fileio.c

@@ -1125,6 +1125,35 @@ typedef struct {
     FIO_rust_compress_status_display_fn display_status;
 } FIO_rust_compress_callbacks_t;
 
+typedef int (*FIO_rust_compress_multiple_file_fn)(
+        void* opaque, const char* dstFileName, const char* srcFileName);
+typedef struct {
+    void* fCtx;
+    const char** inFileNamesTable;
+    const char* outFileName;
+    void* opaque;
+    FIO_rust_compress_multiple_file_fn compressFile;
+} FIO_rust_compress_multiple_projection_t;
+typedef char FIO_rust_compress_multiple_fctx_offset[
+        (offsetof(FIO_rust_compress_multiple_projection_t, fCtx) == 0) ? 1 : -1];
+typedef char FIO_rust_compress_multiple_input_names_offset[
+        (offsetof(FIO_rust_compress_multiple_projection_t, inFileNamesTable)
+         == sizeof(void*)) ? 1 : -1];
+typedef char FIO_rust_compress_multiple_output_name_offset[
+        (offsetof(FIO_rust_compress_multiple_projection_t, outFileName)
+         == 2 * sizeof(void*)) ? 1 : -1];
+typedef char FIO_rust_compress_multiple_opaque_offset[
+        (offsetof(FIO_rust_compress_multiple_projection_t, opaque)
+         == 3 * sizeof(void*)) ? 1 : -1];
+typedef char FIO_rust_compress_multiple_callback_offset[
+        (offsetof(FIO_rust_compress_multiple_projection_t, compressFile)
+         == 4 * sizeof(void*)) ? 1 : -1];
+typedef char FIO_rust_compress_multiple_projection_size[
+        (sizeof(FIO_rust_compress_multiple_projection_t)
+         == 4 * sizeof(void*) + sizeof(FIO_rust_compress_multiple_file_fn)) ? 1 : -1];
+int FIO_rust_compressMultipleFilenames(
+        const FIO_rust_compress_multiple_projection_t* projection);
+
 enum {
     FIO_RUST_ZSTD_OK = 0,
     FIO_RUST_ZSTD_COMPRESS_ERROR = 1,
@@ -2897,6 +2926,24 @@ static unsigned long long FIO_getLargestFileSize(const char** inFileNames, unsig
     return FIO_rust_getLargestFileSize(inFileNames, nbFiles);
 }
 
+typedef struct {
+    FIO_ctx_t* fCtx;
+    FIO_prefs_t* prefs;
+    cRess_t* ress;
+    int compressionLevel;
+} FIO_rust_compress_multiple_context_t;
+
+static int FIO_rust_compressMultipleFileCallback(void* opaque,
+                                                  const char* dstFileName,
+                                                  const char* srcFileName)
+{
+    FIO_rust_compress_multiple_context_t* const context =
+        (FIO_rust_compress_multiple_context_t*)opaque;
+    return FIO_compressFilename_srcFile(
+        context->fCtx, context->prefs, *context->ress,
+        dstFileName, srcFileName, context->compressionLevel);
+}
+
 /* FIO_compressMultipleFilenames() :
  * compress nbFiles files
  * into either one destination (outFileName),
@@ -2930,12 +2977,15 @@ int FIO_compressMultipleFilenames(FIO_ctx_t* const fCtx,
         if (dstFile == NULL) {  /* could not open outFileName */
             error = 1;
         } else {
+            FIO_rust_compress_multiple_context_t callbackContext = {
+                fCtx, prefs, &ress, compressionLevel
+            };
+            FIO_rust_compress_multiple_projection_t projection = {
+                fCtx, inFileNamesTable, outFileName,
+                &callbackContext, FIO_rust_compressMultipleFileCallback
+            };
             AIO_WritePool_setFile(ress.writeCtx, dstFile);
-            for (; fCtx->currFileIdx < fCtx->nbFilesTotal; ++fCtx->currFileIdx) {
+            error = FIO_rust_compressMultipleFilenames(&projection);
-                status = FIO_compressFilename_srcFile(fCtx, prefs, ress, outFileName, inFileNamesTable[fCtx->currFileIdx], compressionLevel);
-                if (!status) fCtx->nbFilesProcessed++;
-                error |= status;
-            }
             if (AIO_WritePool_closeFile(ress.writeCtx))
                 EXM_THROW(29, "Write error (%s) : cannot properly close %s",
                             strerror(errno), outFileName);

rust/src/fileio_asyncio.rs

@@ -259,6 +259,42 @@ pub struct FIO_rust_compress_callbacks_t {
     pub display_status: Option<FIO_rust_compress_status_display_fn>,
 }
 
+pub type FIO_rust_compress_multiple_file_fn =
+    unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char) -> c_int;
+
+/// Rust owns only the shared-destination file iteration and aggregate error
+/// handling.  C retains the private preferences/resources and supplies one
+/// callback for each source file.
+#[repr(C)]
+pub struct FIO_rust_compress_multiple_projection_t {
+    pub f_ctx: *mut c_void,
+    pub input_file_names: *const *const c_char,
+    pub output_file_name: *const c_char,
+    pub opaque: *mut c_void,
+    pub compress_file: Option<FIO_rust_compress_multiple_file_fn>,
+}
+const _: () = {
+    assert!(std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, f_ctx) == 0);
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, input_file_names)
+            == size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, output_file_name)
+            == 2 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, opaque)
+            == 3 * size_of::<usize>()
+    );
+    assert!(
+        std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, compress_file)
+            == 4 * size_of::<usize>()
+    );
+    assert!(size_of::<FIO_rust_compress_multiple_file_fn>() == size_of::<usize>());
+    assert!(size_of::<FIO_rust_compress_multiple_projection_t>() == 5 * size_of::<usize>());
+};
+
 pub const FIO_RUST_ZSTD_OK: c_int = 0;
 pub const FIO_RUST_ZSTD_COMPRESS_ERROR: c_int = 1;
 pub const FIO_RUST_ZSTD_INCOMPLETE_INPUT: c_int = 2;
@@ -2122,6 +2158,48 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal(
     FIO_RUST_COMPRESS_OK
 }
 
+/// Iterate the files that share one already-open destination.  The C
+/// callback retains source validation, resource state, diagnostics, and
+/// compression dispatch; Rust preserves the original non-short-circuiting
+/// loop and its counter updates.
+#[no_mangle]
+pub unsafe extern "C" fn FIO_rust_compressMultipleFilenames(
+    projection: *const FIO_rust_compress_multiple_projection_t,
+) -> c_int {
+    assert!(!projection.is_null());
+    let projection = unsafe { &*projection };
+    assert!(!projection.f_ctx.is_null());
+    assert!(!projection.input_file_names.is_null());
+    assert!(!projection.output_file_name.is_null());
+    let compress_file = projection
+        .compress_file
+        .expect("shared-destination compression callback is required");
+
+    let f_ctx = projection.f_ctx.cast::<FIO_rust_compression_context_t>();
+    let mut error = 0;
+    while unsafe { (*f_ctx).currFileIdx < (*f_ctx).nbFilesTotal } {
+        let file_index = unsafe { (*f_ctx).currFileIdx as usize };
+        let src_file_name = unsafe { *projection.input_file_names.add(file_index) };
+        let status = unsafe {
+            compress_file(
+                projection.opaque,
+                projection.output_file_name,
+                src_file_name,
+            )
+        };
+        if status == 0 {
+            unsafe {
+                (*f_ctx).nbFilesProcessed = (*f_ctx).nbFilesProcessed.wrapping_add(1);
+            }
+        }
+        error |= status;
+        unsafe {
+            (*f_ctx).currFileIdx = (*f_ctx).currFileIdx.wrapping_add(1);
+        }
+    }
+    error
+}
+
 #[inline]
 fn zstd_adapt_level(projection: &FIO_rust_zstd_adapt_projection_t, slower: bool) -> c_int {
     if slower {
@@ -4386,6 +4464,99 @@ mod tests {
         assert_eq!(context.totalBytesOutput, 47);
     }
 
+    #[derive(Default)]
+    struct MultipleCompressionState {
+        sources: Vec<*const c_char>,
+        destinations: Vec<*const c_char>,
+        statuses: Vec<c_int>,
+    }
+
+    unsafe extern "C" fn record_multiple_compression_file(
+        opaque: *mut c_void,
+        destination: *const c_char,
+        source: *const c_char,
+    ) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<MultipleCompressionState>() };
+        let status = state
+            .statuses
+            .get(state.sources.len())
+            .copied()
+            .unwrap_or(0);
+        state.sources.push(source);
+        state.destinations.push(destination);
+        status
+    }
+
+    #[test]
+    fn compression_multiple_shared_destination_preserves_order_and_counters() {
+        let source_names = [c"zero".as_ptr(), c"one".as_ptr(), c"two".as_ptr()];
+        let destination = c"archive.zst";
+        let mut context = FIO_rust_compression_context_t {
+            nbFilesTotal: 3,
+            hasStdinInput: 0,
+            hasStdoutOutput: 0,
+            currFileIdx: 1,
+            nbFilesProcessed: 5,
+            totalBytesInput: 0,
+            totalBytesOutput: 0,
+        };
+        let mut state = MultipleCompressionState {
+            statuses: vec![0, 0],
+            ..MultipleCompressionState::default()
+        };
+        let projection = FIO_rust_compress_multiple_projection_t {
+            f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(),
+            input_file_names: source_names.as_ptr(),
+            output_file_name: destination.as_ptr(),
+            opaque: (&mut state as *mut MultipleCompressionState).cast(),
+            compress_file: Some(record_multiple_compression_file),
+        };
+
+        assert_eq!(
+            unsafe { FIO_rust_compressMultipleFilenames(&projection) },
+            0
+        );
+        assert_eq!(state.sources, vec![source_names[1], source_names[2]]);
+        assert_eq!(state.destinations, vec![destination.as_ptr(); 2]);
+        assert_eq!(context.currFileIdx, 3);
+        assert_eq!(context.nbFilesProcessed, 7);
+    }
+
+    #[test]
+    fn compression_multiple_shared_destination_accumulates_errors_without_short_circuiting() {
+        let source_names = [c"one".as_ptr(), c"two".as_ptr(), c"three".as_ptr()];
+        let destination = c"archive.zst";
+        let mut context = FIO_rust_compression_context_t {
+            nbFilesTotal: 3,
+            hasStdinInput: 0,
+            hasStdoutOutput: 0,
+            currFileIdx: 0,
+            nbFilesProcessed: 4,
+            totalBytesInput: 0,
+            totalBytesOutput: 0,
+        };
+        let mut state = MultipleCompressionState {
+            statuses: vec![1, 4, 2],
+            ..MultipleCompressionState::default()
+        };
+        let projection = FIO_rust_compress_multiple_projection_t {
+            f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(),
+            input_file_names: source_names.as_ptr(),
+            output_file_name: destination.as_ptr(),
+            opaque: (&mut state as *mut MultipleCompressionState).cast(),
+            compress_file: Some(record_multiple_compression_file),
+        };
+
+        assert_eq!(
+            unsafe { FIO_rust_compressMultipleFilenames(&projection) },
+            1 | 4 | 2
+        );
+        assert_eq!(state.sources, source_names);
+        assert_eq!(state.destinations, vec![destination.as_ptr(); 3]);
+        assert_eq!(context.currFileIdx, 3);
+        assert_eq!(context.nbFilesProcessed, 4);
+    }
+
     struct ZstdProjectionState {
         input: [u8; 5],
         input_pos: usize,